legoeducation 1.0.5

High-level Python API for LEGO Education devices via BLE RPC

LEGO® Education Python: What is it?

This is a Python Package for using LEGO® Education hardware via Bluetooth (BLE). The package is designed for communicating with LEGO® Education hardware from LEGO® Education Sciece, and LEGO® Education Computer Science & AI products, using RPC.

The package allows users to connect to LEGO® Education hardware, read sensor data from Color Sensors, Controllers and motors (position, speed, IMU etc.), and send various move commands to the Single Motor and Double Motor.

Relevant links:

• For more information on the products, visit the LEGO® Education website

• For block-based coding instead, visit LEGO® Education Coding Canvas

• For teacher ressources and materials, visit LEGO® Education Teacher Portal

Quick Start and Documentation

Quick start guides, documentation and example code is available on Github

Installation

Binary installers for the latest released version are available at the Python Package Index (PyPI)

pip install legoeducation

Firmware Compatibility

The Python Package requires the firmware version on the connected hardware is using the same RPC version. When connecting to LEGO® Education hardware where there is a discrepancy with the Python Package version and firmware version, an error is thrown, asking to either update the Python Package (package RPC version < firmware RPC version) or update the firmware (firmware RPC version < package RPC version) using the LEGO® Education Coding Canvas

Advanced Usage

Replacing the BLE Transport Layer

The examples found on Github uses the bleak library. bleak is installed by default to give an immediately usable experience:

pip install legoeducation

Advanced: Install without bleak

If you plan to supply your own BLE stack, you can skip dependency resolution:

pip install --no-deps legoeducation

In that mode the default bleak-based transport will NOT register (bleak missing). You must either register a custom transport before creating hubs (see below), or set the LEGOEDUCATION_BLE_IMPL environment variable.

Provide a custom transport class

Implement a subclass of BLETransport with the required async methods (scan_devices, connect, send, device_disconnect, shutdown_all) and register it:

from legoeducation.ble_transport import BLETransport, register_transport

class MyTransport(BLETransport):
	async def scan_devices(self, timeout, filters=None): ...
	async def connect(self, device, notification_callback, disconnect_callback): ...
	async def send(self, device, message: bytes): ...
	async def device_disconnect(self, device): ...
	def shutdown_all(self): ...

register_transport(MyTransport)

Call register_transport before instantiating any hub classes.

Environment variable override

Set LEGOEDUCATION_BLE_IMPL="my_pkg.my_module:MyTransport" before importing the library to auto-load your implementation.

If no transport is registered you will receive a runtime warning and a null transport will be used (no BLE operations). Register or install bleak to proceed.

Method contract details

Below are the required methods with their expected signatures, argument semantics, return values, and behavioral guarantees. Implementations should be robust against spurious calls (e.g. disconnect on an already disconnected device) and must avoid leaking exceptions (log internally instead; unhandled exceptions are treated as fatal by the worker thread).

1. async def scan_devices(timeout: float, filters: dict | None = None) -> list | None

   • Purpose: Perform a BLE scan for up to timeout seconds.
   • timeout: Maximum scan duration in seconds (float). Implementations may clamp or round but should not block longer than this.
   • filters: Optional backend-defined filtering criteria (e.g. service UUIDs, name prefixes). May be ignored if unsupported.
   • Return: A list of backend-specific device descriptors (often objects from your BLE stack) or None if scanning not supported. Empty list means the scan completed but found nothing.
   • Edge cases: Return promptly if timeout <= 0. Handle environments with no adapter gracefully (log and return empty list / None).

2. async def connect(device: Any, notification_callback: Callable, disconnect_callback: Callable) -> bool

   • Purpose: Establish a connection and start notifications for the given device descriptor returned by scan_devices.
   • device: The descriptor/object representing the target device that was returned by scan_devices.
   • notification_callback: Callable invoked for every incoming packet: notification_callback(characteristic_id, data: bytes). Characteristic identifier may be a UUID string or backend object; pick a stable representation and keep it consistent.
   • disconnect_callback: Callable invoked exactly once when the device disconnects (whether initiated locally or due to remote/adapter events).
   • Return: True if connection + notification subscription succeeded; False if the attempt failed (do not raise for routine failures like timeout or permission issues).
   • Edge cases: Must tolerate repeated calls for an already connected device (either no-op returning True or reconnect logic). Ensure disconnect_callback fires even on failed connection attempts if partial resources were allocated.

3. async def send(device: Any, message: bytes) -> None

   • Purpose: Write raw protocol bytes to the device's primary command characteristic.
   • device: Connected device reference.
   • message: Exact bytes to transmit; implementation must not mutate. Fragmentation/retries are transport concerns and should be hidden from caller.
   • Return: None. Raise only on truly unrecoverable programmer errors (e.g. device not connected) preferably converting them into logged warnings instead.
   • Edge cases: If the device disconnects mid-write, swallow backend errors and optionally trigger a disconnect callback if not yet fired.

4. async def device_disconnect(device: Any) -> None

   • Purpose: Gracefully terminate an individual connection.
   • device: Connected device reference.
   • Behavior: Idempotent—calling multiple times should not raise. Should trigger the disconnect_callback once if connected.
   • Return: None.
   • Edge cases: Handle cases where the adapter already reported a disconnect between scheduling and execution of this coroutine.

5. def shutdown_all(self) -> None

   • Purpose: Synchronously (non-async) tear down all connections and release resources (cancel scans, close adapters, stop loops as needed).
   • Behavior: Must invoke the shutdown_callback provided to the transport constructor even if there were no active devices.
   • Return: None.
   • Edge cases: Safe to call while no devices are connected or while a connection attempt is in flight (should abort attempts cleanly).

General expectations:

• Threading/loop: All coroutines run inside the worker's asyncio loop provided by the API. Avoid creating separate event loops unless absolutely necessary.
• Exceptions: Catch backend-specific exceptions; convert to logs and clean disconnects. Unhandled exceptions propagate and can terminate the worker.
• Logging: Prefer using self.logger for consistency.
• Performance: Return control quickly -> long blocking calls should be await-based.
• Resource cleanup: Ensure device objects won't hold references after disconnect (helps GC and avoids stale callbacks).