bricknil.sensor.Button.activate_updates() - Code Metrics - Inspection of "added duplo support" - virantha/bricknil - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( 72bf9f...f85f9f )

by Virantha

created 2019-03-22 04:22 UTC

bricknil.sensor.Button.activate_updates() A

↳ Parent: bricknil.sensor

Complexity

Conditions

Size

Total Lines	8
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	6
nop	1
dl	0
loc	8
rs	10
c	0
b	0
f	0

# Copyright 2019 Virantha N. Ekanayake 
# 
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# 
# http://www.apache.org/licenses/LICENSE-2.0
# 
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Actual sensor and motor peripheral definitions from Boost and PoweredUp
"""
from curio import sleep, current_task, spawn  # Needed for motor speed ramp

from enum import Enum, IntEnum
from .const import Color

from .peripheral import Peripheral, Motor


class InternalMotor(Peripheral):
    """ Access the internal motor(s) in the Boost Move Hub.

        Unlike the train motors, these motors have a built-in sensor for sending back
        the motor's current speed and position.  You don't need to use the sensors, and
        can treat this as strictly an output.

        Examples::

            # Basic connection to the motor on Port A
            @attach(InternalMotor, name='left_motor', port=InternalMotor.Port.A)

            # Basic connection to both motors at the same time (virtual I/O port).
            # Any speed command will cause both motors to rotate at the same speed
            @attach(InternalMotor, name='motors', port=InternalMotor.Port.AB)

            # Report back when motor speed changes. You must have a motors_change method defined 
            @attach(InternalMotor, name='motors', port=InternalMotor.Port.A, capabilities=['sense_speed'])
            # Only report back when speed change exceeds 5 units
            @attach(InternalMotor, name='motors', port=InternalMotor.Port.A, capabilities=[('sense_speed', 5)])

        See Also:
            :class:`TrainMotor` for connecting to a train motor

    """
    _sensor_id = 0x0027
    _DEFAULT_THRESHOLD=2
    """Set to 2 to avoid a lot of updates since the speed seems to oscillate a lot"""

    capability = Enum("capability", {"sense_speed":1, "sense_pos":2})

    Port = Enum('Port', 'A B AB', start=0)
    """Address either motor A or Motor B, or both AB at the same time"""

    # Dict of cap: (num_datasets, bytes_per_dataset)
    datasets = { capability.sense_speed: (1, 1),
                 capability.sense_pos: (1, 4),
                }
    """ Dict of (num_datasets, bytes_per_dataset).
       `sense_speed` (1-byte), and `sense_pos` (uint32)"""

    allowed_combo = [ capability.sense_speed,
                      capability.sense_pos,
                    ]

    def __init__(self, name, port=None, capabilities=[]):
        """Maps the port names `A`, `B`, `AB` to hard-coded port numbers"""
        if port:
            port_map = [55, 56, 57]
            port = port_map[port.value]
        self.speed = 0
        super().__init__(name, port, capabilities)
    
    async def set_speed(self, speed):
        """Sets the speed of the motor, and calls the :func:`bricknil.peripheral.Peripheral._convert_speed_to_val` method
           to do some sanity checking and bounding.

           Args:
               speed (int) : -100 to 100 (I believe this is a percentage)
        """
        self.speed = speed
        speed = self._convert_speed_to_val(speed)
        mode = 0
        await self.set_output(mode, speed)

    async def ramp_speed2(self, target_speed, ramp_time_ms):
        
        # Set acceleration profile
        delta_speed = target_speed - self.speed
        zero_100_ramp_time_ms = int(ramp_time_ms/delta_speed * 100.0) 
        zero_100_ramp_time_ms = zero_100_ramp_time_ms % 10000 # limit time to 10s

        hi = (zero_100_ramp_time_ms >> 8) & 255
        lo = zero_100_ramp_time_ms & 255

        profile = 1
        b = [0x00, 0x81, self.port, 0x01, 0x05, 10, 10, profile]
        await self.send_message(f'set accel profile {zero_100_ramp_time_ms} {hi} {lo} ', b)
        b = [0x00, 0x81, self.port, 0x01, 0x07, self._convert_speed_to_val(target_speed), 80, 1]
        await self.send_message('set speed', b)
        
class VisionSensor(Peripheral):
    """ Access the Boost Vision/Distance Sensor

        Only the sensing capabilities of this sensor is supported right now.

        - *sense_color*: Returns one of the 10 predefined colors
        - *sense_distance*: Distance from 0-7 in roughly inches
        - *sense_count*: Running count of waving your hand/item over the sensor (32-bit)
        - *sense_reflectivity*: Under distances of one inch, the inverse of the distance
        - *sense_ambient*: Distance under one inch (so inverse of the preceeding)
        - *sense_rgb*: R, G, B values (3 sets of uint16)

        Any combination of sense_color, sense_distance, sense_count, sense_reflectivity, 
        and sense_rgb is supported.

        Examples::

            # Basic distance sensor
            @attach(VisionSensor, name='vision', capabilities=['sense_color'])
            # Or use the capability Enum
            @attach(VisionSensor, name='vision', capabilities=[VisionSensor.capability.sense_color])

            # Distance and color sensor
            @attach(VisionSensor, name='vision', capabilities=['sense_color', 'sense_distance'])

            # Distance and rgb sensor with different thresholds to trigger updates
            @attach(VisionSensor, name='vision', capabilities=[('sense_color', 1), ('sense_rgb', 5)])

        The values returned by the sensor will always be available in the instance variable
        `self.value`.  For example, when the `sense_color` and `sense_rgb` capabilities are 
        enabled, the following values will be stored and updated::

            self.value = { VisionSensor.capability.sense_color:  uint8,
                           VisionSensor.capability.sense_rgb: 
                                            [ uint16, uint16, uint16 ]
                         }

        Notes:
            The actual modes supported by the sensor are as follows:

            -  0 = color (0-10)
            -  1 = IR proximity (0-7)
            -  2 = count (32-bit int)
            -  3 = Reflt   (inverse of distance when closer than 1")
            -  4 = Amb  (distance when closer than 1")
            -  5 = COL (output) ?
            -  6 = RGB I
            -  7 = IR tx (output) ?
            -  8 = combined:  Color byte, Distance byte, 0xFF, Reflected light

    """

    _sensor_id = 0x0025
    capability = Enum("capability", 
                      [('sense_color', 0),
                       ('sense_distance', 1),
                       ('sense_count', 2),
                       ('sense_reflectivity', 3),
                       ('sense_ambient', 4),
                       ('sense_rgb', 6),
                       ])

    datasets = { capability.sense_color: (1, 1),
                 capability.sense_distance: (1, 1),
                 capability.sense_count: (1, 4),  # 4-bytes (32-bit)
                 capability.sense_reflectivity: (1, 1),
                 capability.sense_ambient: (1, 1),
                 capability.sense_rgb: (3, 2)   # 3 16-bit values
                }

    allowed_combo = [ capability.sense_color,
                      capability.sense_distance,
                      capability.sense_count,
                      capability.sense_reflectivity,
                      capability.sense_rgb,
                    ]

class InternalTiltSensor(Peripheral):
    """
        Access the internal tilt sensor in the Boost Move Hub.
        
        The various modes are:

        - **sense_angle** - X, Y angles.  Both are 0 if hub is lying flat with button up
        - **sense_tilt** - value from 0-9 if hub is tilted around any of its axis. Seems to be
          a rough mesaure of how much the hub is tilted away from lying flat.
          There is no update for just a translation along an axis
        - **sense_orientation** - returns one of the nine orientations below (0-9)
            - `InternalTiltSensor.orientation`.up = flat with button on top
            - `InternalTiltSensor.orientation`.right - standing up on side closest to button
            - `InternalTiltSensor.orientation`.left - standing up on side furthest from button
            - `InternalTiltSensor.orientation`.far_side - on long face facing away
            - `InternalTiltSensor.orientation`.near_side -  on long face facing you
            - `InternalTiltSensor.orientation`.down - upside down
        - **sense_impact** - 32-bit count of impacts to sensor
        - **sense_acceleration_3_axis** - 3 bytes of raw accelerometer data.

        Any combination of the above modes are allowed.

        Examples::

            # Basic tilt sensor
            @attach(InternalTiltSensor, name='tilt', capabilities=['sense_tilt'])
            # Or use the capability Enum
            @attach(InternalTiltSensor, name='tilt', capabilities=[InternalTiltSensor.sense_tilt])

            # Tilt and orientation sensor
            @attach(InternalTiltSensor, name='tilt', capabilities=['sense_tilt, sense_orientation'])

        The values returned by the sensor will always be available in the
        instance variable `self.value`.  For example, when the `sense_angle`
        and `sense_orientation` capabilities are enabled, the following values
        will be stored and updated::

            self.value = { InternalTiltSensor.capability.sense_angle:  [uint8, uint8],
                           InternalTiltSensor.capability.sense_orientation: 
                                            Enum(InternalTiltSensor.orientation)
                         }
    """
    _sensor_id = 0x0028
    capability = Enum("capability", 
                      [('sense_angle', 0),
                       ('sense_tilt', 1),
                       ('sense_orientation', 2),
                       ('sense_impact', 3),
                       ('sense_acceleration_3_axis', 4),
                       ])

    datasets = { capability.sense_angle: (2, 1),
                 capability.sense_tilt: (1, 1),
                 capability.sense_orientation: (1, 1),  
                 capability.sense_impact: (1, 4),
                 capability.sense_acceleration_3_axis: (3, 1),
                }

    allowed_combo = [ capability.sense_angle,
                      capability.sense_tilt,
                      capability.sense_orientation,
                      capability.sense_impact,
                      capability.sense_acceleration_3_axis,
                    ]

    orientation = Enum('orientation', 
                        {   'up': 0,
                            'right': 1, 
                            'left': 2, 
                            'far_side':3,
                            'near_side':4,
                            'down':5,
                        })


    def update_value(self, msg_bytes):
        """If sense_orientation, then substitute the `IntenalTiltSensor.orientation`
           enumeration value into the self.value dict.  Otherwise, don't do anything
           special to the self.value dict.
        """
        super().update_value(msg_bytes)
        so = self.capability.sense_orientation
        if so in self.value:
            self.value[so] = self.orientation(self.value[so])


class LED(Peripheral):
    """ Changes the LED color on the Hubs::

            @attach(LED, name='hub_led')

            self.hub_led.set_output(Color.red)

        Warnings:
            No support yet for the standalone LEDs that connect to the Hub ports.

    """
    _sensor_id = 0x0017

    async def set_color(self, color: Color):
        """ Converts a Color enumeration to a color value"""

        # For now, only support preset colors
        assert isinstance(color, Color)
        col = color.value
        assert col < 11
        mode = 0
        await self.set_output(mode, col)
        #b = [0x00, 0x81, self.port, 0x11, 0x51, mode, col ]
        #await self.message_info(f'set color to {color}')

class TrainMotor(Motor):
    """
        Connects to the train motors.

        TrainMotor has no sensing capabilities and only supports a single output mode that
        sets the speed.

        Examples::

             @attach(TrainMotor, name='train')

        And then within the run body, use::

            await self.train.set_speed(speed)

        Attributes:
            speed (int) : Keep track of the current speed in order to ramp it

        See Also:
            :class:`InternalMotor`
    """
    _sensor_id = 0x0002



class RemoteButtons(Peripheral):
    """Represents one set of '+', '-', 'red' buttons on the PoweredHub Remote

       Each remote has two sets of buttons, on the left and right side.  Pick the one
       your want to attach to by using the port argument with either Port.L or Port.R.

       There are actually a few different modes that the hardware supports, but we are
       only going to use one of them called 'KEYSD' (see the notes in the documentation on the
       raw values reported by the hub).  This mode makes the remote send three values back
       in a list.  To access each button state, there are three helper methods provided 
       (see below)

       Examples::

            # Basic connection to the left buttons
            @attach(RemoteButtons, name='left_buttons', port=RemoteButtons.Port.L)

            # Getting values back in the handler
            async def left_buttons_change(self):

                is_plus_pressed = self.left_buttons.plus_pressed()
                is_minus_pressed = self.left_buttons.minus_pressed()
                is_red_pressed = self.left_buttons.red_pressed()

    """

    _sensor_id = 0x0037
    Port = Enum('Port', 'L R', start=0)
    Button = IntEnum('Button', 'PLUS RED MINUS', start=0)
    """The button index in the value list returned by the sensor"""

    capability = Enum('capability', {'sense_press':4},)

    datasets = { capability.sense_press: (3,1) }
    allowed_combo = []

    def __init__(self, name, port=None, capabilities=[]):
        """Maps the port names `L`, `R`"""
        if port:
            port = port.value
        super().__init__(name, port, capabilities)

    def plus_pressed(self):
        """Return whether `value` reflects that the PLUS button is pressed"""
        button_list = self.value[self.capability.sense_press]
        return button_list[self.Button.PLUS] == 1
    def minus_pressed(self):
        """Return whether `value` reflects that the MINUS button is pressed"""
        button_list = self.value[self.capability.sense_press]
        return button_list[self.Button.MINUS] == 1
    def red_pressed(self):
        """Return whether `value` reflects that the RED button is pressed"""
        button_list = self.value[self.capability.sense_press]
        return button_list[self.Button.RED] == 1

class Button(Peripheral):
    """ Register to be notified of button presses on the Hub (Boost or PoweredUp)

        This is actually a slight hack, since the Hub button is not a peripheral that is 
        attached like other sensors in the Lego protocol.  Instead, the buttons are accessed
        through Hub property messages.  We abstract away these special messages to make the
        button appear to be like any other peripheral sensor.

        Examples::

            @attach(Button, name='hub_btn')

        Notes:
            Since there is no attach I/O message from the hub to trigger the
            :func:`activate_updates` method, we instead insert a fake
            "attaach" message from this fake sensor on port 255 in the
            `BLEventQ.get_messages` method that is used to register for updates
            from a given sensor.

    """
    _sensor_id = 0x0005
    """Piggy back the hub button off the normal peripheral button id 0x0005.
       Might need to change this in the future"""

    capability = Enum('capability', {'sense_press':0})

    datasets = { capability.sense_press: (1,1)
               }
    allowed_combo = [capability.sense_press]

    def __init__(self, name, port=None, capabilities=[]):
        """Call super-class with port set to 255 """
        super().__init__(name, 255, capabilities)

    async def activate_updates(self):
        """Use a special Hub Properties button message updates activation message"""
        self.value = {}
        for cap in self.capabilities:
            self.value[cap] = [None]*self.datasets[cap][0]

        b = [0x00, 0x01, 0x02, 0x02]  # Button reports from "Hub Properties Message Type"
        await self.send_message(f'Activate button reports: port {self.port}', b) 

class DuploTrainMotor(Motor):
    """Train Motor on Duplo Trains

       Make sure that the train is sitting on the ground (the front wheels need to keep rotating) in 
       order to keep the train motor powered.  If you pick up the train, the motor will stop operating
       withina few seconds.

       Examples::

            @attach(DuploTrainMotor, name='motor')

       And then within the run body, use::

            await self.train.set_speed(speed)

       Attributes:
            speed (int): Keep track of the current speed in order to ramp it

       See Also:
            :class:`TrainMotor` for connecting to a PoweredUp train motor
    """
    _sensor_id = 0x0029

class DuploSpeedSensor(Peripheral):
    """Speedometer on Duplo train base that measures front wheel speed.

       This can measure the following values:

       - *sense_speed*: Returns the speed of the front wheels
       - *sense_count*: Keeps count of the number of revolutions the front wheels have spun

       Either or both can be enabled for measurement. 

       Examples::

            # Report speed changes
            @attach(DuploSpeedSensor, name='speed_sensor', capabilities=['sense_speed'])

            # Report all
            @attach(DuploSpeedSensor, name='speed_sensor', capabilities=['sense_speed', 'sense_count'])

       The values returned by the sensor will be in `self.value`.  For the first example, get the
       current speed by::

            speed = self.speed_sensor.value
        
       For the second example, the two values will be in a dict::

            speed = self.speed_sensor.value[DuploSpeedSensor.sense_speed]
            revs  = self.speed_sensor.value[DuploSpeedSensor.sense_count]

    """
    _sensor_id = 0x002C
    capability = Enum("capability", 
                      [('sense_speed', 0),
                       ('sense_count', 1),
                       ])

    datasets = { capability.sense_speed: (1, 2),
                 capability.sense_count: (1, 4),
                }

    allowed_combo = [ capability.sense_speed,
                      capability.sense_count,
                    ]

class DuploVisionSensor(Peripheral):
    """ Access the Duplo Vision/Distance Sensor

        Only the sensing capabilities of this sensor is supported right now.

        - *sense_color*: Returns one of the 10 predefined colors
        - *sense_ctag*: Returns one of the 10 predefined tags
        - *sense_reflectivity*: Under distances of one inch, the inverse of the distance
        - *sense_rgb*: R, G, B values (3 sets of uint16)

        Any combination of sense_color, sense_ctag, sense_reflectivity, 
        and sense_rgb is supported.

        Examples::

            # Basic color sensor
            @attach(DuploVisionSensor, 'vision', capabilities=['sense_color'])
            # Or use the capability Enum
            @attach(DuploVisionSensor, 'vision', capabilities=[DuploVisionSensor.capability.sense_color])

            # Ctag and reflectivity sensor
            @attach(DuploVisionSensor, 'vision', capabilities=['sense_ctag', 'sense_reflectivity'])

            # Distance and rgb sensor with different thresholds to trigger updates
            @attach(DuploVisionSensor, 'vision', capabilities=[('sense_color', 1), ('sense_rgb', 5)])

        The values returned by the sensor will always be available in the instance variable
        `self.value`.  For example, when the `sense_color` and `sense_rgb` capabilities are 
        enabled, the following values will be stored and updated::

            self.value = { DuploVisionSensor.capability.sense_color:  uint8,
                           DuploVisionSensor.capability.sense_rgb: 
                                            [ uint16, uint16, uint16 ]
                         }

        Notes:
            The actual modes supported by the sensor are as follows:

            -  0 = color (0-10)
            -  1 = ctag (32-bit int)
            -  2 = Reflt   (inverse of distance when closer than 1")
            -  3 = RGB I
    """
    _sensor_id = 0x002B
    capability = Enum("capability", 
                      [('sense_color', 0),
                       ('sense_ctag', 1),
                       ('sense_reflectivity', 2),
                       ('sense_rgb', 3),
                       ])

    datasets = { capability.sense_color: (1, 1),
                 capability.sense_ctag: (1, 1),  # 4-bytes (32-bit)
                 capability.sense_reflectivity: (1, 1),
                 capability.sense_rgb: (3, 2)   # 3 16-bit values
                }

    allowed_combo = [ capability.sense_color,
                      capability.sense_ctag,
                      capability.sense_reflectivity,
                      capability.sense_rgb,
                    ]

class DuploSpeaker(Peripheral):
    """Plays one of five preset sounds through the Duplo built-in speaker

       See :class:`sounds` for the list.


       Examples::

            @attach(DuploSpeaker, name='speaker')
            ...
            await self.speaker.play_sound(DuploSpeaker.sounds.brake)
           
       Notes:
            Uses Mode 1 to play the presets

    """
    _sensor_id = 0x002A
    sounds = Enum('sounds', { 'brake': 3,
                              'station': 5,
                              'water': 7,
                              'horn': 9,
                              'steam': 10,
                              })

    async def activate_updates(self):
        """For some reason, even though the speaker is an output device
           we need to send a Port Input Format Setup command (0x41) to enable
           notifications.  Otherwise, none of the sound output commands will play.
        """
        mode = 1
        b = [0x00, 0x41, self.port, mode, 0x01, 0x00, 0x00, 0x00, 0x01]
        await self.send_message('Activate DUPLO Speaker: port {self.port}', b)

    async def play_sound(self, sound):
        assert isinstance(sound, self.sounds), 'Can only play sounds that are enums (DuploSpeaker.sounds.brake, etc)'
        mode = 1
        self.message_info(f'Playing sound {sound.name}:{sound.value}')
        await self.set_output(mode, sound.value)




1			# Copyright 2019 Virantha N. Ekanayake
2			#
3			# Licensed under the Apache License, Version 2.0 (the "License");
4			# you may not use this file except in compliance with the License.
5			# You may obtain a copy of the License at
6			#
7			# http://www.apache.org/licenses/LICENSE-2.0
8			#
9			# Unless required by applicable law or agreed to in writing, software
10			# distributed under the License is distributed on an "AS IS" BASIS,
11			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12			# See the License for the specific language governing permissions and
13			# limitations under the License.
14
15			"""Actual sensor and motor peripheral definitions from Boost and PoweredUp
16			"""
17			from curio import sleep, current_task, spawn # Needed for motor speed ramp
18
19			from enum import Enum, IntEnum
20			from .const import Color
21
22			from .peripheral import Peripheral, Motor
23
24
25			class InternalMotor(Peripheral):
26			""" Access the internal motor(s) in the Boost Move Hub.
27
28			Unlike the train motors, these motors have a built-in sensor for sending back
29			the motor's current speed and position. You don't need to use the sensors, and
30			can treat this as strictly an output.
31
32			Examples::
33
34			# Basic connection to the motor on Port A
35			@attach(InternalMotor, name='left_motor', port=InternalMotor.Port.A)
36
37			# Basic connection to both motors at the same time (virtual I/O port).
38			# Any speed command will cause both motors to rotate at the same speed
39			@attach(InternalMotor, name='motors', port=InternalMotor.Port.AB)
40
41			# Report back when motor speed changes. You must have a motors_change method defined
42			@attach(InternalMotor, name='motors', port=InternalMotor.Port.A, capabilities=['sense_speed'])
43			# Only report back when speed change exceeds 5 units
44			@attach(InternalMotor, name='motors', port=InternalMotor.Port.A, capabilities=[('sense_speed', 5)])
45
46			See Also:
47			:class:`TrainMotor` for connecting to a train motor
48
49			"""
50			_sensor_id = 0x0027
51			_DEFAULT_THRESHOLD=2
52			"""Set to 2 to avoid a lot of updates since the speed seems to oscillate a lot"""
53
54			capability = Enum("capability", {"sense_speed":1, "sense_pos":2})
55
56			Port = Enum('Port', 'A B AB', start=0)
57			"""Address either motor A or Motor B, or both AB at the same time"""
58
59			# Dict of cap: (num_datasets, bytes_per_dataset)
60			datasets = { capability.sense_speed: (1, 1),
61			capability.sense_pos: (1, 4),
62			}
63			""" Dict of (num_datasets, bytes_per_dataset).
64			`sense_speed` (1-byte), and `sense_pos` (uint32)"""
65
66			allowed_combo = [ capability.sense_speed,
67			capability.sense_pos,
68			]
69
70			def __init__(self, name, port=None, capabilities=[]):
71			"""Maps the port names `A`, `B`, `AB` to hard-coded port numbers"""
72			if port:
73			port_map = [55, 56, 57]
74			port = port_map[port.value]
75			self.speed = 0
76			super().__init__(name, port, capabilities)
77
78			async def set_speed(self, speed):
79			"""Sets the speed of the motor, and calls the :func:`bricknil.peripheral.Peripheral._convert_speed_to_val` method
80			to do some sanity checking and bounding.
81
82			Args:
83			speed (int) : -100 to 100 (I believe this is a percentage)
84			"""
85			self.speed = speed
86			speed = self._convert_speed_to_val(speed)
87			mode = 0
88			await self.set_output(mode, speed)
89
90			async def ramp_speed2(self, target_speed, ramp_time_ms):
91
92			# Set acceleration profile
93			delta_speed = target_speed - self.speed
94			zero_100_ramp_time_ms = int(ramp_time_ms/delta_speed * 100.0)
95			zero_100_ramp_time_ms = zero_100_ramp_time_ms % 10000 # limit time to 10s
96
97			hi = (zero_100_ramp_time_ms >> 8) & 255
98			lo = zero_100_ramp_time_ms & 255
99
100			profile = 1
101			b = [0x00, 0x81, self.port, 0x01, 0x05, 10, 10, profile]
102			await self.send_message(f'set accel profile {zero_100_ramp_time_ms} {hi} {lo} ', b)
103			b = [0x00, 0x81, self.port, 0x01, 0x07, self._convert_speed_to_val(target_speed), 80, 1]
104			await self.send_message('set speed', b)
105
106			class VisionSensor(Peripheral):
107			""" Access the Boost Vision/Distance Sensor
108
109			Only the sensing capabilities of this sensor is supported right now.
110
111			- sense_color: Returns one of the 10 predefined colors
112			- sense_distance: Distance from 0-7 in roughly inches
113			- sense_count: Running count of waving your hand/item over the sensor (32-bit)
114			- sense_reflectivity: Under distances of one inch, the inverse of the distance
115			- sense_ambient: Distance under one inch (so inverse of the preceeding)
116			- sense_rgb: R, G, B values (3 sets of uint16)
117
118			Any combination of sense_color, sense_distance, sense_count, sense_reflectivity,
119			and sense_rgb is supported.
120
121			Examples::
122
123			# Basic distance sensor
124			@attach(VisionSensor, name='vision', capabilities=['sense_color'])
125			# Or use the capability Enum
126			@attach(VisionSensor, name='vision', capabilities=[VisionSensor.capability.sense_color])
127
128			# Distance and color sensor
129			@attach(VisionSensor, name='vision', capabilities=['sense_color', 'sense_distance'])
130
131			# Distance and rgb sensor with different thresholds to trigger updates
132			@attach(VisionSensor, name='vision', capabilities=[('sense_color', 1), ('sense_rgb', 5)])
133
134			The values returned by the sensor will always be available in the instance variable
135			`self.value`. For example, when the `sense_color` and `sense_rgb` capabilities are
136			enabled, the following values will be stored and updated::
137
138			self.value = { VisionSensor.capability.sense_color: uint8,
139			VisionSensor.capability.sense_rgb:
140			[ uint16, uint16, uint16 ]
141			}
142
143			Notes:
144			The actual modes supported by the sensor are as follows:
145
146			- 0 = color (0-10)
147			- 1 = IR proximity (0-7)
148			- 2 = count (32-bit int)
149			- 3 = Reflt (inverse of distance when closer than 1")
150			- 4 = Amb (distance when closer than 1")
151			- 5 = COL (output) ?
152			- 6 = RGB I
153			- 7 = IR tx (output) ?
154			- 8 = combined: Color byte, Distance byte, 0xFF, Reflected light
155
156			"""
157
158			_sensor_id = 0x0025
159			capability = Enum("capability",
160			[('sense_color', 0),
161			('sense_distance', 1),
162			('sense_count', 2),
163			('sense_reflectivity', 3),
164			('sense_ambient', 4),
165			('sense_rgb', 6),
166			])
167
168			datasets = { capability.sense_color: (1, 1),
169			capability.sense_distance: (1, 1),
170			capability.sense_count: (1, 4), # 4-bytes (32-bit)
171			capability.sense_reflectivity: (1, 1),
172			capability.sense_ambient: (1, 1),
173			capability.sense_rgb: (3, 2) # 3 16-bit values
174			}
175
176			allowed_combo = [ capability.sense_color,
177			capability.sense_distance,
178			capability.sense_count,
179			capability.sense_reflectivity,
180			capability.sense_rgb,
181			]
182
183			class InternalTiltSensor(Peripheral):
184			"""
185			Access the internal tilt sensor in the Boost Move Hub.
186
187			The various modes are:
188
189			- sense_angle - X, Y angles. Both are 0 if hub is lying flat with button up
190			- sense_tilt - value from 0-9 if hub is tilted around any of its axis. Seems to be
191			a rough mesaure of how much the hub is tilted away from lying flat.
192			There is no update for just a translation along an axis
193			- sense_orientation - returns one of the nine orientations below (0-9)
194			- `InternalTiltSensor.orientation`.up = flat with button on top
195			- `InternalTiltSensor.orientation`.right - standing up on side closest to button
196			- `InternalTiltSensor.orientation`.left - standing up on side furthest from button
197			- `InternalTiltSensor.orientation`.far_side - on long face facing away
198			- `InternalTiltSensor.orientation`.near_side - on long face facing you
199			- `InternalTiltSensor.orientation`.down - upside down
200			- sense_impact - 32-bit count of impacts to sensor
201			- sense_acceleration_3_axis - 3 bytes of raw accelerometer data.
202
203			Any combination of the above modes are allowed.
204
205			Examples::
206
207			# Basic tilt sensor
208			@attach(InternalTiltSensor, name='tilt', capabilities=['sense_tilt'])
209			# Or use the capability Enum
210			@attach(InternalTiltSensor, name='tilt', capabilities=[InternalTiltSensor.sense_tilt])
211
212			# Tilt and orientation sensor
213			@attach(InternalTiltSensor, name='tilt', capabilities=['sense_tilt, sense_orientation'])
214
215			The values returned by the sensor will always be available in the
216			instance variable `self.value`. For example, when the `sense_angle`
217			and `sense_orientation` capabilities are enabled, the following values
218			will be stored and updated::
219
220			self.value = { InternalTiltSensor.capability.sense_angle: [uint8, uint8],
221			InternalTiltSensor.capability.sense_orientation:
222			Enum(InternalTiltSensor.orientation)
223			}
224			"""
225			_sensor_id = 0x0028
226			capability = Enum("capability",
227			[('sense_angle', 0),
228			('sense_tilt', 1),
229			('sense_orientation', 2),
230			('sense_impact', 3),
231			('sense_acceleration_3_axis', 4),
232			])
233
234			datasets = { capability.sense_angle: (2, 1),
235			capability.sense_tilt: (1, 1),
236			capability.sense_orientation: (1, 1),
237			capability.sense_impact: (1, 4),
238			capability.sense_acceleration_3_axis: (3, 1),
239			}
240
241			allowed_combo = [ capability.sense_angle,
242			capability.sense_tilt,
243			capability.sense_orientation,
244			capability.sense_impact,
245			capability.sense_acceleration_3_axis,
246			]
247
248			orientation = Enum('orientation',
249			{ 'up': 0,
250			'right': 1,
251			'left': 2,
252			'far_side':3,
253			'near_side':4,
254			'down':5,
255			})
256
257
258			def update_value(self, msg_bytes):
259			"""If sense_orientation, then substitute the `IntenalTiltSensor.orientation`
260			enumeration value into the self.value dict. Otherwise, don't do anything
261			special to the self.value dict.
262			"""
263			super().update_value(msg_bytes)
264			so = self.capability.sense_orientation
265			if so in self.value:
266			self.value[so] = self.orientation(self.value[so])
267
268
269			class LED(Peripheral):
270			""" Changes the LED color on the Hubs::
271
272			@attach(LED, name='hub_led')
273
274			self.hub_led.set_output(Color.red)
275
276			Warnings:
277			No support yet for the standalone LEDs that connect to the Hub ports.
278
279			"""
280			_sensor_id = 0x0017
281
282			async def set_color(self, color: Color):
283			""" Converts a Color enumeration to a color value"""
284
285			# For now, only support preset colors
286			assert isinstance(color, Color)
287			col = color.value
288			assert col < 11
289			mode = 0
290			await self.set_output(mode, col)
291			#b = [0x00, 0x81, self.port, 0x11, 0x51, mode, col ]
292			#await self.message_info(f'set color to {color}')
293
294			class TrainMotor(Motor):
295			"""
296			Connects to the train motors.
297
298			TrainMotor has no sensing capabilities and only supports a single output mode that
299			sets the speed.
300
301			Examples::
302
303			@attach(TrainMotor, name='train')
304
305			And then within the run body, use::
306
307			await self.train.set_speed(speed)
308
309			Attributes:
310			speed (int) : Keep track of the current speed in order to ramp it
311
312			See Also:
313			:class:`InternalMotor`
314			"""
315			_sensor_id = 0x0002
316
317
318
319			class RemoteButtons(Peripheral):
320			"""Represents one set of '+', '-', 'red' buttons on the PoweredHub Remote
321
322			Each remote has two sets of buttons, on the left and right side. Pick the one
323			your want to attach to by using the port argument with either Port.L or Port.R.
324
325			There are actually a few different modes that the hardware supports, but we are
326			only going to use one of them called 'KEYSD' (see the notes in the documentation on the
327			raw values reported by the hub). This mode makes the remote send three values back
328			in a list. To access each button state, there are three helper methods provided
329			(see below)
330
331			Examples::
332
333			# Basic connection to the left buttons
334			@attach(RemoteButtons, name='left_buttons', port=RemoteButtons.Port.L)
335
336			# Getting values back in the handler
337			async def left_buttons_change(self):
338
339			is_plus_pressed = self.left_buttons.plus_pressed()
340			is_minus_pressed = self.left_buttons.minus_pressed()
341			is_red_pressed = self.left_buttons.red_pressed()
342
343			"""
344
345			_sensor_id = 0x0037
346			Port = Enum('Port', 'L R', start=0)
347			Button = IntEnum('Button', 'PLUS RED MINUS', start=0)
348			"""The button index in the value list returned by the sensor"""
349
350			capability = Enum('capability', {'sense_press':4},)
351
352			datasets = { capability.sense_press: (3,1) }
353			allowed_combo = []
354
355			def __init__(self, name, port=None, capabilities=[]):
356			"""Maps the port names `L`, `R`"""
357			if port:
358			port = port.value
359			super().__init__(name, port, capabilities)
360
361			def plus_pressed(self):
362			"""Return whether `value` reflects that the PLUS button is pressed"""
363			button_list = self.value[self.capability.sense_press]
364			return button_list[self.Button.PLUS] == 1
365			def minus_pressed(self):
366			"""Return whether `value` reflects that the MINUS button is pressed"""
367			button_list = self.value[self.capability.sense_press]
368			return button_list[self.Button.MINUS] == 1
369			def red_pressed(self):
370			"""Return whether `value` reflects that the RED button is pressed"""
371			button_list = self.value[self.capability.sense_press]
372			return button_list[self.Button.RED] == 1
373
374			class Button(Peripheral):
375			""" Register to be notified of button presses on the Hub (Boost or PoweredUp)
376
377			This is actually a slight hack, since the Hub button is not a peripheral that is
378			attached like other sensors in the Lego protocol. Instead, the buttons are accessed
379			through Hub property messages. We abstract away these special messages to make the
380			button appear to be like any other peripheral sensor.
381
382			Examples::
383
384			@attach(Button, name='hub_btn')
385
386			Notes:
387			Since there is no attach I/O message from the hub to trigger the
388			:func:`activate_updates` method, we instead insert a fake
389			"attaach" message from this fake sensor on port 255 in the
390			`BLEventQ.get_messages` method that is used to register for updates
391			from a given sensor.
392
393			"""
394			_sensor_id = 0x0005
395			"""Piggy back the hub button off the normal peripheral button id 0x0005.
396			Might need to change this in the future"""
397
398			capability = Enum('capability', {'sense_press':0})
399
400			datasets = { capability.sense_press: (1,1)
401			}
402			allowed_combo = [capability.sense_press]
403
404			def __init__(self, name, port=None, capabilities=[]):
405			"""Call super-class with port set to 255 """
406			super().__init__(name, 255, capabilities)
407
408			async def activate_updates(self):
409			"""Use a special Hub Properties button message updates activation message"""
410			self.value = {}
411			for cap in self.capabilities:
412			self.value[cap] = [None]*self.datasets[cap][0]
413
414			b = [0x00, 0x01, 0x02, 0x02] # Button reports from "Hub Properties Message Type"
415			await self.send_message(f'Activate button reports: port {self.port}', b)
416
417			class DuploTrainMotor(Motor):
418			"""Train Motor on Duplo Trains
419
420			Make sure that the train is sitting on the ground (the front wheels need to keep rotating) in
421			order to keep the train motor powered. If you pick up the train, the motor will stop operating
422			withina few seconds.
423
424			Examples::
425
426			@attach(DuploTrainMotor, name='motor')
427
428			And then within the run body, use::
429
430			await self.train.set_speed(speed)
431
432			Attributes:
433			speed (int): Keep track of the current speed in order to ramp it
434
435			See Also:
436			:class:`TrainMotor` for connecting to a PoweredUp train motor
437			"""
438			_sensor_id = 0x0029
439
440			class DuploSpeedSensor(Peripheral):
441			"""Speedometer on Duplo train base that measures front wheel speed.
442
443			This can measure the following values:
444
445			- sense_speed: Returns the speed of the front wheels
446			- sense_count: Keeps count of the number of revolutions the front wheels have spun
447
448			Either or both can be enabled for measurement.
449
450			Examples::
451
452			# Report speed changes
453			@attach(DuploSpeedSensor, name='speed_sensor', capabilities=['sense_speed'])
454
455			# Report all
456			@attach(DuploSpeedSensor, name='speed_sensor', capabilities=['sense_speed', 'sense_count'])
457
458			The values returned by the sensor will be in `self.value`. For the first example, get the
459			current speed by::
460
461			speed = self.speed_sensor.value
462
463			For the second example, the two values will be in a dict::
464
465			speed = self.speed_sensor.value[DuploSpeedSensor.sense_speed]
466			revs = self.speed_sensor.value[DuploSpeedSensor.sense_count]
467
468			"""
469			_sensor_id = 0x002C
470			capability = Enum("capability",
471			[('sense_speed', 0),
472			('sense_count', 1),
473			])
474
475			datasets = { capability.sense_speed: (1, 2),
476			capability.sense_count: (1, 4),
477			}
478
479			allowed_combo = [ capability.sense_speed,
480			capability.sense_count,
481			]
482
483			class DuploVisionSensor(Peripheral):
484			""" Access the Duplo Vision/Distance Sensor
485
486			Only the sensing capabilities of this sensor is supported right now.
487
488			- sense_color: Returns one of the 10 predefined colors
489			- sense_ctag: Returns one of the 10 predefined tags
490			- sense_reflectivity: Under distances of one inch, the inverse of the distance
491			- sense_rgb: R, G, B values (3 sets of uint16)
492
493			Any combination of sense_color, sense_ctag, sense_reflectivity,
494			and sense_rgb is supported.
495
496			Examples::
497
498			# Basic color sensor
499			@attach(DuploVisionSensor, 'vision', capabilities=['sense_color'])
500			# Or use the capability Enum
501			@attach(DuploVisionSensor, 'vision', capabilities=[DuploVisionSensor.capability.sense_color])
502
503			# Ctag and reflectivity sensor
504			@attach(DuploVisionSensor, 'vision', capabilities=['sense_ctag', 'sense_reflectivity'])
505
506			# Distance and rgb sensor with different thresholds to trigger updates
507			@attach(DuploVisionSensor, 'vision', capabilities=[('sense_color', 1), ('sense_rgb', 5)])
508
509			The values returned by the sensor will always be available in the instance variable
510			`self.value`. For example, when the `sense_color` and `sense_rgb` capabilities are
511			enabled, the following values will be stored and updated::
512
513			self.value = { DuploVisionSensor.capability.sense_color: uint8,
514			DuploVisionSensor.capability.sense_rgb:
515			[ uint16, uint16, uint16 ]
516			}
517
518			Notes:
519			The actual modes supported by the sensor are as follows:
520
521			- 0 = color (0-10)
522			- 1 = ctag (32-bit int)
523			- 2 = Reflt (inverse of distance when closer than 1")
524			- 3 = RGB I
525			"""
526			_sensor_id = 0x002B
527			capability = Enum("capability",
528			[('sense_color', 0),
529			('sense_ctag', 1),
530			('sense_reflectivity', 2),
531			('sense_rgb', 3),
532			])
533
534			datasets = { capability.sense_color: (1, 1),
535			capability.sense_ctag: (1, 1), # 4-bytes (32-bit)
536			capability.sense_reflectivity: (1, 1),
537			capability.sense_rgb: (3, 2) # 3 16-bit values
538			}
539
540			allowed_combo = [ capability.sense_color,
541			capability.sense_ctag,
542			capability.sense_reflectivity,
543			capability.sense_rgb,
544			]
545
546			class DuploSpeaker(Peripheral):
547			"""Plays one of five preset sounds through the Duplo built-in speaker
548
549			See :class:`sounds` for the list.
550
551
552			Examples::
553
554			@attach(DuploSpeaker, name='speaker')
555			...
556			await self.speaker.play_sound(DuploSpeaker.sounds.brake)
557
558			Notes:
559			Uses Mode 1 to play the presets
560
561			"""
562			_sensor_id = 0x002A
563			sounds = Enum('sounds', { 'brake': 3,
564			'station': 5,
565			'water': 7,
566			'horn': 9,
567			'steam': 10,
568			})
569
570			async def activate_updates(self):
571			"""For some reason, even though the speaker is an output device
572			we need to send a Port Input Format Setup command (0x41) to enable
573			notifications. Otherwise, none of the sound output commands will play.
574			"""
575			mode = 1
576			b = [0x00, 0x41, self.port, mode, 0x01, 0x00, 0x00, 0x00, 0x01]
577			await self.send_message('Activate DUPLO Speaker: port {self.port}', b)
578
579			async def play_sound(self, sound):
580			assert isinstance(sound, self.sounds), 'Can only play sounds that are enums (DuploSpeaker.sounds.brake, etc)'
581			mode = 1
582			self.message_info(f'Playing sound {sound.name}:{sound.value}')
583			await self.set_output(mode, sound.value)
584
585
586

virantha / bricknil

GitHub Access Token became invalid

Push — master ( 72bf9f...f85f9f )

bricknil.sensor.Button.activate_updates() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like