diff --git a/docs/bledev-internals.md b/docs/bledev-internals.md index c80723e5..675f9e34 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. | +| `midi.py` | BLE-MIDI as a usbif-style MIDI port, both sides, built only on the contract. | +| `midi_codec.py` | The BLE-MIDI packet codec. Pure, no imports, shared by every host. | | `auto.py` | Picks a backend. Nothing imports it, and backends must not. | A backend module imports `bledev` and whatever its host provides, and nothing @@ -181,6 +183,27 @@ no difference to nus throughput. After a disconnect bleak sets `client.services` to `None`, so every client call checks the connection first and raises `DisconnectedError`. +## Reconnecting fast + +The REPL gate once failed one run in five: bleak couldn't subscribe on a +connection made 1.5 s after the last one closed. A loop of rapid reconnects +from the laptop (`tests/bledev_board/reconnect_loop.py`: log in, run a line, +close, wait 1.5 s) caught it at about one cycle in fifty, always the same +way. bleak's connect returned in 70-440 ms where a real one takes 1.7 s or +more, Windows reported the device connected and the GATT session active, the +board never saw a connection (its IRQ log shows it advertising throughout), +and the first GATT operation hung until Windows reported a drop about nine +seconds later. + +It went away with the REPL's output fixes (below): against `bledev-host`'s +`repl.py` the loop failed 7 times in 465 cycles; against the fixed one, 0 in +270, 150 of them with the retry switched off. We haven't found how the +board's old timer handling produced a connection Windows believed in and the +board never saw, so `bledev.connect_and_set_up()` stays as the defence: it +bounds the setup at 4 s and tries again, up to three times. `nus.connect()` +and `midi.connect()` use it, and the contract checks both the retry and the +giving up. + ## How the REPL gets there The REPL has to work when nothing runs asyncio, at the `>>>` prompt, so the @@ -190,9 +213,26 @@ with `gap_advertise`, and does everything else from an IRQ handler it adds to aioble's dispatcher (or sets itself, with no aioble). `os.dupterm` reads a `_Stream` whose `readinto` returns `None` when empty, because 0 means end of stream and detaches it. Output goes out as notifications of `mtu - 3` bytes, -and what the controller can't take yet is retried from a one-shot soft timer. -A program printing faster than the link carries waits in `write()` up to two -seconds, then drops the excess, because `print()` can't fail. +and what the controller can't take yet is retried every 5 ms by one periodic +`machine.Timer`. A program printing faster than the link carries waits in +`write()` up to two seconds, then drops the excess, because `print()` can't +fail. + +Two ways that retry went wrong, both found by printing a 16,890-character +list over the link (the REPL gate now does this): + +- **A doubled chunk and a lost one.** The timer's callback runs between the + main program's bytecodes, so it could land inside a `flush()` that + `print()` had started: both sent the same chunk, then both trimmed it, and + the next chunk never went out. Five prints out of five came back corrupt. + `flush()` now returns at once if it's already running. +- **A panic.** The retry used to be a one-shot timer re-armed on every write. + On the esp32, `Timer.init()` on a virtual timer whose alarm is being + dispatched clears its handler first, so `esp_timer`'s task on the other + core calls a NULL function (`InstrFetchProhibited` in + `timer_process_alarm`). It took the board down twice in about 45 long + prints. The timer is now started once and never re-armed; 60 long prints + afterwards, no panic and every byte intact. Ctrl-C works because the IRQ calls `os.dupterm_notify()` whenever a write contains 0x03; dupterm then reads it and raises `KeyboardInterrupt`, even in @@ -234,6 +274,8 @@ The serving scripts have a `PLANT` switch, and a planted run must fail: | `gatt_peripheral.py` + `gatt_latency.py` | Per-operation latency of reads and writes-with-response, with the stall count | | `nus_server.py` + `nus_client.py` | 16 KB up, down and echoed over nus, byte for byte, timed. On a laptop, `--fast` asks for throughput parameters and `--plant` flips a bit | | `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 | +| `midi_server.py` + `midi_client.py` | BLE-MIDI. `gate`: 1,000 mixed messages up, checked by the server, then down, checked by the client; `--plant` on the client and `PLANT = "drop"` or `"flip"` on the server must FAIL. `latency`: note-on round trips through an echo, `--fast` for a 7.5 ms interval (throughput parameters on a laptop). A board client reads its arguments from `/midi_client_args.py` | +| `reconnect_loop.py` (laptop) against `repl_server.py` | Rapid reconnects: log in, run a line, close, wait 1.5 s, N times; `--no-retry` turns `connect_and_set_up`'s retry 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 | @@ -323,3 +365,48 @@ One early run with BLE advertising read 8.1 KB/s, which looked like a finding until the repeats. Measuring BLE's cost to Wi-Fi needs the board at -60 dBm or better. +### BLE-MIDI + +The LCD-7 ran `midi_server.py` and the T-Embed or the laptop ran +`midi_client.py`, one desk apart, Wi-Fi off, bledev as `.py` source, +2026-09-24. + +**The gate** passed board to board at both intervals and from the laptop: +1,000 messages each way, complete, in order and intact, no decode errors, in +about 0.7 s up and 0.5 s down. The planted runs failed as they must: a bit +flipped in one packet going up (the server reported the first bad message, +546), one packet dropped going down (940 of 1,000 arrived), and a bit flipped +going down. + +**Latency**, half a note-on's round trip, each ping sent 0-20 ms after the +last echo so the phase against the connection events varies: + +| Link | Pings | Median | p90 | p99 | Max | +|---|---|---|---|---|---| +| S3 to S3, default interval (MicroPython asks 30-50 ms) | 400 notes | 44.3 ms | 44.3 | 49.8 | 69.3 | +| the same, four-note chords (to the last note) | 100 | 44.4 ms | 44.4 | 49.7 | 49.7 | +| S3 to S3, 7.5 ms interval | 400 notes | 10.8 ms | 13.6 | 15.9 | 21.1 | +| the same, chords | 100 | 10.9 ms | 13.7 | 18.8 | 18.8 | +| Laptop to S3, Windows' defaults | 400 notes | 52.3 ms | 60.0 | 110.7 | 118.1 | +| the same, chords | 100 | 52.1 ms | 59.8 | 105.7 | 105.7 | +| Laptop to S3, throughput parameters | 400 notes | 13.4 ms | 16.7 | 23.1 | 24.7 | +| the same, chords | 100 | 13.3 ms | 16.5 | 23.2 | 23.2 | + +No echo was lost in any run. **The codec's own cost** on the S3 (240 MHz, +`.py` source), splitting, encoding and decoding: 0.51 ms for a note, 1.27 ms +for a four-note chord, 4.9 ms for a 300-byte SysEx. A round trip crosses the +codec four times, so it accounts for about 1 ms of the 10.8 ms one way at +7.5 ms. What the rest splits into, connection events against asyncio's +scheduling on each board, wasn't measured: that needs a raw GATT echo at the +same interval to compare. + +**Windows** (2026-09-24, Windows 11 build 26200): once the board is paired, +Windows lists it as a MIDI device, `bledev-midi (Bluetooth MIDI IN)` and +`(Bluetooth MIDI OUT)`, found through the WinRT MIDI device interfaces. It +takes pairing: unpaired, nothing appears. winmm doesn't list it, and this +build's Windows MIDI Services has no Bluetooth transport among its +`Midi2.*Transport.dll`s, so a DAW that opens ports through winmm can't see +the board. Pairing needs the board to bond (`ble.config(bond=True, ...)` with +aioble's `security` module loaded); bledev doesn't do that for you yet. +Before `midi.serve()` set the GAP name, Windows called the port `MPY ESP32`. + diff --git a/docs/bledev.md b/docs/bledev.md index daa301f4..7c5eea0b 100644 --- a/docs/bledev.md +++ b/docs/bledev.md @@ -223,6 +223,52 @@ 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. +## MIDI + +`bledev.midi` is BLE-MIDI, the standard every Mac, iPhone, Android phone, +DAW and Bluetooth MIDI controller speaks. The board advertises it; anything +else connects: + +```python +import bledev.midi as midi + +port = await midi.serve(ble, name="synth") # on the board +port = await midi.connect(ble, name="synth") # on a laptop or another board +``` + +What you get is a MIDI port, the same one usbif gives you for a USB MIDI +function: `port.write(b"\x90\x3c\x64")` sends a note, and `port.read(buf)` +returns whatever MIDI bytes have arrived, without blocking. So the loop you +wrote for a USB controller works on a BLE one: + +```python +buf = bytearray(64) +parser = usbif.MidiParser() # running status, SysEx, clock +while True: + n = port.read(buf) + parser.feed(buf, n) + for status, data in parser.drain(): + ... + await asyncio.sleep(0.001) +``` + +When usbif is installed, `port` is a `usbif.MidiPort`. If you'd rather have +messages than bytes, with the sender's timestamp, use `ts, message = await +port.receive()` or `async for ts, message in port`. Messages you write go out +from a task, packed as many to a packet as fit, and `await port.drain()` waits +for them. + +On a board, ask for the shortest connection interval when you connect: +`midi.connect(ble, name="synth", min_conn_interval_us=7500, +max_conn_interval_us=7500)`. The latency that buys is in +[the measurements](#measured-on-two-esp32-s3s). + +The packet format lives in `bledev.midi_codec`, pure Python with no +imports: timestamps, running status, SysEx split across packets, and clock +slipped into a SysEx. Its test vectors (`tests/bledev_midi_vectors.json`) were +worked out by hand from the specification, and every interpreter checks +against the same file. + ## The rules every backend keeps These hold on every backend. The fake enforces the strict version of each, so @@ -310,6 +356,23 @@ two minutes. Turning the radio on costs about 64 KB of the S3's internal RAM; aioble and bledev together cost about 50 KB of Python heap. The details, and how these were measured, are in [bledev-internals.md](bledev-internals.md#measurements). +**MIDI.** How long a note takes to cross the link, one way, as half a +note-on's round trip through an echo (400 single notes at random moments, +then 100 four-note chords, which came out the same): + +| Link | Median | p90 | p99 | Max | +|---|---|---|---|---| +| S3 to S3, default interval | 44.3 ms | 44.3 | 49.8 | 69.3 | +| S3 to S3, 7.5 ms interval | 10.8 ms | 13.6 | 15.9 | 21.1 | +| Laptop (bleak) to S3, Windows' defaults | 52.3 ms | 60.0 | 110.7 | 118.1 | +| Laptop to S3, `priority="throughput"` | 13.4 ms | 16.7 | 23.1 | 24.7 | + +Players start to feel delay at about 10 ms, so a MIDI link wants the shortest +interval, and even then BLE alone uses up that budget. This is BLE's own +delay; the synth's comes on top. 1,000 mixed messages (notes, controllers, +pitch bend, a 300-byte SysEx every hundred, clock in between) crossed both +ways between the boards and to the laptop complete, in order and intact. + ## Writing a backend A backend subclasses the classes in `bledev/__init__.py` and fills in a few diff --git a/lib/bledev/__init__.py b/lib/bledev/__init__.py index 49d81a62..a5a8bac6 100644 --- a/lib/bledev/__init__.py +++ b/lib/bledev/__init__.py @@ -870,3 +870,38 @@ def _want_response(self, response): def is_adapter(obj): """True for a bledev :class:`BLE`; False for a raw ``bluetooth.BLE`` or anything else.""" return isinstance(obj, BLE) + + +async def connect_and_set_up( + ble, setup, *, name=None, service=None, device=None, timeout_ms=10000, attempts=3, setup_timeout_ms=4000, **options +): + """Find a device, connect, and run ``await setup(connection)``; returns + ``(connection, what setup returned)``. + + If the link drops or stalls before ``setup`` finishes, this disconnects + and tries again, up to ``attempts`` times. Profiles use it (``nus``, + ``midi``), because a link can come up dead: about one reconnect in fifty + from Windows, the OS reports a connection the peripheral never saw, and + the first GATT operation on it hangs until Windows gives up nine seconds + later. ``setup_timeout_ms`` bounds that wait. ``options`` go to + ``device.connect()``. + """ + error = None + for _ in range(attempts): + target = device + if target is None: + target = await ble.find(name=name, service=service, timeout_ms=timeout_ms) + connection = await target.connect(timeout_ms=timeout_ms, **options) + try: + return connection, await wait_ms(setup(connection), setup_timeout_ms, "setting up the connection") + except (DisconnectedError, BLETimeoutError) as e: + error = e + except BaseException: + if connection.is_connected(): + await connection.disconnect() + raise + try: + await connection.disconnect() + except BLEError: + pass + raise error diff --git a/lib/bledev/midi.py b/lib/bledev/midi.py new file mode 100644 index 00000000..3c3f5f64 --- /dev/null +++ b/lib/bledev/midi.py @@ -0,0 +1,483 @@ +"""MIDI over Bluetooth LE (BLE-MIDI 1.0), as a MIDI port. + +The board advertises the standard BLE-MIDI service; a laptop, a phone, a DAW +or another board connects:: + + import bledev.midi as midi + + port = await midi.serve(ble, name="rack") # on the board + port = await midi.connect(ble, name="rack") # anywhere that can be a central + +Either way you get a :class:`Port`, which is the same MIDI port usbif hands +out for a USB MIDI function (``usbif.MidiPort``): plain MIDI 1.0 bytes, a +``read(buf)`` that never blocks and a ``write(data)``. So the loop an app +runs over a USB controller runs unchanged over a BLE one:: + + parser = usbif.MidiParser() # or any running-status reader + n = port.read(buf) + parser.feed(buf, n) + for status, data in parser.drain(): + ... + +When usbif is installed, :class:`Port` *is* a ``usbif.MidiPort`` subclass. +BLE-MIDI's packets, timestamps, running status and SysEx fragments stay +inside; the codec is :mod:`bledev.midi_codec`. For the sender's timestamps +as well, ``await port.receive()`` returns ``(timestamp, message)``. + +The characteristic accepts writes with and without response and notifies, as +the BLE-MIDI specification asks, and reads back empty. Every device and host +that speaks BLE-MIDI (macOS and iOS, Android, Windows MIDI Services, BLE-MIDI +controllers) uses the same two UUIDs. +""" + +import sys + +try: + import asyncio +except ImportError: # pragma: no cover + import uasyncio as asyncio + +from . import ( + BLEError, + BLETimeoutError, + BusyError, + Characteristic, + DisconnectedError, + MAX_MTU, + Service, + UnsupportedError, + UUID, + connect_and_set_up, + is_adapter, + sleep_ms, + wait_ms, +) +from .midi_codec import TIMESTAMP_MASK, Decoder, Encoder, Splitter + +SERVICE = UUID("03b80e5a-ede8-4b33-a751-6ce34ec4c700") +CHARACTERISTIC = UUID("7772e5db-3868-4112-a1a9-f2669d106bf3") + +#: The ATT MTU both sides ask for. A note is 5 bytes on the air, so this +#: only matters for SysEx and bursts; it's nus's value, for the same reason. +MTU = 247 + +#: Received messages a port keeps for a reader that isn't reading. +QUEUE_LIMIT = 1024 + +_POLL_MS = 50 +_BUSY_WAIT_MS = 1 + +# ---------------------------------------------------------------- the port contract + +# usbif's MidiPort is the contract (usbif/lib/usbif/__init__.py). When usbif +# is here, subclass it so isinstance() agrees; when it isn't, the same +# contract, written out. +try: + from usbif import INOUT, MidiPort as _PortBase, MidiPortInfo +except Exception: # ImportError, or a usbif that can't load on this host + IN = "in" + OUT = "out" + INOUT = "inout" + + try: + from collections import namedtuple + except ImportError: # pragma: no cover + from ucollections import namedtuple + + MidiPortInfo = namedtuple("MidiPortInfo", "id name direction") + + class _PortBase: + """usbif.MidiPort's contract: a MIDI 1.0 byte stream on one port.""" + + def __init__(self, info): + self.info = info + self.direction = info.direction + self.is_open = True + + def _check(self, want): + if not self.is_open: + raise OSError("MIDI port is closed") + if self.direction not in (want, INOUT): + raise OSError("MIDI port {!r} is {}-only".format(self.info.name, self.direction)) + + def read(self, buf): + """Read waiting MIDI bytes into ``buf``; returns how many. Never blocks.""" + self._check(IN) + return self._read(buf) + + def write(self, data): + """Write MIDI bytes; returns how many were accepted.""" + self._check(OUT) + return self._write(data) + + def close(self): + if not self.is_open: + return + try: + self._close() + finally: + self.is_open = False + + deinit = close + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, traceback): + self.close() + + +if sys.implementation.name == "micropython": + import time + + def now_ms(): + """This host's BLE-MIDI clock: milliseconds, 13 bits.""" + return time.ticks_ms() & TIMESTAMP_MASK + +else: + import time + + def now_ms(): + """This host's BLE-MIDI clock: milliseconds, 13 bits.""" + return int(time.monotonic() * 1000) & TIMESTAMP_MASK + + +def _adapter(ble): + if is_adapter(ble): + return ble + if hasattr(ble, "gap_advertise"): + from . import mpble + + return mpble.get(ble) + raise TypeError("expected a bledev adapter, not {!r}".format(ble)) + + +def _set_mtu(ble, mtu): + try: + ble.config(mtu=mtu) + except UnsupportedError: + pass + + +class Port(_PortBase): + """One BLE-MIDI link, as a MIDI port. Made by :func:`serve` or :func:`connect`. + + ``read(buf)`` and ``write(data)`` are usbif's: plain MIDI bytes, never + blocking. Writes are sent from a task, packed into as few packets as the + link allows, so the app needs a running asyncio loop, as everything in + bledev does. ``await drain()`` waits until they're on the air. + + ``await receive()`` returns the next ``(timestamp, message)``; the + timestamp is the sender's 13-bit millisecond clock. Use it or ``read()``, + not both. ``overflowed`` turns true if more than ``queue_limit`` + messages waited unread; they are counted in ``dropped``, not lost + silently. + """ + + def __init__(self, connection, name, *, queue_limit=QUEUE_LIMIT, running_status=True): + _PortBase.__init__(self, MidiPortInfo("ble:" + connection.device.address, name, INOUT)) + self.connection = connection + self.queue_limit = queue_limit + self.dropped = 0 + self.decode_errors = 0 + self.messages_in = 0 + self.messages_out = 0 + self._messages = [] + self._pending = b"" + self._arrived = asyncio.Event() + self._decoder = Decoder() + self._encoder = Encoder(running_status) + self._splitter = Splitter() + self._wake = asyncio.Event() + self._idle = asyncio.Event() + self._idle.set() + self._send_error = None + self._closed = False + self._sender = asyncio.create_task(self._send_loop()) + + @property + def overflowed(self): + return self.dropped > 0 + + # -- receiving + + def _packet(self, data): + messages = self._decoder.decode(data) + self.decode_errors = self._decoder.errors + for item in messages: + if len(self._messages) >= self.queue_limit: + self.dropped += 1 + else: + self._messages.append(item) + self.messages_in += 1 + if messages: + self._arrived.set() + + def _read(self, buf): + n = 0 + size = len(buf) + while n < size: + if not self._pending: + if not self._messages: + break + self._pending = self._messages.pop(0)[1] + take = min(size - n, len(self._pending)) + buf[n : n + take] = self._pending[:take] + self._pending = self._pending[take:] + n += take + return n + + def any(self): + """How many messages are waiting.""" + return len(self._messages) + (1 if self._pending else 0) + + async def receive(self, timeout_ms=None): + """The next ``(timestamp, message)``. Raises :class:`DisconnectedError` + once the link has closed and everything that arrived is read.""" + while True: + if self._messages: + return self._messages.pop(0) + if self._closed or not self.connection.is_connected(): + raise DisconnectedError("MIDI link closed") + self._arrived.clear() + await wait_ms(self._any_news(), timeout_ms, "receive") + + async def _any_news(self): + await self._arrived.wait() + + def __aiter__(self): + return self + + async def __anext__(self): + try: + return await self.receive() + except DisconnectedError: + raise StopAsyncIteration + + # -- sending + + def _write(self, data): + if self._send_error is not None: + raise self._send_error + if not self.connection.is_connected(): + raise DisconnectedError("MIDI link closed") + ts = now_ms() + count = 0 + for message in self._splitter.feed(data): + self._encoder.put(ts, message) + count += 1 + if count: + self.messages_out += count + self._idle.clear() + self._wake.set() + return len(data) + + def send(self, message, timestamp=None): + """Queue one complete message, optionally with its own timestamp.""" + self._check("out") + self._encoder.put(now_ms() if timestamp is None else timestamp, message) + self.messages_out += 1 + self._idle.clear() + self._wake.set() + + async def drain(self): + """Wait until everything written has been handed to the radio.""" + await self._idle.wait() + if self._send_error is not None: + raise self._send_error + + async def _send_loop(self): + try: + while not self._closed: + await self._wake.wait() + self._wake.clear() + while self._encoder.pending(): + packet = self._encoder.packet(self.connection.mtu - 3) + await self._send_packet(packet) + self._idle.set() + except Exception as e: + self._send_error = e if isinstance(e, BLEError) else BLEError("MIDI send: {}".format(e)) + self._idle.set() + + async def _send_packet(self, packet): + raise NotImplementedError + + # -- closing + + def _close(self): + self._finish() + if self.connection.is_connected(): + try: + asyncio.create_task(self.connection.disconnect()) + except Exception: + pass + + async def aclose(self): + """Close the port and wait for the disconnect.""" + was_open = self.is_open + self.is_open = False + self._finish() + if was_open and self.connection.is_connected(): + await self.connection.disconnect() + + def _finish(self): + self._closed = True + self._wake.set() + self._arrived.set() + + +class _ServerPort(Port): + def __init__(self, connection, server, **options): + Port.__init__(self, connection, server.name, **options) + self._server = server + + async def _send_packet(self, packet): + char = self._server.char + while True: + try: + char.notify(self.connection, packet) + return + except BusyError: + if not self.connection.is_connected(): + raise DisconnectedError("MIDI link closed while sending") + await sleep_ms(_BUSY_WAIT_MS) + + +class _ClientPort(Port): + def __init__(self, connection, char, name, **options): + Port.__init__(self, connection, name, **options) + self._char = char + self._no_response = bool(char.properties & 0x0004) + self._receiver = asyncio.create_task(self._receive()) + + async def _receive(self): + try: + while True: + self._packet(await self._char.notified()) + except Exception: + pass + finally: + self._finish() + + async def _send_packet(self, packet): + await self._char.write(packet, response=not self._no_response) + + +class _Server: + """The BLE-MIDI service on one adapter, shared by every port it serves.""" + + def __init__(self, ble, name): + self.ble = ble + self.name = name + self.service = Service(SERVICE) + self.char = Characteristic( + self.service, + CHARACTERISTIC, + read=True, + write=True, + write_no_response=True, + notify=True, + capture=True, + max_len=MAX_MTU - 3, + initial=b"", + ) + self.ports = [] + self._task = None + + def add(self, port): + self.ports.append(port) + if self._task is None: + self._task = asyncio.create_task(self._run()) + + def _port_for(self, connection): + for port in self.ports: + if port.connection is connection or port.connection == connection: + return port + return None + + def _retire_gone(self): + for port in list(self.ports): + if port._closed or not port.connection.is_connected(): + self.ports.remove(port) + port._finish() + + async def _run(self): + try: + while self.ports: + try: + connection, data = await self.char.written(timeout_ms=_POLL_MS) + except BLETimeoutError: + self._retire_gone() + continue + port = self._port_for(connection) + if port is not None: + port._packet(data) + finally: + self._task = None + + +def _server_for(ble, name, mtu): + server = getattr(ble, "_midi_server", None) + if server is None: + _set_mtu(ble, mtu) + server = _Server(ble, name) + ble.register_services(server.service) + ble._midi_server = server + server.name = name + return server + + +async def serve(ble, name="bledev-midi", *, timeout_ms=None, interval_us=100000, mtu=MTU, **options): + """Advertise BLE-MIDI as ``name`` and return a :class:`Port` when a + central connects. Call it again for the next connection. + + The service is registered once per adapter, the first time; don't register + other services on the same adapter afterwards. + """ + ble = _adapter(ble) + server = _server_for(ble, name, mtu) + # Hosts that pair (Windows, macOS) name the MIDI port after the GAP device + # name, not the advertised one; without this Windows lists "MPY ESP32". + try: + ble.config(gap_name=name) + except (UnsupportedError, ValueError, OSError): + pass + connection = await ble.advertise(interval_us, name=name, services=[SERVICE], timeout_ms=timeout_ms) + port = _ServerPort(connection, server, **options) + server.add(port) + return port + + +async def connect( + ble, name=None, *, device=None, timeout_ms=10000, mtu=MTU, queue_limit=QUEUE_LIMIT, running_status=True, **connect_options +): + """Find a BLE-MIDI device (by ``name``, or the first one advertising the + service), connect, subscribe, and return a :class:`Port`. + + ``connect_options`` go to ``device.connect()``: on a board, + ``min_conn_interval_us=7500, max_conn_interval_us=7500`` asks for the + shortest connection interval, which is what a MIDI link wants. + """ + ble = _adapter(ble) + _set_mtu(ble, mtu) + + async def setup(connection): + try: + await connection.exchange_mtu(mtu) + except (UnsupportedError, BLETimeoutError): + pass + service = await connection.service(SERVICE) + if service is None: + raise BLEError("{!r} has no BLE-MIDI service".format(connection.device)) + char = await service.characteristic(CHARACTERISTIC) + if char is None: + raise BLEError("{!r}: the BLE-MIDI service has no MIDI characteristic".format(connection.device)) + await char.subscribe(notify=True) + return char + + connection, char = await connect_and_set_up( + ble, setup, name=name, service=SERVICE, device=device, timeout_ms=timeout_ms, **connect_options + ) + label = name or getattr(connection.device, "name", None) or connection.device.address + return _ClientPort(connection, char, label, queue_limit=queue_limit, running_status=running_status) + diff --git a/lib/bledev/midi_codec.py b/lib/bledev/midi_codec.py new file mode 100644 index 00000000..3d087d86 --- /dev/null +++ b/lib/bledev/midi_codec.py @@ -0,0 +1,419 @@ +"""The BLE-MIDI packet codec: MIDI 1.0 messages in and out of GATT packets. + +Pure Python with no imports, so it runs anywhere bledev does, and a board, a +laptop and a browser share one implementation. The radio side is +:mod:`bledev.midi`; this module only turns bytes into bytes. + +The format is the MMA/Apple "MIDI over Bluetooth Low Energy" (BLE-MIDI 1.0) +one. Each packet is one characteristic write or notification:: + + header timestamp status data... [timestamp] [status] data... ... + 10hhhhhh 1lllllll + +A timestamp is 13 bits of milliseconds: six high bits in the header, seven low +bits in the byte before each message. A low part smaller than the one before +it in the same packet means the high part rolled over. Beyond that: + +* **Running status.** A message with the same channel status as the one + before may leave the status out, and also the timestamp when it is the + same. The :class:`Decoder` keeps running status across packets, because a + sender may rely on it; the :class:`Encoder` uses it only inside a packet, + so a receiver that lost the previous packet still decodes this one. +* **System exclusive** can span packets. A continuation packet is the header + followed straight by data bytes; the closing ``F7`` has its own timestamp. +* **Real-time messages** (``F8``-``FF``) may sit anywhere, including in the + middle of a SysEx or between a status and its data, each with its own + timestamp. The encoder slips them into a SysEx that spans packets, so a + clock tick doesn't wait for a long dump to finish. +* Any other message fits in one packet; the encoder never splits one, and + the decoder drops one that a sender did split. + +A message's timestamp is the one before its status byte (or before its data, +under running status). A SysEx carries the timestamp before its ``F7``. + +Messages are plain MIDI bytes, complete with their status byte, the same +bytes ``usbif.MidiPort`` carries. A SysEx is one message, ``F0 ... F7``. +""" + +#: 13-bit millisecond timestamps. +TIMESTAMP_MASK = 0x1FFF + +# Data bytes after the status, by high nibble (channel voice) or by status +# (system common). F0 (SysEx) and F7 are handled on their own. +_VOICE_LEN = (2, 2, 2, 2, 1, 1, 2) # 0x8_ .. 0xE_ +_COMMON_LEN = {0xF1: 1, 0xF2: 2, 0xF3: 1, 0xF4: 0, 0xF5: 0, 0xF6: 0} + + +def data_length(status): + """How many data bytes follow ``status``; ``None`` for SysEx and stray bytes.""" + if status < 0x80: + return None + if status < 0xF0: + return _VOICE_LEN[(status >> 4) - 8] + if status >= 0xF8: + return 0 + return _COMMON_LEN.get(status) + + +def is_realtime(message): + return len(message) == 1 and message[0] >= 0xF8 + + +class Splitter: + """A MIDI 1.0 byte stream in, complete messages out, SysEx included. + + What a port's ``write()`` hands over is a stream: it may use running + status, split a message across calls, or drop a real-time byte into the + middle of another message. ``feed()`` returns the messages it completes, + each as ``bytes`` with its status byte. SysEx is kept whole, ``F0`` to + ``F7``. A data byte with no status to attach to is counted in + :attr:`desync` and dropped. + """ + + def __init__(self): + self.desync = 0 + self._running = 0 + self._msg = None + self._want = 0 + self._sysex = None + + def feed(self, data): + out = [] + for b in data: + if b >= 0xF8: + out.append(bytes((b,))) + continue + if self._sysex is not None: + if b < 0x80: + self._sysex.append(b) + continue + if b == 0xF7: + self._sysex.append(b) + out.append(bytes(self._sysex)) + self._sysex = None + continue + self._sysex = None # any other status ends it unfinished + self.desync += 1 + if b >= 0x80: + self._msg = None + if b == 0xF0: + self._sysex = bytearray((b,)) + self._running = 0 + continue + want = data_length(b) + if want is None: # a stray F7 + self.desync += 1 + continue + self._running = b if b < 0xF0 else 0 + if want == 0: + out.append(bytes((b,))) + continue + self._msg = bytearray((b,)) + self._want = want + 1 + continue + if self._msg is None: + if not self._running: + self.desync += 1 + continue + self._msg = bytearray((self._running,)) + self._want = data_length(self._running) + 1 + self._msg.append(b) + if len(self._msg) == self._want: + out.append(bytes(self._msg)) + self._msg = None + return out + + +class Decoder: + """BLE-MIDI packets in, ``(timestamp, message)`` pairs out. + + One decoder per link and direction, because running status and a SysEx + in progress carry from one packet to the next. ``timestamp`` is the + sender's 13-bit millisecond clock. Anything malformed is counted in + :attr:`errors` and skipped rather than guessed at. + """ + + def __init__(self): + self.errors = 0 + self._running = 0 + self._sysex = None + self._ts = 0 + + def reset(self): + """Forget running status and any SysEx in progress (a new link).""" + self._running = 0 + self._sysex = None + + def decode(self, packet): + out = [] + n = len(packet) + if n < 2 or (packet[0] & 0xC0) != 0x80: + self.errors += 1 + return out + high = packet[0] & 0x3F + low = None + # Until a timestamp arrives, a data byte can only continue a SysEx + # (or, from a lax sender, run on from the last packet's status). + ts = (high << 7) | (self._ts & 0x7F) + after_ts = False + msg = None + msg_ts = ts + want = 0 + i = 1 + while i < n: + b = packet[i] + i += 1 + if b & 0x80 and not after_ts: + l = b & 0x7F + if low is not None and l < low: + high = (high + 1) & 0x3F + low = l + ts = (high << 7) | l + after_ts = True + continue + if b & 0x80: + after_ts = False + if b >= 0xF8: + out.append((ts, bytes((b,)))) + continue + if msg is not None: # a status before the last message finished + self.errors += 1 + msg = None + if self._sysex is not None: + if b == 0xF7: + self._sysex.append(b) + out.append((ts, bytes(self._sysex))) + self._sysex = None + continue + self.errors += 1 + self._sysex = None + if b == 0xF0: + self._sysex = bytearray((b,)) + self._running = 0 + continue + want = data_length(b) + if want is None: + self.errors += 1 + continue + self._running = b if b < 0xF0 else 0 + if want == 0: + out.append((ts, bytes((b,)))) + continue + msg = bytearray((b,)) + msg_ts = ts + want += 1 + continue + # A data byte. + after_ts = False + if self._sysex is not None: + self._sysex.append(b) + continue + if msg is None: + if not self._running: + self.errors += 1 + continue + msg = bytearray((self._running,)) + msg_ts = ts + want = data_length(self._running) + 1 + msg.append(b) + if len(msg) == want: + out.append((msg_ts, bytes(msg))) + msg = None + if msg is not None: # only SysEx may continue into the next packet + self.errors += 1 + self._ts = ts + return out + + +class Encoder: + """``(timestamp, message)`` pairs in, BLE-MIDI packets out. + + ``put()`` queues a complete message; ``packet(size)`` returns the next + packet of at most ``size`` bytes (``mtu - 3``), or ``None`` when nothing + is queued. Messages go out in the order they were put, with one + exception MIDI 1.0 also makes: while a SysEx is spread over several + packets, real-time messages queued behind it are slipped into it, so a + clock tick doesn't wait for a long dump to finish. With + ``running_status`` (the default) a repeated channel status is left out + inside a packet, and the timestamp too when it hasn't changed. + """ + + def __init__(self, running_status=True): + self.running_status = running_status + self._queue = [] + self._sysex = None # [timestamp, message, next index] while one spans packets + + def put(self, timestamp, message): + message = bytes(message) + _check_message(message) + self._queue.append((timestamp & TIMESTAMP_MASK, message)) + + def pending(self): + return bool(self._queue or self._sysex) + + def packet(self, size): + if size < 5: + raise ValueError("a BLE-MIDI packet needs at least 5 bytes") + if not self.pending(): + return None + p = _Packet(size) + sysex = self._sysex + if sysex is not None: + # Real-time messages queued behind the SysEx go in first; the + # SysEx's data bytes then run on after them. + i = 0 + while i < len(self._queue): + ts, message = self._queue[i] + if message[0] >= 0xF8: + if not p.stamp(ts, 1): + break + p.put(message) + self._queue.pop(i) + else: + i += 1 + if not self._sysex_body(p, sysex): + return p.finish(sysex[0]) + self._sysex = None + while self._queue: + ts, message = self._queue[0] + status = message[0] + if status == 0xF0: + if not p.stamp(ts, 2): + break + self._queue.pop(0) + p.put(message[:1]) + p.running = 0 + sysex = [ts, message, 1] + if not self._sysex_body(p, sysex): + self._sysex = sysex + break + continue + if self.running_status and status == p.running: + if p.have_ts and ts == p.ts and p.last_was_voice: + if not p.fits(len(message) - 1): + break + elif not p.stamp(ts, len(message) - 1): + break + p.put(message[1:]) + else: + if not p.stamp(ts, len(message)): + break + p.put(message) + if status < 0xF8: + p.running = status if status < 0xF0 else 0 + p.last_was_voice = status < 0xF0 + self._queue.pop(0) + return p.finish(self._queue[0][0] if self._queue else 0) + + def _sysex_body(self, p, sysex): + """Put as much of a SysEx as fits; ``True`` once its F7 is in.""" + ts, message, index = sysex + last = len(message) - 1 # the F7 + take = min(last - index, p.room()) + if take > 0: + p.put(message[index : index + take]) + index += take + p.last_was_voice = False + sysex[2] = index + if index < last: + return False + if p.have_ts: + ts = sysex[0] = p.ts + if not p.stamp(ts, 1): + return False + p.put(message[last:]) + p.last_was_voice = False + return True + + +def _check_message(message): + if not message or message[0] < 0x80: + raise ValueError("a MIDI message starts with its status byte") + status = message[0] + if status == 0xF0: + if message[-1] != 0xF7 or len(message) < 2: + raise ValueError("a SysEx message ends with F7") + body = message[1:-1] + else: + want = data_length(status) + if want is None or len(message) != want + 1: + raise ValueError("not one complete MIDI message: {}".format(message)) + body = message[1:] + for b in body: + if b & 0x80: + raise ValueError("a status byte inside a message: {}".format(message)) + + +class _Packet: + """One packet being built: the header, then stamped messages.""" + + def __init__(self, size): + self.size = size + self.buf = bytearray(1) # the header, filled in by finish() + self.have_ts = False + self.ts = 0 + self.high = 0 + self.low = 0 + self.running = 0 + self.first_high = 0 + self.last_was_voice = False + + def room(self): + return self.size - len(self.buf) + + def fits(self, n): + return len(self.buf) + n <= self.size + + def stamp(self, ts, n): + """Add a timestamp byte followed by ``n`` more bytes, if both fit and + the timestamp is expressible here (it may only move forward, by at + most one rollover of the low seven bits).""" + if not self.fits(n + 1): + return False + high = ts >> 7 + low = ts & 0x7F + if self.have_ts: + if high == self.high and low >= self.low: + pass + elif high == (self.high + 1) & 0x3F and low < self.low: + self.high = high + else: + return False + else: + self.high = high + self.first_high = high + self.low = low + self.ts = ts + self.have_ts = True + self.buf.append(0x80 | low) + return True + + def put(self, data): + self.buf.extend(data) + + def finish(self, fallback_ts): + high = self.first_high if self.have_ts else (fallback_ts >> 7) & 0x3F + self.buf[0] = 0x80 | high + return bytes(self.buf) + + +def encode(messages, size, running_status=True): + """Every packet for ``messages``, a list of ``(timestamp, message)``.""" + e = Encoder(running_status) + for ts, message in messages: + e.put(ts, message) + packets = [] + while True: + p = e.packet(size) + if p is None: + return packets + packets.append(p) + + +def decode(packets): + """Every ``(timestamp, message)`` in ``packets``, with a fresh decoder.""" + d = Decoder() + out = [] + for p in packets: + out.extend(d.decode(p)) + return out diff --git a/lib/bledev/nus.py b/lib/bledev/nus.py index 1d6109b0..07f7388c 100644 --- a/lib/bledev/nus.py +++ b/lib/bledev/nus.py @@ -38,6 +38,7 @@ Service, UnsupportedError, UUID, + connect_and_set_up, is_adapter, sleep_ms, ) @@ -351,24 +352,23 @@ async def connect(ble, name=None, *, device=None, timeout_ms=10000, mtu=MTU, **c """ ble = _adapter(ble) _set_mtu(ble, mtu) - if device is None: - device = await ble.find(name=name, service=SERVICE, timeout_ms=timeout_ms) - connection = await device.connect(timeout_ms=timeout_ms, **connect_options) - try: + + async def setup(connection): try: await connection.exchange_mtu(mtu) except (UnsupportedError, BLETimeoutError): pass service = await connection.service(SERVICE) if service is None: - raise BLEError("{!r} has no Nordic UART service".format(device)) + raise BLEError("{!r} has no Nordic UART service".format(connection.device)) rx = await service.characteristic(RX) tx = await service.characteristic(TX) if rx is None or tx is None: - raise BLEError("{!r}: Nordic UART service is missing RX or TX".format(device)) + raise BLEError("{!r}: Nordic UART service is missing RX or TX".format(connection.device)) await tx.subscribe(notify=True) - except BaseException: - if connection.is_connected(): - await connection.disconnect() - raise + return rx, tx + + connection, (rx, tx) = await connect_and_set_up( + ble, setup, name=name, service=SERVICE, device=device, timeout_ms=timeout_ms, **connect_options + ) return _ClientLink(connection, rx, tx) diff --git a/lib/bledev/repl.py b/lib/bledev/repl.py index b07f6b94..9401f40e 100644 --- a/lib/bledev/repl.py +++ b/lib/bledev/repl.py @@ -34,6 +34,9 @@ from . import nus PROMPT = b"Password: " + +# How often the server retries output the controller couldn't take yet. +_TICK_MS = 5 BANNER = b"\r\nbledev REPL connected\r\n>>> " DENIED = b"\r\nAccess denied\r\n" @@ -190,10 +193,12 @@ def __init__(self, password, name, interval_us, dupterm_index): self.rx = bytearray() self.rx_pos = 0 self.tx = bytearray() + self.flushing = False self.dropped_out = 0 self.prev_term = None self.stream = _Stream(self) self.timer = None + self.hang_up_at = None self.early_mtu = None # -- setup @@ -236,8 +241,12 @@ def start(self): # Appending: several writes between two IRQs arrive together. ble.gatts_set_buffer(self.rx_handle, 1024, True) self._forget_other_services() + # One periodic timer for the life of the server, never re-armed: on + # the esp32, re-initialising a virtual Timer while its alarm is being + # dispatched calls a NULL handler and panics the board (seen as a + # Guru Meditation in esp_timer's task during long prints). self.timer = machine.Timer(-1) - self.one_shot = machine.Timer.ONE_SHOT + self.timer.init(mode=machine.Timer.PERIODIC, period=_TICK_MS, callback=self._tick) self.adv = pack_advertisement(self.name, [nus.SERVICE]) self.running = True self.advertise() @@ -292,6 +301,7 @@ def on_connect(self, conn): self.mtu = self.early_mtu[1] self.early_mtu = None self.login = _Gate(self.password) + self.hang_up_at = None self.authed = False self.rx = bytearray() self.rx_pos = 0 @@ -305,6 +315,7 @@ def on_disconnect(self, conn): return self._detach() self.conn = None + self.hang_up_at = None self.login = None self.authed = False self.rx = bytearray() @@ -321,7 +332,9 @@ def on_write(self, conn, data): self.notify_now(reply) self.login.refused = True # Let "Access denied" reach the client, then hang up. - self.timer.init(mode=self.one_shot, period=300, callback=self._hang_up) + import time + + self.hang_up_at = time.ticks_add(time.ticks_ms(), 300) return if verdict is None: if reply: @@ -373,7 +386,18 @@ def feed_repl(self, data): # -- sending - def _hang_up(self, _timer=None): + def _tick(self, _timer=None): + # Every _TICK_MS while the server runs, as a scheduled callback. + if self.hang_up_at is not None: + import time + + if time.ticks_diff(time.ticks_ms(), self.hang_up_at) >= 0: + self.hang_up_at = None + self._hang_up() + if self.tx and self.conn is not None: + self.flush() + + def _hang_up(self): if self.conn is not None and not self.authed: try: self.ble.gap_disconnect(self.conn) @@ -409,23 +433,26 @@ def send(self, data): break time.sleep_ms(2) self.flush() - if self.tx: - self.timer.init(mode=self.one_shot, period=5, callback=self._flush_later) - - def _flush_later(self, _timer=None): - self.flush() - if self.tx and self.conn is not None: - self.timer.init(mode=self.one_shot, period=5, callback=self._flush_later) def flush(self): - size = self.mtu - 3 - while self.tx and self.conn is not None: - chunk = self.tx[:size] - try: - self.ble.gatts_notify(self.conn, self.tx_handle, chunk) - except OSError: - return # the controller's buffers are full; the timer retries - self.tx = self.tx[len(chunk) :] + # The timer's callback runs between the main program's + # bytecodes, so it can land inside a flush() that print() started. + # Two flushes interleaved sent one chunk twice and dropped the next, + # so the second one leaves the work to the first. + if self.flushing: + return + self.flushing = True + try: + size = self.mtu - 3 + while self.tx and self.conn is not None: + chunk = self.tx[:size] + try: + self.ble.gatts_notify(self.conn, self.tx_handle, chunk) + except OSError: + return # the controller's buffers are full; _tick retries + self.tx = self.tx[len(chunk) :] + finally: + self.flushing = False class _Gate(Login): diff --git a/pydevices-desktop.toml b/pydevices-desktop.toml index a1f8be06..fda4dd22 100644 --- a/pydevices-desktop.toml +++ b/pydevices-desktop.toml @@ -25,6 +25,8 @@ "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/improv.py" = "/lib/bledev/improv.py" +"https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/bledev/midi.py" = "/lib/bledev/midi.py" +"https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/bledev/midi_codec.py" = "/lib/bledev/midi_codec.py" "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/bledev/mpble.py" = "/lib/bledev/mpble.py" "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/bledev/nus.py" = "/lib/bledev/nus.py" "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/bledev/repl.py" = "/lib/bledev/repl.py" diff --git a/tests/bledev_board/midi_client.py b/tests/bledev_board/midi_client.py new file mode 100644 index 00000000..fedc5d74 --- /dev/null +++ b/tests/bledev_board/midi_client.py @@ -0,0 +1,270 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""The BLE-MIDI gate and latency rig, client side. + +Needs ``midi_server.py`` serving on a board. Runs on a board (mpftp probe) +or on a laptop (bleak, Windows Python):: + + PYTHONPATH="$(wslpath -w lib)" python.exe "$(wslpath -w tests/bledev_board/midi_client.py)" [gate] [latency] [--fast] [--plant] + +``gate``: 1,000 mixed messages (notes, CC, pitch bend, program change, clock +and a 300-byte SysEx every hundred) go up and are checked by the server, then +come back down and are checked here. ``--plant`` flips a bit in one packet +of what this side sends; the server's verdict must then be FAIL. + +``latency``: note-on round trips through the server's echo, one note at a +time at a random phase (0-20 ms apart), then chords of four. Half the round +trip is what BLE adds one way. ``--fast`` asks for a 7.5 ms connection +interval (on a laptop, Windows' throughput parameters) instead of the +default. Prints median, p90, p99 and max, and ends ``RESULT PASS`` or +``RESULT FAIL``. +""" +import asyncio +import sys + +import bledev +import bledev.midi as midi + +MICROPYTHON = sys.implementation.name == "micropython" +NAME = "bledev-midi" +PINGS = 400 +CHORDS = 100 + +if MICROPYTHON: + import time + + def now_us(): + return time.ticks_us() + + def elapsed_us(t0): + return time.ticks_diff(time.ticks_us(), t0) + + ARGS = [] + try: + import midi_client_args + + ARGS = midi_client_args.ARGS + except ImportError: + pass +else: + import time + + def now_us(): + return time.perf_counter_ns() // 1000 + + def elapsed_us(t0): + return now_us() - t0 + + ARGS = sys.argv[1:] + +LOG = "/midi_client.log" if MICROPYTHON else None +failures = [] + + +def log(*parts): + line = " ".join(str(p) for p in parts) + print(line) + if LOG: + with open(LOG, "a") as f: + f.write(line + "\n") + + +def check(ok, what, detail=""): + log("PASS" if ok else "FAIL", what, detail) + if not ok: + failures.append(what) + + +def adapter(): + if MICROPYTHON: + import bledev.mpble + + return bledev.mpble.get() + import bledev.bleak + + return bledev.bleak.BleakBLE() + + +def options(fast): + if not fast: + return {} + if MICROPYTHON: + return {"min_conn_interval_us": 7500, "max_conn_interval_us": 7500} + return {"priority": "throughput"} + + +def mix(count): + # The same as midi_server.mix(); kept in step by hand. + out = [] + for i in range(count): + k = i % 10 + if i % 100 == 9: + out.append(bytes([0xF0] + [(i + j) % 128 for j in range(298)] + [0xF7])) + elif k in (0, 1, 2): + out.append(bytes((0x90 | (i % 16), i % 128, 1 + i % 127))) + elif k == 3: + out.append(bytes((0x80 | (i % 16), i % 128, 0))) + elif k in (4, 5): + out.append(bytes((0xB0 | (i % 16), 7, i % 128))) + elif k == 6: + out.append(bytes((0xE0 | (i % 16), i % 128, (i >> 7) % 128))) + elif k in (7, 8): + out.append(b"\xf8") + else: + out.append(bytes((0xC0 | (i % 16), i % 128))) + return out + + +def verify(got, want): + bad = len(got) if len(got) != len(want) else -1 + for kind in (True, False): + g = [m for m in got if (m[0] >= 0xF8) == kind] + w = [m for m in want if (m[0] >= 0xF8) == kind] + for i in range(max(len(g), len(w))): + if i >= len(g) or i >= len(w) or g[i] != w[i]: + bad = i if bad < 0 else min(bad, i) + break + return bad < 0, bad + + +def plant(port): + send = port._send_packet + count = [0] + + async def planted(packet): + count[0] += 1 + if count[0] == 20: + packet = bytearray(packet) + packet[-1] ^= 0x01 + await send(packet) + + port._send_packet = planted + + +async def gate(ble, fast, planted): + port = await midi.connect(ble, name=NAME, timeout_ms=15000, **options(fast)) + log("gate: connected, mtu", port.connection.mtu) + t0 = now_us() + port.send(b"\xf0\x7d\x10\x01\xf7") + await port.drain() + if planted: + plant(port) + want = mix(1000) + for message in want: + port.send(message) + await port.drain() + ts, verdict = await port.receive(timeout_ms=30000) + up_ms = elapsed_us(t0) / 1000 + ok = verdict[:4] == b"\xf0\x7d\x11\x00" + count = (verdict[4] << 7) | verdict[5] if len(verdict) > 7 else -1 + first = (verdict[6] << 7) | verdict[7] if len(verdict) > 7 else -1 + check(ok, "gate up: the server got 1,000 messages complete, in order, intact", "server counted {}, first bad {}".format(count, first if not ok else "-")) + got = [] + t0 = now_us() + try: + while len(got) < 1000: + ts, message = await port.receive(timeout_ms=15000) + got.append(message) + except bledev.BLEError as e: + log("gate down: stopped after", len(got), type(e).__name__, e) + down_ms = elapsed_us(t0) / 1000 + ok, bad = verify(got, want) + check(ok, "gate down: 1,000 messages complete, in order, intact", "got {}, first bad {}, decode errors {}".format(len(got), bad, port.decode_errors)) + log("gate: up took {:.0f} ms, down {:.0f} ms".format(up_ms, down_ms)) + await port.aclose() + + +def stats(samples_us): + s = sorted(samples_us) + n = len(s) + + def pct(p): + return s[min(n - 1, int(p * n))] / 1000 + + return "n={} median {:.1f} ms, p90 {:.1f}, p99 {:.1f}, max {:.1f}".format(n, pct(0.5), pct(0.9), pct(0.99), s[-1] / 1000) + + +class Rand: + def __init__(self, seed): + self.state = seed + + def next(self, n): + self.state = (self.state * 1103515245 + 12345) & 0x7FFFFFFF + return (self.state >> 8) % n + + +async def latency(ble, fast): + port = await midi.connect(ble, name=NAME, timeout_ms=15000, **options(fast)) + label = "7.5 ms interval" if fast and MICROPYTHON else ("throughput parameters" if fast else "default parameters") + log("latency ({}): connected, mtu {}".format(label, port.connection.mtu)) + port.send(b"\xf0\x7d\x10\x02\xf7") + await port.drain() + r = Rand(7) + await bledev.sleep_ms(500) + one_way = [] + lost = 0 + for i in range(PINGS): + await bledev.sleep_ms(r.next(21)) + note = i % 128 + t0 = now_us() + port.write(bytes((0x90, note, 100))) + try: + while True: + ts, message = await port.receive(timeout_ms=2000) + if message == bytes((0x90, note, 100)): + break + except bledev.BLETimeoutError: + lost += 1 + continue + one_way.append(elapsed_us(t0) // 2) + log("latency ({}), one note, half the round trip: {}".format(label, stats(one_way))) + chord = [] + for i in range(CHORDS): + await bledev.sleep_ms(r.next(21)) + base = 36 + (i % 60) + notes = [bytes((0x91, base + k, 90)) for k in (0, 4, 7, 12)] + t0 = now_us() + port.write(b"".join(notes)) + seen = 0 + try: + while seen < 4: + ts, message = await port.receive(timeout_ms=2000) + if message in notes: + seen += 1 + except bledev.BLETimeoutError: + lost += 1 + continue + chord.append(elapsed_us(t0) // 2) + log("latency ({}), a four-note chord, half the round trip to the last note: {}".format(label, stats(chord))) + check(lost == 0, "latency ({}): every echo came back".format(label), "lost {}".format(lost)) + check(port.decode_errors == 0, "latency ({}): no decode errors".format(label)) + await port.aclose() + return one_way + + +async def main(): + if LOG: + open(LOG, "w").close() + ble = adapter() + fast = "--fast" in ARGS + planted = "--plant" in ARGS + if "gate" in ARGS or not any(a in ARGS for a in ("gate", "latency")): + await gate(ble, fast, planted) + await bledev.sleep_ms(1500) + if "latency" in ARGS or not any(a in ARGS for a in ("gate", "latency")): + await latency(ble, fast) + + +async def guarded(): + try: + await main() + except Exception as e: + failures.append("exception") + log("EXCEPTION", type(e).__name__, e) + log("RESULT", "FAIL {}".format(failures) if failures else "PASS") + + +asyncio.run(guarded()) +if not MICROPYTHON: + sys.exit(1 if failures else 0) diff --git a/tests/bledev_board/midi_server.py b/tests/bledev_board/midi_server.py new file mode 100644 index 00000000..86ef24be --- /dev/null +++ b/tests/bledev_board/midi_server.py @@ -0,0 +1,162 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""The BLE-MIDI gate and latency rig, board side: serve BLE-MIDI as ``bledev-midi``. + +Run it from ``/main.py`` on one board (mpftp soft-resets a board before it +runs a script, and a soft reset turns Bluetooth off, so ``mpftp run`` would +lose it). ``midi_client.py`` on another board or a laptop then connects and +asks for one of two sessions with a SysEx command, ``F0 7D 10 F7``: + +* mode 1, the gate: this side reads 1,000 mixed messages (``mix()``) and + checks them, sends back its verdict as ``F0 7D 11 <0 pass, 1 fail> + F7``, then sends the same 1,000 + messages down for the client to check. +* mode 2, echo: every message that arrives goes straight back, until the + link closes. The client times the round trips. + +It serves one connection after another until reset. ``PLANT`` breaks what +this side sends on the air: ``"drop"`` leaves out the 20th packet of the +gate's downstream, ``"flip"`` flips a bit in it. The client must then FAIL. +Logs to /midi_server.log. +""" +import asyncio +import sys + +import bledev.midi as midi +import bledev.mpble + +PLANT = None # None, "drop" or "flip" +LOG = "/midi_server.log" +NAME = "bledev-midi" +COUNT = 1000 + + +def log(*parts): + line = " ".join(str(p) for p in parts) + print(line) + try: + with open(LOG, "a") as f: + f.write(line + "\n") + except OSError: + pass + + +def mix(count): + """Notes, controllers, pitch bend, program changes, clock, and a + 300-byte SysEx every hundred messages, interleaved. Both ends build it.""" + out = [] + for i in range(count): + k = i % 10 + if i % 100 == 9: + out.append(bytes([0xF0] + [(i + j) % 128 for j in range(298)] + [0xF7])) + elif k in (0, 1, 2): + out.append(bytes((0x90 | (i % 16), i % 128, 1 + i % 127))) + elif k == 3: + out.append(bytes((0x80 | (i % 16), i % 128, 0))) + elif k in (4, 5): + out.append(bytes((0xB0 | (i % 16), 7, i % 128))) + elif k == 6: + out.append(bytes((0xE0 | (i % 16), i % 128, (i >> 7) % 128))) + elif k in (7, 8): + out.append(b"\xf8") + else: + out.append(bytes((0xC0 | (i % 16), i % 128))) + return out + + +def verify(got, want): + """``(ok, first bad index)``. Real-time may slip into a SysEx on the air, + so real-time and everything else are each compared in order.""" + if len(got) != len(want): + bad = len(got) + else: + bad = -1 + for kind in (True, False): + g = [m for m in got if (m[0] >= 0xF8) == kind] + w = [m for m in want if (m[0] >= 0xF8) == kind] + for i in range(max(len(g), len(w))): + if i >= len(g) or i >= len(w) or g[i] != w[i]: + bad = i if bad < 0 else min(bad, i) + break + return bad < 0, bad + + +def plant(port): + if not PLANT: + return + send = port._send_packet + count = [0] + + async def planted(packet): + count[0] += 1 + if count[0] == 20: + if PLANT == "drop": + return + packet = bytearray(packet) + packet[-1] ^= 0x01 + await send(packet) + + port._send_packet = planted + + +async def gate(port): + want = mix(COUNT) + got = [] + while len(got) < COUNT: + ts, message = await port.receive(timeout_ms=30000) + got.append(message) + ok, bad = verify(got, want) + log("gate up:", "PASS" if ok else "FAIL", len(got), "messages, first bad", bad, "decode errors", port.decode_errors) + b = max(bad, 0) + port.send(bytes((0xF0, 0x7D, 0x11, 0 if ok else 1, len(got) >> 7, len(got) & 0x7F, b >> 7, b & 0x7F, 0xF7))) + await port.drain() + plant(port) + for message in want: + port.send(message) + await port.drain() + log("gate down: sent", COUNT, "plant", PLANT) + + +async def echo(port): + n = 0 + try: + while True: + ts, message = await port.receive() + port.send(message) + n += 1 + except midi.DisconnectedError: + pass + log("echoed", n) + + +async def main(): + try: + import os + + os.remove(LOG) + except OSError: + pass + ble = bledev.mpble.get() + log("serving BLE-MIDI as", NAME, "plant", PLANT) + while True: + port = await midi.serve(ble, name=NAME) + log("connected, mtu", port.connection.mtu) + try: + ts, command = await port.receive(timeout_ms=20000) + if command[:3] == b"\xf0\x7d\x10" and command[3] == 1: + await gate(port) + elif command[:3] == b"\xf0\x7d\x10" and command[3] == 2: + await echo(port) + else: + log("unknown command", command) + except Exception as e: + log("session ended:", type(e).__name__, e) + try: + await port.connection.disconnected(timeout_ms=30000) + except Exception: + await port.aclose() + + +if sys.implementation.name == "micropython": + asyncio.run(main()) diff --git a/tests/bledev_board/reconnect_loop.py b/tests/bledev_board/reconnect_loop.py new file mode 100644 index 00000000..21ebb308 --- /dev/null +++ b/tests/bledev_board/reconnect_loop.py @@ -0,0 +1,72 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""Rapid reconnects from a laptop against the REPL gate's board. + +Needs ``repl_server.py`` serving (from ``/main.py``). Each cycle logs in +through ``bledev.repl.connect()``, runs one line, closes, and waits ``GAP`` +seconds before the next, which is how the REPL gate once failed: bleak +couldn't subscribe on a connection made 1.5 s after the last one closed. +Prints each cycle, how many connection attempts it took, and a total:: + + PYTHONPATH="$(wslpath -w lib)" python.exe "$(wslpath -w tests/bledev_board/reconnect_loop.py)" 100 1.5 + +``--no-retry`` sets ``attempts=1``, the old behaviour, to show the failure. +""" +import asyncio +import sys +import time + +import bledev +import bledev.bleak +import bledev.nus as nus +import bledev.repl as repl + +args = [a for a in sys.argv[1:] if not a.startswith("--")] +N = int(args[0]) if args else 20 +GAP = float(args[1]) if len(args) > 1 else 1.5 +RETRY = "--no-retry" not in sys.argv + +attempts = [] +_helper = nus.connect_and_set_up + + +async def counting(ble, setup, **kwargs): + async def counted(connection): + attempts[-1] += 1 + return await setup(connection) + + if not RETRY: + kwargs["attempts"] = 1 + return await _helper(ble, counted, **kwargs) + + +nus.connect_and_set_up = counting + + +async def main(): + ble = bledev.bleak.BleakBLE() + fails = 0 + retried = 0 + for i in range(N): + attempts.append(0) + t0 = time.monotonic() + try: + link = await repl.connect(ble, "bledev-gate-pw", name="bledev-repl", timeout_ms=15000) + out = await repl.run(link, "%d * 3" % i) + ok = out.strip() == str(i * 3) + await link.close() + status = "OK" if ok else "BAD OUTPUT" + fails += 0 if ok else 1 + except Exception as e: + fails += 1 + status = "FAIL {} {}".format(type(e).__name__, e) + if attempts[-1] > 1: + retried += 1 + print("cycle", i, status, "attempts", attempts[-1], "%.1fs" % (time.monotonic() - t0), flush=True) + await asyncio.sleep(GAP) + print("RESULT {}/{} failed, {} needed a second attempt".format(fails, N, retried)) + return fails + + +sys.exit(1 if asyncio.run(main()) else 0) diff --git a/tests/bledev_board/repl_client.py b/tests/bledev_board/repl_client.py index c285e272..41c2cdba 100644 --- a/tests/bledev_board/repl_client.py +++ b/tests/bledev_board/repl_client.py @@ -7,6 +7,7 @@ board (mpble); see gatt_central.py for the command line. Checks, in order: 1. The right password gets a prompt, and ``123 * 456`` evaluates to 56088. + A 16,890-character print arrives byte for byte. 2. A wrong password, sent together with a line of code that would create ``/pwned``, gets ``Access denied`` and a disconnect, and never a prompt. 3. Logged in again, ``/pwned`` does not exist: the code never ran. @@ -65,6 +66,11 @@ async def main(): link = await repl.connect(ble, PASSWORD, name=NAME, timeout_ms=15000) out = await repl.run(link, "123 * 456") check(out.strip() == "56088", "the right password gets a working prompt", repr(out)) + # A list prints as hundreds of small writes, which is what once let the + # retry timer interleave two flushes: a chunk sent twice, the next lost. + out = await repl.run(link, "print(list(range(3000)))", timeout_ms=30000) + want = str(list(range(3000))) + check(out.strip() == want, "a long print arrives intact", "{} chars, want {}".format(len(out.strip()), len(want))) await repl.run(link, "import os") await repl.run(link, "exec(\"try:\\n os.remove('/pwned')\\nexcept OSError:\\n pass\")") out = await repl.run(link, "'pwned' in os.listdir('/')") diff --git a/tests/bledev_contract.py b/tests/bledev_contract.py index b671c219..1b33939d 100644 --- a/tests/bledev_contract.py +++ b/tests/bledev_contract.py @@ -46,7 +46,7 @@ UnsupportedError, UUID, ) -from bledev import fake, improv, nus, repl +from bledev import fake, improv, midi, nus, repl from bledev.fake import Air, FakeBLE MICROPYTHON = sys.implementation.name == "micropython" @@ -756,6 +756,154 @@ async def improv_scan(air): # ---------------------------------------------------------------- auto +@check +async def setup_retries_a_link_that_drops(air): + # Windows sometimes reports a connection the peripheral never saw; the + # first GATT operation then fails. connect_and_set_up tries again. + board, laptop = _two(air) + attempts = [] + + async def setup(connection): + attempts.append(connection) + if len(attempts) == 1: + raise bledev.DisconnectedError("the first link came up dead") + return "ready" + + served = asyncio.create_task(_serve_twice(board)) + connection, result = await bledev.connect_and_set_up(laptop, setup, name="board", timeout_ms=2000) + equal(result, "ready", "setup's result") + equal(len(attempts), 2, "attempts") + expect(not attempts[0].is_connected(), "the dead link was closed") + expect(connection.is_connected(), "the second link is up") + served.cancel() + + +def _two(air): + board = FakeBLE(air=air) + laptop = FakeBLE(air=air, peripheral=False) + board.register_services(gatt()[0]) + return board, laptop + + +async def _serve_twice(board): + for _ in range(2): + await board.advertise(name="board", timeout_ms=2000) + + +@check +async def setup_gives_up_after_its_attempts(air): + board, laptop = _two(air) + served = asyncio.create_task(_serve_twice(board)) + + async def setup(connection): + await asyncio.sleep(10) + + await raises( + BLETimeoutError, + bledev.connect_and_set_up(laptop, setup, name="board", timeout_ms=2000, attempts=2, setup_timeout_ms=100), + ) + served.cancel() + + +async def midi_pair(air, *, mtu=midi.MTU, **options): + board = FakeBLE(air=air) + laptop = FakeBLE(air=air, peripheral=False) + served = asyncio.create_task(midi.serve(board, name="synth", mtu=mtu, timeout_ms=2000, **options)) + client = await midi.connect(laptop, name="synth", mtu=mtu, timeout_ms=2000, **options) + server = await served + return server, client + + +async def _collect(port, n, buf=None): + """Read ``port`` as a MidiPort until ``n`` messages have come, as bytes.""" + got = bytearray() + buf = buf or bytearray(64) + splitter = midi.Splitter() + messages = [] + while len(messages) < n: + k = port.read(buf) + if k: + messages.extend(splitter.feed(buf[:k])) + else: + await asyncio.sleep(0.001) + return messages + + +def midi_mix(count): + """Notes, controllers, pitch bend, a 300-byte SysEx and clock, interleaved.""" + out = [] + for i in range(count): + k = i % 10 + if k in (0, 1, 2): + out.append(bytes((0x90 | (i % 16), i % 128, 1 + i % 127))) + elif k == 3: + out.append(bytes((0x80 | (i % 16), i % 128, 0))) + elif k in (4, 5): + out.append(bytes((0xB0 | (i % 16), 7, i % 128))) + elif k == 6: + out.append(bytes((0xE0 | (i % 16), i % 128, (i >> 7) % 128))) + elif k in (7, 8): + out.append(b"\xf8") + elif i % 100 == 9: + out.append(bytes([0xF0] + [(i + j) % 128 for j in range(298)] + [0xF7])) + else: + out.append(bytes((0xC0 | (i % 16), i % 128))) + return out + + +@check +async def midi_notes_both_ways(air): + server, client = await midi_pair(air) + equal(server.connection._ble.config("gap_name"), "synth", "GAP name follows the advertised name") + expect(hasattr(client, "read") and hasattr(client, "write") and client.info.direction == "inout", "a MidiPort") + client.write(b"\x90\x3c\x64\x3e\x64") # running status on the way in + equal(await _collect(server, 2), [b"\x90\x3c\x64", b"\x90\x3e\x64"], "central -> peripheral") + server.write(b"\xb0\x07\x40") + ts, message = await client.receive(timeout_ms=1000) + equal(message, b"\xb0\x07\x40", "peripheral -> central") + expect(0 <= ts <= 0x1FFF, "a 13-bit timestamp") + + +@check +async def midi_a_thousand_mixed_messages(air): + server, client = await midi_pair(air) + sent = midi_mix(1000) + for sender, receiver, way in ((client, server, "up"), (server, client, "down")): + for m in sent: + sender.write(m) + await sender.drain() + got = await _collect(receiver, len(sent)) + equal(len(got), len(sent), way + ": count") + equal([m for m in got if m[0] >= 0xF8], [m for m in sent if m[0] >= 0xF8], way + ": real-time, in order") + equal([m for m in got if m[0] < 0xF8], [m for m in sent if m[0] < 0xF8], way + ": the rest, in order") + equal(receiver.decode_errors, 0, way + ": decode errors") + + +@check +async def midi_sysex_at_the_smallest_mtu(air): + server, client = await midi_pair(air, mtu=23) + sysex = bytes([0xF0] + [j % 128 for j in range(300)] + [0xF7]) + server.write(sysex + b"\xf8") + await server.drain() + first = await client.receive(timeout_ms=1000) + second = await client.receive(timeout_ms=1000) + equal(sorted([first[1], second[1]]), sorted([sysex, b"\xf8"]), "SysEx and clock at MTU 23") + + +@check +async def midi_overflow_is_counted(air): + server, client = await midi_pair(air, queue_limit=10) + for i in range(30): + client.write(bytes((0x90, i, 1))) + await client.drain() + for _ in range(100): + if server.dropped: + break + await asyncio.sleep(0.005) + expect(server.overflowed, "an overflow shows") + equal(server.dropped + server.any(), 30, "dropped + kept") + + @check async def auto_picks_or_explains(air): import bledev.auto as auto @@ -821,11 +969,24 @@ def one_slot(self, item): self.event.set() fake._Inbox.put = one_slot + elif name == "midi": + # One packet in 50 arrives with a bit flipped in its last byte. + packet = midi.Port._packet + count = [0] + + def flipping_packet(self, data): + count[0] += 1 + if count[0] % 50 == 0 and len(data) > 2: + data = bytearray(data) + data[-1] ^= 0x01 + packet(self, data) + + midi.Port._packet = flipping_packet else: raise SystemExit("unknown plant " + name) -PLANTS = ("drop", "corrupt", "reorder", "truncate", "overwrite") +PLANTS = ("drop", "corrupt", "reorder", "truncate", "overwrite", "midi") def _print_exception(e): diff --git a/tests/bledev_midi_codec.py b/tests/bledev_midi_codec.py new file mode 100644 index 00000000..55e24bab --- /dev/null +++ b/tests/bledev_midi_codec.py @@ -0,0 +1,225 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""The BLE-MIDI codec, checked against the shared vectors. + +A plain script, not unittest, so the same file runs on CPython and on +MicroPython:: + + python tests/bledev_midi_codec.py + micropython tests/bledev_midi_codec.py + python tests/bledev_midi_codec.py --plant=rollover # must FAIL + +The vectors are ``tests/bledev_midi_vectors.json``, worked out by hand from +the packet format. Then a seeded round trip: thousands of random messages +(notes, controllers, pitch bend, SysEx, real-time, system common) through the +encoder at random packet sizes and back through the decoder. + +``--plant=NAME`` loads the codec with one fault written into its source (see +``PLANTS``), so the checks are shown to catch it. ``tests/test_bledev_midi.py`` +runs every plant. +""" + +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 "." +_LIB = _dir + _sep + ".." + _sep + "lib" + _sep + "bledev" + _sep + "midi_codec.py" +_VECTORS = _dir + _sep + "bledev_midi_vectors.json" + +import json +from binascii import hexlify, unhexlify + +MICROPYTHON = sys.implementation.name == "micropython" + +#: Faults a check must catch: (text in midi_codec.py, what replaces it). +PLANTS = { + # The decoder misses a rollover of the timestamp's low seven bits. + "rollover": ("high = (high + 1) & 0x3F", "high = high"), + # The decoder forgets running status at each packet. + "running": (" high = packet[0] & 0x3F\n", " high = packet[0] & 0x3F\n self._running = 0\n"), + # The encoder skips one byte of every SysEx fragment after the first. + "sysex": (" index += take\n", " index += take + (1 if index > 1 else 0)\n"), + # The encoder writes a header without the timestamp's high bits. + "header": ("self.buf[0] = 0x80 | high", "self.buf[0] = 0x80"), + # The decoder lets a real-time message end a SysEx. + "realtime": ( + " if b >= 0xF8:\n out.append((ts, bytes((b,))))\n", + " if b >= 0xF8:\n self._sysex = None\n out.append((ts, bytes((b,))))\n", + ), +} + + +def load_codec(plant=None): + with open(_LIB) as f: + source = f.read() + if plant: + old, new = PLANTS[plant] + if old not in source: + raise SystemExit("plant {!r} no longer matches midi_codec.py".format(plant)) + source = source.replace(old, new) + print("PLANTED", plant) + namespace = {"__name__": "midi_codec"} + exec(source, namespace) + + class Codec: + pass + + codec = Codec() + for name in ("Decoder", "Encoder", "Splitter", "encode", "decode", "data_length", "TIMESTAMP_MASK"): + setattr(codec, name, namespace[name]) + return codec + + +def h(text): + return unhexlify(text.replace(" ", "")) + + +def show(data): + return hexlify(data).decode() + + +passed = 0 +failed = [] + + +def check(ok, what, detail=""): + global passed + if ok: + passed += 1 + else: + failed.append(what) + print("FAIL", what, detail) + + +def run_vectors(codec, vectors): + for case in vectors["decode"]: + d = codec.Decoder() + got = [] + for p in case["packets"]: + got.extend(d.decode(h(p))) + want = [(ts, h(m)) for ts, m in case["messages"]] + check(got == want, "decode: " + case["name"], "got {}".format([(ts, show(m)) for ts, m in got])) + check(d.errors == case["errors"], "decode errors: " + case["name"], "got {} errors".format(d.errors)) + for case in vectors["encode"]: + messages = [(ts, h(m)) for ts, m in case["messages"]] + got = codec.encode(messages, case["size"], case["running_status"]) + want = [h(p) for p in case["packets"]] + check(got == want, "encode: " + case["name"], "got {}".format([show(p) for p in got])) + back = codec.decode(got) + check( + [m for _, m in back if m[0] < 0xF8] == [m for _, m in messages if m[0] < 0xF8], + "encode then decode: " + case["name"], + ) + + +def run_splitter(codec): + s = codec.Splitter() + out = [] + for chunk in (b"\x90\x3c", b"\x64\x3e\x64", b"\xf0\x01\xf8\x02", b"\x03\xf7\xb0\x07", b"\xfa\x40\xc0\x05\x06"): + out.extend(s.feed(chunk)) + want = [b"\x90\x3c\x64", b"\x90\x3e\x64", b"\xf8", b"\xf0\x01\x02\x03\xf7", b"\xfa", b"\xb0\x07\x40", b"\xc0\x05", b"\xc0\x06"] + check(out == want, "splitter: running status, split writes, real-time inside SysEx and a message", repr(out)) + check(s.desync == 0, "splitter: no desync on a clean stream") + s.feed(b"\x3c\x64") + s2 = codec.Splitter() + s2.feed(b"\x3c\x64") + check(s2.desync == 2, "splitter: data with no status is counted, not guessed") + + +class Rand: + """A small LCG, so both interpreters make the same stream.""" + + def __init__(self, seed): + self.state = seed + + def next(self, n): + self.state = (self.state * 1103515245 + 12345) & 0x7FFFFFFF + return (self.state >> 8) % n + + +def random_messages(r, count): + """``count`` messages with timestamps that move forward and wrap.""" + ts = r.next(8192) + out = [] + for _ in range(count): + ts = (ts + (0 if r.next(3) == 0 else r.next(300))) & 0x1FFF + kind = r.next(20) + ch = r.next(16) + if kind < 7: + m = bytes((0x90 | ch, r.next(128), r.next(128))) + elif kind < 9: + m = bytes((0x80 | ch, r.next(128), r.next(128))) + elif kind < 11: + m = bytes((0xB0 | ch, r.next(128), r.next(128))) + elif kind < 12: + m = bytes((0xE0 | ch, r.next(128), r.next(128))) + elif kind < 13: + m = bytes((0xC0 | ch, r.next(128))) + elif kind < 14: + m = bytes((0xD0 | ch, r.next(128))) + elif kind < 16: + m = bytes((0xF8 + r.next(8),)) + elif kind < 17: + m = (bytes((0xF1, r.next(128))), bytes((0xF2, r.next(128), r.next(128))), bytes((0xF6,)))[r.next(3)] + else: + m = bytes([0xF0] + [r.next(128) for _ in range(r.next(400))] + [0xF7]) + out.append((ts, m)) + return out + + +def run_round_trip(codec, streams, per_stream): + r = Rand(20260924) + total = 0 + bad = 0 + first = None + for n in range(streams): + messages = random_messages(r, per_stream) + size = 5 + r.next(240) + running = r.next(2) == 0 + packets = codec.encode(messages, size, running) + for p in packets: + if len(p) > size and first is None: + first = "packet of {} bytes at size {}".format(len(p), size) + got = codec.decode(packets) + # Real-time may slip ahead into a SysEx; everything else keeps its + # order, and every message keeps its bytes. Timestamps survive, + # except a SysEx's, which is its F7's. + want_rt = [(ts, m) for ts, m in messages if m[0] >= 0xF8] + got_rt = [(ts, m) for ts, m in got if m[0] >= 0xF8] + want_other = [(ts if m[0] != 0xF0 else None, m) for ts, m in messages if m[0] < 0xF8] + got_other = [(ts if m[0] != 0xF0 else None, m) for ts, m in got if m[0] < 0xF8] + total += len(messages) + if got_rt != want_rt or got_other != want_other: + bad += 1 + if first is None: + first = "stream {} at size {}: {} messages in, {} out".format(n, size, len(messages), len(got)) + check(bad == 0 and first is None, "round trip: {} messages in {} streams".format(total, streams), first or "") + return total + + +def main(): + plant = None + for arg in sys.argv[1:]: + if arg.startswith("--plant="): + plant = arg.split("=", 1)[1] + codec = load_codec(plant) + with open(_VECTORS) as f: + vectors = json.load(f) + run_vectors(codec, vectors) + run_splitter(codec) + total = run_round_trip(codec, 150 if MICROPYTHON else 400, 60) + print( + "bledev.midi codec ({}): {} passed, {} failed; {} decode and {} encode vectors, {} round-trip messages".format( + sys.implementation.name, passed, len(failed), len(vectors["decode"]), len(vectors["encode"]), total + ) + ) + if failed: + sys.exit(1) + + +main() diff --git a/tests/bledev_midi_vectors.json b/tests/bledev_midi_vectors.json new file mode 100644 index 00000000..f5418401 --- /dev/null +++ b/tests/bledev_midi_vectors.json @@ -0,0 +1,54 @@ +{ + "about": "BLE-MIDI 1.0 codec vectors, shared by every interpreter (tests/bledev_midi_codec.py). Hex strings, spaces ignored. Every expected value here was worked out by hand from the BLE-MIDI packet format, not captured from the codec. decode: packets in order through one Decoder -> [timestamp, message] pairs and the error count. encode: messages through one Encoder at a packet size -> the exact packets.", + "decode": [ + {"name": "one note-on", "packets": ["80 80 90 3c 64"], "messages": [[0, "90 3c 64"]], "errors": 0}, + {"name": "timestamp high bits from the header", "packets": ["85 81 90 3c 64"], "messages": [[641, "90 3c 64"]], "errors": 0}, + {"name": "two messages, two timestamps", "packets": ["80 81 90 3c 64 82 80 3c 00"], "messages": [[1, "90 3c 64"], [2, "80 3c 00"]], "errors": 0}, + {"name": "running status with a new timestamp", "packets": ["80 81 90 3c 64 82 3e 64"], "messages": [[1, "90 3c 64"], [2, "90 3e 64"]], "errors": 0}, + {"name": "running status, same timestamp, no timestamp byte", "packets": ["80 81 90 3c 64 3e 64 40 64"], "messages": [[1, "90 3c 64"], [1, "90 3e 64"], [1, "90 40 64"]], "errors": 0}, + {"name": "one-byte running status (program change)", "packets": ["80 81 c0 05 06 82 07"], "messages": [[1, "c0 05"], [1, "c0 06"], [2, "c0 07"]], "errors": 0}, + {"name": "timestamp low rolls over: high part +1", "packets": ["85 fe 90 3c 64 81 90 3e 64"], "messages": [[766, "90 3c 64"], [769, "90 3e 64"]], "errors": 0}, + {"name": "13-bit wrap: high 63 rolls to 0", "packets": ["bf fe b0 07 40 83 b0 07 41"], "messages": [[8190, "b0 07 40"], [3, "b0 07 41"]], "errors": 0}, + {"name": "running status across packets", "packets": ["80 81 b0 07 40", "80 82 0a 20"], "messages": [[1, "b0 07 40"], [2, "b0 0a 20"]], "errors": 0}, + {"name": "pitch bend and channel pressure", "packets": ["80 81 e1 00 40 82 d1 7f"], "messages": [[1, "e1 00 40"], [2, "d1 7f"]], "errors": 0}, + {"name": "SysEx in one packet, F7 has its own timestamp", "packets": ["80 81 f0 7e 7f 06 01 82 f7"], "messages": [[2, "f0 7e 7f 06 01 f7"]], "errors": 0}, + {"name": "SysEx across three packets", "packets": ["80 81 f0 01 02 03", "80 04 05", "80 06 83 f7"], "messages": [[3, "f0 01 02 03 04 05 06 f7"]], "errors": 0}, + {"name": "clock inside a SysEx", "packets": ["80 81 f0 01 02 82 f8 03 04", "80 05 83 f7"], "messages": [[2, "f8"], [3, "f0 01 02 03 04 05 f7"]], "errors": 0}, + {"name": "clock opens a continuation packet", "packets": ["80 81 f0 01 02", "80 84 f8 03 04 85 f7"], "messages": [[4, "f8"], [5, "f0 01 02 03 04 f7"]], "errors": 0}, + {"name": "clock between a status and its data", "packets": ["80 81 90 3c 82 f8 64"], "messages": [[2, "f8"], [1, "90 3c 64"]], "errors": 0}, + {"name": "real-time keeps running status", "packets": ["80 81 90 3c 64 82 fa 83 3e 64"], "messages": [[1, "90 3c 64"], [2, "fa"], [3, "90 3e 64"]], "errors": 0}, + {"name": "system common cancels running status", "packets": ["80 81 90 3c 64 82 f1 20 83 3e 64"], "messages": [[1, "90 3c 64"], [2, "f1 20"]], "errors": 2}, + {"name": "tune request has no data", "packets": ["80 81 f6 82 f2 10 20"], "messages": [[1, "f6"], [2, "f2 10 20"]], "errors": 0}, + {"name": "a bad header drops the packet", "packets": ["00 90 3c 64", "c0 81 90 3c 64", "80 81 90 3c 64"], "messages": [[1, "90 3c 64"]], "errors": 2}, + {"name": "a header alone is not a packet", "packets": ["80"], "messages": [], "errors": 1}, + {"name": "a non-SysEx message split across packets is dropped", "packets": ["80 81 90 3c", "80 82 91 3e 64"], "messages": [[2, "91 3e 64"]], "errors": 1}, + {"name": "a status inside a SysEx ends it unfinished", "packets": ["80 81 f0 01 02 82 90 3c 64"], "messages": [[2, "90 3c 64"]], "errors": 1}, + {"name": "data with no status is an error", "packets": ["80 81 3c 64"], "messages": [], "errors": 2} + ], + "encode": [ + {"name": "running status inside a packet", "size": 20, "running_status": true, + "messages": [[1, "90 3c 64"], [1, "90 3e 64"], [2, "90 40 64"]], + "packets": ["80 81 90 3c 64 3e 64 82 40 64"]}, + {"name": "running status off", "size": 20, "running_status": false, + "messages": [[1, "90 3c 64"], [1, "90 3e 64"], [2, "90 40 64"]], + "packets": ["80 81 90 3c 64 81 90 3e 64 82 90 40 64"]}, + {"name": "a full packet starts a new one, with its status again", "size": 9, "running_status": true, + "messages": [[1, "90 3c 64"], [2, "90 3e 64"], [3, "90 40 64"], [4, "90 41 64"]], + "packets": ["80 81 90 3c 64 82 3e 64", "80 83 90 40 64 84 41 64"]}, + {"name": "timestamp high part in the header, rollover inside a packet", "size": 20, "running_status": true, + "messages": [[766, "b0 07 40"], [769, "b0 07 41"]], + "packets": ["85 fe b0 07 40 81 07 41"]}, + {"name": "a gap of 128 ms or more starts a new packet", "size": 20, "running_status": true, + "messages": [[1, "90 3c 64"], [200, "90 3e 64"]], + "packets": ["80 81 90 3c 64", "81 c8 90 3e 64"]}, + {"name": "SysEx split at the packet size", "size": 8, "running_status": true, + "messages": [[5, "f0 01 02 03 04 05 06 07 08 09 0a f7"]], + "packets": ["80 85 f0 01 02 03 04 05", "80 06 07 08 09 0a 85 f7"]}, + {"name": "a clock slips into a SysEx; a timestamp never runs backwards inside a packet", "size": 8, "running_status": true, + "messages": [[5, "f0 01 02 03 04 05 06 07 08 09 0a f7"], [6, "90 3c 64"], [7, "f8"]], + "packets": ["80 85 f0 01 02 03 04 05", "80 87 f8 06 07 08 09 0a", "80 87 f7", "80 86 90 3c 64"]}, + {"name": "system common is never run on", "size": 20, "running_status": true, + "messages": [[1, "f2 10 20"], [1, "f2 10 21"]], + "packets": ["80 81 f2 10 20 81 f2 10 21"]} + ] +} diff --git a/tests/test_bledev.py b/tests/test_bledev.py index bb56e14a..39800487 100644 --- a/tests/test_bledev.py +++ b/tests/test_bledev.py @@ -22,7 +22,7 @@ _TESTS = Path(__file__).resolve().parent CONTRACT = _TESTS / "bledev_contract.py" -PLANTS = ("drop", "corrupt", "reorder", "truncate", "overwrite") +PLANTS = ("drop", "corrupt", "reorder", "truncate", "overwrite", "midi") def _micropython(): diff --git a/tests/test_bledev_midi.py b/tests/test_bledev_midi.py new file mode 100644 index 00000000..b2249fdb --- /dev/null +++ b/tests/test_bledev_midi.py @@ -0,0 +1,58 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""Run the BLE-MIDI codec checks (``bledev_midi_codec.py``) on CPython and +MicroPython, and once per planted codec fault, each of which must fail. + +The shared vectors are ``bledev_midi_vectors.json``. The port and service +checks over the fake radio are in ``bledev_contract.py`` (``midi_*``), run by +``test_bledev.py``. +""" + +from pathlib import Path +import os +import subprocess +import sys +import unittest + +from test_bledev import _micropython + +_TESTS = Path(__file__).resolve().parent +SCRIPT = _TESTS / "bledev_midi_codec.py" +PLANTS = ("rollover", "running", "sysex", "header", "realtime") + + +def _run(interpreter, *args): + return subprocess.run([interpreter, str(SCRIPT), *args], capture_output=True, text=True, timeout=600) + + +class BledevMidiCodec(unittest.TestCase): + def assertPasses(self, result): + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + self.assertIn(" 0 failed", result.stdout) + + def test_cpython(self): + self.assertPasses(_run(sys.executable)) + + def test_micropython(self): + interpreter = _micropython() + if interpreter is None: + message = "no micropython binary: the BLE-MIDI codec did NOT run on MicroPython" + if os.environ.get("PYDEVICES_REQUIRE_MICROPYTHON") == "1": + self.fail(message) + sys.stderr.write("\n*** " + message + " ***\n") + self.skipTest(message) + result = _run(interpreter) + self.assertPasses(result) + self.assertIn("(micropython)", result.stdout) + + def test_every_plant_fails(self): + for plant in PLANTS: + with self.subTest(plant=plant): + result = _run(sys.executable, "--plant=" + plant) + self.assertIn("PLANTED " + plant, result.stdout) + self.assertNotEqual(result.returncode, 0, "plant {!r} went unnoticed:\n{}".format(plant, result.stdout)) + + +if __name__ == "__main__": + unittest.main()