BrickNil - Control LEGO Bluetooth Sensors and Motors with Python

image_pypi image_downloads image_license passing quality Coverage Status

BrickNil [*] provides an easy way to connect to and program LEGO® Bluetooth hubs (including the PoweredUp Passenger Train 60197 and Cargo Train 60198 sets, and the Lego Duplo Steam Train 10874 and Cargo Train 10875 ) using Python on OS X and Linux. This work was inspired by this EuroBricks thread, and the NodeJS Powered-Up library. Unlike the official apps available from LEGO, BrickNil allows you to control multiple hubs with a single program, which lets you easily scale to programming large interactive systems.

BrickNil requires modern Python (designed and tested for 3.7) and uses asynchronous event programming built on top of the Curio async library. As an aside, the choice of async library is fairly arbitrary; and enabling another library such as asyncio or Trio should be straightforward.

An example BrickNil program for controlling the Train motor speed is shown below:

from curio import sleep
from bricknil import attach, start
from bricknil.hub import PoweredUpHub
from bricknil.sensor import TrainMotor

@attach(TrainMotor, name='motor')
class Train(PoweredUpHub):

    async def run(self):
        for i in range(2):  # Repeat this control two times
            await self.motor.ramp_speed(80,5000) # Ramp speed to 80 over 5 seconds
            await sleep(6)
            await self.motor.ramp_speed(0,1000)  # Brake to 0 over 1 second
            await sleep(2)

async def system():
    train = Train('My train')

if __name__ == '__main__':
    start(system)
[*]BrickNil’s name comes from the word “Nil” (නිල්) in Sinhala which means Blue (as in Bluetooth)

Features

  • Supports the following LEGO® Bluetooth systems:
  • Supports the following actuators/sensors:
    • Internal motors
    • Train motors
    • Hub LED color
    • Boost vision sensor (color, distance)
    • Boost internal tilt/orientation/accelerometer
    • Boost external motor
    • External light
    • Hub buttons
    • Wedo external motor
    • Wedo tiltand motion sensors
  • Fully supports Python asynchronous keywords and coroutines
    • Allows expressive concurrent programming using async/await syntax
    • The current implementation uses the async library Curio by David Beazley
  • Cross-platform
    • Uses the Adafruit Bluefruit BluetoothLE library for Mac OS X
    • Uses the Bleak Bluetooth library for Linux and Win10; also tested on Raspberry Pi.

Building a simple train controller

Let’s build a simple program to control a LEGO® PoweredUp train. The first thing to do is create a class that subclasses the Bluetooth hub that we’ll be using:

from bricknil.hub import PoweredUpHub

class Train(PoweredUpHub):

   async def run(self):
      ...

The run async function (it’s actually a coroutine) will contain the code that will control everything attached to this hub. Speaking of which, because we’ll be wanting to control the train motor connected to this hub, we’d better attach it to the code like so:

from bricknil.hub import PoweredUpHub
from bricknil.sensor import TrainMotor

@attach(TrainMotor, name='motor')
class Train(PoweredUpHub):

   async def run(self):
      ...

Now, we can access the motor functions by calling the object self.motor inside run. For example, let’s say that we wanted to set the motor speed to 50 (the allowed range is -100 to 100 where negative numbers are reverse speeds):

from curio import sleep
from bricknil.hub import PoweredUpHub
from bricknil.sensor import TrainMotor

@attach(TrainMotor, name='motor')
class Train(PoweredUpHub):

   async def run(self):
      await self.motor.set_speed(50)
      await sleep(5)  # Wait 5 seconds before exiting

Notice that we’re using the await keyword in front of all the calls, because those are also asynchronous coroutines that will get run in the event loop. At some point, the bricknil.peripheral.Motor.set_speed() coroutine will finish executing and control will return back to the statement after it. The next statement is a call to the sleep coroutine from the curio library. It’s important to use this, instead of the regular function time.sleep because curio.sleep is a coroutine that will not block other tasks from running.

Note that we can use arbitrary Python to build our controller; suppose that we wanted to ramp the motor speed gradually to 80 over 5 seconds, and then reduce speed back to a stop in 1 second, and then repeat it over again. We could implement the run logic as:

async def run(self):
    for i in range(2):
        await self.motor.ramp_speed(80,5000)
        await sleep(5)
        await self.motor.ramp_speed(0,1000)
        await sleep(2)

The bricknil.peripheral.Motor.ramp_speed() function will ramp the speed from whatever it is currently to the target speed over the millisecond duration given (internally, it will change the train speed every 100ms). Here, you can see how things are running concurrently: we issue the ramp_speed command, that will take 5 seconds to complete in the background, so we need to make sure our control logic sleeps for 5 seconds too, to ensure the train has enough time to get up to speed, before we issue the braking command. Once the train comes to a stop, it will stay stopped for 1 second, then repeat this sequence of speed changes once more before exiting.

It’s also useful to print out what’s happening as we run our program. In order to facilitate that, there is some rudimentary logging capability built-in to bricknil via the bricknil.process.Process class that all of these concurrent processes are sub-classed from. Here’s the run coroutine with logging statements via bricknil.process.Process.message_info() enabled:

async def run(self):
    self.message_info("Running")
    for i in range(2):
        self.message_info('Increasing speed')
        await self.motor.ramp_speed(80,5000)
        await sleep(5)
        self.message_info('Coming to a stop')
        await self.motor.ramp_speed(0,1000)
        await sleep(2)

Of course, just running the above code isn’t quite enough to execute the controller. Once we have the controller logic implemented, we need to define our entire system in a separate top-level coroutine like so:

async def system():
    train = Train('My train')

This coroutine instantiates all the hubs we want to control; once we have that, we can go ahead and implement the full program that calls bricknil.start() with this system coroutine:

from curio import sleep
from bricknil import attach, start
from bricknil.hub import PoweredUpHub
from bricknil.sensor import TrainMotor
from bricknil.process import Process
import logging

@attach(TrainMotor, name='motor')
class Train(PoweredUpHub):

    async def run(self):
        self.message_info("Running")
        for i in range(2):
            self.message_info('Increasing speed')
            await self.motor.ramp_speed(80,5000)
            await sleep(5)
            self.message_info('Coming to a stop')
            await self.motor.ramp_speed(0,1000)
            await sleep(2)

async def system():
    train = Train('My train')

if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)
    start(system)

Running this program will output the following:

BLE Event Q.0: Looking for first matching hub
BLE Event Q.0: Connected to device
BLE Event Q.0: Device name HUB NO.4
BLE Event Q.0: Device id XXXX-XXXX
BLE Event Q.0: Device advertised [UUID('XXXXX')]
My train.2: Waiting for peripheral motor to attach to a port
My train.2: Running
My train.2: Increasing speed
motor.1: Starting ramp of speed: 0 -> 80 (5.0s)
motor.1: Setting speed to 0
motor.1: Setting speed to 1
motor.1: Setting speed to 3
motor.1: Setting speed to 4
...
motor.1: Setting speed to 80
My train.2: Coming to a stop
motor.1: Starting ramp of speed: 76 -> 0 (1.0s)
motor.1: Setting speed to 76
motor.1: Setting speed to 68
motor.1: Setting speed to 60
motor.1: Setting speed to 53
motor.1: Setting speed to 45
motor.1: Setting speed to 38
motor.1: Setting speed to 30
motor.1: Setting speed to 22
motor.1: Setting speed to 15
motor.1: Setting speed to 0
... repeats

Integrating a vision sensor into a simple train controller

Now let’s build a controller that sets the speed of the train depending on how close your hand is to a snensor, and will quit the program if you wave your hand more than three times in front of it.

We’ll be using the Vision Sensor that comes with the LEGO Boost robotics kit; plug the sensor into the second port of the train’s PoweredUP hub. This sensor has a multitude of different sensing abilities including distance and color, but for this example, we’re just going to use the sense_count and sense_distance capabilities. The former measures how many times it sees something pass in front of it at a distance of ~2 inches, while the latter measures roughly how many inches something is from the sensor (from 1-10 inches). For a full list of the supported capabilities, please see the API documentation at bricknil.sensor.VisionSensor.

The full program is listed at the end of this section, but let’s just go through it bit by bit. The first thing we’ll do is attach the sensor to the Train class:

@attach(VisionSensor, name='train_sensor', capabilities=['sense_count', 'sense_distance'])
@attach(TrainMotor, name='motor')
class Train(PoweredUpHub):
     ...

Anytime you attach a sensor to the system (motion, tilt, color, etc), you need to define what capabilities you want to enable; each sensor can physically provide different capabilities depending on which sensor you’re using. As soon as you attach a sensor, you need to provide a call-back coroutine that will be called whenever the sensor detects a change like so:

@attach(VisionSensor, name='train_sensor', capabilities=['sense_count', 'sense_distance'])
@attach(TrainMotor, name='motor')
class Train(PoweredUpHub):

    async def train_sensor_change(self):
        ...

The values will be provided in dictionary called self.value that is indexed by the capability. Let’s look at a practical example, and implement the logic we were discussing above:

@attach(VisionSensor, name='train_sensor', capabilities=['sense_count', 'sense_distance'])
@attach(TrainMotor, name='motor')
class Train(PoweredUpHub):

    async def train_sensor_change(self):
        distance = self.train_sensor.value[VisionSensor.capability.sense_distance]
        count = self.train_sensor.value[VisionSensor.capability.sense_count]

        if count > 3:
            # Wave your hand more than three times in front of the sensor and the program ends
            self.keep_running = False

        # The closer your hand gets to the sensor, the faster the motor runs
        self.motor_speed = (10-distance)*10

        # Flag a change
        self.sensor_change = True

Here, we get the distance and count from the value dict. If the count is greater than 3 (more than 3 hand waves), we set a flag that keeps the system running to False. Next, based on the inverse of the distance, we set a motor_speed instance variable, and then use self.sensor_change to signal to the main run routine that a sensor update has happened. Our run logic can now use these values to implement the controller:

async def run(self):
    self.motor_speed = 0
    self.keep_running = True
    self.sensor_change = False

    while self.keep_running:
        if self.sensor_change:
            await self.motor.ramp_speed(self.motor_speed, 900)  # Ramp to new speed in 0.9 seconds
            self.sensor_change = False
        await sleep(1)

We keep running the train while keep_running flag is True; if a sensor_change is detected, we ramp the train speed to the new target self.motor_speed in 0.9 seconds, and then wait for the next sensor update, whenever that may be, at intervals of 1 second. As soon as you pass your hand more than three times in front of the sensor, the program will exit this while loop and end.

Here’s the full code:

from curio import sleep
from bricknil import attach, start
from bricknil.hub import PoweredUpHub
from bricknil.sensor import TrainMotor, VisionSensor
from bricknil.process import Process
import logging

@attach(VisionSensor, name='train_sensor', capabilities=['sense_count', 'sense_distance'])
@attach(TrainMotor, name='motor')
class Train(PoweredUpHub):

    async def train_sensor_change(self):
        distance = self.train_sensor.value[VisionSensor.capability.sense_distance]
        count = self.train_sensor.value[VisionSensor.capability.sense_count]

        if count > 3:
            # Wave your hand more than three times in front of the sensor and the program ends
            self.keep_running = False

        # The closer your hand gets to the sensor, the faster the motor runs
        self.motor_speed = (10-distance)*10

        # Flag a change
        self.sensor_change = True

    async def run(self):
        self.motor_speed = 0
        self.keep_running = True
        self.sensor_change = False

        while self.keep_running:
            if self.sensor_change:
                await self.motor.ramp_speed(self.motor_speed, 900)  # Ramp to new speed in 0.9 seconds
                self.sensor_change = False
            await sleep(1)

async def system():
    train = Train('My Train')

if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)
    start(system)

More examples

Connecting to a specific hub

If you know the BluetoothLE network address of the hub you want to connect to, then you can force a Hub object to only connect to that hub. This can be useful, for example, for connecting to two trains that need to have different code and can be accomplished by passing in the ble_id argument during instantiation of the Hub.

On Windows and Linux, you will use the 6-byte Bluetooth network address:

async def system():
    hub = Train('train1', ble_id='XX:XX:XX:XX:XX:XX')
    hub = Train('train2', ble_id='YY:YY:YY:YY:YY:YY')

And on OS X systems, you will use the UUID for the Bluetooth hub like so:

async def system():
    hub = Train('train1', ble_id='05c5e50e-XXXX-XXXX-XXXX-XXXXXXXXXXXX')
    hub = Train('train2', ble_id='05c5e50e-YYYY-YYYY-YYYY-YYYYYYYYYYYY')

Hub buttons and LED colors

Here’s an example that enables the train motor, vision sensor, hub button, and hub LED. First, we’ll wait until the hub button is pressed before we do anything; the LED will blink purple and yellow while it’s waiting. Then, we’ll change the speed like the previous example based on the vision distance, while at the same time changing the LED color orange if it’s responding to a distance change.

import logging
from itertools import cycle
from curio import sleep
from bricknil import attach, start
from bricknil.hub import PoweredUpHub
from bricknil.sensor import TrainMotor, VisionSensor, Button, LED
from bricknil.process import Process
from bricknil.const import Color

@attach(Button, name='train_btn', capabilities=['sense_press'])
@attach(LED, name='train_led')
@attach(VisionSensor, name='train_sensor', capabilities=['sense_count', 'sense_distance'])
@attach(TrainMotor, name='motor')
class Train(PoweredUpHub):

    async def train_btn_change(self):
        self.message_info(f'train button push {self.train_btn.value}')
        btn = self.train_btn.value[Button.capability.sense_press]
        if btn == 1:
            # Pushed!
            self.go = True


    async def train_sensor_change(self):
        self.message_info(f'Train sensor value change {self.train_sensor.value}')
        distance = self.train_sensor.value[VisionSensor.capability.sense_distance]
        count = self.train_sensor.value[VisionSensor.capability.sense_count]

        if count > 3:
            # Wave your hand more than three times in front of the sensor and the program ends
            self.keep_running = False

        # The closer your hand gets to the sensor, the faster the motor runs
        self.motor_speed = (10-distance)*10

        # Flag a change
        self.sensor_change = True

    async def run(self):
        self.message_info("Running")
        self.motor_speed = 0
        self.keep_running = True
        self.sensor_change = False
        self.go = False

        # Blink the color  from purple and yellow
        colors = cycle([Color.purple, Color.yellow])
        while not self.go:  # Wait until the hub button is pushed
            await self.train_led.set_color(next(colors))
            await sleep(1)

        colors = cycle([Color.green, Color.orange])
        # Ready to go, let's change the color to green!
        while self.keep_running:
            if self.sensor_change:
                await self.train_led.set_color(next(colors))
                await self.motor.ramp_speed(self.motor_speed, 900)  # Ramp to new speed in 0.9 seconds
                self.sensor_change = False
                await sleep(1)
                await self.train_led.set_color(next(colors))
            else:
                await sleep(1)

async def system():
    train = Train('My Train')

if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)
    start(system)

Controlling Vernie (Boost Hub) with the PoweredUp Remote

Here’s a nice example of controlling two hubs (the remote is also a type of hub) and feeding the button presses of the remote to make Vernie move forward, backward, left, and right.

And here’s the code that’s being used in the video above:

import logging

from curio import sleep, Queue
from bricknil import attach, start
from bricknil.hub import PoweredUpRemote, BoostHub
from bricknil.sensor import InternalMotor, RemoteButtons, LED, Button
from bricknil.process import Process
from bricknil.const import Color


@attach(LED, name='led') 
@attach(RemoteButtons, name='btns_right',  capabilities=['sense_press'])
@attach(RemoteButtons, name='btns_left',  capabilities=['sense_press'])
class Remote(PoweredUpRemote):

    async def btns_left_change(self):
        if self.btns_left.plus_pressed():
            await self.tell_robot.put('forward')
        elif self.btns_left.minus_pressed():
            await self.tell_robot.put('backward')
        else:
            await self.tell_robot.put('stop')

    async def btns_right_change(self):
        if self.btns_right.plus_pressed():
            await self.tell_robot.put('right')
        elif self.btns_right.minus_pressed():
            await self.tell_robot.put('left')
        else:
            await self.tell_robot.put('stop')

    async def run(self):
        self.message('Running')
        # Set the remote LED to green to show we're ready
        await self.led.set_color(Color.green)
        while True:
            await sleep(10)   # Keep the remote running

@attach(LED, name='led') 
@attach(InternalMotor, name='motor_right', port=InternalMotor.Port.B)
@attach(InternalMotor, name='motor_left', port=InternalMotor.Port.A)
class Robot(BoostHub):

    async def run(self):
        self.message("Running")
        speed = 30

        # Set the robot LED to green to show we're ready
        await self.led.set_color(Color.green)
        while True:
            msg = await self.listen_remote.get()
            await self.listen_remote.task_done()
            if msg=='forward':
                self.message('going forward')
                await self.motor_left.set_speed(speed)
                await self.motor_right.set_speed(speed)
            elif msg=='backward':
                self.message('going backward')
                await self.motor_left.set_speed(-speed)
                await self.motor_right.set_speed(-speed)
            elif msg=='stop':
                self.message('stop')
                await self.motor_left.set_speed(0)
                await self.motor_right.set_speed(0)
            elif msg=='left':
                self.message('left')
                await self.motor_left.set_speed(-speed)
                await self.motor_right.set_speed(speed)
            elif msg=='right':
                self.message('right')
                await self.motor_left.set_speed(speed)
                await self.motor_right.set_speed(-speed)


async def system():
    robot = Robot('Vernie')
    remote = Remote('remote')
    
    # Define a message passing queue from the remote to the robot
    remote.tell_robot = Queue()
    robot.listen_remote = remote.tell_robot

if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)
    start(system)

Here, we are using two hubs running in parallel, and we use a curio.Queue to send messages from the remote telling the robot(Boost hub) what to do. Notice that each RemoteButtons instance consists of 3 buttons, so there are some helper methods to check if a particular button is pressed.

Using the Duplo Train and Playing Sounds

The Duplo trains 10874 and 10875 have the ability to play a set of 5 predetermined sounds through their built-in speakers (bricknil.sensor.DuploSpeaker). In addition, there is a speedometer(bricknil.sensor.DuploSpeedSensor) built-in to the front wheels, and a vision sensor(bricknil.sensor.DuploVisionSensor) in the undercarriage pointing straight down. This vision sensor is slightly less capable than the stand-alone Boost Vision Sensor discussed above, but it can still recognize colors, distance, and the special blocks that Lego provides in those sets. Here’s an example that puts everything together:


from itertools import cycle
from curio import sleep
from bricknil import attach, start
from bricknil.hub import DuploTrainHub
from bricknil.sensor import DuploTrainMotor, DuploSpeedSensor, LED, DuploVisionSensor, DuploSpeaker, Button, VoltageSensor
from bricknil.const import Color
import logging

#@attach(DuploSpeaker, name='speaker')
@attach(DuploVisionSensor, name='vision_sensor', capabilities=[('sense_reflectivity', 5)])
@attach(LED, name='led')
@attach(DuploSpeedSensor, name='speed_sensor', capabilities=['sense_speed', 'sense_count'])
#@attach(VoltageSensor, name='voltage', capabilities=[('sense_l', 50)])
@attach(DuploTrainMotor, name='motor')
class Train(DuploTrainHub):

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.go = False     # Only becomes true with hub button is pressed

    async def voltage_change(self):
        pass
    async def speed_sensor_change(self):
        speed = self.speed_sensor.value[DuploSpeedSensor.capability.sense_speed]
        if not self.go and speed > 0:
            self.go = True
            self.message_info('Movement detected: starting...')
        elif self.go:
            #count = self.speed_sensor.value[DuploSpeedSensor.capability.sense_count]
            #self.message_info(f'Speed sensor changed speed: {speed} count: {count}')
            self.message_info(f'Speed sensor changed speed: {speed}')

    async def vision_sensor_change(self):
        cap = DuploVisionSensor.capability
        #color = self.vision_sensor.value[cap.sense_color]
        #ctag  = self.vision_sensor.value[cap.sense_ctag]
        reflt  = self.vision_sensor.value[cap.sense_reflectivity]
        if self.go:
            #self.message_info(f'Vision sensor changed color: {color} ctag: {ctag} reflt: {reflt}')
            self.message_info(f'Vision sensor changed color: reflt: {reflt}')

    async def run(self):
        self.message_info("Running")

        colors = cycle([Color.red, Color.purple, Color.yellow, Color.blue, Color.white])

        snd = DuploSpeaker.sounds
        sounds = cycle([snd.brake, snd.station, snd.water, snd.horn, snd.steam])

        self.message_info('Please move the train to start the program')
        while not self.go:
            await self.led.set_color(next(colors))
            await sleep(0.3)

        for i in range(5):
            await self.led.set_color(next(colors))       # Cycle through the colors
            #await self.speaker.play_sound(next(sounds))  # cycle through the sounds
            tgt_speed = 20 + i*15                        # Keep increasing the speed
            await self.motor.ramp_speed(tgt_speed, 2000)
            self.message_info(f"Set speed to {i}")
            await sleep(3)

        self.message_info("Done")

async def system():
    hub = Train('train', False)

if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)
    start(system)

BrickNil Architecture

This section documents the internal architecture of BrickNil and how all the components communicate with each other.

Run loops

There are actually two threads of execution in the current system architecture. The main Bluetooth radio communication loop is provided by the BluetoothLE library, which manages everything in the background and can callback directly into user code. In parallel with this, inside this library, a separate execution loop is running the Curio event library, which provides the async event loop that executes our user code. Thus, we need to be careful about maintaining thread safety between the Curio async event loop and the background Bluetooth event processing.

_images/run_loops.svg

BrickNil running inside Curio’s event loop, which in turn is run by the Adafruit_BluefruitLE library run loop

I’d much have preferred to have the Bluetooth library be implemented via an async library like Curio, asyncio, or Trio, but I wasn’t able to find any such library. This admitted kludge of nested run loops was the only way I could get everything working.

Installation

On all platforms (OS X, Linux, Win10), it should just be a simple:

$ pip install bricknil

On Linux, you might need to install the BlueZ >= 5.43 libraries.

On a Raspberry Pi (and other Linux boxes should be similar), you can follow my automated setup instructions

Supported Devices

BrickNil supported Lego® devices
Peripheral Sets Support
PoweredUp Hub (88009) 60197, 60198, 76112
PoweredUpHub
PoweredUp Train Motor (88011) 60197, 60198 TrainMotor
PoweredUp Remote (88010) 60197, 60198
PoweredUpRemote
PoweredUp Light (88005) 88005 Light
Boost Hub (88006) 17101
BoostHub
Boost Vision Sensor (88007) 17101 VisionSensor
Boost External Motor (88008) 17101 ExternalMotor
Wedo External Motor (45303) 45300 WedoMotor
Wedo Tilt Sensor (45305) 45300 ExternalTiltSensor
Wedo Motion Sensor (45304) 45300 ExternalMotionSensor
Duplo Train Base 10874, 10875
DuploTrainHub

Credits

  • Virantha N. Ekanayake @virantha - lead developer
  • David Lechner @dlech - contributor
    • Added Windows 10 support
    • Added support for Lego 88005 stand-alone LED peripheral

This project is also greatly indebted to the following persons, as well as their open-sourced libraries, portions of which have been incorporated into BrickNil under the terms of their respective licenses:

  • Tony DiCola for his Adafruit_Python_BluefruitLE library that provides the BluetoothLE communication stack on Mac OS X
  • Henrik Blidh for his Bleak library that provided a pure python way to communicate with BluetoothLE over DBus on Linux.

Disclaimer

The software is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. Licensed under ASL 2.0