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..0f206880 --- /dev/null +++ b/docs/bledev-internals.md @@ -0,0 +1,183 @@ +# 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 | +| 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 | + +**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..f0053c16 --- /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) | 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 +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/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/mpble.py b/lib/bledev/mpble.py new file mode 100644 index 00000000..c9eca2ea --- /dev/null +++ b/lib/bledev/mpble.py @@ -0,0 +1,598 @@ +"""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) + # 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: + 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)) + # 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 + + +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)) + found.sort(key=lambda c: c._achar._value_handle) + 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() + + 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) + self._live("write") + 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") + self._live("subscribe") + 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 new file mode 100644 index 00000000..1d6109b0 --- /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 . import mpble + + return mpble.get(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/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"] } diff --git a/pydevices-desktop.toml b/pydevices-desktop.toml index 9345d4b6..1f3d50ac 100644 --- a/pydevices-desktop.toml +++ b/pydevices-desktop.toml @@ -20,6 +20,11 @@ "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/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" "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/boarddev.py" = "/lib/boarddev.py" 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)) | 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()) 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()