From 1dd587574ce0f057992ebbe627a5c7b530542125 Mon Sep 17 00:00:00 2001 From: Brad Barnett <127794626+bdbarnett@users.noreply.github.com> Date: Thu, 24 Sep 2026 11:03:11 -0500 Subject: [PATCH 1/6] bledev: the contract, the fake loopback, nus and auto bledev/__init__ is the portable BLE contract (UUID, errors, capabilities, adapter, services and characteristics on both sides), async and aioble-shaped. bledev.fake is an in-process loopback that is stricter than a radio: over-MTU packets, over-max_len writes, unsubscribed notifications, transmit bursts and receive overruns all raise instead of truncating or dropping. bledev.nus is the Nordic UART byte stream on top of it, and bledev.auto picks a backend. tests/bledev_contract.py runs on CPython and unix MicroPython; test_bledev runs it both ways and proves five planted faults each fail it. --- lib/bledev/__init__.py | 840 +++++++++++++++++++++++++++++++++++++++ lib/bledev/auto.py | 80 ++++ lib/bledev/fake.py | 566 ++++++++++++++++++++++++++ lib/bledev/nus.py | 374 +++++++++++++++++ pydevices-desktop.toml | 4 + tests/bledev_contract.py | 704 ++++++++++++++++++++++++++++++++ tests/test_bledev.py | 83 ++++ 7 files changed, 2651 insertions(+) create mode 100644 lib/bledev/__init__.py create mode 100644 lib/bledev/auto.py create mode 100644 lib/bledev/fake.py create mode 100644 lib/bledev/nus.py create mode 100644 tests/bledev_contract.py create mode 100644 tests/test_bledev.py diff --git a/lib/bledev/__init__.py b/lib/bledev/__init__.py new file mode 100644 index 00000000..3fed80bc --- /dev/null +++ b/lib/bledev/__init__.py @@ -0,0 +1,840 @@ +"""Portable Bluetooth Low Energy: the contract every bledev backend implements. + +You write a BLE program once against this module, and it runs on a board +(:mod:`bledev.mpble`), on a laptop (``bledev.bleak``), in a browser +(``bledev.webble``), or against the in-process loopback in tests +(:mod:`bledev.fake`). The guide is ``docs/bledev.md``. + +The shape is aioble's, and it is async throughout. The pieces: + +* :class:`UUID`, the errors, and the ``FLAG_*`` property bits. +* :class:`BLE`, the adapter. A backend subclasses it. ``capabilities()`` says + which roles it can take, because roles aren't symmetric: bleak and Web + Bluetooth are central only, and a board can be either. +* The peripheral side: :class:`Service` and :class:`Characteristic` describe a + GATT server, ``ble.register_services()`` installs it, and + ``await ble.advertise()`` returns a :class:`Connection` when a central + connects. +* The central side: ``ble.scan()`` yields :class:`ScanResult`, ``ble.find()`` + returns a :class:`Device`, and ``await device.connect()`` returns a + :class:`Connection`, whose services and characteristics you discover as + :class:`ClientService` and :class:`ClientCharacteristic`. + +Nothing here imports a backend, and nothing here imports :mod:`bledev.auto`. +""" + +try: + import asyncio +except ImportError: # pragma: no cover - very old MicroPython + import uasyncio as asyncio + +import struct + +__version__ = "0.1.0" + +# GATT characteristic property bits. The values are the Bluetooth spec's, and +# the same ones MicroPython's bluetooth module and aioble use. +FLAG_READ = 0x0002 +FLAG_WRITE_NO_RESPONSE = 0x0004 +FLAG_WRITE = 0x0008 +FLAG_NOTIFY = 0x0010 +FLAG_INDICATE = 0x0020 + +#: Every connection starts at this ATT MTU. One ATT packet carries ``mtu - 3`` +#: bytes of value. +DEFAULT_MTU = 23 +MAX_MTU = 517 + +ADDR_PUBLIC = 0 +ADDR_RANDOM = 1 + + +# ---------------------------------------------------------------- errors + + +class BLEError(Exception): + """Base of every error bledev raises.""" + + +class UnsupportedError(BLEError): + """This backend cannot do that: a role it can't take, or a missing feature.""" + + +class DisconnectedError(BLEError): + """The connection dropped, or was already closed, before the operation finished.""" + + +class BLETimeoutError(BLEError): + """An operation given ``timeout_ms`` did not finish in time.""" + + +class BusyError(BLEError): + """The controller's transmit buffers are full. Wait a moment and send again. + + Only :meth:`Characteristic.notify` raises it, because it is synchronous; + every async send waits for room instead. + """ + + +class GattError(BLEError): + """The peer answered a GATT request with an error ``status``.""" + + def __init__(self, status, message=None): + super().__init__(message or "GATT error 0x{:02x}".format(status)) + self.status = status + + +async def sleep_ms(ms): + """``asyncio.sleep_ms`` on every interpreter.""" + await asyncio.sleep(ms / 1000) + + +async def wait_ms(awaitable, timeout_ms, what="operation"): + """Await ``awaitable``, raising :class:`BLETimeoutError` after ``timeout_ms``. + + ``timeout_ms=None`` waits forever. Backends use this so every timeout in + bledev is the same exception. + """ + if timeout_ms is None: + return await awaitable + try: + return await asyncio.wait_for(awaitable, timeout_ms / 1000) + except asyncio.TimeoutError: + raise BLETimeoutError("{} timed out after {} ms".format(what, timeout_ms)) + + +# ---------------------------------------------------------------- UUID + +_BASE_TAIL = bytes((0x00, 0x00, 0x10, 0x00, 0x80, 0x00, 0x00, 0x80, 0x5F, 0x9B, 0x34, 0xFB)) +_HEX = "0123456789abcdef" + + +def _hex(data): + return "".join(_HEX[b >> 4] + _HEX[b & 15] for b in data) + + +def _unhex(text): + if len(text) % 2: + raise ValueError("odd-length hex") + return bytes(int(text[i : i + 2], 16) for i in range(0, len(text), 2)) + + +class UUID: + """A Bluetooth UUID, the same value on every host. + + Accepts a 16- or 32-bit ``int`` (``UUID(0x180F)``), a string in any of the + usual spellings (``"180f"``, ``"0x180F"``, + ``"6e400001-b5a3-f393-e0a9-e50e24dcca9e"``), another UUID, MicroPython's + ``bluetooth.UUID``, or 2, 4 or 16 bytes in Bluetooth's little-endian wire + order. A short UUID equals its 128-bit expansion. + + ``str()`` is always the 128-bit lowercase form, which is what bleak and Web + Bluetooth use. ``short`` is the 16/32-bit value, or ``None``. + ``to_bytes()`` is the wire form, shortest first. + """ + + def __init__(self, value): + if isinstance(value, UUID): + self._be = value._be + return + if isinstance(value, int): + self._be = self._from_short(value) + return + if isinstance(value, str): + text = value.strip().lower() + if text.startswith("0x"): + self._be = self._from_short(int(text[2:], 16)) + return + text = text.replace("-", "") + if len(text) in (4, 8): + self._be = self._from_short(int(text, 16)) + return + if len(text) != 32: + raise ValueError("not a UUID: {!r}".format(value)) + self._be = _unhex(text) + return + try: + raw = bytes(value) + except TypeError: + raise ValueError("not a UUID: {!r}".format(value)) + if len(raw) in (2, 4): + self._be = self._from_short(int.from_bytes(raw, "little")) + elif len(raw) == 16: + self._be = bytes(reversed(raw)) + else: + raise ValueError("a UUID is 2, 4 or 16 bytes, not {}".format(len(raw))) + + @staticmethod + def _from_short(value): + if not 0 <= value <= 0xFFFFFFFF: + raise ValueError("short UUID out of range: {}".format(value)) + return value.to_bytes(4, "big") + _BASE_TAIL + + @property + def short(self): + """The 16- or 32-bit value when this is a Bluetooth-base UUID, else ``None``.""" + if self._be[4:] == _BASE_TAIL: + return int.from_bytes(self._be[:4], "big") + return None + + def to_bytes(self): + """Little-endian wire bytes: 2 for a 16-bit UUID, 4 for 32-bit, else 16.""" + short = self.short + if short is not None: + return short.to_bytes(2 if short <= 0xFFFF else 4, "little") + return bytes(reversed(self._be)) + + def __bytes__(self): + return self.to_bytes() + + def __str__(self): + h = _hex(self._be) + return "-".join((h[:8], h[8:12], h[12:16], h[16:20], h[20:])) + + def __repr__(self): + short = self.short + if short is not None: + return "UUID(0x{:04x})".format(short) + return "UUID('{}')".format(self) + + def __eq__(self, other): + if not isinstance(other, UUID): + try: + other = UUID(other) + except (ValueError, TypeError): + return False + return self._be == other._be + + def __ne__(self, other): + return not self.__eq__(other) + + def __hash__(self): + return hash(self._be) + + +# ---------------------------------------------------------------- advertising payloads + +ADV_TYPE_FLAGS = 0x01 +ADV_TYPE_UUID16_INCOMPLETE = 0x02 +ADV_TYPE_UUID16_COMPLETE = 0x03 +ADV_TYPE_UUID32_INCOMPLETE = 0x04 +ADV_TYPE_UUID32_COMPLETE = 0x05 +ADV_TYPE_UUID128_INCOMPLETE = 0x06 +ADV_TYPE_UUID128_COMPLETE = 0x07 +ADV_TYPE_SHORT_NAME = 0x08 +ADV_TYPE_NAME = 0x09 +ADV_TYPE_APPEARANCE = 0x19 +ADV_TYPE_MANUFACTURER = 0xFF + +ADV_PAYLOAD_MAX = 31 + + +def pack_advertisement(name=None, services=None, appearance=0, manufacturer=None): + """Build ``(adv_data, resp_data)`` exactly the way aioble packs them. + + Fields go into the 31-byte advertisement in this order (flags, services, + name, appearance, manufacturer) and overflow into the scan response, which + only an *active* scan sees. That is why a long name can be invisible to a + passive scan: the fake backend uses this so tests see the same thing. + ``manufacturer`` is ``(company_id, data)``. + """ + adv = bytearray() + resp = bytearray() + + def append(adv_type, value): + field = bytes((len(value) + 1, adv_type)) + bytes(value) + if len(adv) + len(field) < ADV_PAYLOAD_MAX: + adv.extend(field) + elif len(resp) + len(field) < ADV_PAYLOAD_MAX: + resp.extend(field) + else: + raise ValueError("advertising payload too long") + + append(ADV_TYPE_FLAGS, b"\x06") + if services: + uuids = [UUID(s).to_bytes() for s in services] + for size, code in ((2, ADV_TYPE_UUID16_COMPLETE), (4, ADV_TYPE_UUID32_COMPLETE), (16, ADV_TYPE_UUID128_COMPLETE)): + group = b"".join(u for u in uuids if len(u) == size) + if group: + append(code, group) + if name: + append(ADV_TYPE_NAME, name.encode() if isinstance(name, str) else name) + if appearance: + append(ADV_TYPE_APPEARANCE, struct.pack("= 2: + appearance = struct.unpack("= 2: + manufacturer.append((struct.unpack(" (module, class). Each module stands alone and never imports this one. +BACKENDS = { + "mpble": ("bledev.mpble", "MPBLE"), + "bleak": ("bledev.bleak", "BleakBLE"), + "webble": ("bledev.webble", "WebBLE"), + "fake": ("bledev.fake", "FakeBLE"), +} + + +def _importable(name): + try: + __import__(name) + return True + except Exception: + return False + + +def _in_browser(): + if sys.platform in ("emscripten", "webassembly"): + return True + return _importable("_wasm_bridge") or _importable("pyodide") + + +def backend_name(): + """The backend this host would get, or ``None`` if it has no BLE.""" + impl = getattr(sys, "implementation", None) + impl = getattr(impl, "name", "") + if impl in ("micropython", "circuitpython"): + if _in_browser(): + return "webble" + if _importable("bluetooth"): + return "mpble" + return None # CircuitPython's _bleio waits for bledev.cpble + if _in_browser(): + return "webble" + if _importable("bleak"): + return "bleak" + return None + + +def _env_backend(): + try: + import os + + return os.getenv("BLEDEV_BACKEND") + except Exception: + return None + + +def adapter(backend=None, **kwargs): + """Construct the adapter for this host (or for ``backend``).""" + name = backend or _env_backend() or backend_name() + if name is None: + raise UnsupportedError( + "no BLE backend on this host: on CPython, pip install 'pydevices[ble]' " + "for bleak; on MicroPython, use firmware built with bluetooth" + ) + if name not in BACKENDS: + raise UnsupportedError("unknown bledev backend {!r}; one of {}".format(name, sorted(BACKENDS))) + module_name, class_name = BACKENDS[name] + module = __import__(module_name, None, None, [class_name]) + return getattr(module, class_name)(**kwargs) diff --git a/lib/bledev/fake.py b/lib/bledev/fake.py new file mode 100644 index 00000000..49ab8a89 --- /dev/null +++ b/lib/bledev/fake.py @@ -0,0 +1,566 @@ +"""In-process loopback backend: two adapters, one pretend radio, no hardware. + +Every :class:`FakeBLE` built on the same :class:`Air` can see and connect to +the others, so a test runs a peripheral and a central in one event loop:: + + from bledev.fake import Air, FakeBLE + + air = Air() + board = FakeBLE(air=air) # either role, like a board + laptop = FakeBLE(air=air, peripheral=False) # central only, like bleak + +It runs on CPython and MicroPython, and it is stricter than a real radio on +purpose. Where real hardware would quietly truncate or drop data, the fake +raises, so a bug shows up in a test rather than as corruption on a board: + +* A notification or write longer than ``mtu - 3`` raises :class:`BLEError` + (NimBLE truncates it). +* A write longer than the characteristic's ``max_len`` raises (MicroPython + truncates to the buffer, 20 bytes by default). +* Notifications reach a client only after it calls ``subscribe()``. +* Only ``notify_buffers`` notifications can be in flight per connection; + one more raises :class:`BusyError` until the event loop runs, the way a real + controller runs out of buffers. Send loops must await between bursts. +* A receive queue that fills up (``queue_limit``) makes the next + ``notified()`` or ``written()`` raise instead of dropping data silently. +* The name only reaches a *passive* scan when it fits in the 31-byte + advertisement; otherwise it is in the scan response, as on real hardware. +""" + +try: + import asyncio +except ImportError: # pragma: no cover + import uasyncio as asyncio + +from . import ( + BLE, + BLEError, + BusyError, + ClientCharacteristic, + ClientService, + Connection, + DEFAULT_MTU, + Device, + DisconnectedError, + FLAG_INDICATE, + FLAG_NOTIFY, + FLAG_READ, + FLAG_WRITE, + FLAG_WRITE_NO_RESPONSE, + MAX_MTU, + ScanResult, + Scanner, + UnsupportedError, + decode_advertisement, + pack_advertisement, + sleep_ms, + wait_ms, +) + +#: How often a fake scan looks at the air, in milliseconds. +SCAN_TICK_MS = 5 + + +class Air: + """The medium the fake adapters share. Make one per test.""" + + def __init__(self): + self.adverts = [] + self._next = 1 + + def _allocate(self): + n = self._next + self._next += 1 + return "fa:ce:00:00:{:02x}:{:02x}".format((n >> 8) & 0xFF, n & 0xFF) + + +_default_air = Air() + + +def default_air(): + """The :class:`Air` used when an adapter isn't given one.""" + return _default_air + + +class _Advert: + def __init__(self, ble, adv_data, resp_data, connectable): + self.ble = ble + self.adv_data = adv_data + self.resp_data = resp_data + self.connectable = connectable + self.event = asyncio.Event() + self.connection = None + self.closed = False + + +class _Inbox: + """An ordered queue that refuses to lose data quietly.""" + + def __init__(self, limit, what): + self.items = [] + self.event = asyncio.Event() + self.limit = limit + self.what = what + self.overrun = 0 + self.closed = False + + def put(self, item): + if len(self.items) >= self.limit: + self.overrun += 1 + else: + self.items.append(item) + self.event.set() + + def close(self): + self.closed = True + self.event.set() + + async def get(self, timeout_ms): + while True: + if self.overrun: + raise BLEError( + "{} queue overran: {} item(s) dropped because the reader fell " + "{} behind".format(self.what, self.overrun, self.limit) + ) + if self.items: + return self.items.pop(0) + if self.closed: + raise DisconnectedError("{}: disconnected".format(self.what)) + self.event.clear() + await wait_ms(self.event.wait(), timeout_ms, self.what) + + +class FakeBLE(BLE): + """A pretend radio. ``central`` and ``peripheral`` pick the roles it may take. + + ``mtu`` is the ATT MTU it asks for in ``exchange_mtu()``. + ``notify_buffers`` and ``queue_limit`` size the flow control described in + the module docstring. + """ + + backend = "fake" + + def __init__( + self, + *, + air=None, + central=True, + peripheral=True, + mtu=DEFAULT_MTU, + address=None, + rssi=-50, + notify_buffers=8, + queue_limit=128, + gap_name="fake", + ): + self._air = air if air is not None else _default_air + self._central = central + self._peripheral = peripheral + self._mtu = mtu + self.address = address or self._air._allocate() + self.rssi = rssi + self.notify_buffers = notify_buffers + self.queue_limit = queue_limit + self._gap_name = gap_name + self._services = () + self._adverts = [] + self._links = [] + + def capabilities(self): + caps = BLE.capabilities(self) + caps.update( + central=self._central, + peripheral=self._peripheral, + mtu=self._mtu, + max_mtu=MAX_MTU, + indicate=True, + ) + return caps + + def config(self, *names, **settings): + for key, value in settings.items(): + if key == "mtu": + if not 23 <= value <= MAX_MTU: + raise ValueError("mtu must be 23..{}".format(MAX_MTU)) + self._mtu = value + elif key == "gap_name": + self._gap_name = value + else: + raise UnsupportedError("fake has no setting {!r}".format(key)) + if names: + if len(names) != 1: + raise ValueError("read one setting at a time") + if names[0] == "mtu": + return self._mtu + if names[0] == "gap_name": + return self._gap_name + raise UnsupportedError("fake has no setting {!r}".format(names[0])) + return None + + def close(self): + for advert in list(self._adverts): + advert.closed = True + advert.event.set() + for link in list(self._links): + link.drop() + + # -- peripheral + + def register_services(self, *services): + if not self._peripheral: + raise UnsupportedError("this fake adapter is central only") + for service in self._services: + for characteristic in service.characteristics: + characteristic._impl = None + self._services = tuple(services) + for service in services: + for characteristic in service.characteristics: + characteristic._impl = _ServerCharacteristic(self, characteristic) + + async def advertise( + self, + interval_us=100000, + *, + name=None, + services=None, + appearance=0, + manufacturer=None, + connectable=True, + timeout_ms=None, + ): + if not self._peripheral: + raise UnsupportedError("this fake adapter is central only; it cannot advertise") + adv_data, resp_data = pack_advertisement(name, services, appearance, manufacturer) + advert = _Advert(self, adv_data, resp_data, connectable) + self._air.adverts.append(advert) + self._adverts.append(advert) + try: + await wait_ms(advert.event.wait(), timeout_ms, "advertise") + finally: + if advert in self._air.adverts: + self._air.adverts.remove(advert) + self._adverts.remove(advert) + if advert.closed: + raise BLEError("adapter closed while advertising") + return advert.connection + + # -- central + + def scan(self, duration_ms, *, active=False, interval_us=None, window_us=None): + if not self._central: + raise UnsupportedError("this fake adapter is peripheral only; it cannot scan") + return _FakeScanner(self, duration_ms, active, interval_us, window_us) + + async def _connect(self, device, timeout_ms, **options): + if not self._central: + raise UnsupportedError("this fake adapter is peripheral only; it cannot connect") + + async def attempt(): + while True: + for advert in self._air.adverts: + if advert.ble.address == device.address and advert.connectable: + return advert + await sleep_ms(SCAN_TICK_MS) + + advert = await wait_ms(attempt(), timeout_ms, "connect to {}".format(device.address)) + link = _Link(self, advert.ble, device) + advert.connection = link.peripheral + self._air.adverts.remove(advert) + advert.event.set() + return link.central + + +class _FakeScanner(Scanner): + async def _start(self): + self._task = asyncio.create_task(self._run()) + + async def _stop(self): + task = getattr(self, "_task", None) + if task is not None: + task.cancel() + self._task = None + + async def _run(self): + seen = {} + elapsed = 0 + try: + while elapsed < self.duration_ms: + for advert in list(self._ble._air.adverts): + if advert.ble is self._ble: + continue + payloads = (advert.adv_data, advert.resp_data) if self.active else (advert.adv_data,) + key = advert.ble.address + if seen.get(key) == payloads: + continue + seen[key] = payloads + name, services, appearance, manufacturer = decode_advertisement(*payloads) + device = Device(self._ble, advert.ble.address, name=name) + self._push( + ScanResult( + device, + rssi=advert.ble.rssi, + name=name, + services=services, + appearance=appearance, + manufacturer=manufacturer, + connectable=advert.connectable, + ) + ) + await sleep_ms(SCAN_TICK_MS) + elapsed += SCAN_TICK_MS + finally: + self._task = None + self._finish() + + +class _Link: + """Both ends of one fake connection, and what travels between them.""" + + def __init__(self, central_ble, peripheral_ble, device): + self.connected = True + self.central_ble = central_ble + self.peripheral_ble = peripheral_ble + self.dropped = asyncio.Event() + self.central = _FakeConnection(central_ble, device, "central", self) + self.peripheral = _FakeConnection( + peripheral_ble, Device(peripheral_ble, central_ble.address), "peripheral", self + ) + # Keyed by (server characteristic, "notify" | "indicate"). + self.inboxes = {} + self.subscribed = {} + self.in_flight = [] + self._drain_task = None + central_ble._links.append(self) + peripheral_ble._links.append(self) + + def set_mtu(self, mtu): + self.central._mtu = mtu + self.peripheral._mtu = mtu + + def check(self): + if not self.connected: + raise DisconnectedError("the connection is closed") + + def inbox(self, server_char, kind, limit): + key = (id(server_char), kind) + box = self.inboxes.get(key) + if box is None: + box = _Inbox(limit, "{} {}".format(kind, server_char.uuid)) + if not self.connected: + box.close() + self.inboxes[key] = box + return box + + def deliver(self, server_char, kind, data, busy_check): + self.check() + if busy_check and len(self.in_flight) >= self.peripheral_ble.notify_buffers: + raise BusyError("{} notifications already in flight".format(len(self.in_flight))) + self.in_flight.append((server_char, kind, data)) + if self._drain_task is None: + # Held so CPython can't collect the task mid-flight. + self._drain_task = asyncio.create_task(self._drain()) + + async def _drain(self): + # One event-loop turn of "air time", then everything buffered arrives. + await asyncio.sleep(0) + self._drain_task = None + pending, self.in_flight = self.in_flight, [] + if not self.connected: + return + for server_char, kind, data in pending: + notify, indicate = self.subscribed.get(id(server_char), (False, False)) + if (kind == "notify" and notify) or (kind == "indicate" and indicate): + self.inbox(server_char, kind, self.central_ble.queue_limit).put(data) + + def drop(self): + if not self.connected: + return + self.connected = False + for box in self.inboxes.values(): + box.close() + for ble in (self.central_ble, self.peripheral_ble): + if self in ble._links: + ble._links.remove(self) + self.dropped.set() + + +class _FakeConnection(Connection): + def __init__(self, ble, device, role, link): + Connection.__init__(self, ble, device, role) + self._link = link + + def is_connected(self): + return self._link.connected + + async def disconnect(self, timeout_ms=2000): + self._link.drop() + await asyncio.sleep(0) + + async def disconnected(self, timeout_ms=None): + if self._link.connected: + await wait_ms(self._link.dropped.wait(), timeout_ms, "disconnected") + + async def exchange_mtu(self, mtu=None, timeout_ms=1000): + self._link.check() + if mtu: + self._ble.config(mtu=mtu) + link = self._link + agreed = max(DEFAULT_MTU, min(link.central_ble._mtu, link.peripheral_ble._mtu)) + await asyncio.sleep(0) + link.set_mtu(agreed) + return agreed + + def _peer(self): + link = self._link + return link.peripheral_ble if self.role == "central" else link.central_ble + + async def _discover_services(self, uuid, timeout_ms): + self._link.check() + await asyncio.sleep(0) + return [ + _FakeClientService(self, service) + for service in self._peer()._services + if uuid is None or service.uuid == uuid + ] + + +class _FakeClientService(ClientService): + def __init__(self, connection, server_service): + ClientService.__init__(self, connection, server_service.uuid) + self._server = server_service + + async def _discover_characteristics(self, uuid, timeout_ms): + self.connection._link.check() + await asyncio.sleep(0) + return [ + _FakeClientCharacteristic(self, characteristic) + for characteristic in self._server.characteristics + if uuid is None or characteristic.uuid == uuid + ] + + +class _FakeClientCharacteristic(ClientCharacteristic): + def __init__(self, service, server_char): + ClientCharacteristic.__init__(self, service, server_char.uuid, server_char.flags) + self._server = server_char + + def _link(self): + link = self.connection._link + link.check() + return link + + async def read(self, timeout_ms=1000): + self._check(FLAG_READ, "read") + self._link() + await asyncio.sleep(0) + self._link() + return self._server.read() + + async def write(self, data, response=None, timeout_ms=1000): + data = bytes(data) + response = self._want_response(response) + link = self._link() + if len(data) > self.connection.mtu - 3: + raise BLEError( + "write of {} bytes exceeds the ATT payload of {} (mtu {} - 3)".format( + len(data), self.connection.mtu - 3, self.connection.mtu + ) + ) + server = self._server + if len(data) > server.max_len: + raise BLEError( + "write of {} bytes exceeds {!r} max_len={}; real firmware would " + "truncate it".format(len(data), server, server.max_len) + ) + server._impl._remote_write(link.peripheral, data) + await asyncio.sleep(0) + + async def subscribe(self, notify=True, indicate=False): + if notify: + self._check(FLAG_NOTIFY, "notify") + if indicate: + self._check(FLAG_INDICATE, "indicate") + link = self._link() + link.subscribed[id(self._server)] = (bool(notify), bool(indicate)) + await asyncio.sleep(0) + + async def notified(self, timeout_ms=None): + self._check(FLAG_NOTIFY, "notify") + link = self.connection._link + return await link.inbox(self._server, "notify", link.central_ble.queue_limit).get(timeout_ms) + + async def indicated(self, timeout_ms=None): + self._check(FLAG_INDICATE, "indicate") + link = self.connection._link + return await link.inbox(self._server, "indicate", link.central_ble.queue_limit).get(timeout_ms) + + +class _ServerCharacteristic: + """The fake's half of a :class:`bledev.Characteristic`.""" + + def __init__(self, ble, characteristic): + self._ble = ble + self._char = characteristic + self._value = characteristic._initial or b"" + self._writes = _Inbox(ble.queue_limit, "writes to {}".format(characteristic.uuid)) + self._last_writer = None + self._written = asyncio.Event() + + def read(self): + return self._value + + def write(self, data, send_update): + self._value = data + if send_update: + for link in list(self._ble._links): + if link.peripheral_ble is not self._ble or not link.connected: + continue + if self._char.flags & FLAG_NOTIFY: + link.deliver(self._char, "notify", data, False) + if self._char.flags & FLAG_INDICATE: + link.deliver(self._char, "indicate", data, False) + + def _link_for(self, connection): + link = getattr(connection, "_link", None) + if link is None or link.peripheral is not connection: + raise BLEError("{!r} is not a peripheral-side connection of this adapter".format(connection)) + return link + + def _check_size(self, connection, data): + if len(data) > connection.mtu - 3: + raise BLEError( + "notification of {} bytes exceeds the ATT payload of {} (mtu {} - 3)".format( + len(data), connection.mtu - 3, connection.mtu + ) + ) + + def notify(self, connection, data): + link = self._link_for(connection) + data = self._value if data is None else data + self._check_size(connection, data) + link.deliver(self._char, "notify", data, True) + + async def indicate(self, connection, data, timeout_ms): + link = self._link_for(connection) + data = self._value if data is None else data + self._check_size(connection, data) + link.deliver(self._char, "indicate", data, False) + await asyncio.sleep(0) + await asyncio.sleep(0) + link.check() + + def _remote_write(self, connection, data): + self._value = data + if self._char.capture: + self._writes.put((connection, data)) + else: + self._last_writer = connection + self._written.set() + + async def written(self, timeout_ms): + if self._char.capture: + return await self._writes.get(timeout_ms) + if not self._written.is_set(): + await wait_ms(self._written.wait(), timeout_ms, "written") + self._written.clear() + return self._last_writer diff --git a/lib/bledev/nus.py b/lib/bledev/nus.py new file mode 100644 index 00000000..037ddbb6 --- /dev/null +++ b/lib/bledev/nus.py @@ -0,0 +1,374 @@ +"""Nordic UART Service: a two-way byte stream over BLE. + +The board advertises and the other side connects:: + + import bledev.nus as nus + + link = await nus.serve(ble, name="rack") # on the board + link = await nus.connect(ble, name="rack") # on the laptop, phone or another board + +Both return a :class:`Link`, which reads like an asyncio stream:: + + await link.write(b"hello\\n") + line = await link.readline() + async for line in link: # lines until the link closes + ... + await link.close() + +The UUIDs are Nordic's, so nRF Connect, Web Bluetooth serial terminals and +Workbench's Bluetooth transport all speak to it. ``RX`` is what the central +writes; ``TX`` is what the peripheral notifies. + +``ble`` is a bledev adapter. A raw MicroPython ``bluetooth.BLE`` is wrapped in +:class:`bledev.mpble.MPBLE` for you. +""" + +try: + import asyncio +except ImportError: # pragma: no cover + import uasyncio as asyncio + +from . import ( + BLEError, + BLETimeoutError, + BusyError, + Characteristic, + DisconnectedError, + MAX_MTU, + Service, + UnsupportedError, + UUID, + is_adapter, + sleep_ms, +) + +SERVICE = UUID("6e400001-b5a3-f393-e0a9-e50e24dcca9e") +RX = UUID("6e400002-b5a3-f393-e0a9-e50e24dcca9e") +TX = UUID("6e400003-b5a3-f393-e0a9-e50e24dcca9e") + +#: The ATT MTU both sides ask for. 247 fills one data-length-extended link +#: packet (251 bytes), so a notification carries 244 bytes. +MTU = 247 + +# How long the server's receive loop waits before checking for links whose +# peer has gone, in milliseconds. A closed link reports end-of-stream only +# after every write that arrived before the disconnect has been read. +_POLL_MS = 50 + +# How long a send waits after the controller reports its buffers full. +_BUSY_WAIT_MS = 2 + + +def _adapter(ble): + if is_adapter(ble): + return ble + if hasattr(ble, "gap_advertise"): + from .mpble import MPBLE + + return MPBLE(ble) + raise TypeError("expected a bledev adapter, not {!r}".format(ble)) + + +def _set_mtu(ble, mtu): + try: + ble.config(mtu=mtu) + except UnsupportedError: + pass + + +class Link: + """One end of a NUS byte stream. Made by :func:`serve` or :func:`connect`. + + ``read()`` returns ``b""`` once the link has closed and everything that + arrived before has been read. ``bytes_in`` and ``bytes_out`` count + payload bytes. + """ + + def __init__(self, connection): + self.connection = connection + self._buf = bytearray() + self._event = asyncio.Event() + self._closed = False + self._error = None + self.bytes_in = 0 + self.bytes_out = 0 + + def __repr__(self): + return "Link({!r})".format(self.connection) + + @property + def closed(self): + """True once the link is down and nothing is left to read.""" + return self._closed and not self._buf + + # -- fed by the receive side + + def _feed(self, data): + if data: + self._buf.extend(data) + self.bytes_in += len(data) + self._event.set() + + def _finish(self, error=None): + if error is not None and self._error is None: + self._error = error + self._closed = True + self._event.set() + + async def _wait(self): + self._event.clear() + await self._event.wait() + + def _check_error(self): + if self._error is not None: + error, self._error = self._error, None + raise error + + # -- reading + + async def read(self, n=-1): + """Up to ``n`` bytes (all that's buffered when ``n`` < 0), waiting for at least one.""" + while not self._buf: + self._check_error() + if self._closed: + return b"" + await self._wait() + if n < 0 or n >= len(self._buf): + data = bytes(self._buf) + self._buf = bytearray() + else: + data = bytes(self._buf[:n]) + self._buf = self._buf[n:] + return data + + async def readexactly(self, n): + """Exactly ``n`` bytes. Raises :class:`DisconnectedError` if the link closes first.""" + while len(self._buf) < n: + self._check_error() + if self._closed: + raise DisconnectedError( + "link closed after {} of {} bytes".format(len(self._buf), n) + ) + await self._wait() + data = bytes(self._buf[:n]) + self._buf = self._buf[n:] + return data + + async def readline(self): + """Bytes up to and including ``b"\\n"``; at end of stream, whatever is left.""" + while True: + i = self._buf.find(b"\n") + if i >= 0: + data = bytes(self._buf[: i + 1]) + self._buf = self._buf[i + 1 :] + return data + self._check_error() + if self._closed: + data = bytes(self._buf) + self._buf = bytearray() + return data + await self._wait() + + def __aiter__(self): + return self + + async def __anext__(self): + line = await self.readline() + if not line: + raise StopAsyncIteration + return line + + # -- writing + + def _payload(self): + return max(1, self.connection.mtu - 3) + + async def write(self, data): + """Send all of ``data``, split to the link's MTU. Returns when the stack has it.""" + data = memoryview(bytes(data)) + sent = 0 + while sent < len(data): + if not self.connection.is_connected(): + raise DisconnectedError("link closed after sending {} of {} bytes".format(sent, len(data))) + n = min(self._payload(), len(data) - sent) + await self._send(bytes(data[sent : sent + n])) + sent += n + self.bytes_out += n + + async def drain(self): + """For asyncio-stream compatibility; ``write()`` already waited.""" + + async def close(self): + """Disconnect. Anything already received stays readable.""" + if self.connection.is_connected(): + await self.connection.disconnect() + self._finish() + + async def _send(self, chunk): + raise NotImplementedError + + +class _ServerLink(Link): + """The peripheral's end: receives writes on RX, notifies on TX.""" + + def __init__(self, connection, server): + Link.__init__(self, connection) + self._server = server + self._peer_gone = False + self._watch = asyncio.create_task(self._watch_disconnect()) + + async def _watch_disconnect(self): + try: + await self.connection.disconnected() + finally: + # The server loop closes the link once the receive queue is empty, + # so writes that raced the disconnect are still delivered. + self._peer_gone = True + self._server.wake() + + async def _send(self, chunk): + tx = self._server.tx + while True: + try: + tx.notify(self.connection, chunk) + return + except BusyError: + if not self.connection.is_connected(): + raise DisconnectedError("link closed while sending") + await sleep_ms(_BUSY_WAIT_MS) + + +class _Server: + """The GATT service on one adapter, shared by every link it serves.""" + + def __init__(self, ble): + self.ble = ble + self.service = Service(SERVICE) + self.rx = Characteristic( + self.service, RX, write=True, write_no_response=True, capture=True, max_len=MAX_MTU - 3 + ) + self.tx = Characteristic(self.service, TX, notify=True) + self.links = [] + self._task = None + + def add(self, link): + self.links.append(link) + if self._task is None: + self._task = asyncio.create_task(self._run()) + + def wake(self): + pass # the loop polls; kept so a backend with an event can hook it + + def _link_for(self, connection): + for link in self.links: + if link.connection is connection or link.connection == connection: + return link + return None + + def _retire_gone(self): + for link in list(self.links): + if link._peer_gone: + self.links.remove(link) + link._finish() + + async def _run(self): + try: + while self.links: + try: + connection, data = await self.rx.written(timeout_ms=_POLL_MS) + except BLETimeoutError: + self._retire_gone() + continue + link = self._link_for(connection) + if link is not None: + link._feed(data) + except Exception as e: # a receive queue overrun, most likely + for link in self.links: + link._finish(e) + self.links = [] + finally: + self._task = None + + +def _server_for(ble, mtu): + server = getattr(ble, "_nus_server", None) + if server is None: + _set_mtu(ble, mtu) + server = _Server(ble) + ble.register_services(server.service) + ble._nus_server = server + return server + + +async def serve(ble, name="bledev", *, timeout_ms=None, interval_us=100000, mtu=MTU): + """Advertise the Nordic UART service as ``name`` and return a :class:`Link` + when a central connects. + + Call it again for the next connection; the service is registered once per + adapter, so don't register other services on the same adapter afterwards. + """ + ble = _adapter(ble) + server = _server_for(ble, mtu) + connection = await ble.advertise(interval_us, name=name, services=[SERVICE], timeout_ms=timeout_ms) + link = _ServerLink(connection, server) + server.add(link) + return link + + +class _ClientLink(Link): + """The central's end: writes RX, receives TX notifications.""" + + def __init__(self, connection, rx, tx): + Link.__init__(self, connection) + self._rx = rx + self._tx = tx + self._no_response = bool(rx.properties & 0x0004) + self._pump = asyncio.create_task(self._receive()) + + async def _receive(self): + error = None + try: + while True: + self._feed(await self._tx.notified()) + except DisconnectedError: + pass + except Exception as e: + error = e + finally: + self._finish(error) + + async def _send(self, chunk): + await self._rx.write(chunk, response=not self._no_response) + + +async def connect(ble, name=None, *, device=None, timeout_ms=10000, mtu=MTU, **connect_options): + """Find a NUS peripheral (by ``name``, or the first one advertising the + service), connect, and return a :class:`Link`. + + Pass ``device`` to skip the scan. ``connect_options`` go to + ``device.connect()``; on a board, ``min_conn_interval_us`` and + ``max_conn_interval_us`` trade power for throughput. + """ + ble = _adapter(ble) + _set_mtu(ble, mtu) + if device is None: + device = await ble.find(name=name, service=SERVICE, timeout_ms=timeout_ms) + connection = await device.connect(timeout_ms=timeout_ms, **connect_options) + try: + try: + await connection.exchange_mtu(mtu) + except (UnsupportedError, BLETimeoutError): + pass + service = await connection.service(SERVICE) + if service is None: + raise BLEError("{!r} has no Nordic UART service".format(device)) + rx = await service.characteristic(RX) + tx = await service.characteristic(TX) + if rx is None or tx is None: + raise BLEError("{!r}: Nordic UART service is missing RX or TX".format(device)) + await tx.subscribe(notify=True) + except BaseException: + if connection.is_connected(): + await connection.disconnect() + raise + return _ClientLink(connection, rx, tx) diff --git a/pydevices-desktop.toml b/pydevices-desktop.toml index 9345d4b6..251a3539 100644 --- a/pydevices-desktop.toml +++ b/pydevices-desktop.toml @@ -20,6 +20,10 @@ "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/audiodev/wasm_audio.py" = "/lib/audiodev/wasm_audio.py" "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/audiodev/web_audio.py" = "/lib/audiodev/web_audio.py" "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/audiodev/win_audio.py" = "/lib/audiodev/win_audio.py" +"https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/bledev/__init__.py" = "/lib/bledev/__init__.py" +"https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/bledev/auto.py" = "/lib/bledev/auto.py" +"https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/bledev/fake.py" = "/lib/bledev/fake.py" +"https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/bledev/nus.py" = "/lib/bledev/nus.py" "https://raw.githubusercontent.com/PyDevices/pydevices/main/board_configs/desktop/board_config.py" = "/lib/board_config.py" "https://raw.githubusercontent.com/PyDevices/pydevices/main/board_configs/desktop/board_peripherals.py" = "/lib/board_peripherals.py" "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/boarddev.py" = "/lib/boarddev.py" diff --git a/tests/bledev_contract.py b/tests/bledev_contract.py new file mode 100644 index 00000000..a57f1164 --- /dev/null +++ b/tests/bledev_contract.py @@ -0,0 +1,704 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""The bledev contract, checked against ``bledev.fake``. + +A plain script, not unittest, so the same file runs on CPython and on +MicroPython (which has no unittest):: + + python tests/bledev_contract.py + micropython tests/bledev_contract.py + python tests/bledev_contract.py --plant=corrupt # must FAIL + +It prints one PASS/FAIL line per check and exits non-zero on any failure. +``tests/test_bledev.py`` runs it both ways, and runs every plant to prove the +checks can fail: a plant breaks the fake the way a real backend could, and a +contract that still passes under it isn't checking anything. + +A new backend can reuse these checks: most take two adapters and don't care +which backend they are (see ``docs/bledev.md``). +""" + +import sys + +try: + _here = __file__ +except NameError: # pragma: no cover + _here = sys.argv[0] +_sep = "\\" if "\\" in _here else "/" +_dir = _here.rsplit(_sep, 1)[0] if _sep in _here else "." +sys.path.insert(0, _dir + _sep + ".." + _sep + "lib") + +try: + import asyncio +except ImportError: # pragma: no cover + import uasyncio as asyncio + +import bledev +from bledev import ( + BLEError, + BLETimeoutError, + BusyError, + Characteristic, + DisconnectedError, + GattError, + Service, + UnsupportedError, + UUID, +) +from bledev import fake, nus +from bledev.fake import Air, FakeBLE + +MICROPYTHON = sys.implementation.name == "micropython" + +#: A check that hangs (a lost notification, say) fails after this long. +CHECK_TIMEOUT_S = 5 + +SVC = UUID("12345678-1234-5678-1234-56789abcdef0") +CHR_RW = UUID("12345678-1234-5678-1234-56789abcdef1") +CHR_STREAM = UUID("12345678-1234-5678-1234-56789abcdef2") +CHR_IND = UUID("12345678-1234-5678-1234-56789abcdef3") + +TESTS = [] + + +def check(fn): + TESTS.append(fn) + return fn + + +class Failed(Exception): + pass + + +def expect(cond, message="expectation failed"): + if not cond: + raise Failed(message) + + +def equal(a, b, what="value"): + if a != b: + raise Failed("{}: {} != {}".format(what, _short(a), _short(b))) + + +def _short(value): + text = repr(value) + return text if len(text) <= 100 else text[:97] + "..." + + +async def raises(exc_type, awaitable): + try: + await awaitable + except exc_type as e: + return e + raise Failed("expected {}".format(exc_type.__name__)) + + +def raises_sync(exc_type, fn, *args, **kwargs): + try: + fn(*args, **kwargs) + except exc_type as e: + return e + raise Failed("expected {}".format(exc_type.__name__)) + + +def pattern(n, seed=0): + """Bytes nobody gets right by accident: position-dependent, not periodic at 256.""" + return bytes(((i * 7 + (i >> 8) * 13 + seed) & 0xFF) for i in range(n)) + + +def gatt(initial=b"hi"): + service = Service(SVC) + rw = Characteristic(service, CHR_RW, read=True, write=True, initial=initial) + stream = Characteristic( + service, CHR_STREAM, write_no_response=True, notify=True, capture=True, max_len=100 + ) + ind = Characteristic(service, CHR_IND, indicate=True, read=True) + return service, rw, stream, ind + + +async def pair(air, *, board_mtu=23, central_mtu=23, services=None, name="board", **fake_options): + """A peripheral with the test service, and a central connected to it.""" + board = FakeBLE(air=air, mtu=board_mtu, **fake_options) + laptop = FakeBLE(air=air, peripheral=False, mtu=central_mtu, **fake_options) + parts = gatt() + board.register_services(parts[0]) + adv = asyncio.create_task(board.advertise(name=name, services=services or [SVC], timeout_ms=2000)) + device = await laptop.find(name=name, timeout_ms=2000) + central = await device.connect(timeout_ms=2000) + peripheral = await adv + return board, laptop, central, peripheral, parts + + +async def client_chars(central): + service = await central.service(SVC) + expect(service is not None, "service not found") + return ( + await service.characteristic(CHR_RW), + await service.characteristic(CHR_STREAM), + await service.characteristic(CHR_IND), + ) + + +# ---------------------------------------------------------------- values + + +@check +async def uuid_forms_agree(air): + short = UUID(0x180F) + for other in ("180f", "0x180F", "0000180f-0000-1000-8000-00805f9b34fb", b"\x0f\x18", UUID(0x180F)): + equal(UUID(other), short, "UUID({!r})".format(other)) + equal(short.short, 0x180F, "short") + equal(str(short), "0000180f-0000-1000-8000-00805f9b34fb", "str") + equal(short.to_bytes(), b"\x0f\x18", "to_bytes") + long = UUID("6E400001-B5A3-F393-E0A9-E50E24DCCA9E") + equal(str(long), "6e400001-b5a3-f393-e0a9-e50e24dcca9e", "str(long)") + equal(long.short, None, "long.short") + equal(UUID(long.to_bytes()), long, "wire round trip") + equal(long.to_bytes()[0], 0x9E, "little-endian wire order") + expect(hash(UUID("180f")) == hash(short), "hash follows equality") + expect(short != long and not (short == long), "different UUIDs differ") + expect(short in [long, UUID(0x180F)], "list membership") + expect({short: 1}.get(UUID("0x180f")) == 1, "dict key") + raises_sync(ValueError, UUID, "not-a-uuid") + raises_sync(ValueError, UUID, b"\x00\x01\x02") + + +@check +async def errors_share_a_base(air): + for exc in (UnsupportedError, DisconnectedError, BLETimeoutError, BusyError, GattError): + expect(issubclass(exc, BLEError), exc.__name__) + equal(GattError(0x0A).status, 0x0A, "GattError.status") + + +@check +async def advertisement_packing(air): + adv, resp = bledev.pack_advertisement("rack", [nus.SERVICE]) + expect(len(adv) <= 31 and not resp, "short name fits in the advertisement") + name, services, _, _ = bledev.decode_advertisement(adv) + equal(name, "rack", "name") + equal(services, [nus.SERVICE], "services") + adv, resp = bledev.pack_advertisement("a-rather-long-board-name", [nus.SERVICE]) + equal(bledev.decode_advertisement(adv)[0], None, "long name left the advertisement") + equal(bledev.decode_advertisement(adv, resp)[0], "a-rather-long-board-name", "and went to the scan response") + adv, resp = bledev.pack_advertisement(None, [0x180F], appearance=0x03C1, manufacturer=(0xFFFF, b"\x01\x02")) + _, services, appearance, manufacturer = bledev.decode_advertisement(adv, resp) + equal(services, [UUID(0x180F)], "16-bit service") + equal(appearance, 0x03C1, "appearance") + equal(manufacturer, [(0xFFFF, b"\x01\x02")], "manufacturer") + raises_sync(ValueError, bledev.pack_advertisement, "x" * 40, [nus.SERVICE]) + + +# ---------------------------------------------------------------- roles + + +@check +async def capabilities_report_roles(air): + board = FakeBLE(air=air) + laptop = FakeBLE(air=air, peripheral=False) + for ble in (board, laptop): + caps = ble.capabilities() + for key in ("backend", "central", "peripheral", "mtu", "max_mtu", "indicate", "pairing", "l2cap"): + expect(key in caps, "capabilities lacks " + key) + equal(caps["backend"], "fake", "backend") + expect(board.capabilities()["peripheral"], "board can be a peripheral") + expect(not laptop.capabilities()["peripheral"], "laptop cannot") + expect(laptop.capabilities()["central"], "laptop can be central") + + +@check +async def central_only_cannot_advertise(air): + laptop = FakeBLE(air=air, peripheral=False) + await raises(UnsupportedError, laptop.advertise(name="x", timeout_ms=10)) + raises_sync(UnsupportedError, laptop.register_services, Service(SVC)) + await raises(UnsupportedError, nus.serve(laptop, name="x", timeout_ms=10)) + board = FakeBLE(air=air, central=False) + raises_sync(UnsupportedError, board.scan, 10) + + +@check +async def config_mtu(air): + ble = FakeBLE(air=air) + ble.config(mtu=185) + equal(ble.config("mtu"), 185, "config('mtu')") + raises_sync(UnsupportedError, ble.config, flux=1) + + +# ---------------------------------------------------------------- discovery + + +@check +async def find_by_name_and_service(air): + board = FakeBLE(air=air) + laptop = FakeBLE(air=air, peripheral=False) + adv = asyncio.create_task(board.advertise(name="rack", services=[SVC], timeout_ms=1000)) + device = await laptop.find(name="rack", timeout_ms=1000) + equal(device.name, "rack", "device.name") + expect(isinstance(device.address, str), "address is a string") + device2 = await laptop.find(service=SVC, timeout_ms=1000) + equal(device2, device, "same device by service") + adv.cancel() + + +@check +async def scan_yields_results(air): + board = FakeBLE(air=air) + laptop = FakeBLE(air=air, peripheral=False) + adv = asyncio.create_task(board.advertise(name="rack", services=[SVC], manufacturer=(0x1234, b"z"))) + await asyncio.sleep(0) + seen = [] + async with laptop.scan(100, active=True) as scanner: # manufacturer data overflows to the scan response + async for result in scanner: + seen.append(result) + break + expect(seen, "scan found nothing") + result = seen[0] + equal(result.name(), "rack", "name()") + expect(SVC in result.services(), "services()") + equal(result.manufacturer(0x1234), [(0x1234, b"z")], "manufacturer()") + expect(result.rssi is not None, "rssi") + adv.cancel() + + +@check +async def passive_scan_misses_a_long_name(air): + board = FakeBLE(air=air) + laptop = FakeBLE(air=air, peripheral=False) + adv = asyncio.create_task(board.advertise(name="a-rather-long-board-name", services=[SVC])) + await raises(BLETimeoutError, laptop.find(name="a-rather-long-board-name", timeout_ms=60, active=False)) + device = await laptop.find(name="a-rather-long-board-name", timeout_ms=200, active=True) + expect(device is not None, "active scan finds it") + adv.cancel() + + +@check +async def timeouts_raise(air): + board = FakeBLE(air=air) + laptop = FakeBLE(air=air, peripheral=False) + await raises(BLETimeoutError, board.advertise(name="lonely", timeout_ms=30)) + await raises(BLETimeoutError, laptop.find(name="nobody", timeout_ms=30)) + ghost = bledev.Device(laptop, "fa:ce:ff:ff:ff:ff") + await raises(BLETimeoutError, ghost.connect(timeout_ms=30)) + + +@check +async def connect_and_discover(air): + board, laptop, central, peripheral, parts = await pair(air) + equal(central.role, "central", "central.role") + equal(peripheral.role, "peripheral", "peripheral.role") + expect(central.is_connected() and peripheral.is_connected(), "both connected") + equal(central.mtu, 23, "default mtu") + names = [] + async for service in central.services(): + names.append(service.uuid) + equal(names, [SVC], "services()") + equal(await central.service(UUID(0x180F)), None, "missing service is None") + service = await central.service(SVC) + chars = [] + async for c in service.characteristics(): + chars.append(c.uuid) + equal(chars, [CHR_RW, CHR_STREAM, CHR_IND], "characteristics()") + rw = await service.characteristic(CHR_RW) + expect(rw.properties & bledev.FLAG_READ and rw.properties & bledev.FLAG_WRITE, "properties") + equal(await service.characteristic(UUID(0x2A19)), None, "missing characteristic is None") + + +# ---------------------------------------------------------------- GATT + + +@check +async def read_and_write(air): + board, laptop, central, peripheral, (svc, rw, stream, ind) = await pair(air) + crw, cstream, cind = await client_chars(central) + equal(await crw.read(), b"hi", "initial value") + rw.write(b"local") + equal(await crw.read(), b"local", "after local write") + waiter = asyncio.create_task(rw.written(timeout_ms=1000)) + await crw.write(b"remote", response=True) + writer = await waiter + expect(writer is peripheral, "written() returns the writer's connection") + equal(rw.read(), b"remote", "server sees the write") + await raises(UnsupportedError, crw.write(b"x", response=False)) + await raises(UnsupportedError, cstream.read()) + + +@check +async def capture_keeps_every_write_in_order(air): + board, laptop, central, peripheral, (svc, rw, stream, ind) = await pair(air) + crw, cstream, cind = await client_chars(central) + sent = [pattern(20, seed=i) for i in range(60)] + for chunk in sent: + await cstream.write(chunk) + got = [] + for _ in sent: + connection, data = await stream.written(timeout_ms=500) + expect(connection is peripheral, "capture reports the connection") + got.append(data) + equal(got, sent, "captured writes") + await raises(BLETimeoutError, stream.written(timeout_ms=0)) + + +@check +async def notifications_need_a_subscription(air): + board, laptop, central, peripheral, (svc, rw, stream, ind) = await pair(air) + crw, cstream, cind = await client_chars(central) + stream.notify(peripheral, b"early") + await raises(BLETimeoutError, cstream.notified(timeout_ms=30)) + await cstream.subscribe(notify=True) + stream.notify(peripheral, b"late") + equal(await cstream.notified(timeout_ms=500), b"late", "after subscribe") + + +@check +async def notifications_arrive_in_order(air): + board, laptop, central, peripheral, (svc, rw, stream, ind) = await pair(air) + crw, cstream, cind = await client_chars(central) + await cstream.subscribe() + sent = [pattern(20, seed=i) for i in range(300)] + + async def send(): + for chunk in sent: + while True: + try: + stream.notify(peripheral, chunk) + break + except BusyError: + await asyncio.sleep(0) + + sender = asyncio.create_task(send()) + got = [] + for _ in sent: + got.append(await cstream.notified(timeout_ms=1000)) + await sender + equal(len(got), len(sent), "count") + equal(got, sent, "notifications") + + +@check +async def a_burst_reports_busy(air): + board, laptop, central, peripheral, (svc, rw, stream, ind) = await pair(air, notify_buffers=4) + crw, cstream, cind = await client_chars(central) + await cstream.subscribe() + for i in range(4): + stream.notify(peripheral, bytes((i,))) + raises_sync(BusyError, stream.notify, peripheral, b"x") + await asyncio.sleep(0) + await asyncio.sleep(0) + stream.notify(peripheral, b"room again") + + +@check +async def mtu_bounds_every_packet(air): + board, laptop, central, peripheral, (svc, rw, stream, ind) = await pair(air, board_mtu=185, central_mtu=247) + crw, cstream, cind = await client_chars(central) + await cstream.subscribe() + raises_sync(BLEError, stream.notify, peripheral, bytes(21)) + await raises(BLEError, cstream.write(bytes(21))) + agreed = await central.exchange_mtu() + equal(agreed, 185, "the smaller MTU wins") + equal(central.mtu, 185, "central.mtu") + equal(peripheral.mtu, 185, "peripheral.mtu follows") + await cstream.write(bytes(100)) + await raises(BLEError, cstream.write(bytes(101))) # max_len=100 + stream.notify(peripheral, bytes(182)) + equal(len(await cstream.notified(timeout_ms=500)), 182, "a full-MTU notification") + + +@check +async def indications(air): + board, laptop, central, peripheral, (svc, rw, stream, ind) = await pair(air) + crw, cstream, cind = await client_chars(central) + await cind.subscribe(notify=False, indicate=True) + await ind.indicate(peripheral, b"ack me") + equal(await cind.indicated(timeout_ms=500), b"ack me", "indicated()") + raises_sync(UnsupportedError, ind.notify, peripheral, b"x") + + +@check +async def overrun_is_loud(air): + board, laptop, central, peripheral, (svc, rw, stream, ind) = await pair(air, queue_limit=10) + crw, cstream, cind = await client_chars(central) + await cstream.subscribe() + for i in range(30): + while True: + try: + stream.notify(peripheral, bytes((i,))) + break + except BusyError: + await asyncio.sleep(0) + await asyncio.sleep(0) + await asyncio.sleep(0) + await raises(BLEError, cstream.notified(timeout_ms=100)) + + +# ---------------------------------------------------------------- lifecycle + + +@check +async def disconnect_reaches_both_sides(air): + board, laptop, central, peripheral, (svc, rw, stream, ind) = await pair(air) + crw, cstream, cind = await client_chars(central) + await cstream.subscribe() + waiting = asyncio.create_task(cstream.notified()) + await asyncio.sleep(0) + await peripheral.disconnect() + await central.disconnected(timeout_ms=500) + expect(not central.is_connected() and not peripheral.is_connected(), "both down") + await raises(DisconnectedError, waiting) + await raises(DisconnectedError, crw.read()) + await raises(DisconnectedError, crw.write(b"x")) + raises_sync(DisconnectedError, stream.notify, peripheral, b"x") + + +@check +async def data_sent_before_a_disconnect_is_kept(air): + board, laptop, central, peripheral, (svc, rw, stream, ind) = await pair(air) + crw, cstream, cind = await client_chars(central) + await cstream.subscribe() + stream.notify(peripheral, b"last words") + await asyncio.sleep(0) + await asyncio.sleep(0) + await central.disconnect() + equal(await cstream.notified(timeout_ms=100), b"last words", "queued before the drop") + await raises(DisconnectedError, cstream.notified(timeout_ms=100)) + + +@check +async def async_with_disconnects(air): + board, laptop, central, peripheral, parts = await pair(air) + async with central: + expect(central.is_connected(), "inside") + expect(not central.is_connected(), "after") + await peripheral.disconnected(timeout_ms=100) + + +@check +async def advertise_again_after_a_connection(air): + board, laptop, central, peripheral, parts = await pair(air) + adv = asyncio.create_task(board.advertise(name="board", timeout_ms=1000)) + device = await laptop.find(name="board", timeout_ms=1000) + second = await device.connect(timeout_ms=1000) + other = await adv + expect(other is not peripheral, "a new connection") + expect(central.is_connected() and second.is_connected(), "both links up") + + +@check +async def close_drops_everything(air): + board, laptop, central, peripheral, parts = await pair(air) + board.close() + expect(not central.is_connected(), "closed") + + +# ---------------------------------------------------------------- nus + + +async def nus_pair(air, *, name="rack", mtu=nus.MTU, **options): + board = FakeBLE(air=air, **options) + laptop = FakeBLE(air=air, peripheral=False, **options) + served = asyncio.create_task(nus.serve(board, name=name, mtu=mtu, timeout_ms=2000)) + client = await nus.connect(laptop, name=name, mtu=mtu, timeout_ms=2000) + server = await served + return board, laptop, server, client + + +@check +async def nus_round_trip(air): + board, laptop, server, client = await nus_pair(air) + equal(client.connection.mtu, nus.MTU, "negotiated MTU") + await client.write(b"ping\n") + equal(await server.readline(), b"ping\n", "central -> peripheral") + await server.write(b"pong\n") + equal(await client.readline(), b"pong\n", "peripheral -> central") + + +@check +async def nus_bulk_both_ways(air): + board, laptop, server, client = await nus_pair(air) + down = pattern(10240, seed=1) + up = pattern(10240, seed=2) + sender = asyncio.create_task(server.write(down)) + got = await client.readexactly(len(down)) + await sender + equal(len(got), len(down), "downstream length") + expect(got == down, "downstream bytes differ") + sender = asyncio.create_task(client.write(up)) + got = await server.readexactly(len(up)) + await sender + expect(got == up, "upstream bytes differ") + equal(server.bytes_in, len(up), "bytes_in") + equal(server.bytes_out, len(down), "bytes_out") + + +@check +async def nus_small_mtu_still_intact(air): + board, laptop, server, client = await nus_pair(air, mtu=23) + equal(client.connection.mtu, 23, "mtu") + data = pattern(2000, seed=3) + sender = asyncio.create_task(client.write(data)) + expect(await server.readexactly(len(data)) == data, "bytes differ") + await sender + + +@check +async def nus_lines_and_end_of_stream(air): + board, laptop, server, client = await nus_pair(air) + await client.write(b"one\ntwo\nthree") + await asyncio.sleep(0) + await client.close() + lines = [] + async for line in server: + lines.append(line) + equal(lines, [b"one\n", b"two\n", b"three"], "lines, then the unterminated tail") + equal(await server.read(), b"", "end of stream") + expect(server.closed, "server.closed") + await raises(DisconnectedError, client.write(b"x")) + + +@check +async def nus_serves_two_links(air): + board = FakeBLE(air=air) + a = FakeBLE(air=air, peripheral=False) + b = FakeBLE(air=air, peripheral=False) + served = asyncio.create_task(nus.serve(board, name="hub", timeout_ms=2000)) + link_a = await nus.connect(a, name="hub", timeout_ms=2000) + server_a = await served + served = asyncio.create_task(nus.serve(board, name="hub", timeout_ms=2000)) + link_b = await nus.connect(b, name="hub", timeout_ms=2000) + server_b = await served + await link_a.write(b"from a\n") + await link_b.write(b"from b\n") + equal(await server_b.readline(), b"from b\n", "b's bytes go to b's link") + equal(await server_a.readline(), b"from a\n", "a's bytes go to a's link") + + +@check +async def nus_readexactly_short_stream(air): + board, laptop, server, client = await nus_pair(air) + await server.write(b"abc") + await asyncio.sleep(0) + await asyncio.sleep(0) + await server.close() + await raises(DisconnectedError, client.readexactly(10)) + + +# ---------------------------------------------------------------- auto + + +@check +async def auto_picks_or_explains(air): + import bledev.auto as auto + + name = auto.backend_name() + expect(name in (None, "mpble", "bleak", "webble", "fake"), "backend_name: {!r}".format(name)) + ble = auto.adapter(backend="fake", air=air) + equal(ble.backend, "fake", "forced backend") + if name is None: + await raises(UnsupportedError, _call(auto.adapter)) + + +async def _call(fn, *args, **kwargs): + return fn(*args, **kwargs) + + +# ---------------------------------------------------------------- plants + + +def _plant(name): + """Break the fake on purpose. Each plant must make at least one check fail.""" + if name == "drop": + put = fake._Inbox.put + count = [0] + + def dropping_put(self, item): + count[0] += 1 + if count[0] % 97 == 0: + return # silently lost, the way aioble's one-slot queue loses data + put(self, item) + + fake._Inbox.put = dropping_put + elif name == "corrupt": + feed = nus.Link._feed + + def corrupting_feed(self, data): + if len(data) > 8: + data = bytearray(data) + data[5] ^= 0x01 + feed(self, data) + + nus.Link._feed = corrupting_feed + elif name == "reorder": + drain = fake._Link._drain + + async def reordering_drain(self): + if len(self.in_flight) > 1: + self.in_flight[0], self.in_flight[1] = self.in_flight[1], self.in_flight[0] + await drain(self) + + fake._Link._drain = reordering_drain + elif name == "truncate": + check_size = fake._ServerCharacteristic._check_size + + def lax(self, connection, data): + pass # what NimBLE does: no error, the tail is cut on the air + + fake._ServerCharacteristic._check_size = lax + elif name == "overwrite": + # aioble's client queue holds one notification; a second replaces it. + def one_slot(self, item): + self.items[:] = [item] + self.event.set() + + fake._Inbox.put = one_slot + else: + raise SystemExit("unknown plant " + name) + + +PLANTS = ("drop", "corrupt", "reorder", "truncate", "overwrite") + + +def _print_exception(e): + if MICROPYTHON: + sys.print_exception(e) + else: + import traceback + + traceback.print_exception(type(e), e, e.__traceback__) + + +def run(only=None, verbose=False): + failed = 0 + for fn in TESTS: + if only and fn.__name__ not in only: + continue + if MICROPYTHON: + asyncio.new_event_loop() + try: + asyncio.run(asyncio.wait_for(fn(Air()), CHECK_TIMEOUT_S)) + except Exception as e: + failed += 1 + print("FAIL", fn.__name__, "-", type(e).__name__, e) + if verbose: + _print_exception(e) + else: + print("PASS", fn.__name__) + print("{} checks, {} failed ({})".format(len(TESTS) if not only else len(only), failed, sys.implementation.name)) + return failed + + +def main(argv): + only = [] + verbose = False + for arg in argv: + if arg.startswith("--plant="): + _plant(arg.split("=", 1)[1]) + print("PLANTED", arg.split("=", 1)[1]) + elif arg in ("-v", "--verbose"): + verbose = True + else: + only.append(arg) + sys.exit(1 if run(only, verbose) else 0) + + +if __name__ == "__main__": + main(sys.argv[1:]) diff --git a/tests/test_bledev.py b/tests/test_bledev.py new file mode 100644 index 00000000..bb56e14a --- /dev/null +++ b/tests/test_bledev.py @@ -0,0 +1,83 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""Run the bledev contract (``bledev_contract.py``) on CPython and MicroPython. + +The contract is a plain script so MicroPython can run it too. This file runs +it under CPython, under a unix ``micropython`` when one is available, and once +per planted fault, each of which must make it fail. + +The MicroPython run looks for ``PYDEVICES_MICROPYTHON``, then ``micropython`` +on PATH, then the workspace's ``bin/micropython`` beside this checkout. When +none exists the skip is loud, and ``PYDEVICES_REQUIRE_MICROPYTHON=1`` turns it +into a failure. +""" + +from pathlib import Path +import os +import shutil +import subprocess +import sys +import unittest + +_TESTS = Path(__file__).resolve().parent +CONTRACT = _TESTS / "bledev_contract.py" +PLANTS = ("drop", "corrupt", "reorder", "truncate", "overwrite") + + +def _micropython(): + explicit = os.environ.get("PYDEVICES_MICROPYTHON") + if explicit: + return explicit + found = shutil.which("micropython") + if found: + return found + # The workspace layout: ~/gh/pydevices/bin next to the pydevices checkout + # (or its worktrees). + for parent in _TESTS.parents: + candidate = parent / "bin" / "micropython" + if candidate.is_file() and os.access(str(candidate), os.X_OK): + return str(candidate) + return None + + +def _run(interpreter, *args): + return subprocess.run( + [interpreter, str(CONTRACT), *args], + capture_output=True, + text=True, + timeout=600, + ) + + +class BledevContract(unittest.TestCase): + def assertPasses(self, result): + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + self.assertIn(" 0 failed", result.stdout) + + def test_cpython(self): + self.assertPasses(_run(sys.executable)) + + def test_micropython(self): + interpreter = _micropython() + if interpreter is None: + message = "no micropython binary: the bledev contract did NOT run on MicroPython" + if os.environ.get("PYDEVICES_REQUIRE_MICROPYTHON") == "1": + self.fail(message) + sys.stderr.write("\n*** " + message + " ***\n") + self.skipTest(message) + result = _run(interpreter) + self.assertPasses(result) + self.assertIn("(micropython)", result.stdout) + + def test_every_plant_fails(self): + for plant in PLANTS: + with self.subTest(plant=plant): + result = _run(sys.executable, "--plant=" + plant) + self.assertIn("PLANTED " + plant, result.stdout) + self.assertNotEqual(result.returncode, 0, "plant {!r} went unnoticed:\n{}".format(plant, result.stdout)) + self.assertIn("FAIL", result.stdout) + + +if __name__ == "__main__": + unittest.main() From 958805528c9bdc776bd6c0f4a35aa41a17e8969e Mon Sep 17 00:00:00 2001 From: Brad Barnett <127794626+bdbarnett@users.noreply.github.com> Date: Thu, 24 Sep 2026 11:23:20 -0500 Subject: [PATCH 2/6] bledev.mpble: the MicroPython backend, over aioble Both roles, on the one bluetooth.BLE(). Three places aioble loses data are closed here: a client characteristic keeps a queue instead of one notification slot (and notified() waits only on an empty queue, since aioble's wait-when-one-left logic strands the last item of a longer queue); captured writes are copied in the IRQ into a per-characteristic queue instead of a shared ten-entry deque; and payloads over mtu - 3 are refused instead of truncated. Scans default to full duty: aioble's 1 % default found a board a few inches away once in six seconds. Proven board to board on two ESP32-S3s: 16 KB each way and echoed, byte-for-byte, and a planted one-bit flip fails every phase. --- lib/bledev/mpble.py | 581 +++++++++++++++++++++++++++++++++++++++++ lib/bledev/nus.py | 4 +- pydevices-desktop.toml | 1 + 3 files changed, 584 insertions(+), 2 deletions(-) create mode 100644 lib/bledev/mpble.py diff --git a/lib/bledev/mpble.py b/lib/bledev/mpble.py new file mode 100644 index 00000000..2c73270e --- /dev/null +++ b/lib/bledev/mpble.py @@ -0,0 +1,581 @@ +"""MicroPython backend, over aioble. Both roles. + + from bledev.mpble import MPBLE + ble = MPBLE() # or MPBLE(bluetooth.BLE()) from a board config + +Needs aioble, which isn't frozen into our firmware: install it with +``mip.install("aioble")``, or install ``bledev`` from the PyDevices index, which +brings aioble with it. + +aioble owns the one ``bluetooth.BLE()`` object and its IRQ handler, so don't +set ``ble.irq()`` yourself while this backend is in use. + +Where aioble would lose data, this backend doesn't: + +* aioble keeps one notification per client characteristic and replaces it + when the next arrives. Here each gets a queue of ``queue_limit`` entries, + and an overflow raises from ``notified()``. +* aioble's ``capture=True`` shares a ten-entry queue across every + characteristic and drops the oldest. Here captured writes are read straight + from the IRQ into a per-characteristic queue of ``queue_limit``. +* ``notify()`` and ``write()`` refuse a payload over ``mtu - 3`` instead of + letting NimBLE truncate it. +""" + +try: + import asyncio +except ImportError: # pragma: no cover + import uasyncio as asyncio + +import bluetooth +import aioble +from aioble import core as _aioble_core +from aioble.device import DeviceConnection as _AioConnection + +from . import ( + BLE, + BLEError, + BLETimeoutError, + BusyError, + ClientCharacteristic, + ClientService, + Connection, + DEFAULT_MTU, + Device, + DisconnectedError, + FLAG_INDICATE, + FLAG_NOTIFY, + FLAG_READ, + GattError, + ScanResult, + Scanner, + UnsupportedError, + UUID, + sleep_ms, +) + +_IRQ_GATTS_WRITE = 3 + +# errno values NimBLE uses when its buffers are full (ENOMEM, EAGAIN, EBUSY, +# ENOBUFS). A send that sees one of these waits and tries again. +_BUSY_ERRNOS = (12, 11, 16, 105) + +# The largest ATT MTU MicroPython's NimBLE will negotiate. +_MAX_MTU = 512 + +# Default scan interval and window: listen all the time. +_SCAN_US = 30000 + +# How long a write without response keeps retrying a full buffer by default. +_WRITE_RETRY_MS = 2000 + + +def _buuid(uuid): + """bledev UUID -> bluetooth.UUID.""" + uuid = UUID(uuid) + short = uuid.short + if short is not None and short <= 0xFFFF: + return bluetooth.UUID(short) + return bluetooth.UUID(str(uuid)) + + +def _uuid(buuid): + """bluetooth.UUID -> bledev UUID.""" + return UUID(bytes(buuid)) + + +def _errno(e): + return e.args[0] if e.args else None + + +class _Queue: + """Stands in for aioble's one-slot deque: same three operations, but it + keeps ``limit`` entries and turns an overflow into an error.""" + + def __init__(self, limit, what): + self._items = [] + self._limit = limit + self._what = what + self.overrun = 0 + + def __len__(self): + return len(self._items) + + def append(self, item): + if len(self._items) >= self._limit: + self.overrun += 1 + else: + self._items.append(item) + + def popleft(self): + if self.overrun: + raise BLEError( + "{} queue overran: {} dropped because the reader fell {} behind".format( + self._what, self.overrun, self._limit + ) + ) + return self._items.pop(0) + + +class _Translate: + """Maps aioble's exceptions onto bledev's.""" + + def __init__(self, what): + self._what = what + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + if exc_type is None: + return False + if exc_type is asyncio.TimeoutError or issubclass(exc_type, asyncio.TimeoutError): + raise BLETimeoutError("{} timed out".format(self._what)) + if issubclass(exc_type, aioble.DeviceDisconnectedError): + raise DisconnectedError("{}: disconnected".format(self._what)) + if issubclass(exc_type, aioble.GattError): + raise GattError(getattr(exc, "_status", 0)) + if exc_type is ValueError and exc.args and exc.args[0] == "Not connected": + raise DisconnectedError("{}: not connected".format(self._what)) + return False + + +# Captured characteristics by value handle, for the IRQ handler. Module-level +# because there is one radio however many MPBLE objects wrap it. +_captures = {} +_irq_registered = False +_shared = None + + +def _irq(event, data): + # Runs beside aioble's own handlers. A captured write is copied here, in + # the IRQ, before the next write can overwrite the value buffer, and tied + # to its connection while that connection still exists. + if event == _IRQ_GATTS_WRITE: + conn_handle, value_handle = data + impl = _captures.get(value_handle) + if impl is not None: + impl._captured( + _AioConnection._connected.get(conn_handle), + bytes(_aioble_core.ble.gatts_read(value_handle)), + ) + return None + + +def get(ble=None): + """The shared :class:`MPBLE` for this board's radio, made on first use.""" + global _shared + if _shared is None: + _shared = MPBLE(ble) + return _shared + + +class MPBLE(BLE): + """The board's radio. ``ble`` may be a ``bluetooth.BLE()``; there is only one. + + ``queue_limit`` bounds each notification and capture queue. Prefer + :func:`get`, which returns one shared adapter. + """ + + backend = "mpble" + + def __init__(self, ble=None, *, queue_limit=128, mtu=None): + global _irq_registered + if ble is not None and ble is not _aioble_core.ble: + raise BLEError("MicroPython has one bluetooth.BLE(); pass that one or nothing") + self.raw = _aioble_core.ble + self.queue_limit = queue_limit + self._connections = {} + self._services = () + if not _irq_registered: + _aioble_core.register_irq_handler(_irq, None) + _irq_registered = True + if mtu: + self.config(mtu=mtu) + + def capabilities(self): + caps = BLE.capabilities(self) + caps.update( + central=True, + peripheral=True, + mtu=self._config_mtu(), + max_mtu=_MAX_MTU, + indicate=True, + pairing=True, + ) + return caps + + def _config_mtu(self): + try: + return aioble.config("mtu") + except Exception: + return DEFAULT_MTU + + def config(self, *names, **settings): + if settings: + aioble.config(**settings) + if names: + if len(names) != 1: + raise ValueError("read one setting at a time") + return aioble.config(names[0]) + return None + + def close(self): + aioble.stop() + self._connections = {} + + # -- connections, one wrapper per aioble connection + + def _wrap(self, aconn, role, device=None): + # Keep recent wrappers after a disconnect: captured writes that raced + # it still name their connection by this object. + if len(self._connections) > 8: + for key in list(self._connections): + if not key.is_connected(): + del self._connections[key] + conn = self._connections.get(aconn) + if conn is None: + if device is None: + device = Device(self, aconn.device.addr, aconn.device.addr_type, native=aconn.device) + conn = _MPConnection(self, device, role, aconn) + self._connections[aconn] = conn + return conn + + def _lookup(self, aconn): + if aconn is None: + return None + return self._connections.get(aconn) or self._wrap(aconn, "peripheral") + + # -- peripheral + + def register_services(self, *services): + aservices = [] + for service in self._services: + for characteristic in service.characteristics: + characteristic._impl = None + for service in services: + aservice = aioble.Service(_buuid(service.uuid)) + for c in service.characteristics: + f = c.flags + achar = aioble.BufferedCharacteristic( + aservice, + _buuid(c.uuid), + read=bool(f & FLAG_READ), + write=bool(f & 0x0008), + write_no_response=bool(f & 0x0004), + notify=bool(f & FLAG_NOTIFY), + indicate=bool(f & FLAG_INDICATE), + initial=c._initial, + max_len=c.max_len, + ) + c._impl = _MPServerCharacteristic(self, c, achar) + aservices.append(aservice) + aioble.register_services(*aservices) + self._services = tuple(services) + _captures.clear() + for service in services: + for c in service.characteristics: + if c.capture: + _captures[c._impl._achar._value_handle] = c._impl + + async def advertise( + self, + interval_us=100000, + *, + name=None, + services=None, + appearance=0, + manufacturer=None, + connectable=True, + timeout_ms=None, + ): + with _Translate("advertise"): + aconn = await aioble.advertise( + interval_us, + name=name.encode() if isinstance(name, str) else name, + services=[_buuid(s) for s in services] if services else None, + appearance=appearance, + manufacturer=manufacturer, + connectable=connectable, + timeout_ms=timeout_ms, + ) + if aconn is None: + # aioble swallows the cancellation; pass it on. + raise asyncio.CancelledError() + return self._wrap(aconn, "peripheral") + + # -- central + + def scan(self, duration_ms, *, active=False, interval_us=None, window_us=None): + # aioble's defaults (1.28 s interval, 11.25 ms window) listen less + # than 1 % of the time, and on the bench found one advertiser in six + # seconds where a full-duty scan found the board 40 times. Scans here + # are short and foreground, so listen continuously unless told not to. + if interval_us is None: + interval_us = _SCAN_US + if window_us is None: + window_us = interval_us + return _MPScanner(self, duration_ms, active, interval_us, window_us) + + async def _connect(self, device, timeout_ms, **options): + adevice = device.native + if adevice is None: + adevice = aioble.Device(device.addr_type, device.addr or device.address) + with _Translate("connect"): + aconn = await adevice.connect( + timeout_ms=timeout_ms, + min_conn_interval_us=options.get("min_conn_interval_us"), + max_conn_interval_us=options.get("max_conn_interval_us"), + ) + return self._wrap(aconn, "central", device) + + +class _MPScanner(Scanner): + async def __aenter__(self): + self._scan = aioble.scan( + self.duration_ms, + interval_us=self.interval_us, + window_us=self.window_us, + active=self.active, + ) + await self._scan.__aenter__() + return self + + async def __aexit__(self, exc_type, exc, tb): + await self._scan.__aexit__(exc_type, exc, tb) + + async def cancel(self): + await self._scan.cancel() + + def __aiter__(self): + return self + + async def __anext__(self): + r = await self._scan.__anext__() + name = r.name() + adevice = r.device + device = Device(self._ble, adevice.addr, adevice.addr_type, name=name, native=adevice) + return ScanResult( + device, + rssi=r.rssi, + name=name, + services=[_uuid(u) for u in r.services()], + manufacturer=list(r.manufacturer()), + connectable=r.connectable, + ) + + +class _MPConnection(Connection): + def __init__(self, ble, device, role, aconn): + Connection.__init__(self, ble, device, role) + self._aconn = aconn + + @property + def mtu(self): + return self._aconn.mtu or DEFAULT_MTU + + def is_connected(self): + return self._aconn.is_connected() + + async def disconnect(self, timeout_ms=2000): + with _Translate("disconnect"): + await self._aconn.disconnect(timeout_ms) + + async def disconnected(self, timeout_ms=None): + with _Translate("disconnected"): + await self._aconn.disconnected(timeout_ms) + + async def exchange_mtu(self, mtu=None, timeout_ms=1000): + with _Translate("exchange_mtu"): + return await self._aconn.exchange_mtu(mtu, timeout_ms) + + async def _discover_services(self, uuid, timeout_ms): + found = [] + with _Translate("service discovery"): + async for aservice in self._aconn.services(None if uuid is None else _buuid(uuid), timeout_ms): + found.append(_MPClientService(self, aservice)) + return found + + +class _MPClientService(ClientService): + def __init__(self, connection, aservice): + ClientService.__init__(self, connection, _uuid(aservice.uuid)) + self._aservice = aservice + + async def _discover_characteristics(self, uuid, timeout_ms): + found = [] + with _Translate("characteristic discovery"): + async for achar in self._aservice.characteristics(None if uuid is None else _buuid(uuid), timeout_ms): + found.append(_MPClientCharacteristic(self, achar)) + return found + + +class _MPClientCharacteristic(ClientCharacteristic): + def __init__(self, service, achar): + ClientCharacteristic.__init__(self, service, _uuid(achar.uuid), achar.properties) + self._achar = achar + limit = service.connection._ble.queue_limit + if achar.properties & FLAG_NOTIFY: + achar._notify_queue = _Queue(limit, "notify {}".format(self.uuid)) + if achar.properties & FLAG_INDICATE: + achar._indicate_queue = _Queue(limit, "indicate {}".format(self.uuid)) + # Register now, so notifications that arrive before the first + # notified() are queued rather than ignored. + achar._register_with_connection() + + async def read(self, timeout_ms=1000): + self._check(FLAG_READ, "read") + with _Translate("read"): + return bytes(await self._achar.read(timeout_ms)) + + async def write(self, data, response=None, timeout_ms=1000): + data = bytes(data) + response = self._want_response(response) + if not self.connection.is_connected(): + raise DisconnectedError("write: not connected") + if len(data) > self.connection.mtu - 3: + raise BLEError( + "write of {} bytes exceeds the ATT payload of {} (mtu {} - 3)".format( + len(data), self.connection.mtu - 3, self.connection.mtu + ) + ) + waited = 0 + limit = timeout_ms if timeout_ms is not None else _WRITE_RETRY_MS + while True: + try: + with _Translate("write"): + await self._achar.write(data, response, timeout_ms) + return + except OSError as e: + if _errno(e) not in _BUSY_ERRNOS: + if not self.connection.is_connected(): + raise DisconnectedError("write: disconnected") + raise + if not self.connection.is_connected(): + raise DisconnectedError("write: disconnected") + if waited >= limit: + raise BLETimeoutError("write: buffers stayed full for {} ms".format(waited)) + await sleep_ms(2) + waited += 2 + + async def subscribe(self, notify=True, indicate=False): + if notify: + self._check(FLAG_NOTIFY, "notify") + if indicate: + self._check(FLAG_INDICATE, "indicate") + with _Translate("subscribe"): + try: + await self._achar.subscribe(notify, indicate) + except ValueError as e: + raise UnsupportedError(str(e)) + + async def _next(self, queue, event, timeout_ms, what): + # Not aioble's notified(): it waits only when the queue holds <= 1 + # item and assumes the flag is then set, which holds for its one-slot + # queue but strands the last item of a longer one. Wait only when the + # queue is empty; the IRQ sets the flag on every empty -> non-empty. + aconn = self._achar._connection() + while True: + if len(queue): + return bytes(queue.popleft()) + if not aconn.is_connected(): + raise DisconnectedError("{}: disconnected".format(what)) + with _Translate(what): + with aconn.timeout(timeout_ms): + await event.wait() + + async def notified(self, timeout_ms=None): + self._check(FLAG_NOTIFY, "notify") + a = self._achar + return await self._next(a._notify_queue, a._notify_event, timeout_ms, "notified") + + async def indicated(self, timeout_ms=None): + self._check(FLAG_INDICATE, "indicate") + a = self._achar + return await self._next(a._indicate_queue, a._indicate_event, timeout_ms, "indicated") + + +class _MPServerCharacteristic: + """mpble's half of a :class:`bledev.Characteristic`.""" + + def __init__(self, ble, characteristic, achar): + self._ble = ble + self._char = characteristic + self._achar = achar + if characteristic.capture: + self._queue = [] + self._overrun = 0 + self._flag = asyncio.ThreadSafeFlag() + + def read(self): + return bytes(self._achar.read()) + + def write(self, data, send_update): + self._achar.write(data, send_update) + + def _aconn(self, connection): + aconn = getattr(connection, "_aconn", None) + if aconn is None: + raise BLEError("{!r} is not an mpble connection".format(connection)) + if not aconn.is_connected(): + raise DisconnectedError("the connection is closed") + return aconn + + def _check_size(self, connection, data): + if data is not None and len(data) > connection.mtu - 3: + raise BLEError( + "notification of {} bytes exceeds the ATT payload of {} (mtu {} - 3)".format( + len(data), connection.mtu - 3, connection.mtu + ) + ) + + def notify(self, connection, data): + aconn = self._aconn(connection) + self._check_size(connection, data) + try: + self._achar.notify(aconn, data) + except OSError as e: + if _errno(e) in _BUSY_ERRNOS: + raise BusyError("notify: transmit buffers full") + if not aconn.is_connected(): + raise DisconnectedError("notify: disconnected") + raise + + async def indicate(self, connection, data, timeout_ms): + aconn = self._aconn(connection) + self._check_size(connection, data) + with _Translate("indicate"): + await self._achar.indicate(aconn, data, timeout_ms) + + def _captured(self, aconn, data): + if len(self._queue) >= self._ble.queue_limit: + self._overrun += 1 + else: + self._queue.append((aconn, data)) + self._flag.set() + + async def written(self, timeout_ms): + if not self._char.capture: + with _Translate("written"): + aconn = await self._achar.written(timeout_ms) + return self._ble._lookup(aconn) + while True: + if self._overrun: + raise BLEError( + "writes to {} overran: {} dropped because the reader fell {} behind".format( + self._char.uuid, self._overrun, self._ble.queue_limit + ) + ) + if self._queue: + aconn, data = self._queue.pop(0) + return self._ble._lookup(aconn), data + if timeout_ms is not None and timeout_ms <= 0: + raise BLETimeoutError("written timed out") + try: + if timeout_ms is None: + await self._flag.wait() + else: + await asyncio.wait_for_ms(self._flag.wait(), timeout_ms) + except asyncio.TimeoutError: + if not self._queue: + raise BLETimeoutError("written timed out after {} ms".format(timeout_ms)) diff --git a/lib/bledev/nus.py b/lib/bledev/nus.py index 037ddbb6..1d6109b0 100644 --- a/lib/bledev/nus.py +++ b/lib/bledev/nus.py @@ -63,9 +63,9 @@ def _adapter(ble): if is_adapter(ble): return ble if hasattr(ble, "gap_advertise"): - from .mpble import MPBLE + from . import mpble - return MPBLE(ble) + return mpble.get(ble) raise TypeError("expected a bledev adapter, not {!r}".format(ble)) diff --git a/pydevices-desktop.toml b/pydevices-desktop.toml index 251a3539..1f3d50ac 100644 --- a/pydevices-desktop.toml +++ b/pydevices-desktop.toml @@ -23,6 +23,7 @@ "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/bledev/__init__.py" = "/lib/bledev/__init__.py" "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/bledev/auto.py" = "/lib/bledev/auto.py" "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/bledev/fake.py" = "/lib/bledev/fake.py" +"https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/bledev/mpble.py" = "/lib/bledev/mpble.py" "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/bledev/nus.py" = "/lib/bledev/nus.py" "https://raw.githubusercontent.com/PyDevices/pydevices/main/board_configs/desktop/board_config.py" = "/lib/board_config.py" "https://raw.githubusercontent.com/PyDevices/pydevices/main/board_configs/desktop/board_peripherals.py" = "/lib/board_peripherals.py" From 5ae0b83dbbd055d4350333fc03fed1444e03b5df Mon Sep 17 00:00:00 2001 From: Brad Barnett <127794626+bdbarnett@users.noreply.github.com> Date: Thu, 24 Sep 2026 11:38:04 -0500 Subject: [PATCH 3/6] bledev: board-to-board checks, and three mpble fixes they found tests/bledev_board holds the over-the-air checks run on two ESP32-S3s: gatt_peripheral/gatt_central (the contract over a real radio), gatt_latency (per-operation latency) and nus_server/nus_client (the byte-stream gate). Each has a PLANT switch whose run must fail. They found three things in mpble: aioble returns discovered services and characteristics in IRQ-race order (now sorted by handle), a BufferedCharacteristic loses its initial value when aioble sizes the buffer after writing it (now written again), and an operation on a dead connection raised TypeError from inside aioble (now DisconnectedError). --- lib/bledev/mpble.py | 21 ++++- tests/bledev_board/gatt_central.py | 131 ++++++++++++++++++++++++++ tests/bledev_board/gatt_latency.py | 78 +++++++++++++++ tests/bledev_board/gatt_peripheral.py | 111 ++++++++++++++++++++++ tests/bledev_board/nus_client.py | 117 +++++++++++++++++++++++ tests/bledev_board/nus_server.py | 94 ++++++++++++++++++ 6 files changed, 550 insertions(+), 2 deletions(-) create mode 100644 tests/bledev_board/gatt_central.py create mode 100644 tests/bledev_board/gatt_latency.py create mode 100644 tests/bledev_board/gatt_peripheral.py create mode 100644 tests/bledev_board/nus_client.py create mode 100644 tests/bledev_board/nus_server.py diff --git a/lib/bledev/mpble.py b/lib/bledev/mpble.py index 2c73270e..c9eca2ea 100644 --- a/lib/bledev/mpble.py +++ b/lib/bledev/mpble.py @@ -271,6 +271,12 @@ def register_services(self, *services): c._impl = _MPServerCharacteristic(self, c, achar) aservices.append(aservice) aioble.register_services(*aservices) + # aioble writes the initial value and then sizes the buffer, which + # empties it again; write it once more now the buffer is final. + for service in services: + for c in service.characteristics: + if c._initial is not None: + c._impl._achar.write(c._initial) self._services = tuple(services) _captures.clear() for service in services: @@ -394,6 +400,9 @@ async def _discover_services(self, uuid, timeout_ms): with _Translate("service discovery"): async for aservice in self._aconn.services(None if uuid is None else _buuid(uuid), timeout_ms): found.append(_MPClientService(self, aservice)) + # aioble yields results in whatever order its IRQs and its task + # interleave; the contract promises the server's order. + found.sort(key=lambda s: s._aservice._start_handle) return found @@ -407,6 +416,7 @@ async def _discover_characteristics(self, uuid, timeout_ms): with _Translate("characteristic discovery"): async for achar in self._aservice.characteristics(None if uuid is None else _buuid(uuid), timeout_ms): found.append(_MPClientCharacteristic(self, achar)) + found.sort(key=lambda c: c._achar._value_handle) return found @@ -423,16 +433,22 @@ def __init__(self, service, achar): # notified() are queued rather than ignored. achar._register_with_connection() + def _live(self, what): + # aioble hands a dead connection's None handle to the radio, which + # raises TypeError; say what actually happened. + if not self.connection.is_connected(): + raise DisconnectedError("{}: not connected".format(what)) + async def read(self, timeout_ms=1000): self._check(FLAG_READ, "read") + self._live("read") with _Translate("read"): return bytes(await self._achar.read(timeout_ms)) async def write(self, data, response=None, timeout_ms=1000): data = bytes(data) response = self._want_response(response) - if not self.connection.is_connected(): - raise DisconnectedError("write: not connected") + self._live("write") if len(data) > self.connection.mtu - 3: raise BLEError( "write of {} bytes exceeds the ATT payload of {} (mtu {} - 3)".format( @@ -463,6 +479,7 @@ async def subscribe(self, notify=True, indicate=False): self._check(FLAG_NOTIFY, "notify") if indicate: self._check(FLAG_INDICATE, "indicate") + self._live("subscribe") with _Translate("subscribe"): try: await self._achar.subscribe(notify, indicate) diff --git a/tests/bledev_board/gatt_central.py b/tests/bledev_board/gatt_central.py new file mode 100644 index 00000000..539859d8 --- /dev/null +++ b/tests/bledev_board/gatt_central.py @@ -0,0 +1,131 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""Board side of the over-the-air contract check: the central. + +Needs ``gatt_peripheral.py`` running on another board. Writes one line per +check to /gatt_central.log and ends with ``RESULT PASS`` or ``RESULT FAIL``. +""" + +import asyncio +import struct +import sys + +import bledev +import bledev.mpble + +LOG = "/gatt_central.log" +SVC = bledev.UUID("12345678-1234-5678-1234-56789abcdef0") +CHR_RW = bledev.UUID("12345678-1234-5678-1234-56789abcdef1") +CHR_STREAM = bledev.UUID("12345678-1234-5678-1234-56789abcdef2") +CHR_IND = bledev.UUID("12345678-1234-5678-1234-56789abcdef3") + +failures = [] + + +def log(*parts): + line = " ".join(str(p) for p in parts) + print(line) + with open(LOG, "a") as f: + f.write(line + "\n") + + +def check(ok, what, detail=""): + log("PASS" if ok else "FAIL", what, detail) + if not ok: + failures.append(what) + + +async def main(): + ble = bledev.mpble.get() + caps = ble.capabilities() + check(caps["central"] and caps["peripheral"], "capabilities: both roles", caps) + + device = await ble.find(service=SVC, timeout_ms=15000) + # The name may arrive in a later scan response than the service UUID, so + # a find by service can return before the name is known. + check(device is not None, "find by service", repr(device)) + connection = await device.connect(timeout_ms=10000) + check(connection.is_connected() and connection.role == "central", "connect") + + mtu = await connection.exchange_mtu(247) + check(mtu == 185, "exchange_mtu: the smaller side wins", mtu) + + check(await connection.service(bledev.UUID(0x180F)) is None, "missing service is None") + service = await connection.service(SVC) + uuids = [] + async for c in service.characteristics(): + uuids.append(c.uuid) + check(uuids == [CHR_RW, CHR_STREAM, CHR_IND], "characteristic discovery", uuids) + rw = await service.characteristic(CHR_RW) + stream = await service.characteristic(CHR_STREAM) + ind = await service.characteristic(CHR_IND) + + check(await rw.read() == b"hi", "read the initial value") + + await stream.subscribe(notify=True) + await rw.write(b"notify 200", response=True) + seqs = [] + for _ in range(200): + try: + data = await stream.notified(timeout_ms=3000) + except bledev.BLETimeoutError: + break + seqs.append(struct.unpack(" 100), + "over 500 ms", sum(1 for x in s if x > 500)) + + +async def main(): + ble = bledev.mpble.get() + device = await ble.find(service=SVC, timeout_ms=15000) + connection = await device.connect(timeout_ms=10000, **CONN) + log("connected", CONN, "mtu", await connection.exchange_mtu(185)) + rw = await (await connection.service(SVC)).characteristic(CHR_RW) + reads = [] + writes = [] + slow = [] + t_start = time.ticks_ms() + while time.ticks_diff(time.ticks_ms(), t_start) < DURATION_S * 1000: + t0 = time.ticks_ms() + await rw.read(timeout_ms=5000) + t1 = time.ticks_ms() + await rw.write(b"ping", response=True, timeout_ms=5000) + t2 = time.ticks_ms() + r, w = time.ticks_diff(t1, t0), time.ticks_diff(t2, t1) + reads.append(r) + writes.append(w) + for kind, ms, at in (("read", r, t0), ("write", w, t1)): + if ms > 500: + slow.append((kind, ms, time.ticks_diff(at, t_start))) + summary("read", reads) + summary("write", writes) + summary("all", reads + writes) + log("stalls over 500 ms (kind, ms, at ms):", slow) + await connection.disconnect() + log("RESULT done") + + +open(LOG, "w").close() +asyncio.run(main()) diff --git a/tests/bledev_board/gatt_peripheral.py b/tests/bledev_board/gatt_peripheral.py new file mode 100644 index 00000000..d4804e19 --- /dev/null +++ b/tests/bledev_board/gatt_peripheral.py @@ -0,0 +1,111 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""Board side of the over-the-air contract check: a GATT server driven by commands. + +Run it on one board (``mpftp run gatt_peripheral.py -d COMx``), then run +``gatt_central.py`` on another board, or point a host backend's central at it. +It advertises as ``bledev-radio`` with the contract's test service, and the +central steers it by writing commands to the read/write characteristic: + +``notify N`` send N notifications on the stream characteristic, each 20 bytes + starting with its sequence number +``count`` put " " (captured stream writes so far, and how many + were out of sequence) in the read/write characteristic +``indicate`` send an indication, b"ack me", on the indicate characteristic +``drop`` disconnect from this side +``ping`` nothing (the latency probe's write; not logged) + +It logs to /gatt_peripheral.log and serves connections until reset. +""" + +import asyncio +import struct + +import bledev +import bledev.mpble + +PLANT = False # the planted run skips notification 7, which the central must catch +LOG = "/gatt_peripheral.log" + +SVC = bledev.UUID("12345678-1234-5678-1234-56789abcdef0") +CHR_RW = bledev.UUID("12345678-1234-5678-1234-56789abcdef1") +CHR_STREAM = bledev.UUID("12345678-1234-5678-1234-56789abcdef2") +CHR_IND = bledev.UUID("12345678-1234-5678-1234-56789abcdef3") + + +def log(*parts): + line = " ".join(str(p) for p in parts) + print(line) + with open(LOG, "a") as f: + f.write(line + "\n") + + +async def notify_all(stream, connection, n): + for i in range(n): + if PLANT and i == 7: + continue + data = struct.pack("> 8) * 13 + seed) & 0xFF) for i in range(n)) + + +def compare(got, want): + bad = 0 + first = -1 + for i in range(min(len(got), len(want))): + if got[i] != want[i]: + bad += 1 + if first < 0: + first = i + return bad, first + + +def kbps(n, ms): + return "{:.1f} KB/s".format(n / 1024 / (ms / 1000)) if ms else "inf" + + +async def main(): + gc.collect() + ble = bledev.mpble.get() + t0 = time.ticks_ms() + link = await nus.connect(ble, name="bledev-gate", timeout_ms=20000, **CONN) + log("connected in", time.ticks_diff(time.ticks_ms(), t0), "ms, mtu", link.connection.mtu, "conn", CONN, "plant", PLANT) + ok = True + + # UP: we send, the server checks. + data = pattern(N, 1) + if PLANT: + data = bytearray(data) + data[5000] ^= 0x01 + data = bytes(data) + await link.write("UP {}\n".format(N).encode()) + t0 = time.ticks_ms() + await link.write(data) + sent_ms = time.ticks_diff(time.ticks_ms(), t0) + reply = (await link.readline()).decode().split() + ms = int(reply[2]) + log("UP", reply[1], N, "bytes: server received in", ms, "ms =", kbps(N, ms), "(our send", sent_ms, "ms) mismatches", reply[3], "first", reply[4]) + ok = ok and reply[1] == "OK" + + # DOWN: the server sends, we check. + await link.write("DOWN {}\n".format(N).encode()) + t0 = time.ticks_ms() + try: + got = await asyncio.wait_for_ms(link.readexactly(N), 15000) + except asyncio.TimeoutError: + log("DOWN stalled: buffered", len(link._buf), "bytes_in", link.bytes_in, "connected", link.connection.is_connected()) + raise + ms = time.ticks_diff(time.ticks_ms(), t0) + bad, first = compare(got, pattern(N, 2)) + verdict = "OK" if bad == 0 and len(got) == N else "BAD" + log("DOWN", verdict, len(got), "bytes in", ms, "ms =", kbps(N, ms), "mismatches", bad, "first", first) + ok = ok and verdict == "OK" + + # ECHO: round trip through the server. + await link.write("ECHO {}\n".format(N).encode()) + data = pattern(N, 3) + t0 = time.ticks_ms() + sender = asyncio.create_task(link.write(data)) + got = await link.readexactly(N) + ms = time.ticks_diff(time.ticks_ms(), t0) + await sender + bad, first = compare(got, data) + verdict = "OK" if bad == 0 and len(got) == N else "BAD" + log("ECHO", verdict, len(got), "bytes round trip in", ms, "ms =", kbps(N, ms), "each way; mismatches", bad, "first", first) + ok = ok and verdict == "OK" + + await link.write(b"BYE\n") + await asyncio.sleep_ms(300) + await link.close() + log("RESULT", "PASS" if ok else "FAIL") + + +async def guarded(): + try: + await main() + except BaseException as e: + import sys, io + buf = io.StringIO() + sys.print_exception(e, buf) + log("EXCEPTION", buf.getvalue()) + log("RESULT", "FAIL") + + +open(LOG, "w").close() +log("start") +asyncio.run(guarded()) diff --git a/tests/bledev_board/nus_server.py b/tests/bledev_board/nus_server.py new file mode 100644 index 00000000..b600821a --- /dev/null +++ b/tests/bledev_board/nus_server.py @@ -0,0 +1,94 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""The nus byte-stream gate, board side: serve Nordic UART as ``bledev-gate``. + +Run it on one board, then ``nus_client.py`` on another. The client drives three +phases over one link (UP: it sends, this side checks; DOWN: this side sends, +it checks; ECHO: this side sends back what it gets) and prints the verdict. +Set PLANT = True to flip one bit at offset 5000 of everything this side sends; +the client must then report FAIL. Logs to /gate_server.log. +""" +import asyncio, time +import bledev.mpble +import bledev.nus as nus + +PLANT = False # the planted run sets this: flip one bit at offset 5000 of what we send +LOG = "/gate_server.log" + + +def log(*parts): + line = " ".join(str(p) for p in parts) + print(line) + with open(LOG, "a") as f: + f.write(line + "\n") + + +def pattern(n, seed): + return bytes(((i * 7 + (i >> 8) * 13 + seed) & 0xFF) for i in range(n)) + + +def plant(data): + if PLANT and len(data) > 5000: + data = bytearray(data) + data[5000] ^= 0x01 + data = bytes(data) + return data + + +def compare(got, want): + bad = 0 + first = -1 + for i in range(min(len(got), len(want))): + if got[i] != want[i]: + bad += 1 + if first < 0: + first = i + return bad, first + + +async def main(): + open(LOG, "w").close() + ble = bledev.mpble.get() + log("serving as bledev-gate, plant", PLANT) + link = await nus.serve(ble, name="bledev-gate", timeout_ms=120000) + log("connected, mtu", link.connection.mtu) + while True: + cmd = (await link.readline()).decode().strip() + if not cmd or cmd == "BYE": + break + op, n = cmd.split() + n = int(n) + if op == "UP": + t0 = time.ticks_ms() + got = await link.readexactly(n) + ms = time.ticks_diff(time.ticks_ms(), t0) + bad, first = compare(got, pattern(n, 1)) + verdict = "OK" if bad == 0 and len(got) == n else "BAD" + log("UP", verdict, n, "bytes", ms, "ms mismatches", bad, "first", first, "mtu", link.connection.mtu) + await link.write("UP {} {} {} {}\n".format(verdict, ms, bad, first).encode()) + elif op == "DOWN": + data = plant(pattern(n, 2)) + t0 = time.ticks_ms() + await link.write(data) + log("DOWN sent", n, "bytes in", time.ticks_diff(time.ticks_ms(), t0), "ms") + elif op == "ECHO": + left = n + sent = 0 + while left: + chunk = await link.read(left) + if not chunk: + break + if PLANT and sent <= 5000 < sent + len(chunk): + chunk = bytearray(chunk) + chunk[5000 - sent] ^= 0x01 + chunk = bytes(chunk) + await link.write(chunk) + sent += len(chunk) + left -= len(chunk) + log("ECHO echoed", sent, "of", n) + log("done; bytes_in", link.bytes_in, "bytes_out", link.bytes_out) + await link.close() + + +asyncio.run(main()) From 2566ef5e44cd9c1fb3a63b3cb66b6b0d9e4f5516 Mon Sep 17 00:00:00 2001 From: Brad Barnett <127794626+bdbarnett@users.noreply.github.com> Date: Thu, 24 Sep 2026 11:42:00 -0500 Subject: [PATCH 4/6] mip-split.toml: bledev publishes to MIP on its own, requiring aioble Declares the packaging from docs/ble.md section 3: bledev leaves the pydevices MIP package so a board without a radio doesn't carry it, one mip.install("bledev") brings aioble, and on PyPI bledev stays in the pydevices wheel with bleak behind pydevices[ble]. The publisher learns these keys in PyDevices/.github (branch bledev-packaging); until that lands, the current publisher reads only host-only and ships bledev inside pydevices as before. --- mip-split.toml | 22 +++++++++++++++++++++- 1 file changed, 21 insertions(+), 1 deletion(-) diff --git a/mip-split.toml b/mip-split.toml index c739c576..80633a62 100644 --- a/mip-split.toml +++ b/mip-split.toml @@ -1,4 +1,6 @@ -# Which modules in lib/ cannot run on a microcontroller. +# Which modules in lib/ cannot run on a microcontroller, and which packages in +# lib/ publish to MIP on their own instead of inside `pydevices` (`own-package`, +# at the end: bledev). # # `displaydev`, `audiodev` and `multimer` each carry a full set of platform # backends and pick one at import time through their `auto` module. That design @@ -66,3 +68,21 @@ host-only = [ "wasm", "win32", ] + +[bledev] +# BLE is its own MIP package rather than part of `pydevices`, so a board +# without a radio doesn't carry it (docs/ble.md §3): +# mip.install("bledev", index="https://PyDevices.github.io/mip") +# `requires` is included into the package at index build time, which is how +# micropython-lib resolves require(): one install brings aioble with it. On a +# firmware that freezes aioble, the frozen copy wins and the installed one is +# never imported. +own-package = true +requires = ["aioble"] +# The central-only host backends go in pydevices-desktop, as for the other +# packages here. List `bleak` and `webble` when those modules land; a name +# listed before its file exists fails the sync. +host-only = [] +# On PyPI bledev stays inside the pydevices wheel. bleak is optional, behind +# `pip install "pydevices[ble]"`. +pypi-extras = { ble = ["bleak>=1.0"] } From f1ba26d524326e2e3b28e8188022f247ab8bf379 Mon Sep 17 00:00:00 2001 From: Brad Barnett <127794626+bdbarnett@users.noreply.github.com> Date: Thu, 24 Sep 2026 11:43:54 -0500 Subject: [PATCH 5/6] docs: bledev.md for apps, bledev-internals.md for backend authors bledev.md says what you can do (a nus stream first, then serving and connecting), where each backend runs, how to install and test, and the rules every backend keeps. bledev-internals.md is the next agents' brief: the hooks to fill in, what to translate, how mpble works around aioble, the checks to run, and the S3 measurements. --- AGENTS.md | 5 +- docs/README.md | 2 + docs/bledev-internals.md | 182 ++++++++++++++++++++++++++++++ docs/bledev.md | 231 +++++++++++++++++++++++++++++++++++++++ tests/README.md | 1 + 5 files changed, 420 insertions(+), 1 deletion(-) create mode 100644 docs/bledev-internals.md create mode 100644 docs/bledev.md diff --git a/AGENTS.md b/AGENTS.md index a9c257ed..523418ae 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -2,7 +2,7 @@ Canonical PyDevices product/source repository. Owns **board configs**, **hardware drivers**, portable libraries (`displaydev`, `audiodev`, -`appdev`, `multimer`, `events`, `keys`), and pip/MIP publishing. +`appdev`, `multimer`, `events`, `keys`, `bledev`), and pip/MIP publishing. Docs are markdown under `docs/`, published only via GitHub Pages ([docs/](docs/README.md); the Pages site is the landing page only) — not Read the Docs. Build locally with `./scripts/build_pages.sh` (needs @@ -46,6 +46,9 @@ Docs are markdown under `docs/`, published only via GitHub Pages Every library component in `utils/` is bundled automatically into `pydevices-desktop` without becoming a separate package. The desktop package depends on `pydevices`. + `bledev` is the one `lib/` package declared to publish to MIP on its own + (`mip-split.toml`, `own-package`), requiring aioble; `bledev.auto` is + never imported by `bledev/__init__.py` or a backend. `AutoDisplay` is `displaydev.auto` only — never re-exported from `displaydev/__init__.py`. Backends must not import `.auto`. Likewise, synchronous `Timer` providers are explicit `multimer` modules diff --git a/docs/README.md b/docs/README.md index e45c2acc..fce7fe38 100644 --- a/docs/README.md +++ b/docs/README.md @@ -21,6 +21,7 @@ - [appdev.md](appdev.md) — the event poller and device mux (`App`) - [multimer.md](multimer.md) — portable timers - [audio.md](audio.md) — `audiodev` interfaces +- [bledev.md](bledev.md) — Bluetooth Low Energy: one async API on boards, laptops and browsers - [app-and-board-config.md](app-and-board-config.md) — the application loop ## Drivers @@ -41,6 +42,7 @@ - [displaydev-internals.md](displaydev-internals.md) - [multimer-internals.md](multimer-internals.md) +- [bledev-internals.md](bledev-internals.md) — writing a bledev backend, and what was measured ## Maintainers diff --git a/docs/bledev-internals.md b/docs/bledev-internals.md new file mode 100644 index 00000000..ed5e72ca --- /dev/null +++ b/docs/bledev-internals.md @@ -0,0 +1,182 @@ +# bledev internals: writing a backend + +This page is for you if you're adding a backend (`bledev.bleak`, +`bledev.webble`, later `bledev.cpble`) or changing one. What an app can rely +on is in [bledev.md](bledev.md#the-rules-every-backend-keeps); this is how a +backend delivers it. + +## The layout + +`lib/bledev/` follows `audiodev` and `displaydev`: a portable base, one +standalone module per backend, and an optional selector. + +| Module | What it is | +|---|---| +| `__init__.py` | The contract: `UUID`, the errors, `FLAG_*`, advertising pack/unpack, and the base classes below. Imports no backend. | +| `fake.py` | In-process loopback. The reference implementation of every hook. | +| `mpble.py` | MicroPython, over aioble. | +| `nus.py` | Nordic UART byte stream, built only on the contract. | +| `auto.py` | Picks a backend. Nothing imports it, and backends must not. | + +A backend module imports `bledev` and whatever its host provides, and nothing +else in bledev except the base classes. `auto.BACKENDS` expects +`bledev.bleak.BleakBLE` and `bledev.webble.WebBLE`; if you name your class +differently, change that table. + +Anything under `lib/` must also be listed in `pydevices-desktop.toml` (a test +checks it). A central-only host module belongs in `mip-split.toml`'s +`[bledev] host-only` list once its file exists, so it doesn't ship to boards. + +## The hooks + +Subclass these and fill in the hooks. Everything else in the base classes is +built on them. + +**`BLE`**, the adapter: + +- `backend` (class attribute): the module's short name. +- `capabilities()`: start from `BLE.capabilities(self)` and set `central`, + `peripheral`, `mtu`, `max_mtu`, `indicate`, `pairing`, `l2cap`. +- `config(*names, **settings)`: at least `mtu`; raise `UnsupportedError` for + anything else you don't have. +- `close()`. +- Peripheral backends: `register_services(*services)`, which sets each + `Characteristic._impl` (see below), and + `advertise(interval_us, *, name, services, appearance, manufacturer, connectable, timeout_ms)`, + which returns your `Connection` with `role="peripheral"`. A central-only + backend leaves both alone; the base raises `UnsupportedError`. +- Central backends: `scan(duration_ms, *, active, interval_us, window_us)`, + returning a `Scanner`, and `_connect(device, timeout_ms, **options)`, + returning your `Connection` with `role="central"`. `find()` is already + written on top of `scan()`. + +**`Scanner`**: either override `__aenter__`/`__anext__`/`cancel()` (mpble +wraps aioble's scanner that way), or keep the base's queue and call +`_push(ScanResult(...))` and `_finish()` from the event loop, starting and +stopping your scan in `_start()`/`_stop()`. bleak's detection callback runs on +the event loop, so `_push` from it is fine. From another thread it isn't: the +base uses `asyncio.Event`. + +**`Device`**: construct it with your adapter, the address (six bytes, or the +host's string), and `native=` your own object, which `_connect()` gets back. + +**`Connection`**: `is_connected()`, `disconnect(timeout_ms)`, +`disconnected(timeout_ms)`, `exchange_mtu(mtu, timeout_ms)` (return the +current MTU if the host negotiates on its own), the `mtu` property (override +it, or keep `self._mtu` current), and `_discover_services(uuid, timeout_ms)`, +returning a list of your `ClientService` in handle order. + +**`ClientService`**: `_discover_characteristics(uuid, timeout_ms)`, a list of +your `ClientCharacteristic` in handle order. + +**`ClientCharacteristic`**: `read`, `write`, `subscribe`, `notified`, +`indicated`. `self._want_response(response)` resolves `response=None` and +checks the flags for you; `self._check(FLAG_..., what)` checks the rest. + +**The server side**: `Characteristic` is backend-neutral. Your +`register_services()` gives each one an `_impl` object with `read()`, +`write(data, send_update)`, `notify(connection, data)` (synchronous), +`indicate(connection, data, timeout_ms)` and `written(timeout_ms)`. The public +methods check the flags before they call it. + +## What a backend must translate + +- Every timeout to `BLETimeoutError`. `bledev.wait_ms(awaitable, timeout_ms)` + does it for you. +- A dead link to `DisconnectedError`, including an operation started after + the drop. aioble raised `TypeError` there; bleak raises its own errors. +- A full transmit buffer to `BusyError` from `notify()`, or a wait-and-retry + inside an async send. +- A payload over `mtu - 3` to `BLEError`, checked before it reaches the host + stack, because NimBLE and some OS stacks truncate silently. +- A queue overflow to `BLEError` from the next read of that queue, never a + silent drop. +- Discovery order to handle order. + +## How mpble gets there + +aioble is a good base with three places it loses data, and mpble works +around each rather than forking it: + +- **Notifications.** aioble keeps one notification per client characteristic + and replaces it when the next arrives. mpble swaps in a `_Queue` of + `queue_limit` entries. It also doesn't use aioble's `notified()`, which waits + only when the queue holds one item or fewer and assumes the flag is then + set: true for a one-slot queue, but with a longer one it strands the last + item of a burst. That hung the first board-to-board stream at its tail. +- **Captured writes.** aioble shares one ten-entry deque across every + characteristic and drops the oldest. mpble registers its own IRQ handler + beside aioble's, copies each captured write out of the value buffer in the + IRQ, and ties it to its connection while the connection still exists. +- **Scans.** aioble's defaults (1.28 s interval, 11.25 ms window) listen less + than 1 % of the time. On the bench a six-second scan saw one advertiser, + while a full-duty scan saw the board forty times, so mpble defaults to + 30 ms interval and window. + +Two more turned up on hardware: discovery results come back in IRQ-race +order, and a `BufferedCharacteristic` loses its initial value, because aioble +writes it and then sizes the buffer. mpble sorts the first and rewrites the +second. From reading aioble rather than from a failure: its `advertise()` +returns `None` when cancelled, and mpble turns that back into +`CancelledError`. + +aioble owns the one `bluetooth.BLE()` and its IRQ, so `mpble.get()` shares +one adapter, and the capture handler is registered once per boot. + +## The checks + +**Against the fake**, on both interpreters: `tests/bledev_contract.py`. +Its `pair()` and `nus_pair()` helpers build two fake adapters. A host backend +that has a fake peripheral to talk to can reuse most checks by building its +adapters there instead. `tests/test_bledev.py` also runs five planted faults +(a dropped, corrupted, reordered, truncated or overwritten packet), and each +must fail the contract; if you change the fake, keep that true. + +**Over a real radio**, `tests/bledev_board/`. Run the serving script with +`mpftp run` on one board and the other with `mpftp probe --capture` on a +second. `gatt_peripheral.py`, `nus_server.py` and `nus_client.py` each have a +`PLANT` switch, and a planted run must fail: + +| Scripts | What they check | +|---|---| +| `gatt_peripheral.py` + `gatt_central.py` | The contract over the air: find, MTU exchange, discovery order, read, 200 notifications in order, 100 captured writes in order, the MTU refusal, an indication, a disconnect from the far side | +| `gatt_peripheral.py` + `gatt_latency.py` | Per-operation latency of reads and writes-with-response, with the stall count | +| `nus_server.py` + `nus_client.py` | 16 KB up, down and echoed over nus, byte for byte, timed | + +`gatt_peripheral.py` is also the peripheral to point a host backend at: it +advertises as `bledev-radio`, and a central steers it by writing commands. + +## Measurements + +Two ESP32-S3 boards (the Waveshare ESP32-S3-Touch-LCD-7 serving, the LilyGo +T-Embed connecting), MicroPython 1.29, on one desk (RSSI about -50 dBm), Wi-Fi +off, 2026-09-24. + +**nus throughput**, 16 KB per phase, MTU 247, every byte checked: + +| Run | Up (writes) | Down (notifications) | Echo, each way | +|---|---|---|---| +| default interval | 81.2 KB/s | 35.9 KB/s | 25.1 KB/s | +| default interval | 86.5 KB/s | 40.7 KB/s | 23.6 KB/s | +| 7.5-15 ms interval | 69.0 KB/s | 38.6 KB/s | 43.5 KB/s | +| 7.5-15 ms interval | 78.8 KB/s | 37.0 KB/s | 44.0 KB/s | +| planted bit flip at offset 5000 | FAIL, 1 mismatch at 5000 | FAIL, same | FAIL, same | + +**GATT latency**, alternating a read and a write-with-response: + +| Link | Operations | Median | p99 | Max | Over 500 ms | +|---|---|---|---|---|---| +| S3 to S3, default interval, 60 s | 598 | 100 ms | 107 ms | 151 ms | 0 | +| S3 to S3, 7.5-15 ms, 120 s | 3966 | 30 ms | 40 ms | 80 ms | 0 | +| Laptop (bleak, Windows) to S3, 120 s | 1004 | 120 ms | 183 ms | 241 ms | 0 | + +**Memory**, `gc.mem_free()` and `esp32.idf_heap_info()` after a hard reset: + +| Step | Internal RAM | Python heap | +|---|---|---| +| `bluetooth.BLE().active(True)` | 63.9 KB (both boards); the largest free block drops to 31.7 KB | 0.7 KB | +| `import aioble` (.mpy) | none | 18.5 KB | +| `import bledev, bledev.mpble, bledev.nus` as .mpy (what mip installs) | none | 31.6 KB | +| the same, as .py source | none | 98.8 KB | +| registering nus, then advertising | 0.2 KB | 1.8 KB | +| `active(False)` | all returned | | diff --git a/docs/bledev.md b/docs/bledev.md new file mode 100644 index 00000000..57154330 --- /dev/null +++ b/docs/bledev.md @@ -0,0 +1,231 @@ +# bledev: Bluetooth Low Energy + +You write one BLE program and it runs on a board, on a laptop, in a browser, or +against a pretend radio in your tests. `bledev` is async throughout and shaped +like [aioble](https://github.com/micropython/micropython-lib/tree/master/micropython/bluetooth/aioble), +so if you know aioble you already know most of it. + +The quickest useful thing is a byte stream between a board and anything else: + +```python +import asyncio +import bledev.mpble +import bledev.nus as nus + +async def main(): + ble = bledev.mpble.get() + link = await nus.serve(ble, name="rack") # the board advertises; you connect + await link.write(b"hello\n") + async for line in link: # lines until the other side hangs up + await link.write(b"you said " + line) + +asyncio.run(main()) +``` + +The other side runs the same code with `nus.connect(ble, name="rack")`. That's +the Nordic UART service, so nRF Connect on a phone and Workbench's Bluetooth +transport can talk to it too. + +## What runs where + +| Backend | Host | Central (scan, connect) | Peripheral (advertise, serve) | +|---|---|---|---| +| `bledev.mpble` | MicroPython with `bluetooth` (ESP32, S3, P4 + C6, Pico W) | yes | yes | +| `bledev.bleak` | CPython on Windows, Linux, macOS | yes | no | +| `bledev.webble` | Browsers: PyScript, the Workbench simulator | yes | no | +| `bledev.fake` | Anywhere: tests | yes | yes | + +`bledev.bleak` and `bledev.webble` are being written; the table is the plan. + +Roles aren't symmetric. A laptop or a browser can only scan and connect, so +when a board talks to one, the board advertises. Only board-to-board links put +a board in the central role. `ble.capabilities()` tells you which roles you +have, and anything a backend can't do raises `bledev.UnsupportedError` rather +than doing nothing. + +## Installing it + +On a board, once `bledev` is published to the PyDevices index: + +```python +import mip +mip.install("bledev", index="https://PyDevices.github.io/mip") +``` + +That one install brings aioble with it; our firmware has `bluetooth` but +doesn't freeze aioble. Until then, `mip.install("aioble", index=...)` and copy +`lib/bledev/` to `/lib/bledev/` yourself. + +On a laptop, bledev is inside the `pydevices` wheel. `pip install "pydevices[ble]"` +adds bleak, which the laptop backend needs. + +## Getting an adapter + +The adapter is one radio. On a board: + +```python +import bledev.mpble +ble = bledev.mpble.get() # the board has one radio, so get() shares one adapter +``` + +Anywhere, if you don't want to name the backend: + +```python +import bledev.auto +ble = bledev.auto.adapter() # mpble, webble or bleak, whichever this host has +``` + +`bledev.auto` is optional, and nothing else in bledev imports it. Set +`BLEDEV_BACKEND=fake` to force the fake one. + +## Serving a service (peripheral) + +Describe the service, register it, then advertise. `advertise()` returns when +a central connects: + +```python +import bledev + +SENSOR = bledev.UUID("12345678-1234-5678-1234-56789abcdef0") + +service = bledev.Service(SENSOR) +level = bledev.Characteristic(service, 0x2A19, read=True, notify=True, initial=b"\x64") +command = bledev.Characteristic(service, "12345678-1234-5678-1234-56789abcdef1", + write=True, capture=True, max_len=64) +ble.register_services(service) + +connection = await ble.advertise(name="sensor", services=[SENSOR]) +level.notify(connection, b"\x63") +peer, data = await command.written() # capture=True: every write, in order +``` + +`register_services()` replaces whatever was registered before, so register +everything at once, before you advertise. Call `advertise()` again to accept +another connection. + +## Connecting to one (central) + +```python +device = await ble.find(name="sensor") # or find(service=SENSOR) +async with await device.connect() as connection: + service = await connection.service(SENSOR) + level = await service.characteristic(0x2A19) + print(await level.read()) + await level.subscribe() + while True: + print(await level.notified()) +``` + +`find()` scans actively, so it sees a name that only fits in the scan +response. For more control, scan yourself: + +```python +async with ble.scan(5000, active=True) as scanner: + async for result in scanner: + print(result.device, result.rssi, result.name(), result.services()) +``` + +A device can come back more than once as its advertisement and scan response +arrive, and `device.name` is whatever the scan had seen by then. + +## Byte streams with nus + +`bledev.nus` gives you a `Link` from either end, and it reads like an asyncio +stream: `read(n)`, `readexactly(n)`, `readline()`, `write(data)`, `close()`, +and `async for line in link`. `write()` splits your data to the link's MTU and +handles a full transmit buffer for you. `read()` returns `b""` at end of +stream, and only after everything that arrived before the disconnect has been +read. + +`nus.serve()` registers the Nordic UART service on the adapter the first time +you call it, so don't register other services on that adapter afterwards. It +can serve several links at once; each gets only its own central's bytes. + +On a board, `nus.connect(ble, name=..., min_conn_interval_us=7500, +max_conn_interval_us=15000)` asks for a short connection interval, which +nearly doubles a round trip's throughput (the numbers are in +[the measurements](#measured-on-two-esp32-s3s)). + +## The rules every backend keeps + +These hold on every backend. The fake enforces the strict version of each, so +code that passes its tests keeps them: + +- **One packet is `connection.mtu - 3` bytes.** A notification or write + longer than that raises `BLEError`. Real NimBLE would cut the tail off and + say nothing. `exchange_mtu()` asks for more, and the smaller side wins. +- **A characteristic holds `max_len` bytes**, 20 by default, as on + MicroPython. Raise it for anything that takes MTU-sized writes. The fake + raises on a longer write; a real board keeps the first `max_len` bytes and + says nothing (measured: 29 bytes sent, 20 kept). +- **Nothing is dropped quietly.** Notifications and captured writes queue in + order. If a queue overflows because you stopped reading, the next + `notified()` or `written()` raises `BLEError` instead. +- **`notify()` can say `BusyError`.** It's synchronous, like aioble's, and + the transmit buffers are finite. Wait a moment and send again, or use nus, + which does. Every async send waits for room itself. +- **Subscribe before you expect notifications.** Browsers and bleak deliver + nothing until you do, and the fake matches them. +- **Discovery comes back in the server's order** (by handle). +- **A dropped link ends pending client calls with `DisconnectedError`,** and + anything that arrived before the drop is still delivered first. A server's + `written()` serves every connection, so it doesn't end on a disconnect; give + it a `timeout_ms` or cancel it. With `capture=True`, `written(timeout_ms=0)` + returns a write that's already queued, or raises `BLETimeoutError` at once. +- **Every timeout is `BLETimeoutError`,** and every error is a `BLEError`: + `UnsupportedError`, `DisconnectedError`, `BLETimeoutError`, `BusyError`, + `GattError` (with `.status`). +- **`UUID` is one value everywhere.** `UUID(0x180F)`, `UUID("180f")` and the + 128-bit string are equal. `str()` is always the 128-bit form, which bleak + and browsers use, and `to_bytes()` is the little-endian wire form. + +## Testing without a radio + +`bledev.fake` is an in-process loopback that runs on CPython and MicroPython. +Adapters on the same `Air` see each other: + +```python +from bledev.fake import Air, FakeBLE + +air = Air() +board = FakeBLE(air=air) # either role, like a board +laptop = FakeBLE(air=air, peripheral=False) # central only, like bleak +``` + +It's stricter than a radio on purpose (see the rules above), and it models +what trips people up on hardware: a name that only reaches an active scan, a +transmit buffer that fills (`notify_buffers`), and a reader that falls behind +(`queue_limit`). + +The contract's own checks are `tests/bledev_contract.py`, a plain script so it +runs on both interpreters. `tests/test_bledev.py` runs it on CPython and on a +unix `micropython`, and then once per planted fault, each of which must make +it fail: + +```bash +python tests/bledev_contract.py +micropython tests/bledev_contract.py +python -m unittest tests.test_bledev +``` + +## Measured on two ESP32-S3s + +Board to board over nus (16 KB each way, verified byte for byte), with Wi-Fi off: + +| Direction | Default interval | 7.5-15 ms interval | +|---|---|---| +| Central to peripheral (writes) | 80-87 KB/s | 69-79 KB/s | +| Peripheral to central (notifications) | 36-41 KB/s | 37-39 KB/s | +| Echo round trip, each way | 24-25 KB/s | 44 KB/s | + +A single GATT read or write takes two connection intervals: a median of 100 ms +at the default interval, and 30 ms at 7.5-15 ms, with nothing over 100 ms in +two minutes. Turning the radio on costs about 64 KB of the S3's internal RAM; +aioble and bledev together cost about 50 KB of Python heap. The details, and +how these were measured, are in [bledev-internals.md](bledev-internals.md#measurements). + +## Writing a backend + +A backend subclasses the classes in `bledev/__init__.py` and fills in a few +hooks. What each hook must do, how the fake and mpble do it, and the checks +to run are in [bledev-internals.md](bledev-internals.md). diff --git a/tests/README.md b/tests/README.md index 612ae98a..026f5056 100644 --- a/tests/README.md +++ b/tests/README.md @@ -36,3 +36,4 @@ Do not run audiodev tests while another process is playing audio — see | `test_mip_portable.py` | portable `utils/mip.py` | | `test_audiodev*.py`, `test_*_audio.py`, `test_auto.py` | audiodev (see audio README) | | `test_portability.py` / `test_contract_proof.py` | portable-module constraints | +| `test_bledev.py` | runs `bledev_contract.py` on CPython and unix MicroPython, and proves five planted faults fail it (see [docs/bledev-internals.md](../docs/bledev-internals.md#the-checks)) | From 1497d869c0f039f9eab800079b7dd49abcaab905 Mon Sep 17 00:00:00 2001 From: Brad Barnett <127794626+bdbarnett@users.noreply.github.com> Date: Thu, 24 Sep 2026 11:49:00 -0500 Subject: [PATCH 6/6] docs/bledev: add the final-code gate run to the measurements --- docs/bledev-internals.md | 1 + docs/bledev.md | 4 ++-- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/docs/bledev-internals.md b/docs/bledev-internals.md index ed5e72ca..0f206880 100644 --- a/docs/bledev-internals.md +++ b/docs/bledev-internals.md @@ -158,6 +158,7 @@ off, 2026-09-24. |---|---|---|---| | default interval | 81.2 KB/s | 35.9 KB/s | 25.1 KB/s | | default interval | 86.5 KB/s | 40.7 KB/s | 23.6 KB/s | +| default interval, final code | 87.0 KB/s | 31.4 KB/s | 29.7 KB/s | | 7.5-15 ms interval | 69.0 KB/s | 38.6 KB/s | 43.5 KB/s | | 7.5-15 ms interval | 78.8 KB/s | 37.0 KB/s | 44.0 KB/s | | planted bit flip at offset 5000 | FAIL, 1 mismatch at 5000 | FAIL, same | FAIL, same | diff --git a/docs/bledev.md b/docs/bledev.md index 57154330..f0053c16 100644 --- a/docs/bledev.md +++ b/docs/bledev.md @@ -215,8 +215,8 @@ Board to board over nus (16 KB each way, verified byte for byte), with Wi-Fi off | Direction | Default interval | 7.5-15 ms interval | |---|---|---| | Central to peripheral (writes) | 80-87 KB/s | 69-79 KB/s | -| Peripheral to central (notifications) | 36-41 KB/s | 37-39 KB/s | -| Echo round trip, each way | 24-25 KB/s | 44 KB/s | +| Peripheral to central (notifications) | 31-41 KB/s | 37-39 KB/s | +| Echo round trip, each way | 24-30 KB/s | 44 KB/s | A single GATT read or write takes two connection intervals: a median of 100 ms at the default interval, and 30 ms at 7.5-15 ms, with nothing over 100 ms in