diff --git a/docs/bledev-internals.md b/docs/bledev-internals.md index c80723e5..5fe71afe 100644 --- a/docs/bledev-internals.md +++ b/docs/bledev-internals.md @@ -19,6 +19,8 @@ standalone module per backend, and an optional selector. | `nus.py` | Nordic UART byte stream, built only on the contract. | | `repl.py` | The REPL over nus: MicroPython's raw `bluetooth` API from the IRQ on the board, `nus` on the client. | | `improv.py` | Improv Wi-Fi setup, both sides, built only on the contract. | +| `hidreport.py` | HID report descriptors and reports to `events`. No Bluetooth in it. | +| `hid.py` | HID over GATT, host and device, built on the contract and `hidreport`. | | `auto.py` | Picks a backend. Nothing imports it, and backends must not. | A backend module imports `bledev` and whatever its host provides, and nothing @@ -72,6 +74,13 @@ returning a list of your `ClientService` in handle order. **`ClientService`**: `_discover_characteristics(uuid, timeout_ms)`, a list of your `ClientCharacteristic` in handle order. +**Pairing and descriptors** (for HID): `Connection.encrypted` and +`Connection.pair(bond, timeout_ms)`; `ClientCharacteristic._discover_descriptors(uuid, timeout_ms)` +returning your `ClientDescriptor`s, whose `read()` you fill in; and on the +server, each `Characteristic`'s `descriptors` (fixed values) and `encrypted` +flag. Report `long_read` in `capabilities()`: False if a read returns only +one packet. A backend without pairing leaves the base's `UnsupportedError`. + **`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. @@ -144,6 +153,18 @@ The laptop found two more (2026-09-24): case never showed it because the central there asks for the MTU after connecting. +The bless probe found two more on the P4 (2026-09-24), with the board as the +central and Windows as the peripheral: + +- **The same early MTU, the other way round.** Windows exchanges the MTU the + moment the board connects, before aioble has recorded the connection, so + the board believed 23 on a 256 link. mpble's IRQ now keeps that value too, + and the connection's `mtu` picks it up. +- **`exchange_mtu()` after the peer already exchanged** raised `EALREADY`. + NimBLE exchanges once per link, so mpble now returns the settled MTU. + `bledev.hid`'s host exchanges before reading the Report Map, so any + peripheral that exchanges first (a real keyboard may) would have failed. + **Code that uses the raw API.** A board config's `ble` is now an `MPBLE`, and `MPBLE.__getattr__` hands every name it doesn't define to the raw `bluetooth.BLE`, so `ble.active()` or `ble.gap_advertise()` still work. @@ -209,6 +230,66 @@ without using up the attempt. The ESP32 has one dupterm slot. The REPL takes it on login and gives back whatever held it (WebREPL, say) on disconnect. +## How HID gets there + +`bledev.hid` is two halves over one parser. `bledev.hidreport` reads a report +descriptor into `Report`s of `Field`s (bit offset, size, count, flags, +usages, logical range) and a `Decoder` diffs successive input reports into +events. It imports `events` and `keys` and nothing from Bluetooth. + +**The keyboard decoder is usbif's, event for event.** Same order (modifier +releases, key releases, key presses, modifier presses, the new modifier mask +on all of them), same fields (`scancode` is the HID usage, `window` is +`None`), same rollover rule. `tests/test_bledev_hid.py` feeds 3,000 random +boot reports through both decoders when usbif's checkout is beside this one, +and they must agree. It found one difference on its first run: the ISO "# ~" +key (usage 0x32), which SDL calls `#` and usbif leaves unmapped. hidreport +now leaves it unmapped too; the two tables should become one. + +**Axes, hats and buttons** come only from joystick, gamepad and multi-axis +collections, so a mouse's X, Y and buttons produce nothing yet. Axes are +numbered by usage (desktop X to wheel, then the simulation page), so an Xbox +controller's sticks are 0 to 3 and its triggers 4 and 5. Every axis starts +at 0.0 and moves only when its value changes, as SDL's do. + +**Descriptors, encryption and pairing** were added to the contract for HID. +A Report characteristic says which report it carries in its Report Reference +descriptor (0x2908), and HOGP gives the report ID nowhere else, so the host +needs `ClientCharacteristic.descriptor()`. A `Characteristic` can be +`encrypted=True`, and `Connection.pair()` encrypts a link. The host pairs +only after a read fails with insufficient encryption or authentication, +which is how a real keyboard asks. mpble pairs through aioble's security +module ("just works", `io=3`); `ble.enable_bonding()` loads aioble's +`ble_secrets.json` and turns bonding on, and it has to run on both sides, +because the side that didn't start pairing stores keys too. + +**MicroPython has no long read.** `gattc_read()` is one ATT Read, so a board +central gets at most `mtu - 1` bytes of any value. The host exchanges the MTU +up to 247 before reading the Report Map (246 bytes fit), and a map that fails +to parse at that length raises an error that says why. Our own map is 162 +bytes; an Xbox Series controller's is 283, so a board can't host one until +MicroPython grows a long read. bleak reads long values itself, and the +fake's `long_reads=False` models the board. + +**The peripheral's identity comes from its functions.** Name, appearance and +PnP product ID are derived from which of keyboard, consumer control and +gamepad it serves, so a host that cached one set never sees another set +under the same identity. The PnP vendor is the Bluetooth SIG's test company +ID (0xFFFF), because we have no vendor ID of our own. + +**A laptop can't see a HID service.** Windows hides 0x1812 from apps: bleak +lists the board's GAP, GATT, Battery and Device Information services and a +22-handle gap where the HID service sits (measured against the P4, +2026-09-24). Chrome's Web Bluetooth blocklist hides it too. So +`Peripheral(service_uuid=hid.INSPECT_SERVICE)` serves the same +characteristics under a vendor UUID, and `Host(service_uuid=...)` reads them +there; that is how the laptop checks a board's HID side, with the same code +a board host runs. No host treats a board serving that UUID as a keyboard. + +The host writes the keyboard's LED report once it has subscribed, as Windows +does. `Peripheral.leds()` returns those writes, which is how the gate knows +the host is listening, and how it times a round trip. + ## The checks **Against the fake**, on both interpreters: `tests/bledev_contract.py`. @@ -236,6 +317,8 @@ The serving scripts have a `PLANT` switch, and a planted run must fail: | `repl_server.py` + `repl_client.py` | The REPL: the right password evaluates `123 * 456`, a wrong one sent with code that would create `/pwned` is refused and hung up on and the code never runs, and Ctrl-C stops `while True`. `repl_server.py` runs from `/main.py`, because mpftp soft-resets the board, which turns Bluetooth off | | `improv_server.py` + `improv_client.py` | Improv: `--wrong` must get "unable to connect" and a return to "authorized"; without it, on a board, the network comes from that board's own `secrets.py` and must end "provisioned" with a URL | | `coex_server.py` + `tcp_pull.py` + `nus_client.py` | Wi-Fi and BLE on one S3: a TCP source on Wi-Fi beside the nus gate | +| `hid_peripheral.py` + `hid_central.py` | HID: the device types `hid_script.py`'s text, chords, media keys and gamepad moves; the host's events must equal `hid_script.expected()` and spell the text. Then key-press round trips. `PLANT` drops one Shift. The host also runs on a laptop, where it refuses to pair | +| `bless_peripheral.py` + `bless_probe.py` | The laptop as a peripheral, through bless, probed by a board | `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. @@ -276,6 +359,42 @@ off, 2026-09-24. | registering nus, then advertising | 0.2 KB | 1.8 KB | | `active(False)` | all returned | | +### HID, board to board and from the laptop + +`hid_peripheral.py` typed `hid_script.py` (30 characters with Shift, Ctrl+C, +Right Alt + Shift + F5, two media keys, five gamepad reports: 112 events) and +`hid_central.py` compared every event and the text they spell. MicroPython +1.29, aioble from mip, 2026-09-24. The planted run drops Shift from the first +character. + +| Device | Host | Plain | Planted | 7.5-15 ms interval | Round trip, default (median / max) | Round trip, 7.5-15 ms | +|---|---|---|---|---|---|---| +| LCD-7 (S3) | T-Embed (S3) | PASS, 112/112 | FAIL, event 0 | PASS | 69 / 119 ms | 29 / 49 ms | +| T-Embed (S3) | LCD-7 (S3) | PASS | FAIL, event 0 | PASS | 68 / 118 ms | 30 / 69 ms | +| P4 | LCD-7 (S3) | PASS | FAIL, event 0 | PASS | 70 / 220 ms | 30 / 80 ms | +| LCD-7 (S3) | P4 | PASS 2 of 5 | FAIL, event 0 | FAIL 6 of 6 | 68 / 178 ms | 29 / 69 ms | +| P4, vendor UUID | laptop (bleak) | PASS | FAIL, event 0 | PASS | 90 / 270 ms | 60 / 90 ms | + +The round trip is a key press on the device to the host's LED write arriving +back, so one way is roughly half: 35 ms at the default interval, 15 ms at +7.5-15 ms. Decoding on the host adds 1.7 ms (P4) to 2.8 ms (S3) from the +report's arrival to the app's `events()`, with a p99 of 10 to 23 ms. + +**The P4 as host loses reports** below Python, always from around the 26th +report on (pydevices#87). The S3 never did. `burst_source.py` and +`burst_sink.py` reproduce it without HID. + +**Pairing and bonding** (`ENCRYPTED = True`: every HID characteristic +encrypted, both sides call `enable_bonding()`), each direction between the S3s +and with the P4 as device: + +| Run | Result | +|---|---| +| First connect, no keys on either side | The Report Map read fails, the host pairs "just works" (encrypted, not authenticated, 16-byte key, bonded), 112/112 | +| Both boards reset, then reconnect | Encryption from the stored keys: the key store's checksum unchanged, connect-to-started 0.8 to 1.4 s faster, 112/112 | +| The host's `ble_secrets.json` deleted (as an erase would) | The device accepts a fresh pairing (NimBLE's repeat-pairing path drops the old bond), new keys, 112/112 | +| A host that refuses to pair (planted) | Fails: the device's encrypted Report Map can't be read | + ### The laptop to an S3 Windows 11, bleak 3.0.2, the laptop's Intel radio, to the LCD-7 running diff --git a/docs/bledev.md b/docs/bledev.md index daa301f4..40404409 100644 --- a/docs/bledev.md +++ b/docs/bledev.md @@ -223,6 +223,62 @@ boot is up to you: pass `on_join=lambda ssid, password, ip: ...` to `serve()`. (from a button, say) before it accepts credentials. The credentials cross the air unencrypted; that's the standard. +## Keyboards and gamepads (HID) + +`bledev.hid` turns a BLE keyboard, media remote or game controller into the +same `events` a USB keyboard gives you through usbif, or SDL gives you on the +desktop: `events.Key` with `keys.K_*` codes and modifiers, and +`events.JoyAxisMotion`, `JoyHatMotion`, `JoyButtonDown` and `JoyButtonUp`. +Your app can't tell which one it's talking to. + +```python +import bledev.hid as hid + +host = await hid.connect(ble, name="Keyboard K380") # or no name: any HID device +while True: + for event in await host.events(): # waits for at least one + print(event) +``` + +`host.poll()` returns whatever has arrived without waiting, the way usbif's +`poll()` does, so it fits a frame loop. Media keys arrive as `Key` events +too (`keys.K_VOLUMEUP`, `K_AUDIOPLAY`). Axes run from -1.0 to 1.0 and are +numbered X, Y, Z, Rx, Ry, Rz, then triggers; buttons count from 0; a hat is +an `(x, y)` tuple with y up. + +A keyboard that wants an encrypted link gets one: the first read it refuses +pairs the link ("just works", no passkey) and tries again. On a board, call +`ble.enable_bonding()` once before connecting so the keys are kept, and the +next connect doesn't pair from scratch. The keys live in `ble_secrets.json` +on the board's filesystem, so erasing the board forgets them, and a keyboard +that remembers the board may then need putting back into pairing mode. + +**A board can be the keyboard, too.** It advertises as a keyboard, media +remote and gamepad (or any one or two of them) and sends what you tell it: + +```python +kb = hid.Peripheral(ble) # keyboard=True, consumer=True, gamepad=True +await kb.serve() # returns when a host connects +await kb.keyboard.type("Hello\n") # US layout +await kb.keyboard.tap(0x06, hid.MOD_LCTRL) # Ctrl+C +await kb.consumer.tap(hid.VOLUME_UP) +await kb.gamepad.send(axes=(0, 127, 0, 0), buttons=0b101, hat=2) +``` + +Its name, appearance and Report Map come from which functions you ask for, +the same way every time, because hosts cache a device against its identity. +Pass `encrypted=True` to make hosts pair first, as real keyboards do. + +**Don't pair a test board with the computer you're testing from.** Once +paired, Windows and phones use it as a real keyboard, and it types into +them. + +The parser underneath, `bledev.hidreport`, knows nothing about Bluetooth. +Give it any HID report descriptor and it decodes that device's reports, so a +USB host can use it the same way. How it numbers axes, what it does with a +rollover, and what a board central can't read are in +[bledev-internals.md](bledev-internals.md#how-hid-gets-there). + ## The rules every backend keeps These hold on every backend. The fake enforces the strict version of each, so diff --git a/lib/bledev/__init__.py b/lib/bledev/__init__.py index 49d81a62..a925d50b 100644 --- a/lib/bledev/__init__.py +++ b/lib/bledev/__init__.py @@ -76,6 +76,11 @@ class BusyError(BLEError): """ +#: ATT error codes a :class:`GattError` can carry that apps act on. +INSUFFICIENT_AUTHENTICATION = 0x05 +INSUFFICIENT_ENCRYPTION = 0x0F + + class GattError(BLEError): """The peer answered a GATT request with an error ``status``.""" @@ -353,7 +358,8 @@ def capabilities(self): ``backend`` (its module name), ``central`` and ``peripheral`` (which roles it can take), ``mtu`` (the ATT MTU it asks for), ``max_mtu``, - ``indicate``, ``pairing`` and ``l2cap`` (bools). + ``indicate``, ``pairing`` and ``l2cap`` (bools), and ``long_read``: + whether a read returns values longer than one packet (``mtu - 1``). """ return { "backend": self.backend, @@ -364,6 +370,7 @@ def capabilities(self): "indicate": False, "pairing": False, "l2cap": False, + "long_read": True, } def config(self, *names, **settings): @@ -468,7 +475,8 @@ class Characteristic: write; MicroPython's default buffer is 20 bytes, so raise it for anything that takes whole MTU-sized writes. With ``capture=True``, ``written()`` returns every write as ``(connection, data)``, in order, instead of just - the last writer. + the last writer. ``encrypted=True`` refuses a central that hasn't paired + (``GattError`` with status ``INSUFFICIENT_ENCRYPTION``) until it does. """ def __init__( @@ -483,6 +491,7 @@ def __init__( initial=None, capture=False, max_len=20, + encrypted=False, ): service.characteristics.append(self) self.service = service @@ -502,6 +511,11 @@ def __init__( self.capture = bool(capture) and bool(flags & (FLAG_WRITE | FLAG_WRITE_NO_RESPONSE)) self.max_len = max(max_len, len(initial) if initial else 0) self._initial = bytes(initial) if initial is not None else None + #: True when a central must pair (an encrypted link) before it may + #: read, write or subscribe. HID keyboards serve their reports this way. + self.encrypted = bool(encrypted) + #: :class:`Descriptor` objects, in the order they were made. + self.descriptors = [] # Set by the backend's register_services(); it does the real work. self._impl = None @@ -555,6 +569,25 @@ async def written(self, timeout_ms=None): return await self._bound().written(timeout_ms) +class Descriptor: + """A read-only descriptor on a :class:`Characteristic` you serve. + + HID uses these: each Report characteristic carries a Report Reference + (0x2908) saying which report it is. The value is fixed once registered. + Don't make a CCCD (0x2902); every backend adds that itself for a + characteristic with ``notify`` or ``indicate``. + """ + + def __init__(self, characteristic, uuid, initial=b""): + characteristic.descriptors.append(self) + self.characteristic = characteristic + self.uuid = UUID(uuid) + self.value = bytes(initial) + + def __repr__(self): + return "Descriptor({!r})".format(self.uuid) + + # ---------------------------------------------------------------- central side @@ -754,6 +787,23 @@ async def disconnected(self, timeout_ms=None): """Wait until the link drops, from either side.""" raise NotImplementedError + @property + def encrypted(self): + """True once the link is encrypted (after :meth:`pair`, or a bonded reconnect).""" + return False + + async def pair(self, bond=True, timeout_ms=20000): + """Pair with the peer and encrypt the link; with ``bond``, keep the keys. + + Either side may call it; in practice the central does, when a read + fails with ``INSUFFICIENT_ENCRYPTION``. Pairing here is "just works" + (no passkey), which is what a keyboard without a screen does. A bonded + peer reconnects without pairing again, for as long as both sides keep + their keys. Raises :class:`UnsupportedError` on a backend without + pairing. + """ + raise UnsupportedError("{} cannot pair".format(self._ble.backend)) + async def exchange_mtu(self, mtu=None, timeout_ms=1000): """Negotiate a larger ATT MTU and return the result. @@ -836,8 +886,27 @@ def _check(self, flag, what): raise UnsupportedError("{!r} does not support {}".format(self, what)) async def read(self, timeout_ms=1000): + """The value. A backend without long reads (``capabilities()["long_read"]`` + is False: MicroPython) returns at most ``connection.mtu - 1`` bytes.""" raise NotImplementedError + def descriptors(self, uuid=None, timeout_ms=2000): + """Async-iterate this characteristic's descriptors, optionally only ``uuid``.""" + uuid = None if uuid is None else UUID(uuid) + return _Discovery(lambda: self._discover_descriptors(uuid, timeout_ms)) + + async def descriptor(self, uuid, timeout_ms=2000): + """The descriptor ``uuid`` on this characteristic, or ``None``.""" + uuid = UUID(uuid) + found = None + async for d in self.descriptors(uuid, timeout_ms): + if found is None and d.uuid == uuid: + found = d + return found + + async def _discover_descriptors(self, uuid, timeout_ms): + raise UnsupportedError("{} cannot discover descriptors".format(self.connection._ble.backend)) + async def write(self, data, response=None, timeout_ms=1000): """Write ``data`` (at most ``connection.mtu - 3`` bytes). @@ -867,6 +936,21 @@ def _want_response(self, response): return bool(response) +class ClientDescriptor: + """A descriptor on the peer, found by :meth:`ClientCharacteristic.descriptors`.""" + + def __init__(self, characteristic, uuid): + self.characteristic = characteristic + self.connection = characteristic.connection + self.uuid = UUID(uuid) + + def __repr__(self): + return "ClientDescriptor({!r})".format(self.uuid) + + async def read(self, timeout_ms=1000): + raise NotImplementedError + + def is_adapter(obj): """True for a bledev :class:`BLE`; False for a raw ``bluetooth.BLE`` or anything else.""" return isinstance(obj, BLE) diff --git a/lib/bledev/bleak.py b/lib/bledev/bleak.py index df106544..0d004383 100644 --- a/lib/bledev/bleak.py +++ b/lib/bledev/bleak.py @@ -41,6 +41,7 @@ BleakGATTProtocolError = None from . import ( + ClientDescriptor, BLE, BLEError, ClientCharacteristic, @@ -275,6 +276,7 @@ def __init__(self, ble, device, client): self._dropped = asyncio.Event() self._inboxes = [] self._wwr = 0 # the largest write-without-response size seen in discovery + self._paired = False def _drop(self): if self._dropped.is_set(): @@ -316,6 +318,20 @@ async def disconnected(self, timeout_ms=None): if self.is_connected(): await wait_ms(self._dropped.wait(), timeout_ms, "disconnected") + @property + def encrypted(self): + return self._paired + + async def pair(self, bond=True, timeout_ms=20000): + """Pair through the OS. On Windows that makes a bond the OS keeps (and, + for a HID device, a keyboard it uses), so the tests here never call it.""" + self._live("pair") + try: + await wait_ms(self._client.pair(), timeout_ms, "pair") + except (BleakError, OSError) as e: + raise _translate(e, self, "pair") + self._paired = True + async def exchange_mtu(self, mtu=None, timeout_ms=1000): """The OS negotiates on its own; wait briefly for it and return the result.""" self._live("exchange_mtu") @@ -401,6 +417,15 @@ async def read(self, timeout_ms=1000): except (BleakError, OSError) as e: raise _translate(e, self.connection, "read") + async def _discover_descriptors(self, uuid, timeout_ms): + self._live("descriptor discovery") + found = [] + for native in sorted(self._native.descriptors, key=lambda d: d.handle): + d = _BleakClientDescriptor(self, native) + if uuid is None or d.uuid == uuid: + found.append(d) + return found + async def write(self, data, response=None, timeout_ms=1000): data = bytes(data) response = self._want_response(response) @@ -447,3 +472,17 @@ async def notified(self, timeout_ms=None): async def indicated(self, timeout_ms=None): self._check(FLAG_INDICATE, "indicate") return await self._indicate_inbox.get(timeout_ms) + + +class _BleakClientDescriptor(ClientDescriptor): + def __init__(self, characteristic, native): + ClientDescriptor.__init__(self, characteristic, native.uuid) + self._native = native + + async def read(self, timeout_ms=1000): + client = self.connection._client + self.connection._live("descriptor read") + try: + return bytes(await wait_ms(client.read_gatt_descriptor(self._native.handle), timeout_ms, "descriptor read")) + except (BleakError, OSError) as e: + raise _translate(e, self.connection, "descriptor read") diff --git a/lib/bledev/fake.py b/lib/bledev/fake.py index 71eb66f5..256404c5 100644 --- a/lib/bledev/fake.py +++ b/lib/bledev/fake.py @@ -25,6 +25,12 @@ ``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. +* An ``encrypted`` characteristic refuses an unpaired central with + ``GattError(INSUFFICIENT_ENCRYPTION)``. ``pair()`` encrypts the link, and + with ``bond`` both adapters remember each other (``bonds``) until + ``forget()``. +* ``long_reads=False`` makes a central read at most ``mtu - 1`` bytes, as + MicroPython does (it has no long read). """ try: @@ -37,6 +43,7 @@ BLEError, BusyError, ClientCharacteristic, + ClientDescriptor, ClientService, Connection, DEFAULT_MTU, @@ -47,7 +54,10 @@ FLAG_READ, FLAG_WRITE, FLAG_WRITE_NO_RESPONSE, + GattError, + INSUFFICIENT_ENCRYPTION, MAX_MTU, + UUID, ScanResult, Scanner, UnsupportedError, @@ -153,6 +163,7 @@ def __init__( notify_buffers=8, queue_limit=128, gap_name="fake", + long_reads=True, ): self._air = air if air is not None else _default_air self._central = central @@ -163,6 +174,9 @@ def __init__( self.notify_buffers = notify_buffers self.queue_limit = queue_limit self._gap_name = gap_name + self._long_reads = long_reads + #: Addresses of the peers this adapter has bonded with. + self.bonds = set() self._services = () self._adverts = [] self._links = [] @@ -175,9 +189,18 @@ def capabilities(self): mtu=self._mtu, max_mtu=MAX_MTU, indicate=True, + pairing=True, + long_read=self._long_reads, ) return caps + def forget(self, address=None): + """Drop the bond with ``address``, or every bond (a board's filesystem erased).""" + if address is None: + self.bonds.clear() + else: + self.bonds.discard(address) + def config(self, *names, **settings): for key, value in settings.items(): if key == "mtu": @@ -333,6 +356,7 @@ def __init__(self, central_ble, peripheral_ble, device): self.subscribed = {} self.in_flight = [] self._drain_task = None + self.encrypted = False central_ble._links.append(self) peripheral_ble._links.append(self) @@ -403,6 +427,20 @@ async def disconnected(self, timeout_ms=None): if self._link.connected: await wait_ms(self._link.dropped.wait(), timeout_ms, "disconnected") + @property + def encrypted(self): + return self._link.encrypted + + async def pair(self, bond=True, timeout_ms=20000): + link = self._link + link.check() + await asyncio.sleep(0) + link.check() + link.encrypted = True + if bond: + link.central_ble.bonds.add(link.peripheral_ble.address) + link.peripheral_ble.bonds.add(link.central_ble.address) + async def exchange_mtu(self, mtu=None, timeout_ms=1000): self._link.check() if mtu: @@ -450,14 +488,29 @@ def __init__(self, service, server_char): def _link(self): link = self.connection._link link.check() + if self._server.encrypted and not link.encrypted: + raise GattError(INSUFFICIENT_ENCRYPTION, "{} needs an encrypted link; pair() first".format(self.uuid)) 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() + link = self._link() + value = self._server.read() + if not link.central_ble._long_reads: + value = value[: self.connection.mtu - 1] + return value + + async def _discover_descriptors(self, uuid, timeout_ms): + self.connection._link.check() + await asyncio.sleep(0) + found = [] + if self._server.flags & (FLAG_NOTIFY | FLAG_INDICATE): + found.append(_FakeClientDescriptor(self, UUID(0x2902), None)) + for d in self._server.descriptors: + found.append(_FakeClientDescriptor(self, d.uuid, d)) + return [d for d in found if uuid is None or d.uuid == uuid] async def write(self, data, response=None, timeout_ms=1000): data = bytes(data) @@ -498,6 +551,22 @@ async def indicated(self, timeout_ms=None): return await link.inbox(self._server, "indicate", link.central_ble.queue_limit).get(timeout_ms) +class _FakeClientDescriptor(ClientDescriptor): + def __init__(self, characteristic, uuid, server): + ClientDescriptor.__init__(self, characteristic, uuid) + self._server = server + + async def read(self, timeout_ms=1000): + link = self.connection._link + link.check() + await asyncio.sleep(0) + link.check() + if self._server is None: # the CCCD + notify, indicate = link.subscribed.get(id(self.characteristic._server), (False, False)) + return bytes((notify | (indicate << 1), 0)) + return self._server.value + + class _ServerCharacteristic: """The fake's half of a :class:`bledev.Characteristic`.""" diff --git a/lib/bledev/hid.py b/lib/bledev/hid.py new file mode 100644 index 00000000..5da23571 --- /dev/null +++ b/lib/bledev/hid.py @@ -0,0 +1,547 @@ +"""BLE keyboards and game controllers, both ways: HID over GATT (HOGP). + +**As a host** (central), a board or a laptop connects to a BLE keyboard, +remote or gamepad and gets PyDevices events, the same ``events.Key`` and +``events.Joy*`` records a USB keyboard gives through usbif or SDL gives on the +desktop:: + + import bledev.hid as hid + + host = await hid.connect(ble, name="Keyboard K380") + while True: + for event in await host.events(): + print(event) + +It reads the device's Report Map, parses it (:mod:`bledev.hidreport`), +subscribes to every input report and decodes each one as it arrives. A +keyboard that insists on pairing gets it: the first read that fails for want +of encryption pairs the link and tries again, and with ``bond=True`` (the +default) the keys are kept, so the next connect is quick. + +**As a device** (peripheral), a board advertises as a BLE keyboard, media +remote and gamepad, and types, taps and moves what you tell it:: + + kb = hid.Peripheral(ble) # keyboard + consumer + gamepad + await kb.serve() # waits for a host + await kb.keyboard.type("Hello\\n") + await kb.consumer.tap(hid.VOLUME_UP) + await kb.gamepad.send(axes=(0, 127, 0, 0), buttons=0b101, hat=2) + +Its name, appearance and Report Map follow from which of the three it +offers, the same way every time, because hosts cache a device's services +against its identity (``docs/ble.md`` in the anchor repo, section 7). + +A laptop (bleak) or a browser can't advertise, so they are hosts only. +Windows keeps HID devices for itself once paired; don't pair a board you are +testing with the PC you are testing from, or it will type into it. +""" + +try: + import asyncio +except ImportError: # pragma: no cover + import uasyncio as asyncio + +import struct + +try: + from time import ticks_diff, ticks_us +except ImportError: # CPython + import time as _time + + def ticks_us(): + return int(_time.perf_counter() * 1000000) + + def ticks_diff(a, b): + return a - b + + +from . import ( + BLEError, + BusyError, + Characteristic, + Descriptor, + GattError, + INSUFFICIENT_AUTHENTICATION, + INSUFFICIENT_ENCRYPTION, + Service, + UUID, + sleep_ms, + wait_ms, +) +from .hidreport import INPUT, OUTPUT, Decoder, HIDParseError, ReportMap + +# GATT numbers from the HID Service and HOGP specifications. +HID_SERVICE = UUID(0x1812) +HID_INFORMATION = UUID(0x2A4A) +REPORT_MAP = UUID(0x2A4B) +HID_CONTROL_POINT = UUID(0x2A4C) +REPORT = UUID(0x2A4D) +PROTOCOL_MODE = UUID(0x2A4E) +REPORT_REFERENCE = UUID(0x2908) +BATTERY_SERVICE = UUID(0x180F) +BATTERY_LEVEL = UUID(0x2A19) +DEVICE_INFORMATION = UUID(0x180A) +PNP_ID = UUID(0x2A50) +MANUFACTURER_NAME = UUID(0x2A29) + +#: The HID characteristics under a vendor UUID, for inspection from a laptop. +#: Windows hides the HID service (0x1812) from apps entirely, and Chrome's +#: Web Bluetooth blocklist does too, so a laptop can only see a board's HID +#: characteristics if the board serves them under another UUID. +INSPECT_SERVICE = UUID("b1ed0000-4849-4400-8000-00805f9b34fb") + +APPEARANCE_HID = 0x03C0 +APPEARANCE_KEYBOARD = 0x03C1 +APPEARANCE_GAMEPAD = 0x03C4 + +# The report IDs a board serves. +KEYBOARD_ID = 1 +CONSUMER_ID = 2 +GAMEPAD_ID = 3 + +# Consumer usages worth a name. +VOLUME_UP = 0xE9 +VOLUME_DOWN = 0xEA +MUTE = 0xE2 +PLAY_PAUSE = 0xCD +NEXT_TRACK = 0xB5 +PREV_TRACK = 0xB6 + +# Modifier bits in a keyboard report. +MOD_LCTRL = 0x01 +MOD_LSHIFT = 0x02 +MOD_LALT = 0x04 +MOD_LGUI = 0x08 +MOD_RCTRL = 0x10 +MOD_RSHIFT = 0x20 +MOD_RALT = 0x40 +MOD_RGUI = 0x80 + +KEYBOARD_MAP = bytes(( + 0x05, 0x01, 0x09, 0x06, 0xA1, 0x01, 0x85, KEYBOARD_ID, + 0x05, 0x07, 0x19, 0xE0, 0x29, 0xE7, 0x15, 0x00, 0x25, 0x01, + 0x75, 0x01, 0x95, 0x08, 0x81, 0x02, # 8 modifier bits + 0x95, 0x01, 0x75, 0x08, 0x81, 0x01, # reserved byte + 0x95, 0x05, 0x75, 0x01, 0x05, 0x08, 0x19, 0x01, 0x29, 0x05, + 0x91, 0x02, 0x95, 0x01, 0x75, 0x03, 0x91, 0x01, # LEDs out + 0x95, 0x06, 0x75, 0x08, 0x15, 0x00, 0x25, 0x65, + 0x05, 0x07, 0x19, 0x00, 0x29, 0x65, 0x81, 0x00, # 6 keys + 0xC0, +)) + +CONSUMER_MAP = bytes(( + 0x05, 0x0C, 0x09, 0x01, 0xA1, 0x01, 0x85, CONSUMER_ID, + 0x75, 0x10, 0x95, 0x01, 0x15, 0x01, 0x26, 0x8C, 0x02, + 0x19, 0x01, 0x2A, 0x8C, 0x02, 0x81, 0x00, # one 16-bit usage + 0xC0, +)) + +GAMEPAD_MAP = bytes(( + 0x05, 0x01, 0x09, 0x05, 0xA1, 0x01, 0x85, GAMEPAD_ID, + 0x05, 0x09, 0x19, 0x01, 0x29, 0x10, 0x15, 0x00, 0x25, 0x01, + 0x75, 0x01, 0x95, 0x10, 0x81, 0x02, # 16 buttons + 0x05, 0x01, 0x09, 0x39, 0x15, 0x01, 0x25, 0x08, + 0x35, 0x00, 0x46, 0x3B, 0x01, 0x65, 0x14, + 0x75, 0x04, 0x95, 0x01, 0x81, 0x42, # hat, null = centred + 0x35, 0x00, 0x45, 0x00, 0x65, 0x00, 0x81, 0x03, # 4 bits padding + 0x09, 0x30, 0x09, 0x31, 0x09, 0x32, 0x09, 0x35, + 0x15, 0x81, 0x25, 0x7F, 0x75, 0x08, 0x95, 0x04, 0x81, 0x02, # X Y Z Rz, -127..127 + 0xC0, +)) + + +def report_map(keyboard=True, consumer=True, gamepad=True): + """The Report Map a board serves for this set of functions, always the same bytes.""" + return (KEYBOARD_MAP if keyboard else b"") + (CONSUMER_MAP if consumer else b"") + (GAMEPAD_MAP if gamepad else b"") + + +def identity(keyboard=True, consumer=True, gamepad=True): + """``(name, appearance, product_id)`` for a set of functions, derived the same way every time.""" + parts = [] + if keyboard: + parts.append("Keyboard") + if gamepad: + parts.append("Gamepad") + if consumer and not keyboard: + parts.append("Remote") + name = "PyDevices " + "+".join(parts) if parts else "PyDevices HID" + if keyboard and not gamepad: + appearance = APPEARANCE_KEYBOARD + elif gamepad and not keyboard and not consumer: + appearance = APPEARANCE_GAMEPAD + else: + appearance = APPEARANCE_HID + product = (1 if keyboard else 0) | (2 if consumer else 0) | (4 if gamepad else 0) + return name, appearance, product + + +# US layout: character -> (usage, shift). +def _us_layout(): + table = {"\n": (0x28, False), "\t": (0x2B, False), " ": (0x2C, False)} + for i in range(26): + table[chr(ord("a") + i)] = (0x04 + i, False) + table[chr(ord("A") + i)] = (0x04 + i, True) + for i, (plain, shifted) in enumerate(zip("1234567890", "!@#$%^&*()")): + table[plain] = (0x1E + i, False) + table[shifted] = (0x1E + i, True) + for usage, plain, shifted in ( + (0x2D, "-", "_"), (0x2E, "=", "+"), (0x2F, "[", "{"), (0x30, "]", "}"), + (0x31, "\\", "|"), (0x33, ";", ":"), (0x34, "'", '"'), (0x35, "`", "~"), + (0x36, ",", "<"), (0x37, ".", ">"), (0x38, "/", "?"), + ): + table[plain] = (usage, False) + table[shifted] = (usage, True) + return table + + +US_LAYOUT = _us_layout() + + +# ---------------------------------------------------------------- the device (peripheral) + + +class _Sender: + def __init__(self, owner, characteristic, size): + self._owner = owner + self._char = characteristic + self._size = size + + async def _send(self, report): + report = bytes(report) + if len(report) != self._size: + raise ValueError("report is {} bytes, not {}".format(len(report), self._size)) + conn = self._owner.connection + if conn is None or not conn.is_connected(): + raise BLEError("no host connected") + self._char.write(report) + waited = 0 + while True: + try: + self._char.notify(conn, report) + self._owner.last_send_us = ticks_us() + return + except BusyError: + if waited >= 2000: + raise + await sleep_ms(2) + waited += 2 + + +class Keyboard(_Sender): + """The keyboard half of a :class:`Peripheral`. Usages are HID keyboard usages.""" + + async def send(self, modifiers=0, keys=()): + """Send one report: ``modifiers`` bits held, and up to six usages held.""" + keys = list(keys)[:6] + await self._send(bytes([modifiers, 0] + keys + [0] * (6 - len(keys)))) + + async def tap(self, usage, modifiers=0, hold_ms=0): + """Press and release one key (with ``modifiers`` held around it).""" + await self.send(modifiers, (usage,)) + if hold_ms: + await sleep_ms(hold_ms) + await self.send(0, ()) + + async def type(self, text, gap_ms=0): + """Type ``text`` on a US layout: each character a press and a release.""" + for ch in text: + usage, shift = US_LAYOUT[ch] + await self.tap(usage, MOD_LSHIFT if shift else 0) + if gap_ms: + await sleep_ms(gap_ms) + + +class Consumer(_Sender): + """Media keys: consumer-page usages such as :data:`VOLUME_UP`.""" + + async def send(self, usage=0): + await self._send(struct.pack("= conn.mtu - 1: + raise BLEError( + "the Report Map is at least {} bytes and this host reads one packet " + "(mtu {} - 1); MicroPython has no long read".format(len(data), conn.mtu) + ) + raise BLEError("the Report Map doesn't parse: {}".format(e)) + self.decoder = Decoder(self.report_map, self.instance_id) + for c in chars: + if c.uuid == PROTOCOL_MODE: + # Report protocol (1), which is also the default after connecting. + try: + await self._secure(lambda: c.write(b"\x01", response=False)) + except BLEError: + pass + inputs = [] + for c in chars: + if c.uuid != REPORT: + continue + ref = await c.descriptor(REPORT_REFERENCE, timeout_ms) + if ref is None: + continue + value = await self._secure(lambda: ref.read(timeout_ms)) + if len(value) < 2: + continue + report_id, kind = value[0], value[1] + if kind == INPUT: + inputs.append((c, report_id)) + elif kind == OUTPUT and report_id and self.report_map.report(OUTPUT, report_id): + self._leds = (c, report_id) + if not inputs: + raise BLEError("the HID service has no input reports") + for c, report_id in inputs: + await self._secure(lambda: c.subscribe(notify=True)) + for c, report_id in inputs: + self._tasks.append(asyncio.create_task(self._reader(c, report_id))) + if self._leds is not None: + # Hosts set a keyboard's LEDs once they're listening; it also + # tells our own Peripheral that someone is (Peripheral.leds()). + await self.set_leds(0) + return self + + async def _reader(self, characteristic, report_id): + try: + while True: + data = await characteristic.notified() + self.last_report_us = ticks_us() + for event in self.decoder.feed(data, report_id): + if len(self._queue) >= self._limit: + self._overrun += 1 + else: + self._queue.append(event) + self._flag.set() + except asyncio.CancelledError: + raise + except Exception as e: # a disconnect, or an overrun below us + self._error = e + self._flag.set() + + def _check(self): + if self._overrun: + n, self._overrun = self._overrun, 0 + raise BLEError("{} HID event(s) dropped: nobody read them".format(n)) + + def poll(self): + """Every event decoded so far, as a list (possibly empty). Never waits.""" + self._check() + out, self._queue = self._queue, [] + if not out and self._error is not None: + raise self._error + return out + + async def events(self, timeout_ms=None): + """Wait for at least one event, and return every one waiting.""" + while not self._queue: + self._check() + if self._error is not None: + raise self._error + self._flag.clear() + await wait_ms(self._flag.wait(), timeout_ms, "HID events") + return self.poll() + + async def set_leds(self, bits): + """Write the keyboard's LED report (Num Lock is bit 0), if it has one.""" + if self._leds is None: + raise BLEError("this device has no LED report") + c, _ = self._leds + await self._secure(lambda: c.write(bytes((bits,)), response=False)) + + async def close(self): + for t in self._tasks: + t.cancel() + self._tasks = [] + if self.connection.is_connected(): + await self.connection.disconnect() + + +async def connect(ble, name=None, *, device=None, timeout_ms=10000, bond=True, instance_id=0, + service_uuid=HID_SERVICE, **options): + """Find a BLE HID device (by ``name``, or any advertising the HID service), + connect, and return a started :class:`Host`.""" + if device is None: + device = await ble.find(name=name, service=None if name else service_uuid, timeout_ms=timeout_ms) + connection = await device.connect(timeout_ms=timeout_ms, **options) + host = Host(connection, bond=bond, instance_id=instance_id, service_uuid=service_uuid) + try: + await host.start() + except BaseException: + await host.close() + raise + return host + + +__all__ = [ + "Host", "Peripheral", "Keyboard", "Consumer", "Gamepad", "connect", "report_map", "identity", + "US_LAYOUT", "ticks_us", "ticks_diff", +] diff --git a/lib/bledev/hidreport.py b/lib/bledev/hidreport.py new file mode 100644 index 00000000..8ce4a11b --- /dev/null +++ b/lib/bledev/hidreport.py @@ -0,0 +1,658 @@ +"""HID report descriptors, and the reports they describe, as PyDevices events. + +A HID device describes its own reports: the *report descriptor* (Bluetooth +calls it the Report Map) says which bits of which report carry which control. +This module reads one, and turns the device's input reports into the same +``events.Key``, ``events.JoyAxisMotion``, ``events.JoyHatMotion`` and +``events.JoyButtonDown``/``Up`` records a keyboard or a joystick produces on +the desktop through SDL:: + + from bledev.hidreport import ReportMap, Decoder + + rmap = ReportMap(descriptor_bytes) + decoder = Decoder(rmap) + for event in decoder.feed(report, report_id=1): + ... + +Nothing here knows about Bluetooth. A USB host that has the descriptor can +use it the same way; ``feed(data)`` without ``report_id`` takes the ID from +the report's first byte when the map numbers its reports, as USB sends them. + +**Keyboards** follow usbif's boot-keyboard decoder exactly (same event order, +same fields), because an app must not be able to tell a BLE keyboard from a +USB one. The keyboard says which keys are held; events come from the +difference between successive reports: modifier releases, key releases, key +presses, then modifier presses. An ErrorRollOver report is ignored, so the +eventual releases still balance. + +**Consumer controls** (media keys, remotes) become ``events.Key`` as well, +with ``keys.K_VOLUMEUP`` and friends, and ``scancode`` set to the extended +usage (``0x000C00E9`` for volume up). A usage with no key constant arrives +with ``key=None`` and name ``"Unknown"``, as usbif does. + +**Game controllers** (the joystick, gamepad and multi-axis collections) +produce SDL-style joystick events. Axes are numbered by usage, X, Y, Z, Rx, +Ry, Rz, slider, dial, wheel, then the simulation page (accelerator, brake), +and their value is a float from -1.0 to 1.0 across the logical range. +Buttons are numbered from 0 (Button 1 is button 0). A hat is an ``(x, y)`` +tuple, y up, ``(0, 0)`` when centred, like pygame. +""" + +import events +import keys + +# Main item kinds, numbered as the HOGP Report Reference descriptor numbers them. +INPUT = 1 +OUTPUT = 2 +FEATURE = 3 + +# Main-item flag bits. +CONSTANT = 0x01 +VARIABLE = 0x02 +RELATIVE = 0x04 +NULL_STATE = 0x40 + +# Usage pages. +PAGE_DESKTOP = 0x01 +PAGE_SIMULATION = 0x02 +PAGE_KEYBOARD = 0x07 +PAGE_LED = 0x08 +PAGE_BUTTON = 0x09 +PAGE_CONSUMER = 0x0C + +# Application collections, as extended usages (page << 16 | usage). +APP_POINTER = 0x00010001 +APP_MOUSE = 0x00010002 +APP_JOYSTICK = 0x00010004 +APP_GAMEPAD = 0x00010005 +APP_KEYBOARD = 0x00010006 +APP_KEYPAD = 0x00010007 +APP_MULTI_AXIS = 0x00010008 +APP_CONSUMER = 0x000C0001 + +_GAME_APPS = (APP_JOYSTICK, APP_GAMEPAD, APP_MULTI_AXIS) +_HAT = 0x00010039 + +ERROR_ROLLOVER = 0x01 + + +class HIDParseError(ValueError): + """The descriptor is malformed: truncated, unbalanced, or impossible.""" + + +def _signed(raw, size): + if size == 1 and raw & 0x80: + return raw - 0x100 + if size == 2 and raw & 0x8000: + return raw - 0x10000 + if size == 4 and raw & 0x80000000: + return raw - 0x100000000 + return raw + + +class Field: + """One main item: ``count`` values of ``size`` bits at bit ``offset``. + + ``offset`` counts from the start of the report *after* its report ID, the + way HOGP delivers it. ``usages`` is a list of extended usages + (``page << 16 | id``) and ``(first, last)`` ranges, in descriptor order. + ``application`` is the extended usage of the top-level collection the + field sits in. + """ + + def __init__(self, kind, report_id, offset, size, count, flags, usages, + logical_min, logical_max, physical_min, physical_max, + unit, unit_exponent, application): + self.kind = kind + self.report_id = report_id + self.offset = offset + self.size = size + self.count = count + self.flags = flags + self.usages = usages + self.logical_min = logical_min + self.logical_max = logical_max + self.physical_min = physical_min + self.physical_max = physical_max + self.unit = unit + self.unit_exponent = unit_exponent + self.application = application + + def __repr__(self): + return "Field(kind={}, id={}, offset={}, size={}, count={}, flags=0x{:02x}, usages={})".format( + self.kind, self.report_id, self.offset, self.size, self.count, self.flags, + ["0x{:08x}".format(u) if isinstance(u, int) else "0x{:08x}-0x{:08x}".format(*u) for u in self.usages], + ) + + @property + def constant(self): + return bool(self.flags & CONSTANT) + + @property + def variable(self): + return bool(self.flags & VARIABLE) + + @property + def relative(self): + return bool(self.flags & RELATIVE) + + def usage(self, n): + """The ``n``-th usage in the list, ranges expanded; past the end, the last one. + + For a variable field that's the usage of element ``n``; for an array + it's what the value ``logical_min + n`` means. ``None`` if the field + has no usages at all (padding). + """ + last = None + for entry in self.usages: + if isinstance(entry, int): + if n == 0: + return entry + n -= 1 + last = entry + else: + first, end = entry + span = end - first + 1 + if n < span: + return first + n + n -= span + last = end + return last + + def raw(self, data, n=0): + """Element ``n``'s raw value from report ``data`` (without its ID), sign-extended if signed.""" + bit = self.offset + n * self.size + first = bit >> 3 + last = (bit + self.size + 7) >> 3 + if last > len(data): + raise ValueError("report of {} bytes is too short for bits {}..{}".format( + len(data), bit, bit + self.size - 1)) + value = (int.from_bytes(bytes(data[first:last]), "little") >> (bit & 7)) & ((1 << self.size) - 1) + if self.logical_min < 0 and value & (1 << (self.size - 1)): + value -= 1 << self.size + return value + + def values(self, data): + """Every element's raw value, as a list.""" + return [self.raw(data, n) for n in range(self.count)] + + +class Report: + """The fields of one report: its ``kind``, its ``report_id`` (0 if unnumbered) and its size.""" + + def __init__(self, kind, report_id): + self.kind = kind + self.report_id = report_id + self.fields = [] + self.bits = 0 + + @property + def size(self): + """Bytes, without the report ID.""" + return (self.bits + 7) >> 3 + + def __repr__(self): + return "Report(kind={}, id={}, size={}, fields={})".format( + self.kind, self.report_id, self.size, len(self.fields)) + + +class ReportMap: + """A parsed report descriptor. + + ``reports`` maps ``(kind, report_id)`` to :class:`Report`; ``numbered`` + says whether the device uses report IDs; ``applications`` lists the + top-level collections' extended usages in order. + """ + + def __init__(self, descriptor): + self.descriptor = bytes(descriptor) + self.reports = {} + self.numbered = False + self.applications = [] + self._parse() + + def report(self, kind, report_id=0): + return self.reports.get((kind, report_id)) + + def inputs(self): + """The input reports, in report ID order.""" + return [r for (k, _), r in sorted(self.reports.items()) if k == INPUT] + + def _parse(self): + d = self.descriptor + i = 0 + # Global state: page, lmin, lmax, pmin, pmax, unit_exp, unit, size, id, count. + page = 0 + lmin = lmax = pmin = pmax = 0 + lmax_n = 1 + unit = unit_exp = 0 + size = 0 + report_id = 0 + count = 0 + stack = [] + # Local state. + usages = [] + umin = None + collections = [] # extended usage of each open collection + application = 0 + seen_ids = False + while i < len(d): + prefix = d[i] + if prefix == 0xFE: # long item: skip + if i + 2 >= len(d): + raise HIDParseError("truncated long item at {}".format(i)) + i += 3 + d[i + 1] + continue + n = prefix & 0x03 + n = 4 if n == 3 else n + if i + 1 + n > len(d): + raise HIDParseError("item at byte {} runs past the end".format(i)) + raw = int.from_bytes(d[i + 1:i + 1 + n], "little") if n else 0 + kind = (prefix >> 2) & 0x03 + tag = prefix >> 4 + i += 1 + n + if kind == 1: # global + if tag == 0x0: + page = raw + elif tag == 0x1: + lmin = _signed(raw, n) + elif tag == 0x2: + lmax = _signed(raw, n) + lmax_n = n + elif tag == 0x3: + pmin = _signed(raw, n) + elif tag == 0x4: + pmax = _signed(raw, n) + elif tag == 0x5: + unit_exp = _signed(raw, n) if n > 1 else (raw - 16 if raw & 0x08 else raw) + elif tag == 0x6: + unit = raw + elif tag == 0x7: + size = raw + elif tag == 0x8: + if raw == 0: + raise HIDParseError("report ID 0 is reserved") + report_id = raw + seen_ids = True + elif tag == 0x9: + count = raw + elif tag == 0xA: + stack.append((page, lmin, lmax, lmax_n, pmin, pmax, unit_exp, unit, size, report_id, count)) + elif tag == 0xB: + if not stack: + raise HIDParseError("Pop with nothing pushed") + page, lmin, lmax, lmax_n, pmin, pmax, unit_exp, unit, size, report_id, count = stack.pop() + elif kind == 2: # local + # A 4-byte usage carries its own page; a shorter one takes + # the page in force when the main item is reached. + ext = raw if n == 4 else None + if tag == 0x0: + usages.append(ext if ext is not None else ("p", raw)) + elif tag == 0x1: + umin = ext if ext is not None else ("p", raw) + elif tag == 0x2: + if umin is None: + raise HIDParseError("Usage Maximum without Usage Minimum") + usages.append((umin, ext if ext is not None else ("p", raw))) + umin = None + # Designators, strings and delimiters don't change the layout. + elif kind == 0: # main + resolved = [] + for u in usages: + if isinstance(u, int): + resolved.append(u) + elif u[0] == "p": + resolved.append((page << 16) | u[1]) + else: + a, b = u + a = (page << 16) | a[1] if isinstance(a, tuple) else a + b = (page << 16) | b[1] if isinstance(b, tuple) else b + resolved.append((a, b)) + if tag in (0x8, 0x9, 0xB): # input, output, feature + mkind = INPUT if tag == 0x8 else (OUTPUT if tag == 0x9 else FEATURE) + if size == 0 and count: + raise HIDParseError("main item with Report Size 0 at byte {}".format(i)) + key = (mkind, report_id) + rep = self.reports.get(key) + if rep is None: + rep = self.reports[key] = Report(mkind, report_id) + # A signed range read as unsigned: 25 FF with 15 00 means 255. + hi = lmax + if lmin >= 0 and hi < 0: + hi += 1 << (8 * lmax_n) + rep.fields.append(Field( + mkind, report_id, rep.bits, size, count, raw, resolved, + lmin, hi, pmin, pmax, unit, unit_exp, application, + )) + rep.bits += size * count + elif tag == 0xA: # collection + usage = resolved[0] if resolved else 0 + if isinstance(usage, tuple): + usage = usage[0] + if not collections: + application = usage + self.applications.append(usage) + collections.append(usage) + elif tag == 0xC: # end collection + if not collections: + raise HIDParseError("End Collection with none open") + collections.pop() + if not collections: + application = 0 + usages = [] + umin = None + if collections: + raise HIDParseError("{} collection(s) left open".format(len(collections))) + self.numbered = seen_ids + if seen_ids and any(rid == 0 for (_, rid) in self.reports): + raise HIDParseError("a report outside any Report ID in a numbered descriptor") + + +# ---------------------------------------------------------------- key tables + +# HID keyboard usage -> SDL keycode. SDL's scancodes *are* HID usages, and +# SDL makes the keycode of every key without a character ``usage | 1 << 30``, +# so only the character keys need a table. It agrees with usbif's table +# (tests/test_bledev_hid.py compares the two), so 0x32, the ISO keyboard's +# "# ~" key, stays unmapped here as it is there, though SDL calls it '#'. +_CHARS = { + 0x28: keys.K_RETURN, 0x29: keys.K_ESCAPE, 0x2A: keys.K_BACKSPACE, + 0x2B: keys.K_TAB, 0x2C: keys.K_SPACE, + 0x2D: ord("-"), 0x2E: ord("="), 0x2F: ord("["), 0x30: ord("]"), + 0x31: ord("\\"), 0x33: ord(";"), 0x34: ord("'"), + 0x35: ord("`"), 0x36: ord(","), 0x37: ord("."), 0x38: ord("/"), + 0x4C: keys.K_DELETE, +} + +_SCANCODE_KEY = 1 << 30 + + +def keycode(usage): + """The SDL-style keycode (``keys.K_*``) for a keyboard-page usage, or ``None``.""" + if 0x04 <= usage <= 0x1D: + return keys.K_a + usage - 0x04 + if 0x1E <= usage <= 0x26: + return ord("1") + usage - 0x1E + if usage == 0x27: + return ord("0") + code = _CHARS.get(usage) + if code is not None: + return code + if usage >= 0x39: + code = usage | _SCANCODE_KEY + if keys.keyname(code) != "Unknown": + return code + return None + + +# Consumer page usage -> keys constant name. +_CONSUMER = { + 0x30: "K_POWER", 0x32: "K_SLEEP", 0x6F: "K_BRIGHTNESSUP", 0x70: "K_BRIGHTNESSDOWN", + 0xB5: "K_AUDIONEXT", 0xB6: "K_AUDIOPREV", 0xB7: "K_AUDIOSTOP", 0xB8: "K_EJECT", + 0xCD: "K_AUDIOPLAY", 0xE2: "K_MUTE", 0xE9: "K_VOLUMEUP", 0xEA: "K_VOLUMEDOWN", + 0x183: "K_MEDIASELECT", 0x18A: "K_MAIL", 0x192: "K_CALCULATOR", 0x194: "K_COMPUTER", + 0x196: "K_WWW", 0x221: "K_AC_SEARCH", 0x223: "K_AC_HOME", 0x224: "K_AC_BACK", + 0x225: "K_AC_FORWARD", 0x226: "K_AC_STOP", 0x227: "K_AC_REFRESH", 0x22A: "K_AC_BOOKMARKS", +} + + +def consumer_keycode(usage): + """The ``keys.K_*`` code for a consumer-page usage (``0xE9``, not extended), or ``None``.""" + name = _CONSUMER.get(usage) + return getattr(keys, name, None) if name else None + + +# Modifier usage 0xE0..0xE7 -> (KMOD mask, keycode), in report bit order. +_MODIFIERS = ( + (keys.KMOD_LCTRL, keys.K_LCTRL), (keys.KMOD_LSHIFT, keys.K_LSHIFT), + (keys.KMOD_LALT, keys.K_LALT), (keys.KMOD_LGUI, keys.K_LGUI), + (keys.KMOD_RCTRL, keys.K_RCTRL), (keys.KMOD_RSHIFT, keys.K_RSHIFT), + (keys.KMOD_RALT, keys.K_RALT), (keys.KMOD_RGUI, keys.K_RGUI), +) + + +def modifier_mask(bits): + """``keys.KMOD_*`` for a byte of modifier bits (bit 0 is left Ctrl).""" + mask = keys.KMOD_NONE + for i in range(8): + if bits & (1 << i): + mask |= _MODIFIERS[i][0] + return mask + + +# Axis order: desktop X..Wheel, then the simulation page. +_AXES_DESKTOP = (0x30, 0x31, 0x32, 0x33, 0x34, 0x35, 0x36, 0x37, 0x38) + +# Hat directions for 8 positions, clockwise from north; y is up. +_HAT8 = ((0, 1), (1, 1), (1, 0), (1, -1), (0, -1), (-1, -1), (-1, 0), (-1, 1)) +_HAT4 = ((0, 1), (1, 0), (0, -1), (-1, 0)) + + +def _is_axis(usage): + page = usage >> 16 + if page == PAGE_DESKTOP: + return (usage & 0xFFFF) in _AXES_DESKTOP + return page == PAGE_SIMULATION + + +def _axis_order(usage): + page = usage >> 16 + if page == PAGE_DESKTOP: + return _AXES_DESKTOP.index(usage & 0xFFFF) + return 0x100 + (usage & 0xFFFF) + + +class Decoder: + """Turns one device's input reports into PyDevices events. + + One instance per device: it holds that device's state (held keys, axis + positions) and diffs each report against it. ``instance_id`` is the + joystick instance its joystick events carry. + """ + + def __init__(self, report_map, instance_id=0): + self.map = report_map + self.instance_id = instance_id + # Per input report ID: (keyboard, consumer, axes, hats, buttons) element lists. + self._plan = {} + axes = [] + hats = [] + for report in report_map.inputs(): + kb, cc, ax, ht, bt = [], [], [], [], [] + for f in report.fields: + if f.constant or not f.usages: + continue + game = f.application in _GAME_APPS + if not f.variable: + page = f.usage(0) >> 16 + if page == PAGE_KEYBOARD: + kb.append((f, None)) + elif page == PAGE_CONSUMER: + cc.append((f, None)) + continue + for n in range(f.count): + u = f.usage(n) + page = u >> 16 + if page == PAGE_KEYBOARD: + kb.append((f, n)) + elif page == PAGE_CONSUMER: + cc.append((f, n)) + elif game and u == _HAT: + hats.append((report.report_id, f.offset, n, f)) + elif game and page == PAGE_BUTTON and (u & 0xFFFF): + bt.append((f, n, (u & 0xFFFF) - 1)) + elif game and not f.relative and _is_axis(u): + axes.append((_axis_order(u), report.report_id, f.offset, n, f)) + self._plan[report.report_id] = [kb, cc, [], [], bt] + # Number axes and hats across the whole device, by usage then position. + axes.sort(key=lambda a: a[:4]) + for index, (_, rid, _, n, f) in enumerate(axes): + self._plan[rid][2].append((f, n, index)) + hats.sort(key=lambda h: h[:3]) + for index, (rid, _, n, f) in enumerate(hats): + self._plan[rid][3].append((f, n, index)) + self.num_axes = len(axes) + self.num_hats = len(hats) + self.num_buttons = 0 + for plan in self._plan.values(): + for _, _, b in plan[4]: + self.num_buttons = max(self.num_buttons, b + 1) + self._held = frozenset() + self._mod_bits = 0 + self._consumer = frozenset() + self._axes = {} + self._hats = {} + self._buttons = {} + + @property + def held(self): + """Keyboard usages held now.""" + return self._held + + @property + def modifiers(self): + return modifier_mask(self._mod_bits) + + def axis(self, n): + """Axis ``n``'s last value, -1.0..1.0 (0.0 before any report).""" + return self._axes.get(n, 0.0) + + def button(self, n): + return self._buttons.get(n, False) + + def hat(self, n): + return self._hats.get(n, (0, 0)) + + def feed(self, data, report_id=None): + """Decode one input report and return its events, as a tuple. + + ``report_id`` is the ID HOGP gives in the Report Reference; leave it + out for a USB-style report, whose first byte is the ID when the map + numbers its reports. + """ + data = bytes(data) + if report_id is None: + if self.map.numbered: + if not data: + return () + report_id, data = data[0], data[1:] + else: + report_id = 0 + plan = self._plan.get(report_id) + if plan is None: + return () + kb, cc, ax, ht, bt = plan + out = [] + if kb: + self._keyboard(kb, data, out) + if cc: + self._consumer_keys(cc, data, out) + for f, n, index in ax: + raw = f.raw(data, n) + lo, hi = f.logical_min, f.logical_max + if hi <= lo: + continue + raw = lo if raw < lo else (hi if raw > hi else raw) + value = 2.0 * (raw - lo) / (hi - lo) - 1.0 + if self._axes.get(index, 0.0) != value: + self._axes[index] = value + out.append(events.JoyAxisMotion(events.JOYAXISMOTION, self.instance_id, index, value)) + for f, n, index in ht: + raw = f.raw(data, n) + span = f.logical_max - f.logical_min + 1 + pos = raw - f.logical_min + table = _HAT8 if span == 8 else (_HAT4 if span == 4 else None) + value = table[pos] if table is not None and 0 <= pos < span else (0, 0) + if self._hats.get(index, (0, 0)) != value: + self._hats[index] = value + out.append(events.JoyHatMotion(events.JOYHATMOTION, self.instance_id, index, value)) + ups, downs = [], [] + for f, n, button in bt: + pressed = f.raw(data, n) != 0 + if self._buttons.get(button, False) != pressed: + self._buttons[button] = pressed + (downs if pressed else ups).append(button) + for b in sorted(ups): + out.append(events.JoyButtonUp(events.JOYBUTTONUP, self.instance_id, b)) + for b in sorted(downs): + out.append(events.JoyButtonDown(events.JOYBUTTONDOWN, self.instance_id, b)) + return tuple(out) + + def _keyboard(self, elements, data, out): + now = set() + mod_bits = 0 + for f, n in elements: + if n is not None: # variable: one bit per usage + u = f.usage(n) & 0xFFFF + if f.raw(data, n): + if 0xE0 <= u <= 0xE7: + mod_bits |= 1 << (u - 0xE0) + elif u: + now.add(u) + continue + for k in range(f.count): # array: each value names a usage + v = f.raw(data, k) + if v < f.logical_min or v > f.logical_max: + continue + u = f.usage(v - f.logical_min) + if u is None: + continue + u &= 0xFFFF + if u == ERROR_ROLLOVER: + # More keys held than the report carries: leave the held + # set alone so the eventual releases balance. + return + if 0xE0 <= u <= 0xE7: + mod_bits |= 1 << (u - 0xE0) + elif u > 0x03: + now.add(u) + now = frozenset(now) + mod = modifier_mask(mod_bits) + was = self._mod_bits + for i in range(8): + if was & (1 << i) and not mod_bits & (1 << i): + code = _MODIFIERS[i][1] + out.append(events.Key(events.KEYUP, keys.keyname(code), code, mod, 0xE0 + i, None)) + for u in sorted(self._held - now): + out.append(_key(events.KEYUP, u, keycode(u), mod)) + for u in sorted(now - self._held): + out.append(_key(events.KEYDOWN, u, keycode(u), mod)) + for i in range(8): + if mod_bits & (1 << i) and not was & (1 << i): + code = _MODIFIERS[i][1] + out.append(events.Key(events.KEYDOWN, keys.keyname(code), code, mod, 0xE0 + i, None)) + self._held = now + self._mod_bits = mod_bits + + def _consumer_keys(self, elements, data, out): + now = set() + for f, n in elements: + if n is not None: + if f.raw(data, n): + now.add(f.usage(n) & 0xFFFF) + continue + for k in range(f.count): + v = f.raw(data, k) + if v < f.logical_min or v > f.logical_max: + continue + u = f.usage(v - f.logical_min) + if u is not None and u & 0xFFFF: + now.add(u & 0xFFFF) + now = frozenset(now) + mod = modifier_mask(self._mod_bits) + for u in sorted(self._consumer - now): + out.append(_key(events.KEYUP, (PAGE_CONSUMER << 16) | u, consumer_keycode(u), mod)) + for u in sorted(now - self._consumer): + out.append(_key(events.KEYDOWN, (PAGE_CONSUMER << 16) | u, consumer_keycode(u), mod)) + self._consumer = now + + +def _key(kind, scancode, code, mod): + return events.Key( + kind, + keys.keyname(code) if code is not None else "Unknown", + code, + mod, + scancode, + None, + ) diff --git a/lib/bledev/mpble.py b/lib/bledev/mpble.py index a0df68dc..65ca1ae9 100644 --- a/lib/bledev/mpble.py +++ b/lib/bledev/mpble.py @@ -42,6 +42,7 @@ BLETimeoutError, BusyError, ClientCharacteristic, + ClientDescriptor, ClientService, Connection, DEFAULT_MTU, @@ -63,11 +64,16 @@ _IRQ_CENTRAL_CONNECT = 1 _IRQ_CENTRAL_DISCONNECT = 2 _IRQ_GATTS_WRITE = 3 +_IRQ_PERIPHERAL_DISCONNECT = 8 _IRQ_MTU_EXCHANGED = 21 +_FLAG_READ_ENCRYPTED = 0x0200 +_FLAG_WRITE_ENCRYPTED = 0x1000 + # 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) +_EALREADY = 120 # The largest ATT MTU MicroPython's NimBLE will negotiate. _MAX_MTU = 512 @@ -150,7 +156,11 @@ def __exit__(self, exc_type, exc, tb): if issubclass(exc_type, aioble.DeviceDisconnectedError): raise DisconnectedError("{}: disconnected".format(self._what)) if issubclass(exc_type, aioble.GattError): - raise GattError(getattr(exc, "_status", 0)) + status = getattr(exc, "_status", 0) or 0 + # NimBLE reports an ATT error as 0x100 + the ATT code. + if 0x100 <= status < 0x200: + status -= 0x100 + raise GattError(status) if exc_type is ValueError and exc.args and exc.args[0] == "Not connected": raise DisconnectedError("{}: not connected".format(self._what)) return False @@ -199,6 +209,8 @@ def _irq(event, data): aconn = _AioConnection._connected.get(conn_handle) if mtu and aconn is not None and not aconn.mtu: aconn.mtu = mtu + elif event == _IRQ_PERIPHERAL_DISCONNECT: + _early_mtu.pop(data[0], None) elif event == _IRQ_CENTRAL_DISCONNECT: conn_handle = data[0] _early_mtu.pop(conn_handle, None) @@ -273,9 +285,24 @@ def capabilities(self): max_mtu=_MAX_MTU, indicate=True, pairing=True, + long_read=False, ) return caps + def enable_bonding(self, path=None): + """Keep pairing keys in ``path`` (aioble's ``ble_secrets.json``) and bond from now on. + + Call it before any connection that may pair, on both sides: the side + that doesn't start pairing still has to store the keys. The file + doesn't survive an erase of the filesystem, and a peer that still + holds its half of the bond will then refuse to reconnect encrypted + until it's paired again. + """ + from aioble import security + + security.load_secrets(path) + aioble.config(bond=True, le_secure=True, mitm=False, io=3) + def _config_mtu(self): try: return aioble.config("mtu") @@ -353,6 +380,10 @@ def register_services(self, *services): initial=c._initial, max_len=c.max_len, ) + if c.encrypted: + achar.flags |= _FLAG_READ_ENCRYPTED | _FLAG_WRITE_ENCRYPTED + for d in c.descriptors: + aioble.Descriptor(achar, _buuid(d.uuid), read=True, initial=d.value) c._impl = _MPServerCharacteristic(self, c, achar) aservices.append(aservice) aioble.register_services(*aservices) @@ -466,7 +497,15 @@ def __init__(self, ble, device, role, aconn): @property def mtu(self): - return self._aconn.mtu or DEFAULT_MTU + a = self._aconn + if not a.mtu and a._conn_handle is not None: + # The same race as a central's early exchange, with this board as + # the central: a peripheral that exchanges at once (Windows, + # through bless) is heard before aioble records the connection. + early = _early_mtu.pop(a._conn_handle, None) + if early: + a.mtu = early + return a.mtu or DEFAULT_MTU def is_connected(self): return self._aconn.is_connected() @@ -480,8 +519,28 @@ async def disconnected(self, timeout_ms=None): 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) + try: + with _Translate("exchange_mtu"): + return await self._aconn.exchange_mtu(mtu, timeout_ms) + except OSError as e: + # NimBLE exchanges the MTU once per link. When the peer already + # did (Windows as a peripheral through bless does, at once), a + # second exchange is EALREADY: the MTU is settled, so return it. + if _errno(e) != _EALREADY: + raise + return self.mtu + + @property + def encrypted(self): + return bool(self._aconn.encrypted) + + async def pair(self, bond=True, timeout_ms=20000): + if not self.is_connected(): + raise DisconnectedError("pair: not connected") + with _Translate("pair"): + await self._aconn.pair(bond=bond, le_secure=True, mitm=False, io=3, timeout_ms=timeout_ms) + if not self._aconn.encrypted: + raise BLEError("pair: the link is not encrypted") async def _discover_services(self, uuid, timeout_ms): found = [] @@ -533,6 +592,17 @@ async def read(self, timeout_ms=1000): with _Translate("read"): return bytes(await self._achar.read(timeout_ms)) + async def _discover_descriptors(self, uuid, timeout_ms): + self._live("descriptor discovery") + found = [] + with _Translate("descriptor discovery"): + async for adesc in self._achar.descriptors(timeout_ms): + d = _MPClientDescriptor(self, adesc) + if uuid is None or d.uuid == uuid: + found.append(d) + found.sort(key=lambda d: d._adesc._value_handle) + return found + async def write(self, data, response=None, timeout_ms=1000): data = bytes(data) response = self._want_response(response) @@ -604,6 +674,18 @@ async def indicated(self, timeout_ms=None): return await self._next(a._indicate_queue, a._indicate_event, timeout_ms, "indicated") +class _MPClientDescriptor(ClientDescriptor): + def __init__(self, characteristic, adesc): + ClientDescriptor.__init__(self, characteristic, _uuid(adesc.uuid)) + self._adesc = adesc + + async def read(self, timeout_ms=1000): + if not self.connection.is_connected(): + raise DisconnectedError("read: not connected") + with _Translate("descriptor read"): + return bytes(await self._adesc.read(timeout_ms)) + + class _MPServerCharacteristic: """mpble's half of a :class:`bledev.Characteristic`.""" diff --git a/pydevices-desktop.toml b/pydevices-desktop.toml index a1f8be06..0e353dfa 100644 --- a/pydevices-desktop.toml +++ b/pydevices-desktop.toml @@ -24,6 +24,8 @@ "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/bledev/auto.py" = "/lib/bledev/auto.py" "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/bledev/bleak.py" = "/lib/bledev/bleak.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/hid.py" = "/lib/bledev/hid.py" +"https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/bledev/hidreport.py" = "/lib/bledev/hidreport.py" "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/bledev/improv.py" = "/lib/bledev/improv.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" diff --git a/tests/bledev_board/bless_peripheral.py b/tests/bledev_board/bless_peripheral.py new file mode 100644 index 00000000..8f39c3cf --- /dev/null +++ b/tests/bledev_board/bless_peripheral.py @@ -0,0 +1,86 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""Can the laptop be a BLE peripheral? A bless GATT server for a board to probe. + +bless (https://github.com/kevincar/bless) serves GATT from CPython. This +serves one service with a readable, writable, notifying characteristic, for +``bless_probe.py`` on a board to find, read, write and subscribe to. Run it +with a Windows Python that has bless, kept out of mpftp's sidecar Python. +On Python 3.14 (2026-09) PyPI's bless 0.3.0 won't install (it pins winrt +2.0.0b1), so this took bless's GitHub master with ``--no-deps``, then +``coloredlogs`` and pysetupdi's GitHub master by hand; the shim below covers +a winrt method bless still calls by its old name, and one more below for its +first subscription:: + + python.exe -m venv %USERPROFILE%\\bless-venv + ...\\Scripts\\python.exe -m pip install bleak pywin32 coloredlogs winrt-Windows.Devices.Bluetooth ... + ...\\Scripts\\python.exe -m pip install --no-deps https://github.com/kevincar/bless/archive/refs/heads/master.zip + ...\\Scripts\\python.exe -m pip install https://github.com/gwangyi/pysetupdi/archive/refs/heads/master.zip + ...\\Scripts\\python.exe bless_peripheral.py + +It prints each read and write, notifies ``tick N`` every second, and stops +after ``SECONDS``. +""" +import asyncio +import sys + +from bless import BlessServer, GATTAttributePermissions, GATTCharacteristicProperties + +SERVICE = "8a1f0000-2b4d-4c5e-9f60-1a2b3c4d5e6f" +CHAR = "8a1f0001-2b4d-4c5e-9f60-1a2b3c4d5e6f" +SECONDS = int(sys.argv[1]) if len(sys.argv) > 1 else 90 + + +def on_read(characteristic, *args, **kwargs): + print("read", bytes(characteristic.value), flush=True) + return characteristic.value + + +def on_write(characteristic, value, *args, **kwargs): + characteristic.value = value + print("write", bytes(value), flush=True) + + +class _Provider: + """bless (master, 2026-09) calls ``start_advertising(parameters)``, an + overload pywinrt 3.x names ``start_advertising_with_parameters``.""" + + def __init__(self, native): + self._native = native + + def start_advertising(self, parameters=None): + if parameters is None: + return self._native.start_advertising() + return self._native.start_advertising_with_parameters(parameters) + + def __getattr__(self, name): + return getattr(self._native, name) + + +async def main(): + server = BlessServer(name="bless-check") + server.read_request_func = on_read + server.write_request_func = on_write + await server.add_new_service(SERVICE) + props = GATTCharacteristicProperties.read | GATTCharacteristicProperties.write | GATTCharacteristicProperties.notify + perms = GATTAttributePermissions.readable | GATTAttributePermissions.writable + await server.add_new_characteristic(SERVICE, CHAR, props, bytearray(b"hello from bless"), perms) + for service in server.services.values(): + native = getattr(service, "service_provider", None) + if native is not None and not hasattr(native, "_native"): + service.service_provider = _Provider(native) + await server.start() + # bless master's first subscription indexes this dict before it has the key. + if hasattr(server, "_subscribed_clients"): + server._subscribed_clients.setdefault(CHAR, set()) + print("advertising", SERVICE, flush=True) + for n in range(SECONDS): + await asyncio.sleep(1) + server.get_characteristic(CHAR).value = bytearray(b"tick %d" % n) + server.update_value(SERVICE, CHAR) + await server.stop() + print("stopped", flush=True) + + +asyncio.run(main()) diff --git a/tests/bledev_board/bless_probe.py b/tests/bledev_board/bless_probe.py new file mode 100644 index 00000000..1f49a7ce --- /dev/null +++ b/tests/bledev_board/bless_probe.py @@ -0,0 +1,65 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""A board probes the laptop's bless peripheral (``bless_peripheral.py``). + +Finds the service by UUID (Windows advertises the service, not a name), +connects, reads, writes, and waits for two notifications. Logs to +/bless_probe.log and ends with ``RESULT PASS`` or ``RESULT FAIL``. +""" +import asyncio +import bledev.mpble + +SERVICE = "8a1f0000-2b4d-4c5e-9f60-1a2b3c4d5e6f" +CHAR = "8a1f0001-2b4d-4c5e-9f60-1a2b3c4d5e6f" +LOG = "/bless_probe.log" + + +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 main(): + ok = False + ble = bledev.mpble.get() + try: + device = await ble.find(service=SERVICE, timeout_ms=30000) + log("found", device) + conn = None + for attempt in range(3): + await asyncio.sleep_ms(500) + try: + conn = await device.connect(timeout_ms=15000) + break + except OSError as e: + log("connect attempt", attempt, "failed:", e) + if conn is None: + raise RuntimeError("could not connect") + log("connected, mtu", await conn.exchange_mtu(247)) + svc = await conn.service(SERVICE) + log("service", svc) + ch = await svc.characteristic(CHAR) + log("characteristic", ch) + log("read", await ch.read()) + await ch.write(b"hi from the board", response=True) + log("wrote") + log("read back", await ch.read()) + await ch.subscribe(notify=True) + for _ in range(2): + log("notified", await ch.notified(5000)) + ok = True + await conn.disconnect() + except Exception as e: + log("error", type(e).__name__, e) + log("RESULT", "PASS" if ok else "FAIL") + + +try: + import os + os.remove(LOG) +except OSError: + pass +asyncio.run(main()) diff --git a/tests/bledev_board/burst_sink.py b/tests/bledev_board/burst_sink.py new file mode 100644 index 00000000..71aea80a --- /dev/null +++ b/tests/bledev_board/burst_sink.py @@ -0,0 +1,47 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""Notification bursts, the sink: asks burst_source.py for bursts and counts what arrives. + +``WORK_US`` busies Python for that long per packet, which is what made the P4 +lose packets and the S3 not (pydevices#87). ``FAST`` asks for a 7.5-15 ms +interval. Logs to /burst_cen.log. +""" +import asyncio, struct, time +import bledev, bledev.mpble +SVC = bledev.UUID("b1ed1000-4849-4400-8000-00805f9b34fb") +CH = bledev.UUID("b1ed1001-4849-4400-8000-00805f9b34fb") +CTL = bledev.UUID("b1ed1002-4849-4400-8000-00805f9b34fb") +FAST = True +WORK_US = 0 +LOG = "/burst_cen.log" +def log(*a): + line = " ".join(str(x) for x in a); print(line) + with open(LOG, "a") as f: f.write(line + "\n") +async def main(): + ble = bledev.mpble.get() + dev = await ble.find(name="burst-src", timeout_ms=20000) + kw = {"min_conn_interval_us": 7500, "max_conn_interval_us": 15000} if FAST else {} + conn = await dev.connect(timeout_ms=10000, **kw) + await conn.exchange_mtu(247) + s = await conn.service(SVC) + ch = await s.characteristic(CH); ctl = await s.characteristic(CTL) + await ch.subscribe() + for n, gap in ((20, 0), (40, 0), (80, 0), (200, 0), (200, 2), (200, 5), (200, 10)): + await ctl.write(struct.pack("= 0 else (ch, False) + + +def _code(usage): + # The keycode for the usages this script uses. + for ch, u in _USAGE.items(): + if u == usage: + return {"\n": keys.K_RETURN, "\t": keys.K_TAB, " ": keys.K_SPACE}.get(ch, ord(ch)) + return {0x3E: keys.K_F5}[usage] + + +_MODS = ( # report bit -> (KMOD, keycode, usage) + (0x01, keys.KMOD_LCTRL, keys.K_LCTRL, 0xE0), (0x02, keys.KMOD_LSHIFT, keys.K_LSHIFT, 0xE1), + (0x04, keys.KMOD_LALT, keys.K_LALT, 0xE2), (0x08, keys.KMOD_LGUI, keys.K_LGUI, 0xE3), + (0x10, keys.KMOD_RCTRL, keys.K_RCTRL, 0xE4), (0x20, keys.KMOD_RSHIFT, keys.K_RSHIFT, 0xE5), + (0x40, keys.KMOD_RALT, keys.K_RALT, 0xE6), (0x80, keys.KMOD_RGUI, keys.K_RGUI, 0xE7), +) + + +def _key(kind, code, mod, scancode): + return events.Key(kind, keys.keyname(code) if code is not None else "Unknown", code, mod, scancode, None) + + +def _chord(bits, usage): + kmod = 0 + for bit, km, _, _ in _MODS: + if bits & bit: + kmod |= km + code = _code(usage) + # Down: the key, then its modifiers. Up: the modifiers, then the key. + out = [_key(events.KEYDOWN, code, kmod, usage)] + out += [_key(events.KEYDOWN, kc, kmod, u) for bit, _, kc, u in _MODS if bits & bit] + out += [_key(events.KEYUP, kc, 0, u) for bit, _, kc, u in _MODS if bits & bit] + out.append(_key(events.KEYUP, code, 0, usage)) + return out + + +_HAT = ((0, 1), (1, 1), (1, 0), (1, -1), (0, -1), (-1, -1), (-1, 0), (-1, 1)) +CONSUMER_KEYS = {0xE9: keys.K_VOLUMEUP, 0xCD: keys.K_AUDIOPLAY} + + +def expected(steps=STEPS, instance_id=0): + out = [] + axes = [0.0] * 4 + buttons = 0 + hat = (0, 0) + for step in steps: + kind = step[0] + if kind == "type": + for ch in step[1]: + plain, shift = _plain(ch) + out += _chord(0x02 if shift else 0, _USAGE[plain]) + elif kind == "chord": + out += _chord(step[1], step[2]) + elif kind == "consumer": + u = step[1] + out.append(_key(events.KEYDOWN, CONSUMER_KEYS[u], 0, 0x000C0000 | u)) + out.append(_key(events.KEYUP, CONSUMER_KEYS[u], 0, 0x000C0000 | u)) + elif kind == "pad": + new_axes, new_buttons, new_hat = step[1], step[2], step[3] + for i in range(4): + v = 2.0 * (new_axes[i] + 127) / 254 - 1.0 + if v != axes[i]: + axes[i] = v + out.append(events.JoyAxisMotion(events.JOYAXISMOTION, instance_id, i, v)) + h = (0, 0) if new_hat is None else _HAT[new_hat] + if h != hat: + hat = h + out.append(events.JoyHatMotion(events.JOYHATMOTION, instance_id, 0, h)) + for b in range(16): + if buttons & (1 << b) and not new_buttons & (1 << b): + out.append(events.JoyButtonUp(events.JOYBUTTONUP, instance_id, b)) + for b in range(16): + if new_buttons & (1 << b) and not buttons & (1 << b): + out.append(events.JoyButtonDown(events.JOYBUTTONDOWN, instance_id, b)) + buttons = new_buttons + return out + + +def typed_text(evs): + """What the key events spell, reading Shift the way a US keyboard does.""" + text = "" + for e in evs: + if e.type != events.KEYDOWN or e.key is None: + continue + if e.mod & ~keys.KMOD_SHIFT: + continue # a chord, not typing + if e.key == keys.K_RETURN: + ch = "\n" + elif e.key == keys.K_TAB: + ch = "\t" + elif 32 <= e.key < 127: + ch = chr(e.key) + else: + continue + if e.mod & keys.KMOD_SHIFT: + i = _UNSHIFTED.find(ch) + ch = _SHIFTED[i] if i >= 0 else ch + text += ch + return text + + +async def perform(peripheral, steps=STEPS, gap_ms=0, sleep_ms=None): + """Send ``steps`` from a :class:`bledev.hid.Peripheral`.""" + for step in steps: + kind = step[0] + if kind == "type": + await peripheral.keyboard.type(step[1], gap_ms) + elif kind == "chord": + await peripheral.keyboard.tap(step[2], step[1]) + elif kind == "consumer": + await peripheral.consumer.tap(step[1]) + elif kind == "pad": + await peripheral.gamepad.send(axes=step[1], buttons=step[2], hat=step[3]) + if gap_ms and sleep_ms is not None: + await sleep_ms(gap_ms) + + +def compare(got, want): + """``None`` if equal, else a one-line description of the first difference.""" + for i in range(min(len(got), len(want))): + if got[i] != want[i]: + return "event {}: got {!r}, want {!r}".format(i, got[i], want[i]) + if len(got) != len(want): + return "got {} events, want {}".format(len(got), len(want)) + return None diff --git a/tests/bledev_hid_checks.py b/tests/bledev_hid_checks.py new file mode 100644 index 00000000..151419a8 --- /dev/null +++ b/tests/bledev_hid_checks.py @@ -0,0 +1,246 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""``bledev.hid`` end to end over ``bledev.fake``: a peripheral types, a host decodes. + +A plain script, like ``bledev_contract.py``, so it runs on CPython and +MicroPython:: + + python tests/bledev_hid_checks.py + micropython tests/bledev_hid_checks.py + python tests/bledev_hid_checks.py --plant=drop_report # must FAIL + +The scripted input and its expected events are ``bledev_board/hid_script.py``, +the same file the board-to-board gate uses. +""" + +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") +sys.path.insert(0, _dir + _sep + "bledev_board") + +try: + import asyncio +except ImportError: # pragma: no cover + import uasyncio as asyncio + +import bledev +from bledev import BLEError, GattError, INSUFFICIENT_ENCRYPTION, fake, hid, hidreport +from bledev.fake import Air, FakeBLE +from bledev.hidreport import INPUT, OUTPUT, ReportMap +import hid_script + +MICROPYTHON = sys.implementation.name == "micropython" +CHECK_TIMEOUT_S = 10 + + +class Failed(Exception): + pass + + +def equal(got, want, what): + if got != want: + raise Failed("{}: got {!r}, want {!r}".format(what, got, want)) + + +async def _pair_up(air, *, encrypted=False, central_kw=None, peripheral_kw=None, **pkw): + board = FakeBLE(air=air, mtu=247, **(peripheral_kw or {})) + host_ble = FakeBLE(air=air, mtu=247, **(central_kw or {})) + per = hid.Peripheral(board, encrypted=encrypted, **pkw) + serving = asyncio.create_task(per.serve()) + host = await hid.connect(host_ble, name=per.name, timeout_ms=2000) + await serving + return board, host_ble, per, host + + +async def _collect(host, n, timeout_ms=3000): + got = [] + while len(got) < n: + got += await host.events(timeout_ms) + return got + + +async def c_served_map_parses(air): + # Our own map, by hand: keyboard 8 bytes, consumer 2, gamepad 16 + 4 + 4 + 32 bits = 7. + m = ReportMap(hid.report_map()) + equal(m.applications, [0x00010006, 0x000C0001, 0x00010005], "applications") + equal([(r.report_id, r.size) for r in m.inputs()], [(1, 8), (2, 2), (3, 7)], "input reports") + equal(m.report(OUTPUT, 1).size, 1, "LED report") + equal(hid.identity(), ("PyDevices Keyboard+Gamepad", 0x03C0, 7), "identity, all three") + equal(hid.identity(True, False, False), ("PyDevices Keyboard", 0x03C1, 1), "identity, keyboard") + equal(hid.identity(False, False, True), ("PyDevices Gamepad", 0x03C4, 4), "identity, gamepad") + equal(hid.identity(False, True, False), ("PyDevices Remote", 0x03C0, 2), "identity, remote") + + +async def c_script_round_trip(air): + board, host_ble, per, host = await _pair_up(air) + want = hid_script.expected() + await per.leds(2000) # the host writes the LEDs once it has subscribed + await hid_script.perform(per) + got = await _collect(host, len(want)) + problem = hid_script.compare(got, want) + if problem: + raise Failed(problem) + equal(hid_script.typed_text(got), hid_script.TEXT, "typed text") + await asyncio.sleep(0) + equal(host.poll(), [], "nothing extra") + await host.close() + + +async def c_encrypted_pairs_and_bonds(air): + board, host_ble, per, host = await _pair_up(air, encrypted=True) + equal(host.paired, True, "the host had to pair") + check_bond = board.address in host_ble.bonds and host_ble.address in board.bonds + equal(check_bond, True, "both sides bonded") + await per.keyboard.tap(0x04) + got = await _collect(host, 2) + equal([e.key for e in got], [ord("a"), ord("a")], "a key through an encrypted link") + await host.close() + + +async def c_unpaired_read_refused(air): + board = FakeBLE(air=air, mtu=247) + other = FakeBLE(air=air, mtu=247) + per = hid.Peripheral(board, encrypted=True) + serving = asyncio.create_task(per.serve()) + device = await other.find(name=per.name, timeout_ms=2000) + conn = await device.connect() + await serving + svc = await conn.service(hid.HID_SERVICE) + rmap = await svc.characteristic(hid.REPORT_MAP) + try: + await rmap.read() + except GattError as e: + equal(e.status, INSUFFICIENT_ENCRYPTION, "status") + else: + raise Failed("an encrypted Report Map was read without pairing") + await conn.disconnect() + + +async def c_short_read_host_says_why(air): + # MicroPython reads one packet. At MTU 23 the map (162 bytes) is cut to + # 22, and the host must say so instead of decoding garbage. + board = FakeBLE(air=air, mtu=23) + host_ble = FakeBLE(air=air, mtu=23, long_reads=False) + per = hid.Peripheral(board) + serving = asyncio.create_task(per.serve()) + try: + await hid.connect(host_ble, name=per.name, timeout_ms=2000) + except BLEError as e: + if "long read" not in str(e): + raise Failed("wrong error: {}".format(e)) + else: + raise Failed("a truncated Report Map was accepted") + await serving + + +async def c_map_fits_one_packet_at_247(air): + # The same host at MTU 247 reads the whole map in one packet. + board = FakeBLE(air=air, mtu=247) + host_ble = FakeBLE(air=air, mtu=247, long_reads=False) + per = hid.Peripheral(board) + serving = asyncio.create_task(per.serve()) + host = await hid.connect(host_ble, name=per.name, timeout_ms=2000) + await serving + equal(host.report_map.descriptor, hid.report_map(), "whole map") + await host.close() + + +async def c_disconnect_ends_events(air): + board, host_ble, per, host = await _pair_up(air) + await per.connection.disconnect() + try: + await host.events(1000) + except bledev.DisconnectedError: + pass + else: + raise Failed("events() didn't end on a disconnect") + + +TESTS = [ + c_served_map_parses, + c_script_round_trip, + c_encrypted_pairs_and_bonds, + c_unpaired_read_refused, + c_short_read_host_says_why, + c_map_fits_one_packet_at_247, + c_disconnect_ends_events, +] + + +def _plant(name): + if name == "drop_report": + # The link loses one notification in twenty. + orig = fake._Inbox.put + count = [0] + + def put(self, item): + count[0] += 1 + if count[0] % 20 == 7: + self.event.set() + return + orig(self, item) + + fake._Inbox.put = put + elif name == "shift_lost": + # The peripheral forgets Shift. + orig = hid.Keyboard.send + + async def send(self, modifiers=0, keys=()): + await orig(self, modifiers & ~0x02, keys) + + hid.Keyboard.send = send + elif name == "axis_sign": + # The decoder reads axes unsigned. + orig = hidreport.Field.raw + + def raw(self, data, n=0): + v = orig(self, data, n) + return v & ((1 << self.size) - 1) + + hidreport.Field.raw = raw + elif name == "no_pair": + # The host never pairs. + async def pair(self, bond=True, timeout_ms=20000): + return None + + fake._FakeConnection.pair = pair + else: + raise SystemExit("unknown plant " + name) + + +PLANTS = ("drop_report", "shift_lost", "axis_sign", "no_pair") + + +def run(): + failed = 0 + for fn in TESTS: + 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) + else: + print("PASS", fn.__name__) + print("{} checks, {} failed ({})".format(len(TESTS), failed, sys.implementation.name)) + return failed + + +def main(argv): + for arg in argv: + if arg.startswith("--plant="): + _plant(arg.split("=", 1)[1]) + print("PLANTED", arg.split("=", 1)[1]) + sys.exit(1 if run() else 0) + + +if __name__ == "__main__": + main(sys.argv[1:]) diff --git a/tests/hidreport_vectors.py b/tests/hidreport_vectors.py new file mode 100644 index 00000000..4e838e01 --- /dev/null +++ b/tests/hidreport_vectors.py @@ -0,0 +1,485 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""Test vectors for ``bledev.hidreport``: real report descriptors, real reports. + +A plain script, so it runs on CPython and MicroPython alike:: + + python tests/hidreport_vectors.py + micropython tests/hidreport_vectors.py + python tests/hidreport_vectors.py --plant=bitorder # must FAIL + +Every expected value below was worked out by hand from the descriptor's own +item list (the comments show the arithmetic), not by running the parser, so a +parser bug can't write its own answer key. The descriptors are shipped by real +devices and libraries: + +* ``BOOT_KEYBOARD``: the boot keyboard from the HID 1.11 specification, + appendix E.6, which every USB and BLE boot keyboard follows. +* ``ESP32_BLE_KEYBOARD``: T-vK/ESP32-BLE-Keyboard's report map, keyboard + (report 1) plus a 16-bit consumer bitmap (report 2); a common BLE remote. +* ``CP_CONSUMER``: CircuitPython's ``usb_hid`` consumer control, a 16-bit + array (``shared-module/usb_hid/report_descriptors.c``). +* ``ADAFRUIT_BLE``: adafruit_ble's default BLE HID map: keyboard, mouse, + consumer control. +* ``ADAFRUIT_GAMEPAD``: the gamepad adafruit_ble carries (commented out) in + the same file: 16 buttons and four signed 8-bit axes, report 5. +* ``XBOX_1914``: the Xbox Series controller (model 1914) over Bluetooth LE, as + dumped in github.com/DJm00n/ControllersInfo. 283 bytes, which is more than + one ATT read carries at MTU 247. +""" + +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") + +from binascii import unhexlify + +import events +import keys +from bledev import hidreport +from bledev.hidreport import INPUT, OUTPUT, Decoder, HIDParseError, ReportMap + +MICROPYTHON = sys.implementation.name == "micropython" + + +def _h(text): + return unhexlify("".join(text.split())) + + +BOOT_KEYBOARD = _h(""" +05 01 09 06 a1 01 05 07 19 e0 29 e7 15 00 25 01 75 01 95 08 81 02 95 01 +75 08 81 01 95 05 75 01 05 08 19 01 29 05 91 02 95 01 75 03 91 01 95 06 +75 08 15 00 25 65 05 07 19 00 29 65 81 00 c0 +""") + +ESP32_BLE_KEYBOARD = _h(""" +05010906a1018501050719e029e71500250175019508810295017508810195057501050819 +012905910295017503910195067508150025650507190029658100c0050c0901a101850205 +0c150025017501951009b509b609b709cd09e209e909ea0a23020a94010a92010a2a020a21 +020a26020a24020a83010a8a018102c0 +""") + +CP_CONSUMER = _h("05 0c 09 01 a1 01 85 03 75 10 95 01 15 01 26 8c 02 19 01 2a 8c 02 81 00 c0") + +ADAFRUIT_BLE = _h(""" +05010906a1018501050719e029e715002501750195088102810119002989150025897508 +950681000508190129051500250175019505910295039101c005010902a1010901a10085 +02050919012905150025019505750181029501750381010501093009311581257f750895 +02810609381581257f750895018106c0c0050c0901a1018503751095011501268c021901 +2a8c028100c0 +""") + +ADAFRUIT_GAMEPAD = _h(""" +05 01 09 05 a1 01 85 05 05 09 19 01 29 10 15 00 25 01 75 01 95 10 81 02 +05 01 15 81 25 7f 09 30 09 31 09 32 09 35 75 08 95 04 81 02 c0 +""") + +XBOX_1914 = _h(""" +05010905a10185010901a10009300931150027ffff0000950275108102c00901a1000932 +0935150027ffff0000950275108102c0050209c5150026ff039501750a81021500250075 +0695018103050209c4150026ff039501750a8102150025007506950181030501093915012508 +3500463b016614007504950181427504950115002500350045006500810305091901290f15 +0025017501950f810215002500750195018103050c0ab2001500250195017501810215002500 +750795018103050f09218503a102099715002501750495019102150025007504950191030970 +150025647508950491020950660110550e150026ff0075089501910209a7150026ff00750895 +01910265005500097c150026ff00750895019102c0c0 +""") + +KD, KU = events.KEYDOWN, events.KEYUP +LSHIFT = keys.KMOD_LSHIFT + + +def key(kind, code, mod, scancode): + return events.Key(kind, keys.keyname(code) if code is not None else "Unknown", code, mod, scancode, None) + + +class Failed(Exception): + pass + + +def check(cond, what): + if not cond: + raise Failed(what) + + +def equal(got, want, what): + if got != want: + raise Failed("{}: got {!r}, want {!r}".format(what, got, want)) + + +# ---------------------------------------------------------------- the parser + + +def t_boot_keyboard_layout(): + m = ReportMap(BOOT_KEYBOARD) + equal(len(BOOT_KEYBOARD), 63, "spec descriptor length") + equal(m.numbered, False, "numbered") + equal(m.applications, [0x00010006], "applications") + r = m.report(INPUT, 0) + # 8 modifier bits + 8 constant + 6 x 8 keys = 64 bits. + equal(r.size, 8, "input size") + equal(len(r.fields), 3, "input fields") + mods, pad, arr = r.fields + equal((mods.offset, mods.size, mods.count, mods.variable), (0, 1, 8, True), "modifier field") + equal(mods.usages, [(0x000700E0, 0x000700E7)], "modifier usages") + equal((pad.offset, pad.size, pad.constant), (8, 8, True), "reserved byte") + equal((arr.offset, arr.size, arr.count, arr.variable), (16, 8, 6, False), "key array") + equal((arr.logical_min, arr.logical_max), (0, 0x65), "key array range") + equal(arr.usage(4), 0x00070004, "array value 4 is usage 'a'") + o = m.report(OUTPUT, 0) + # 5 LED bits + 3 padding. + equal(o.size, 1, "LED report size") + equal(o.fields[0].usages, [(0x00080001, 0x00080005)], "LED usages") + + +def t_esp32_ble_keyboard_layout(): + m = ReportMap(ESP32_BLE_KEYBOARD) + equal(m.numbered, True, "numbered") + equal(m.applications, [0x00010006, 0x000C0001], "applications") + equal(m.report(INPUT, 1).size, 8, "keyboard report") + c = m.report(INPUT, 2) + equal(c.size, 2, "media report: 16 x 1 bit") + f = c.fields[0] + # USAGE (Volume Increment) is the sixth usage, bit 5; a 2-byte usage + # (0A 23 02) is 0x223 on the consumer page, bit 7. + equal(f.usage(5), 0x000C00E9, "bit 5") + equal(f.usage(7), 0x000C0223, "bit 7") + equal(f.usage(15), 0x000C018A, "bit 15") + + +def t_cp_consumer_layout(): + m = ReportMap(CP_CONSUMER) + r = m.report(INPUT, 3) + equal(r.size, 2, "one 16-bit value") + f = r.fields[0] + equal((f.logical_min, f.logical_max, f.variable), (1, 652, False), "array range") + # Usage Minimum 1: value 1 is usage 1, so value 0xE9 is usage 0xE9. + equal(f.usage(0xE9 - 1), 0x000C00E9, "array index") + + +def t_adafruit_layout(): + m = ReportMap(ADAFRUIT_BLE) + equal(m.applications, [0x00010006, 0x00010002, 0x000C0001], "applications") + equal(m.report(INPUT, 1).size, 8, "keyboard") + # 5 buttons + 3 pad + X, Y (8 bit each) + wheel = 4 bytes. + mouse = m.report(INPUT, 2) + equal(mouse.size, 4, "mouse") + equal((mouse.fields[2].logical_min, mouse.fields[2].logical_max), (-127, 127), "signed mouse axes") + check(mouse.fields[2].relative, "mouse X/Y are relative") + equal(m.report(INPUT, 3).size, 2, "consumer") + # X = 1, Y = 0xFB (-5), wheel 0, after the button byte. + equal(mouse.fields[2].values(bytes([0x01, 0x01, 0xFB, 0x00])), [1, -5], "signed X, Y") + + +def t_xbox_layout(): + m = ReportMap(XBOX_1914) + equal(len(XBOX_1914), 283, "dump length") + equal(m.numbered, True, "numbered") + equal(m.applications, [0x00010005], "one gamepad collection") + r = m.report(INPUT, 1) + # 4 x 16 + (10 + 6) + (10 + 6) + (4 + 4) + (15 + 1) + (1 + 7) = 128 bits. + equal(r.size, 16, "input report size") + xy = r.fields[0] + equal((xy.offset, xy.size, xy.count, xy.logical_max), (0, 16, 2, 65535), "X and Y") + brake = r.fields[2] + equal((brake.offset, brake.size, brake.usages), (64, 10, [0x000200C5]), "brake") + hat = r.fields[6] + equal((hat.offset, hat.size, hat.logical_min, hat.logical_max), (96, 4, 1, 8), "hat") + equal((hat.physical_max, hat.unit), (315, 0x14), "hat physical range and unit") + buttons = r.fields[8] + # The hat is followed by 4 bits of padding, so buttons start at bit 104. + equal((buttons.offset, buttons.count), (104, 15), "buttons") + # Output 3: 4 + 4 bits, 4 magnitudes, duration, start delay, loop count. + o = m.report(OUTPUT, 3) + equal(o.size, 8, "rumble report") + dur = [f for f in o.fields if f.usages == [0x000F0050]][0] + equal((dur.unit, dur.unit_exponent), (0x1001, -2), "duration unit, 10^-2 s") + + +def t_long_and_signed_items(): + # 26 FF FF with 15 00 is 65535, not -1; 25 FF with 15 00 is 255. + m = ReportMap(_h("05 01 09 04 a1 01 15 00 26 ff ff 75 10 95 01 09 30 81 02 25 ff 75 08 09 31 81 02 c0")) + f = m.report(INPUT, 0).fields + equal((f[0].logical_max, f[1].logical_max), (65535, 255), "unsigned maxima") + # A long item (FE) is skipped whole. + m = ReportMap(_h("fe 02 00 aa bb") + BOOT_KEYBOARD) + equal(m.report(INPUT, 0).size, 8, "after a long item") + # Push/Pop restore the globals. + m = ReportMap(_h("05 01 09 04 a1 01 75 08 95 01 a4 75 10 09 30 81 02 b4 09 31 81 02 c0")) + f = m.report(INPUT, 0).fields + equal((f[0].size, f[1].size, f[1].offset), (16, 8, 16), "push/pop") + + +def t_malformed(): + for text, what in ( + ("05 01 09 06 a1 01 75", "truncated item"), + ("05 01 09 06 a1 01", "open collection"), + ("c0", "unbalanced end"), + ("85 00", "report ID 0"), + ("b4", "pop without push"), + ): + try: + ReportMap(_h(text)) + except HIDParseError: + continue + raise Failed("accepted a malformed descriptor: " + what) + + +# ---------------------------------------------------------------- reports to events + + +def t_boot_keyboard_events(): + d = Decoder(ReportMap(BOOT_KEYBOARD)) + # "H": Left Shift (bit 1) and usage 0x0B ('h'). The key goes down before + # the modifier, as usbif and SDL order a chord. + equal(d.feed(bytes([0x02, 0, 0x0B, 0, 0, 0, 0, 0])), ( + key(KD, keys.K_h, LSHIFT, 0x0B), + key(KD, keys.K_LSHIFT, LSHIFT, 0xE1), + ), "shift+h") + # Release both: modifier up first, then the key. + equal(d.feed(bytes(8)), ( + key(KU, keys.K_LSHIFT, 0, 0xE1), + key(KU, keys.K_h, 0, 0x0B), + ), "release") + # "1" is usage 0x1E and "0" is 0x27, last in HID and first in ASCII. + equal(d.feed(bytes([0, 0, 0x27, 0x1E, 0, 0, 0, 0])), ( + key(KD, ord("1"), 0, 0x1E), + key(KD, ord("0"), 0, 0x27), + ), "digits, sorted by usage") + # Rollover: six ErrorRollOver slots are ignored and nothing is released. + equal(d.feed(bytes([0, 0, 1, 1, 1, 1, 1, 1])), (), "rollover") + equal(sorted(d.held), [0x1E, 0x27], "held after rollover") + # Enter (0x28), a function key (F5 = 0x3E) and an arrow (Up = 0x52). + equal(d.feed(bytes([0, 0, 0x28, 0x3E, 0x52, 0, 0, 0])), ( + key(KU, ord("1"), 0, 0x1E), + key(KU, ord("0"), 0, 0x27), + key(KD, keys.K_RETURN, 0, 0x28), + key(KD, keys.K_F5, 0, 0x3E), + key(KD, keys.K_UP, 0, 0x52), + ), "enter, F5, up") + # Right Alt is bit 6. + equal(d.feed(bytes([0x40, 0, 0, 0, 0, 0, 0, 0]))[-1], key(KD, keys.K_RALT, keys.KMOD_RALT, 0xE6), "right alt") + + +def t_esp32_media_keys(): + d = Decoder(ReportMap(ESP32_BLE_KEYBOARD)) + # Bit 5 of report 2 is Volume Increment. + equal(d.feed(bytes([0x20, 0x00]), report_id=2), (key(KD, keys.K_VOLUMEUP, 0, 0x000C00E9),), "volume up") + # Bit 3 (Play/Pause) instead: volume up is released first. + equal(d.feed(bytes([0x08, 0x00]), report_id=2), ( + key(KU, keys.K_VOLUMEUP, 0, 0x000C00E9), + key(KD, keys.K_AUDIOPLAY, 0, 0x000C00CD), + ), "play") + # Byte 1 bit 0 is My Computer (0x194); bit 7 is Mail (0x18A). + equal(d.feed(bytes([0x00, 0x81]), report_id=2), ( + key(KU, keys.K_AUDIOPLAY, 0, 0x000C00CD), + key(KD, keys.K_MAIL, 0, 0x000C018A), + key(KD, keys.K_COMPUTER, 0, 0x000C0194), + ), "computer and mail, sorted by usage") + # The keyboard report on the same map, HOGP-style (no ID byte). + equal(d.feed(bytes([0, 0, 0x04, 0, 0, 0, 0, 0]), report_id=1), (key(KD, keys.K_a, 0, 0x04),), "keyboard 'a'") + # A report ID the map doesn't have is ignored. + equal(d.feed(bytes([0xFF, 0xFF]), report_id=9), (), "unknown report id") + + +def t_cp_consumer_events(): + d = Decoder(ReportMap(CP_CONSUMER)) + # USB style: the ID leads the report. + equal(d.feed(bytes([0x03, 0xE9, 0x00])), (key(KD, keys.K_VOLUMEUP, 0, 0x000C00E9),), "volume up") + equal(d.feed(bytes([0x03, 0xEA, 0x00])), ( + key(KU, keys.K_VOLUMEUP, 0, 0x000C00E9), + key(KD, keys.K_VOLUMEDOWN, 0, 0x000C00EA), + ), "volume down") + # 0 is below Logical Minimum 1: nothing held. + equal(d.feed(bytes([0x03, 0x00, 0x00])), (key(KU, keys.K_VOLUMEDOWN, 0, 0x000C00EA),), "release") + # AC Send (0x28C) has no key constant. + equal(d.feed(bytes([0x03, 0x8C, 0x02])), (key(KD, None, 0, 0x000C028C),), "unmapped usage") + + +def t_adafruit_events(): + d = Decoder(ReportMap(ADAFRUIT_BLE)) + # Shift + 'a' through report 1, with the USB-style leading ID. + equal(d.feed(bytes([0x01, 0x02, 0x00, 0x04, 0, 0, 0, 0, 0])), ( + key(KD, keys.K_a, LSHIFT, 0x04), + key(KD, keys.K_LSHIFT, LSHIFT, 0xE1), + ), "shift+a") + # A mouse report is not a joystick: no events. + equal(d.feed(bytes([0x02, 0x01, 0x05, 0xFB, 0x00])), (), "mouse report") + + +def t_adafruit_gamepad_events(): + d = Decoder(ReportMap(ADAFRUIT_GAMEPAD)) + equal((d.num_axes, d.num_hats, d.num_buttons), (4, 0, 16), "counts") + # Buttons 2 and 9 (bits 1 and 8); X = 0x81 (-127), Y = 64, Z = 0, Rz = 127. + ev = d.feed(bytes([0x02, 0x01, 0x81, 0x40, 0x00, 0x7F]), report_id=5) + JA = events.JoyAxisMotion + equal(ev, ( + JA(events.JOYAXISMOTION, 0, 0, -1.0), + # Logical -127..127: (64 + 127) / 254 * 2 - 1 = 64 / 127. + JA(events.JOYAXISMOTION, 0, 1, 2.0 * (64 + 127) / 254 - 1.0), + # Z stays at 0.0, where every axis starts: no event. + JA(events.JOYAXISMOTION, 0, 3, 1.0), + events.JoyButtonDown(events.JOYBUTTONDOWN, 0, 1), + events.JoyButtonDown(events.JOYBUTTONDOWN, 0, 8), + ), "gamepad report") + + +def t_xbox_events(): + d = Decoder(ReportMap(XBOX_1914), instance_id=7) + equal((d.num_axes, d.num_hats, d.num_buttons), (6, 1, 15), "counts") + r = bytearray(16) + # X = 65535 (1.0), Y = 0 (-1.0), Z = 32768, Rz = 32767. + r[0:2] = b"\xff\xff" + r[4:6] = b"\x00\x80" + r[6:8] = b"\xff\x7f" + # Brake (bits 64..73) = 1023, accelerator (80..89) = 512. + r[8] = 0xFF + r[9] = 0x03 + r[10] = 0x00 + r[11] = 0x02 + # Hat (bits 96..99) = 3: logical 1 is north, so 3 is east. + r[12] = 0x03 + # Buttons 1 and 15 (bits 104 and 118), and Record (bit 120). + r[13] = 0x01 + r[14] = 0x40 + r[15] = 0x01 + ev = d.feed(bytes(r), report_id=1) + JA, JH = events.JoyAxisMotion, events.JoyHatMotion + # Axes by usage: X 0, Y 1, Z 2, Rz 3, accelerator 4, brake 5. + want = ( + key(KD, None, 0, 0x000C00B2), + JA(events.JOYAXISMOTION, 7, 0, 1.0), + JA(events.JOYAXISMOTION, 7, 1, -1.0), + JA(events.JOYAXISMOTION, 7, 2, 2.0 * 32768 / 65535 - 1.0), + JA(events.JOYAXISMOTION, 7, 3, 2.0 * 32767 / 65535 - 1.0), + JA(events.JOYAXISMOTION, 7, 4, 2.0 * 512 / 1023 - 1.0), + JA(events.JOYAXISMOTION, 7, 5, 1.0), + JH(events.JOYHATMOTION, 7, 0, (1, 0)), + events.JoyButtonDown(events.JOYBUTTONDOWN, 7, 0), + events.JoyButtonDown(events.JOYBUTTONDOWN, 7, 14), + ) + equal(ev, want, "first report") + # Same report again: nothing changed, nothing to say. + equal(d.feed(bytes(r), report_id=1), (), "unchanged") + # Hat to 0 (the null state: centred), button 1 released. + r[12] = 0x00 + r[13] = 0x00 + ev = d.feed(bytes(r), report_id=1) + equal(ev, ( + JH(events.JOYHATMOTION, 7, 0, (0, 0)), + events.JoyButtonUp(events.JOYBUTTONUP, 7, 0), + ), "hat centred, button up") + equal(d.axis(5), 1.0, "brake held") + + +TESTS = [ + t_boot_keyboard_layout, + t_esp32_ble_keyboard_layout, + t_cp_consumer_layout, + t_adafruit_layout, + t_xbox_layout, + t_long_and_signed_items, + t_malformed, + t_boot_keyboard_events, + t_esp32_media_keys, + t_cp_consumer_events, + t_adafruit_events, + t_adafruit_gamepad_events, + t_xbox_events, +] + + +# ---------------------------------------------------------------- plants + + +def _plant(name): + """Break the parser or the decoder on purpose; each must fail a check.""" + Field = hidreport.Field + if name == "bitorder": + # Forget the bit shift: every field that doesn't start on a byte + # boundary reads the wrong bits. + def raw(self, data, n=0): + bit = self.offset + n * self.size + first = bit >> 3 + last = (bit + self.size + 7) >> 3 + return int.from_bytes(bytes(data[first:last]), "little") & ((1 << self.size) - 1) + + Field.raw = raw + elif name == "unsigned": + # No sign extension: -127 reads as 129. + orig = Field.raw + + def raw(self, data, n=0): + lo = self.logical_min + self.logical_min = 0 + try: + return orig(self, data, n) + finally: + self.logical_min = lo + + Field.raw = raw + elif name == "range": + # Off by one in usage-range expansion. + orig = Field.usage + + def usage(self, n): + return orig(self, n + 1) + + Field.usage = usage + elif name == "order": + # Presses before releases. + orig = hidreport.Decoder._keyboard + + def keyboard(self, elements, data, out): + mark = len(out) + orig(self, elements, data, out) + out[mark:] = list(reversed(out[mark:])) + + hidreport.Decoder._keyboard = keyboard + elif name == "rollover": + hidreport.ERROR_ROLLOVER = -1 + elif name == "hat": + hidreport._HAT8 = hidreport._HAT8[1:] + hidreport._HAT8[:1] + else: + raise SystemExit("unknown plant " + name) + + +PLANTS = ("bitorder", "unsigned", "range", "order", "rollover", "hat") + + +def run(verbose=False): + failed = 0 + for fn in TESTS: + try: + fn() + except Exception as e: + failed += 1 + print("FAIL", fn.__name__, "-", type(e).__name__, e) + if verbose and MICROPYTHON: + sys.print_exception(e) + else: + print("PASS", fn.__name__) + print("{} checks, {} failed ({})".format(len(TESTS), failed, sys.implementation.name)) + return failed + + +def main(argv): + 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 + sys.exit(1 if run(verbose) else 0) + + +if __name__ == "__main__": + main(sys.argv[1:]) diff --git a/tests/test_bledev_hid.py b/tests/test_bledev_hid.py new file mode 100644 index 00000000..f986ef89 --- /dev/null +++ b/tests/test_bledev_hid.py @@ -0,0 +1,115 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""Run bledev.hid's two plain-script suites on CPython and MicroPython, with plants. + +``hidreport_vectors.py`` checks the report-descriptor parser and decoder +against real descriptors; ``bledev_hid_checks.py`` runs a HID peripheral and +host over the fake radio. Each runs under CPython, under a unix +``micropython`` when one is found (the same search as ``test_bledev.py``), and +once per planted fault, each of which must make it fail. + +When usbif's checkout sits beside this one, the keyboard decoder is also +compared with usbif's own boot-keyboard decoder over random reports, since +the promise is that a BLE keyboard and a USB one look the same to an app. +""" + +from pathlib import Path +import os +import random +import subprocess +import sys +import unittest + +_TESTS = Path(__file__).resolve().parent +sys.path.insert(0, str(_TESTS)) +sys.path.insert(0, str(_TESTS.parent / "lib")) + +from test_bledev import _micropython # noqa: E402 + +SUITES = { + "hidreport_vectors.py": ("bitorder", "unsigned", "range", "order", "rollover", "hat"), + "bledev_hid_checks.py": ("drop_report", "shift_lost", "axis_sign", "no_pair"), +} + + +def _run(interpreter, script, *args): + return subprocess.run( + [interpreter, str(_TESTS / script), *args], capture_output=True, text=True, timeout=600 + ) + + +class HidSuites(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): + for script in SUITES: + with self.subTest(script=script): + self.assertPasses(_run(sys.executable, script)) + + def test_micropython(self): + interpreter = _micropython() + if interpreter is None: + message = "no micropython binary: the bledev.hid suites 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) + for script in SUITES: + with self.subTest(script=script): + result = _run(interpreter, script) + self.assertPasses(result) + self.assertIn("(micropython)", result.stdout) + + def test_every_plant_fails(self): + for script, plants in SUITES.items(): + for plant in plants: + with self.subTest(script=script, plant=plant): + result = _run(sys.executable, script, "--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) + + +def _usbif_decoder(): + for parent in _TESTS.parents: + lib = parent.parent / "usbif" / "lib" + if (lib / "usbif" / "hid_keyboard.py").is_file(): + sys.path.insert(0, str(lib)) + try: + from usbif.hid_keyboard import KeyboardDecoder + except Exception: + return None + return KeyboardDecoder + return None + + +class SameAsUsbif(unittest.TestCase): + def test_boot_reports_decode_like_usbif(self): + usbif_decoder = _usbif_decoder() + if usbif_decoder is None: + self.skipTest("usbif's checkout isn't beside this one") + from bledev.hidreport import Decoder, ReportMap + from hidreport_vectors import BOOT_KEYBOARD + + ours = Decoder(ReportMap(BOOT_KEYBOARD)) + theirs = usbif_decoder() + rng = random.Random(1234) + usages = list(range(0x04, 0x53)) # the boot range usbif maps + events = 0 + for i in range(3000): + held = rng.sample(usages, rng.randint(0, 6)) + report = bytes([rng.randint(0, 255), 0] + held + [0] * (6 - len(held))) + if i % 97 == 0: + report = bytes([report[0], 0, 1, 1, 1, 1, 1, 1]) # rollover + a = ours.feed(report) + b = theirs.feed(report) + self.assertEqual(a, b, "report {} {}".format(i, report.hex())) + events += len(a) + self.assertGreater(events, 10000) + + +if __name__ == "__main__": + unittest.main()