Metadata-Version: 2.1
Name: legoeducation
Version: 1.0.6
Summary: High-level Python API for LEGO Education devices via BLE RPC
Home-page: https://github.com/LEGO/legoeducation
Author: LEGO Education
License: BUSL-1.1
Project-URL: Homepage, https://education.lego.com/
Project-URL: Repository, https://github.com/LEGO/LEGOEducation
Project-URL: Issues, https://legoeducation.atlassian.net/servicedesk/customer/portals
Keywords: LEGO,Education,BLE,RPC
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Education
Classifier: Intended Audience :: End Users/Desktop
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3.11
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: Education
Requires-Python: >=3.11
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: bleak>=0.19.0
Provides-Extra: test
Requires-Dist: pytest-asyncio>=0.21; extra == "test"
Requires-Dist: pytest>=7.0; extra == "test"

![LEGO&reg; Education logo](https://github.com/LEGO/LEGOEducation/blob/main/img/LEGOEducation.png?raw=true "LEGO&reg; Education logo")

# LEGO&reg; Education Python: What is it?

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

The package allows users to connect to LEGO&reg; 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&reg; Education website](https://education.lego.com/)

- For block-based coding instead, visit [LEGO&reg; Education Coding Canvas](https://code.legoeducation.com/)

- For teacher ressources and materials, visit [LEGO&reg; Education Teacher Portal](https://teach.legoeducation.com/) 

## Quick Start and Documentation

Quick start guides, documentation and example code is available on [Github](https://github.com/LEGO/LEGOEducation)

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

```bash
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&reg; 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&reg; Education [Coding Canvas](https://code.legoeducation.com/)

![LEGO&reg; Education hardware](https://github.com/LEGO/LEGOEducation/blob/main/img/Hardware.png?raw=true "LEGO&reg; Education hardware")

# 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:

```bash
pip install legoeducation
```

### Advanced: Install without bleak
If you plan to supply your own BLE stack, you can skip dependency resolution:
```bash
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:
```python
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).
