diff --git a/docs/bledev-internals.md b/docs/bledev-internals.md
index 7af66e9d..b8014d81 100644
--- a/docs/bledev-internals.md
+++ b/docs/bledev-internals.md
@@ -17,6 +17,7 @@ standalone module per backend, and an optional selector.
| `mpble.py` | MicroPython, over aioble. |
| `bleak.py` | CPython on Windows, Linux and macOS, over bleak. Central only. |
| `cpble.py` | CircuitPython, over `_bleio`. Both roles. |
+| `webble.py` | Web Bluetooth in a browser, central only. |
| `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. |
@@ -522,6 +523,61 @@ never sees, for a few seconds. From the host the two look the same, so
`connect_and_set_up` backs off between attempts and, when every attempt
drops after encrypting with stored keys, says both.
+## webble
+
+`webble` reaches the page the way `audiodev.web_audio` does, through the `js`
+module, which Pyodide and MicroPython's WebAssembly build both have. The
+runtimes disagree on how Python values cross into JavaScript, so everything
+goes through a few helpers made once with `js.Function.new()`: strings,
+numbers and opaque objects cross, and bytes cross as hex. JS errors come back
+as each runtime's `JsException`, and `_translate()` maps them by name.
+
+What it takes to meet the contract in a browser, all found on hardware:
+
+- **Chrome on Android never matches a `name` filter against a scan
+ response.** A board advertising nus's 128-bit UUID has its name pushed
+ there, so `find(name=, service=)` filters on the service alone and checks
+ the pick's name afterwards. The check caught two wrong picks while the S3s
+ were also advertising nus.
+- **The MTU is invisible.** Chrome on Android asks for 517 right after
+ connecting (logcat: `configureMTU() mtu: 517`, answered with 247 by a
+ board), and Windows negotiates on its own, but neither tells the page.
+ Writes are sized to `WebBLE(mtu=)`, default 23. A write longer than the
+ link carries is cut short on Android without an error, which is what the
+ mock models and `test_bledev_web.py` plants.
+- **First connects fail.** Android gives status 133 / HCI 0x3E ("failed to be
+ established") on about half the first attempts to the P4, so `_connect()`
+ tries three times. One gate run needed two retries.
+- **One GATT operation at a time.** Chrome rejects a second while one is in
+ flight, so each connection serialises them behind a lock.
+- **Discovery is slow the first time.** The browser walks the whole GATT
+ table on the first request, which took longer than nus's 2 s from Windows,
+ so discovery timeouts have a 10 s floor.
+- **MicroPython's asyncio can't `wait_for()` a JS promise**, only a
+ coroutine, so promises with a timeout are wrapped in one.
+
+**The gate** is `tests/bledev_web/`: `gate.py` runs nus's three 16 KB phases
+against `nus_gate_server.py` on a board, on three pages (`pyodide.html`,
+`mpy.html`, `wasm.html`) served by `serve.py`, which also collects what the
+pages log. `?mock=1` swaps in `mock_bluetooth.js`, and
+`tests/test_bledev_web.py` runs that in headless Chromium with two planted
+faults. Against a board, the chooser is the only human step, and both ends
+of it can be automated:
+
+- **Android:** `adb reverse tcp:8765 tcp:8765` makes the page `localhost` on
+ the phone, `adb shell input tap` presses the page's button (a real user
+ gesture), and `uiautomator dump` finds the device's row and the Pair
+ button in the chooser. Chrome needs the Nearby devices permission, and
+ Android's `DeviceAccess` events never fire.
+- **Desktop Chrome:** start it with `--remote-debugging-port` and its own
+ `--user-data-dir`, click the button with `Runtime.evaluate(userGesture=true)`,
+ and answer `DeviceAccess.deviceRequestPrompted` with
+ `DeviceAccess.selectPrompt`. No person is needed at all.
+
+Restart the board's server with a hard reset: after a Ctrl-C, a new
+`nus.serve()` on the P4 twice said it was advertising while nothing was on
+the air.
+
## The checks
**Pairing, without a radio**, on both interpreters: `tests/bledev_pairing.py`
@@ -569,6 +625,8 @@ The serving scripts have a `PLANT` switch, and a planted run must fail:
| `bless_peripheral.py` + `bless_probe.py` | The laptop as a peripheral, through bless, probed by a board |
| `pair_server.py` + `pair_client.py` (laptop) | Pairing. `pair`: a passkey read off the board's console, then the REPL and 20 KB through the file service on the paired link. `reconnect`: from the bond, nobody asked for a passkey. `wrong`: a passkey one off must fail pairing and leave Windows unpaired. `unpaired`: every protected characteristic refuses reads and writes. Plants on the server: `forget`, `justworks`, `open`. `uart_say.py` reads the board's state without mpftp's soft reset |
| `pair_server.py` + `files_pair_client.py` (board) | Board to board: the client pairs on its own, 20 KB both ways byte for byte, then reconnects from the bond. `plant="flip"` and `plant="nopair"` must FAIL |
+| `nus_server.py` + `nus_client.py` | 16 KB up, down and echoed over nus, byte for byte, timed |
+| `tests/bledev_web/nus_gate_server.py` + a gate page | The same, from a browser through webble |
`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.
@@ -852,3 +910,23 @@ mpftp's `ble://` (PyDevices/mpftp `ble-pairing`) used the bond with no password
set: `exec` and 20 KB `put`/`get` byte for byte. Against a just-works board
with a password it paired by itself; against a passkey board with the laptop
unpaired it said to pair once and left no pairing behind.
+**webble to the P4** (through its C6), nus, 16 KB per phase, every byte
+checked, 2026-09-24. Chrome 153 on a Galaxy S21 about a metre away, and on
+Windows 11 through the laptop's Intel radio. KB/s:
+
+| Browser, runtime, page MTU | Up (writes) | Down (notifications) | Echo, each way |
+|---|---|---|---|
+| S21, Pyodide, 247 | 22.8 | 15.8 | 8.6 |
+| S21, PyScript MicroPython, 247 | 15.6 | 19.9 | 9.2 |
+| S21, MicroPython WebAssembly, 247 | 22.3 | 8.5 | 9.8 |
+| S21, Pyodide, 23 | 3.3-4.4 | 10.7-14.1 | 3.9 |
+| S21, Pyodide, 247, planted bit flips | FAIL, 1 at 5000 | FAIL, 1 at 5000 | FAIL, 1 at 5000 |
+| Windows, Pyodide and PyScript MicroPython, 247 | 37-61 | 4-7 | 3.7-4.8 |
+| Windows, MicroPython WebAssembly, 247 | 50.6 | 3.6 | 3.6 |
+
+From Windows the P4 sends 20-byte notifications, because it never learns the
+MTU Windows negotiated ([#80](https://github.com/PyDevices/pydevices/issues/80)).
+Two of three ECHO runs on the phone at MTU 23 (none of five at 247), and
+two of eleven from Windows, lost writes the P4 never received, always while it was notifying
+back ([#79](https://github.com/PyDevices/pydevices/issues/79)). The gate
+reports those as a stall, not a pass.
diff --git a/docs/bledev.md b/docs/bledev.md
index 24f0bb25..8b335bd1 100644
--- a/docs/bledev.md
+++ b/docs/bledev.md
@@ -36,8 +36,6 @@ transport can talk to it too.
| `bledev.webble` | Browsers: PyScript, the Workbench simulator | yes | no |
| `bledev.fake` | Anywhere: tests | yes | yes |
-`bledev.webble` is being written; the other four are here.
-
Roles aren't symmetric. A laptop or a browser can only scan and connect, so
when a board talks to one, the board advertises. Only board-to-board links put
a board in the central role. `ble.capabilities()` tells you which roles you
@@ -154,6 +152,38 @@ async with ble.scan(5000, active=True) as scanner:
A device can come back more than once as its advertisement and scan response
arrive, and `device.name` is whatever the scan had seen by then.
+## In a browser
+
+`bledev.webble` runs the same code in Chrome or Edge, on desktop or Android,
+under PyScript (Pyodide or MicroPython) or the MicroPython WebAssembly build
+Workbench uses. The page must be served from `https://` or `http://localhost`.
+
+```python
+from bledev.webble import WebBLE
+import bledev.nus as nus
+
+ble = WebBLE(mtu=247) # see below for the MTU
+link = await nus.connect(ble, name="rack") # call this from a click or tap
+```
+
+Three things differ from a board:
+
+- **The browser's chooser is the scan.** `find()` and `nus.connect()` open
+ it, and the person picks the device. It only opens during a click or tap,
+ so start from one. `scan()` raises `UnsupportedError`.
+- **Only services you name are reachable.** The one in `find()`'s filter is;
+ list any others as `WebBLE(services=[...])`.
+- **The browser won't tell you the MTU.** `connection.mtu` is 23 unless you
+ pass `WebBLE(mtu=...)`, and writes are sized to it. Chrome asks for a large
+ MTU on every connect (a board running nus ends up at 247), but a write
+ longer than the link carries can be cut short without an error, so only
+ raise it when you know. At 23 everything still works, about five times
+ slower.
+
+A Web Bluetooth page can be tested without a person: Chrome's DevTools
+protocol can answer the chooser. How the gate was run on a phone and a
+laptop is in [bledev-internals.md](bledev-internals.md#webble).
+
## Byte streams with nus
`bledev.nus` gives you a `Link` from either end, and it reads like an asyncio
diff --git a/lib/bledev/mpble.py b/lib/bledev/mpble.py
index dfd3b9ba..6e701ad0 100644
--- a/lib/bledev/mpble.py
+++ b/lib/bledev/mpble.py
@@ -69,6 +69,11 @@
_IRQ_GATTS_WRITE = 3
_IRQ_PERIPHERAL_DISCONNECT = 8
_IRQ_MTU_EXCHANGED = 21
+_PASSKEY_INPUT = 2 # gap_passkey's action for typing in the peer's passkey
+_EINVAL = 22
+# As central with a passkey to type: how long to wait for the stack to ask
+# before asking anyway (see _MPConnection._type_passkey).
+_ASK_WITHIN_MS = 1500
_FLAG_READ_ENCRYPTED = 0x0200
_FLAG_WRITE_ENCRYPTED = 0x1000
@@ -575,7 +580,9 @@ def on_change(conn, encrypted, authenticated, bonded, key_size):
try:
with _Translate("pair"):
_aioble_core.ble.gap_pair(handle)
- deadline = time.ticks_add(time.ticks_ms(), timeout_ms)
+ started = time.ticks_ms()
+ deadline = time.ticks_add(started, timeout_ms)
+ typed = passkey is None
# The stack reports the outcome as an encryption update, success
# or not, so a wrong passkey fails here at once, not at a timeout.
while not updates:
@@ -583,13 +590,44 @@ def on_change(conn, encrypted, authenticated, bonded, key_size):
raise PairingError("pair: the peer hung up")
if time.ticks_diff(deadline, time.ticks_ms()) <= 0:
raise BLETimeoutError("pair: timed out")
+ if not typed and handle not in security._asked and time.ticks_diff(time.ticks_ms(), started) > _ASK_WITHIN_MS:
+ typed = True
+ await self._type_passkey(handle, deadline, updates)
await sleep_ms(20)
finally:
security.remove_listener(on_change)
security.expect_passkey(handle, None)
+ security._asked.discard(handle)
if not security.state(handle)[0]:
raise PairingError("pair: refused (wrong passkey, or the peer lost its keys)")
+ async def _type_passkey(self, handle, deadline, updates):
+ # MicroPython's NimBLE bindings raise _IRQ_PASSKEY_ACTION only on
+ # connections the board accepted, never on one it made
+ # (peripheral_gap_event_cb in extmod/nimble/modbluetooth_nimble.c
+ # doesn't handle it; micropython/micropython#19470 moves it), so a
+ # board as central is never asked for the passkey and pairing stalls
+ # until the stack's 30 s timeout. When the stack hasn't asked by now,
+ # ask for the passkey here and hand it to the stack ourselves: NimBLE
+ # takes it once its pairing reaches the confirm step and answers
+ # EINVAL before that, so keep offering it until it's taken.
+ from . import security
+
+ given = security._inputs.get(handle)
+ if callable(given):
+ given = given()
+ given = int(given)
+ # If the stack does ask after all, it answers from this, not by asking twice.
+ security._inputs[handle] = given
+ while not updates and handle not in security._asked and self.is_connected():
+ try:
+ _aioble_core.ble.gap_passkey(handle, _PASSKEY_INPUT, given)
+ return
+ except OSError as e:
+ if e.errno != _EINVAL or time.ticks_diff(deadline, time.ticks_ms()) <= 0:
+ return # taken already (EALREADY), or the pairing is over
+ await sleep_ms(50)
+
async def unpair(self):
from . import security
diff --git a/lib/bledev/security.py b/lib/bledev/security.py
index 49d8fbc3..8a67b61e 100644
--- a/lib/bledev/security.py
+++ b/lib/bledev/security.py
@@ -60,6 +60,9 @@ def const(x):
# conn_handle -> what to type in when this board, as central, is asked for the
# peer's passkey: an int, or a callable returning one.
_inputs = {}
+# Connections the stack has asked, as central, for the peer's passkey. Stock
+# MicroPython's NimBLE never asks on a connection it made (see mpble's pair()).
+_asked = set()
# The server's side: what to do when the stack wants a passkey shown or a
# number confirmed. Set by configure().
_show = None
@@ -438,7 +441,9 @@ def _irq(event, data):
core.ble.gap_passkey(conn, action, passkey)
if _show is not None:
_defer(_show, passkey)
- elif action == _ACTION_INPUT and not callable(_inputs.get(conn)):
+ if action == _ACTION_INPUT:
+ _asked.add(conn)
+ if action == _ACTION_INPUT and not callable(_inputs.get(conn)):
given = _inputs.get(conn)
# No passkey to give: answer with one that can't be relied on to
# match, so the pairing fails now rather than at a timeout.
@@ -451,6 +456,7 @@ def _irq(event, data):
conn = data[0]
had = _links.pop(conn, None)
_inputs.pop(conn, None)
+ _asked.discard(conn)
if had is None and _hide is not None:
_defer(_call_hide, None)
return None
diff --git a/lib/bledev/webble.py b/lib/bledev/webble.py
new file mode 100644
index 00000000..5e6af6f5
--- /dev/null
+++ b/lib/bledev/webble.py
@@ -0,0 +1,638 @@
+"""Web Bluetooth backend for browsers. Central only.
+
+ from bledev.webble import WebBLE
+ ble = WebBLE()
+ link = await nus.connect(ble, name="rack") # from a click handler
+
+It runs wherever Python can reach the page's JavaScript: PyScript on Pyodide
+or MicroPython, and the direct MicroPython WebAssembly build that Workbench
+uses. It needs Chrome or Edge (desktop or Android) and a secure context,
+``https://`` or ``http://localhost``. Firefox and iOS Safari have no Web
+Bluetooth.
+
+A browser differs from a board in three ways, and this backend says so
+rather than pretending:
+
+* **You don't scan; the browser's chooser does.** ``find()`` opens the
+ device chooser filtered by ``name`` and ``service``, and the person picks.
+ The chooser only opens during a click or a tap (transient user
+ activation), so call ``find()`` or ``nus.connect()`` from one. ``scan()``
+ raises :class:`bledev.UnsupportedError`.
+* **Only services you named are reachable.** The browser grants the
+ services in the ``find()`` filter and those passed as
+ ``WebBLE(services=[...])``, and hides the rest from discovery.
+* **The browser never reports the ATT MTU.** It negotiates one on its own,
+ and a write longer than that can be cut short without an error, so
+ ``connection.mtu`` is 23 (the one value every link has) unless you pass
+ ``WebBLE(mtu=...)`` knowing the link negotiated more. The peripheral can
+ see the real value.
+"""
+
+try:
+ import asyncio
+except ImportError: # pragma: no cover
+ import uasyncio as asyncio
+
+import json
+
+from . import (
+ BLE,
+ BLEError,
+ BLETimeoutError,
+ ClientCharacteristic,
+ ClientService,
+ Connection,
+ DEFAULT_MTU,
+ Device,
+ DisconnectedError,
+ FLAG_INDICATE,
+ FLAG_NOTIFY,
+ FLAG_READ,
+ FLAG_WRITE,
+ FLAG_WRITE_NO_RESPONSE,
+ GattError,
+ MAX_MTU,
+ UnsupportedError,
+ UUID,
+ wait_ms,
+)
+
+try:
+ import js
+except ImportError: # not in a browser; importing still works for tooling
+ js = None
+
+
+def _find_create_proxy():
+ for name in ("pyodide.ffi", "jsffi", "pyscript.ffi"):
+ try:
+ module = __import__(name, None, None, ["create_proxy"])
+ return module.create_proxy
+ except (ImportError, AttributeError):
+ pass
+ return None
+
+
+create_proxy = _find_create_proxy() if js is not None else None
+
+# Everything that crosses into JavaScript goes through these few helpers, so
+# Pyodide and MicroPython see the same thing: strings, numbers, booleans and
+# opaque objects. Bytes cross as hex, which both runtimes convert the same way
+# and which costs nothing next to a radio.
+_HELPERS_JS = """
+const hex = (v) => {
+ let s = "";
+ for (let i = 0; i < v.byteLength; i++) {
+ const b = v.getUint8(i);
+ s += (b < 16 ? "0" : "") + b.toString(16);
+ }
+ return s;
+};
+return {
+ available: () => !!(globalThis.navigator && navigator.bluetooth),
+ toBytes: (h) => {
+ const a = new Uint8Array(h.length >> 1);
+ for (let i = 0; i < a.length; i++) a[i] = parseInt(h.substr(i * 2, 2), 16);
+ return a;
+ },
+ hex: hex,
+ props: (c) => {
+ const p = c.properties;
+ return (p.read ? 0x02 : 0) | (p.writeWithoutResponse ? 0x04 : 0) | (p.write ? 0x08 : 0)
+ | (p.notify ? 0x10 : 0) | (p.indicate ? 0x20 : 0);
+ },
+ listen: (target, type, cb) => {
+ const f = (e) => cb(type === "characteristicvaluechanged" ? hex(e.target.value) : "");
+ target.addEventListener(type, f);
+ return f;
+ },
+ unlisten: (target, type, f) => target.removeEventListener(type, f),
+ errorName: (e) => (e && e.name) ? String(e.name) : "",
+ errorMessage: (e) => (e && e.message) ? String(e.message) : String(e),
+ isNull: (v) => v === null || v === undefined,
+ length: (a) => a.length,
+ item: (a, i) => a[i],
+};
+"""
+
+_helpers = None
+
+
+def _h():
+ global _helpers
+ if _helpers is None:
+ if js is None:
+ raise UnsupportedError("bledev.webble needs a browser: PyScript, Pyodide or MicroPython WebAssembly")
+ _helpers = js.Function.new(_HELPERS_JS)()
+ return _helpers
+
+
+def available():
+ """True when this page can use Web Bluetooth."""
+ if js is None:
+ return False
+ try:
+ return bool(_h().available())
+ except Exception:
+ return False
+
+
+def _to_js_bytes(data):
+ return _h().toBytes(bytes(data).hex())
+
+
+def _from_hex(text):
+ return bytes.fromhex(str(text))
+
+
+def _items(array):
+ h = _h()
+ return [h.item(array, i) for i in range(int(h.length(array)))]
+
+
+def _proxy(fn):
+ return create_proxy(fn) if create_proxy is not None else fn
+
+
+# ---------------------------------------------------------------- errors
+
+
+def _error_parts(e):
+ """(name, message) of a JavaScript error, from either runtime's exception."""
+ name = ""
+ message = ""
+ for holder in (e, getattr(e, "js_error", None)):
+ if holder is None:
+ continue
+ try:
+ n = getattr(holder, "name", None)
+ if n and isinstance(n, str) and not name:
+ name = n
+ m = getattr(holder, "message", None)
+ if m and isinstance(m, str) and not message:
+ message = m
+ except Exception:
+ pass
+ if not name and getattr(e, "args", None):
+ # MicroPython: JsException(error, name, message)
+ args = e.args
+ if len(args) >= 2 and isinstance(args[1], str):
+ name = args[1]
+ if len(args) >= 3 and isinstance(args[2], str):
+ message = args[2]
+ text = str(e)
+ if not name:
+ head = text.split(":", 1)[0].strip()
+ if head.endswith("Error") and " " not in head:
+ name = head
+ return name, message or text
+
+
+def _is_js_error(e):
+ return type(e).__name__ in ("JsException", "JsError") or hasattr(e, "js_error")
+
+
+def _translate(e, what, connection=None):
+ """A bledev error for the JavaScript error ``e``, or ``None`` to re-raise it."""
+ if isinstance(e, BLEError):
+ return e
+ if not _is_js_error(e):
+ return None
+ name, message = _error_parts(e)
+ detail = "{}: {} ({})".format(what, message, name or "Error")
+ if connection is not None and not connection.is_connected():
+ return DisconnectedError(detail)
+ if name == "NetworkError":
+ # Chrome's word for a dropped link, and for a GATT request that failed
+ # on the air. The disconnect event may lag the rejection by a moment.
+ if "disconnect" in message.lower():
+ return DisconnectedError(detail)
+ return GattError(0, detail)
+ if name == "NotSupportedError":
+ return UnsupportedError(detail)
+ if name == "NotAllowedError" and "gatt" in message.lower():
+ return GattError(0, detail)
+ return BLEError(detail)
+
+
+async def _settle(promise):
+ # MicroPython's asyncio only wraps coroutines in tasks, and a JS promise
+ # isn't one, so wait_for() needs this around it.
+ return await promise
+
+
+async def _await(promise, what, connection=None, timeout_ms=None):
+ try:
+ if timeout_ms is None:
+ return await promise
+ return await wait_ms(_settle(promise), timeout_ms, what)
+ except BLETimeoutError:
+ raise
+ except Exception as e:
+ err = _translate(e, what, connection)
+ if err is None:
+ raise
+ raise err
+
+
+# ---------------------------------------------------------------- the adapter
+
+
+class WebBLE(BLE):
+ """The browser's Bluetooth, through ``navigator.bluetooth``. Central only.
+
+ ``services`` lists the service UUIDs your page will use beyond those it
+ passes to ``find()``: the browser hides every service it wasn't told
+ about. ``mtu`` is the ATT MTU writes are sized to (see the module notes);
+ leave it at 23 unless you know the link negotiated more. ``queue_limit``
+ bounds each notification queue.
+ """
+
+ backend = "webble"
+
+ #: How many times ``connect()`` tries before a failed connection attempt
+ #: is raised. ``connect_retries`` counts the extra tries made.
+ CONNECT_ATTEMPTS = 3
+
+ def __init__(self, *, services=(), mtu=DEFAULT_MTU, queue_limit=512):
+ if not available():
+ raise UnsupportedError(
+ "this browser has no Web Bluetooth: use Chrome or Edge on desktop or "
+ "Android, from https:// or http://localhost"
+ )
+ self.bluetooth = js.navigator.bluetooth
+ self.optional_services = [UUID(s) for s in services]
+ self.link_mtu = int(mtu)
+ self.queue_limit = queue_limit
+ self._requested_mtu = DEFAULT_MTU
+ self._gap_name = None
+ self._connections = []
+ self.connect_retries = 0
+
+ def capabilities(self):
+ caps = BLE.capabilities(self)
+ caps.update(
+ central=True,
+ peripheral=False,
+ mtu=self.link_mtu,
+ max_mtu=MAX_MTU,
+ indicate=True,
+ )
+ return caps
+
+ def config(self, *names, **settings):
+ # The browser negotiates the MTU on its own and never says what it
+ # got, so an MTU set here is recorded, not used; see WebBLE(mtu=...).
+ for key, value in settings.items():
+ if key == "mtu":
+ self._requested_mtu = int(value)
+ else:
+ raise UnsupportedError("webble has no setting {!r}".format(key))
+ if names:
+ if len(names) != 1:
+ raise ValueError("read one setting at a time")
+ if names[0] == "mtu":
+ return self._requested_mtu
+ raise UnsupportedError("webble has no setting {!r}".format(names[0]))
+ return None
+
+ def close(self):
+ for connection in list(self._connections):
+ connection._drop()
+ self._connections = []
+
+ # -- central
+
+ def scan(self, duration_ms, *, active=False, interval_us=None, window_us=None):
+ raise UnsupportedError(
+ "browsers don't scan for pages; find() opens the browser's device chooser instead"
+ )
+
+ def _request_options(self, name, service):
+ filters = []
+ if name is not None or service is not None:
+ f = {}
+ # Chrome on Android matches a name filter only against the
+ # advertisement itself, never the scan response, and a board that
+ # advertises a 128-bit service (nus does) has its name pushed into
+ # the scan response. So with a service to filter on, filter on that
+ # alone; the chooser still shows names, and find() checks the pick.
+ if name is not None and service is None:
+ f["name"] = name
+ if service is not None:
+ f["services"] = [str(UUID(service))]
+ filters.append(f)
+ optional = [str(u) for u in self.optional_services]
+ if service is not None and str(UUID(service)) not in optional:
+ optional.append(str(UUID(service)))
+ options = {"optionalServices": optional}
+ if filters:
+ options["filters"] = filters
+ else:
+ options["acceptAllDevices"] = True
+ return js.JSON.parse(json.dumps(options))
+
+ async def find(self, name=None, service=None, *, timeout_ms=10000, active=True):
+ """Open the browser's device chooser and return the :class:`Device` picked.
+
+ The chooser lists devices matching ``name`` and/or ``service`` (all
+ devices if neither), and ``service`` becomes reachable once connected.
+ Given both, the chooser filters on ``service`` alone and the pick must
+ carry ``name``, because Chrome on Android can't match a name that
+ only a scan response carries. A name-only ``find()`` works there only
+ if the board's name fits in its advertisement.
+ Call it from a click or a tap: the browser refuses otherwise.
+ ``timeout_ms`` doesn't apply, because a person is choosing. Closing
+ the chooser without choosing raises :class:`BLETimeoutError`, as a
+ scan that found nothing does.
+ """
+ options = self._request_options(name, service)
+ try:
+ native = await self.bluetooth.requestDevice(options)
+ except Exception as e:
+ ename, message = _error_parts(e)
+ if ename == "NotFoundError":
+ raise BLETimeoutError("no device chosen: {}".format(message))
+ if ename == "SecurityError":
+ raise BLEError(
+ "the browser refused to open its device chooser ({}). Call find() "
+ "from a click or tap handler, on https:// or localhost".format(message)
+ )
+ err = _translate(e, "requestDevice")
+ if err is None:
+ raise
+ raise err
+ dev_name = native.name
+ dev_name = None if _h().isNull(dev_name) else str(dev_name)
+ if name is not None and service is not None and dev_name != name:
+ raise BLEError("the chooser returned {!r}, not {!r}".format(dev_name, name))
+ return Device(self, str(native.id), name=dev_name, native=native)
+
+ async def _connect(self, device, timeout_ms, **options):
+ native = device.native
+ if native is None:
+ raise BLEError("a browser can only connect to a device its chooser returned")
+ # A first connection attempt often fails at the link layer on Android
+ # (status 133, HCI 0x3E "failed to be established"); measured on the
+ # P4 through its C6 it was about half of them, and the next attempt
+ # usually works. So try again, up to CONNECT_ATTEMPTS in all.
+ attempt = 0
+ while True:
+ attempt += 1
+ connection = _WebConnection(self, device, native)
+ try:
+ await connection._open(timeout_ms)
+ except GattError as e:
+ if attempt >= self.CONNECT_ATTEMPTS or "Connection attempt failed" not in str(e):
+ raise
+ self.connect_retries += 1
+ continue
+ self._connections.append(connection)
+ return connection
+
+
+# The first discovery request makes the browser walk the peer's whole GATT
+# table, which took over 2 s (nus's default) from Chrome on Windows to the P4.
+# Discovery timeouts shorter than this are raised to it.
+DISCOVERY_FLOOR_MS = 10000
+
+
+def _discovery_timeout(timeout_ms):
+ return None if timeout_ms is None else max(timeout_ms, DISCOVERY_FLOOR_MS)
+
+
+class _WebConnection(Connection):
+ def __init__(self, ble, device, native):
+ Connection.__init__(self, ble, device, "central")
+ self._native = native
+ self._server = None
+ self._down = asyncio.Event()
+ self._closed = False
+ self._characteristics = []
+ self._lock = asyncio.Lock()
+ self._mtu = ble.link_mtu
+ self._on_disconnect = _proxy(self._disconnected_event)
+ self._listener = None
+
+ def _disconnected_event(self, *args):
+ self._drop()
+
+ def _drop(self):
+ if self._closed:
+ return
+ self._closed = True
+ self._down.set()
+ for characteristic in self._characteristics:
+ characteristic._wake()
+ try:
+ _h().unlisten(self._native, "gattserverdisconnected", self._listener)
+ except Exception:
+ pass
+ try:
+ self._ble._connections.remove(self)
+ except ValueError:
+ pass
+
+ async def _open(self, timeout_ms):
+ self._listener = _h().listen(self._native, "gattserverdisconnected", self._on_disconnect)
+ try:
+ self._server = await _await(self._native.gatt.connect(), "connect", None, timeout_ms)
+ except BLETimeoutError:
+ try:
+ self._native.gatt.disconnect()
+ except Exception:
+ pass
+ self._drop()
+ raise
+ except BaseException:
+ self._drop()
+ raise
+
+ def is_connected(self):
+ if self._closed or self._server is None:
+ return False
+ try:
+ return bool(self._native.gatt.connected)
+ except Exception:
+ return False
+
+ async def disconnect(self, timeout_ms=2000):
+ if not self._closed:
+ try:
+ self._native.gatt.disconnect()
+ except Exception:
+ pass
+ # Chrome fires gattserverdisconnected for a local disconnect too,
+ # but don't depend on it.
+ try:
+ await wait_ms(self._down.wait(), timeout_ms, "disconnect")
+ except BLETimeoutError:
+ pass
+ self._drop()
+
+ async def disconnected(self, timeout_ms=None):
+ if not self._closed:
+ await wait_ms(self._down.wait(), timeout_ms, "disconnected")
+
+ def _live(self, what):
+ if not self.is_connected():
+ raise DisconnectedError("{}: not connected".format(what))
+
+ async def _gatt(self, promise_fn, what, timeout_ms=None):
+ # Chrome rejects a second GATT operation while one is in flight
+ # ("GATT operation already in progress"), so run them one at a time.
+ self._live(what)
+ async with self._lock:
+ self._live(what)
+ return await _await(promise_fn(), what, self, timeout_ms)
+
+ async def _discover_services(self, uuid, timeout_ms):
+ timeout_ms = _discovery_timeout(timeout_ms)
+ server = self._server
+ try:
+ if uuid is None:
+ found = await self._gatt(lambda: server.getPrimaryServices(), "service discovery", timeout_ms)
+ else:
+ found = await self._gatt(
+ lambda: server.getPrimaryServices(str(uuid)), "service discovery", timeout_ms
+ )
+ except BLEError as e:
+ # Nothing matching, or a service the page didn't ask for: the
+ # browser's two ways of saying there is none it will show us.
+ if "NotFoundError" in str(e):
+ return []
+ if "SecurityError" in str(e):
+ raise BLEError(
+ "{} (list the service in find()'s filter or WebBLE(services=[...]))".format(e)
+ )
+ raise
+ # Browsers expose no handles. Chrome returns services in discovery
+ # order, which is the server's.
+ return [_WebClientService(self, s) for s in _items(found)]
+
+
+class _WebClientService(ClientService):
+ def __init__(self, connection, native):
+ ClientService.__init__(self, connection, str(native.uuid))
+ self._native = native
+
+ async def _discover_characteristics(self, uuid, timeout_ms):
+ timeout_ms = _discovery_timeout(timeout_ms)
+ native = self._native
+ try:
+ if uuid is None:
+ found = await self.connection._gatt(
+ lambda: native.getCharacteristics(), "characteristic discovery", timeout_ms
+ )
+ else:
+ found = await self.connection._gatt(
+ lambda: native.getCharacteristics(str(uuid)), "characteristic discovery", timeout_ms
+ )
+ except BLEError as e:
+ if "NotFoundError" in str(e):
+ return []
+ raise
+ return [_WebClientCharacteristic(self, c) for c in _items(found)]
+
+
+class _WebClientCharacteristic(ClientCharacteristic):
+ def __init__(self, service, native):
+ ClientCharacteristic.__init__(self, service, str(native.uuid), int(_h().props(native)))
+ self._native = native
+ self._notify_queue = []
+ self._indicate_queue = []
+ self._overrun = 0
+ self._event = asyncio.Event()
+ self._listener = None
+ self._on_value = _proxy(self._value_event)
+ self._route_indications = False
+ service.connection._characteristics.append(self)
+
+ def _value_event(self, text):
+ queue = self._indicate_queue if self._route_indications else self._notify_queue
+ if len(queue) >= self.connection._ble.queue_limit:
+ self._overrun += 1
+ else:
+ queue.append(_from_hex(text))
+ self._event.set()
+
+ def _wake(self):
+ self._event.set()
+
+ async def read(self, timeout_ms=1000):
+ self._check(FLAG_READ, "read")
+ native = self._native
+ view = await self.connection._gatt(lambda: native.readValue(), "read", timeout_ms)
+ return _from_hex(_h().hex(view))
+
+ async def write(self, data, response=None, timeout_ms=1000):
+ data = bytes(data)
+ response = self._want_response(response)
+ self.connection._live("write")
+ limit = self.connection.mtu - 3
+ if len(data) > limit:
+ raise BLEError(
+ "write of {} bytes exceeds the ATT payload of {} (mtu {} - 3)".format(
+ len(data), limit, self.connection.mtu
+ )
+ )
+ native = self._native
+ value = _to_js_bytes(data)
+ if response:
+ await self.connection._gatt(lambda: native.writeValueWithResponse(value), "write", timeout_ms)
+ else:
+ await self.connection._gatt(lambda: native.writeValueWithoutResponse(value), "write", timeout_ms)
+
+ async def subscribe(self, notify=True, indicate=False):
+ if notify:
+ self._check(FLAG_NOTIFY, "notify")
+ if indicate:
+ self._check(FLAG_INDICATE, "indicate")
+ if notify and indicate:
+ raise UnsupportedError(
+ "a browser subscribes to notifications or indications, not both"
+ )
+ native = self._native
+ if not notify and not indicate:
+ await self.connection._gatt(lambda: native.stopNotifications(), "unsubscribe")
+ if self._listener is not None:
+ _h().unlisten(native, "characteristicvaluechanged", self._listener)
+ self._listener = None
+ return
+ if notify is False and self.properties & FLAG_NOTIFY:
+ # startNotifications() picks notifications whenever the
+ # characteristic has them, so it can't be asked for indications.
+ raise UnsupportedError("a browser can't choose indications on a characteristic that also notifies")
+ self._route_indications = not notify
+ # Listen first, so nothing that arrives as the subscription lands is lost.
+ if self._listener is None:
+ self._listener = _h().listen(native, "characteristicvaluechanged", self._on_value)
+ await self.connection._gatt(lambda: native.startNotifications(), "subscribe")
+
+ async def _next(self, queue, timeout_ms, what):
+ if timeout_ms is None:
+ return await self._take(queue, what)
+ return await wait_ms(self._take(queue, what), timeout_ms, what)
+
+ async def _take(self, queue, what):
+ while True:
+ if self._overrun:
+ raise BLEError(
+ "{} queue for {} overran: {} dropped because the reader fell {} behind".format(
+ what, self.uuid, self._overrun, self.connection._ble.queue_limit
+ )
+ )
+ if queue:
+ return queue.pop(0)
+ if not self.connection.is_connected():
+ raise DisconnectedError("{}: disconnected".format(what))
+ self._event.clear()
+ await self._event.wait()
+
+ async def notified(self, timeout_ms=None):
+ self._check(FLAG_NOTIFY, "notify")
+ return await self._next(self._notify_queue, timeout_ms, "notified")
+
+ async def indicated(self, timeout_ms=None):
+ self._check(FLAG_INDICATE, "indicate")
+ return await self._next(self._indicate_queue, timeout_ms, "indicated")
diff --git a/mip-split.toml b/mip-split.toml
index 10bcb5d8..0d3f4004 100644
--- a/mip-split.toml
+++ b/mip-split.toml
@@ -80,9 +80,8 @@ host-only = [
own-package = true
requires = ["aioble"]
# The central-only host backends go in pydevices-desktop, as for the other
-# packages here. List `webble` when that module lands; a name listed before
-# its file exists fails the sync.
-host-only = ["bleak"]
+# packages here. A name listed before its file exists fails the sync.
+host-only = ["bleak", "webble"]
# On PyPI bledev stays inside the pydevices wheel. bleak is optional, behind
# `pip install "pydevices[ble]"`.
pypi-extras = { ble = ["bleak>=1.0"] }
diff --git a/pydevices-desktop.toml b/pydevices-desktop.toml
index 500cecdb..4d5f0e11 100644
--- a/pydevices-desktop.toml
+++ b/pydevices-desktop.toml
@@ -35,6 +35,7 @@
"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"
"https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/bledev/security.py" = "/lib/bledev/security.py"
+"https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/bledev/webble.py" = "/lib/bledev/webble.py"
"https://raw.githubusercontent.com/PyDevices/pydevices/main/board_configs/desktop/board_config.py" = "/lib/board_config.py"
"https://raw.githubusercontent.com/PyDevices/pydevices/main/board_configs/desktop/board_peripherals.py" = "/lib/board_peripherals.py"
"https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/boarddev.py" = "/lib/boarddev.py"
diff --git a/tests/bledev_board/connect_loop_client.py b/tests/bledev_board/connect_loop_client.py
new file mode 100644
index 00000000..1aa49701
--- /dev/null
+++ b/tests/bledev_board/connect_loop_client.py
@@ -0,0 +1,61 @@
+# SPDX-FileCopyrightText: 2026 Brad Barnett
+#
+# SPDX-License-Identifier: MIT
+"""Connect reliability, the central: find ``connect_loop_server.py`` and
+connect to it ``N`` times, counting connects that time out.
+
+Each attempt scans for the board (up to ``FIND_MS``), then connects with a
+``CONNECT_MS`` timeout at the default connection interval (``FAST`` asks
+for 7.5-15 ms), holds the link 300 ms and hangs up. Ends with a ``CONNECTS``
+line: attempts, connected, timed out, other errors, not found, and the
+connect times in ms.
+"""
+import asyncio
+import time
+
+import bledev
+import bledev.mpble
+
+NAME = "bledev-conn-loop"
+N = 10
+FIND_MS = 20000
+CONNECT_MS = 10000
+FAST = False
+
+
+async def main():
+ ble = bledev.mpble.get()
+ conn_kw = {"min_conn_interval_us": 7500, "max_conn_interval_us": 15000} if FAST else {}
+ ok = timeouts = other = missing = 0
+ times = []
+ for i in range(N):
+ try:
+ device = await ble.find(name=NAME, timeout_ms=FIND_MS)
+ except bledev.BLETimeoutError:
+ missing += 1
+ print("attempt", i + 1, "not found")
+ continue
+ t = time.ticks_ms()
+ try:
+ conn = await device.connect(timeout_ms=CONNECT_MS, **conn_kw)
+ except bledev.BLETimeoutError as e:
+ timeouts += 1
+ print("attempt", i + 1, "connect timed out after", time.ticks_diff(time.ticks_ms(), t), "ms:", e)
+ await asyncio.sleep_ms(500)
+ continue
+ except bledev.BLEError as e:
+ other += 1
+ print("attempt", i + 1, "connect failed:", repr(e))
+ await asyncio.sleep_ms(500)
+ continue
+ dt = time.ticks_diff(time.ticks_ms(), t)
+ ok += 1
+ times.append(dt)
+ print("attempt", i + 1, "connected in", dt, "ms")
+ await asyncio.sleep_ms(300)
+ await conn.disconnect()
+ await asyncio.sleep_ms(1000)
+ print("CONNECTS", N, "ok", ok, "timeouts", timeouts, "other", other, "not found", missing, "times", sorted(times))
+
+
+asyncio.run(main())
diff --git a/tests/bledev_board/connect_loop_server.py b/tests/bledev_board/connect_loop_server.py
new file mode 100644
index 00000000..dd005d90
--- /dev/null
+++ b/tests/bledev_board/connect_loop_server.py
@@ -0,0 +1,31 @@
+# SPDX-FileCopyrightText: 2026 Brad Barnett
+#
+# SPDX-License-Identifier: MIT
+"""Connect reliability, the peripheral: advertise, take a connection, wait for
+it to end, and advertise again, forever. Pair with ``connect_loop_client.py``.
+
+Prints ``accepted N`` for each connection.
+"""
+import asyncio
+
+import bledev
+import bledev.mpble
+
+NAME = "bledev-conn-loop"
+
+
+async def main():
+ ble = bledev.mpble.get()
+ n = 0
+ while True:
+ conn = await ble.advertise(name=NAME)
+ n += 1
+ print("accepted", n)
+ try:
+ await conn.disconnected(30000)
+ except bledev.BLEError as e:
+ print("still connected after 30 s:", e)
+ await conn.disconnect()
+
+
+asyncio.run(main())
diff --git a/tests/bledev_board/files_pair_client.py b/tests/bledev_board/files_pair_client.py
index c38087ad..e08709d0 100644
--- a/tests/bledev_board/files_pair_client.py
+++ b/tests/bledev_board/files_pair_client.py
@@ -22,6 +22,13 @@
``run(plant="flip")`` flips one bit in one notification the client receives,
and ``run(plant="nopair")`` never pairs; each must fail. Ends with
``RESULT PASS`` or ``RESULT FAIL`` (also written to /files_pair_result.txt).
+
+``run(passkey="console")`` is for a serving board started with
+``pair_server.start("passkey")``: when the other board shows its passkey, this
+board prints ``PASSKEY?`` and reads the digits from its own console, the way a
+person would type what they see. The link must then also be authenticated, and
+the reconnect must not ask again. ``passkey_gate.py`` carries the digits from
+one board's console to the other's.
"""
import asyncio
import hashlib
@@ -61,7 +68,17 @@ def pattern(n):
return bytes(out)
-async def _main(plant):
+asked = []
+
+
+def _typed():
+ # Called when the peer shows a passkey: read it off this board's console.
+ asked.append(time.ticks_ms())
+ print("PASSKEY?")
+ return int(input().strip())
+
+
+async def _main(plant, passkey):
ble = bledev.mpble.get()
ble.enable_bonding()
log("store", security.store().kind, "bonds before", security.bonds())
@@ -79,14 +96,18 @@ def feed(self, data):
ft.Client._feed = feed
t = time.ticks_ms()
try:
- client = await ft.connect(ble, PASSWORD, name=NAME, pair=False if plant == "nopair" else None, timeout_ms=15000)
+ client = await ft.connect(ble, PASSWORD, name=NAME, pair=False if plant == "nopair" else None, timeout_ms=15000,
+ passkey=_typed if passkey == "console" else None)
except bledev.BLEError as e:
check(False, "connect and pair", repr(e))
+ log("passkey asked", len(asked), "bonds after", security.bonds())
return
conn = client.connection
log("connected and logged in in", time.ticks_diff(time.ticks_ms(), t), "ms, mtu", conn.mtu)
enc = security.state(conn._aconn._conn_handle)
check(enc[0] and enc[2], "the link is encrypted and bonded", enc)
+ if passkey:
+ check(enc[1] and len(asked) == 1, "the link is authenticated by the passkey", (enc, len(asked)))
data = pattern(20 * 1024)
t = time.ticks_ms()
await client.write(PATH, data)
@@ -107,17 +128,21 @@ def feed(self, data):
log("reconnected from the bond in", time.ticks_diff(time.ticks_ms(), t), "ms")
enc = security.state(conn._aconn._conn_handle)
check(enc[0] and enc[2], "the second link is encrypted from the bond", enc)
+ if passkey:
+ check(enc[1] and len(asked) == 1, "the second link is authenticated with no passkey asked", (enc, len(asked)))
entries = await client.listdir("/")
check(any(e[0] == PATH[1:] and e[1] == len(data) for e in entries), "the file is on the board, 20 KB")
await client.close()
+ log("bonds after", security.bonds())
-def run(plant=None):
+def run(plant=None, passkey=None):
_lines.clear()
_failures.clear()
+ asked.clear()
log("PLANTED", plant) if plant else None
try:
- asyncio.run(_main(plant))
+ asyncio.run(_main(plant, passkey))
except Exception as e:
check(False, "no exception", repr(e))
log("RESULT", "FAIL " + "; ".join(_failures) if _failures else "PASS")
diff --git a/tests/bledev_board/p4_hid_gate.py b/tests/bledev_board/p4_hid_gate.py
new file mode 100644
index 00000000..0275478e
--- /dev/null
+++ b/tests/bledev_board/p4_hid_gate.py
@@ -0,0 +1,146 @@
+# SPDX-FileCopyrightText: 2026 Brad Barnett
+#
+# SPDX-License-Identifier: MIT
+"""The HID gate, repeated, with a board as the device and the P4 as the host.
+
+Runs on Windows Python with pyserial, driving both boards' friendly REPLs over
+their UARTs (``uart_rig.py``). Set up once with mpftp:
+
+* the device board (an S3): ``/hidp.py`` = ``hid_peripheral.py``, and
+ ``/hidp_plant.py`` = the same with ``PLANT = True``;
+* the P4: ``/hidc.py`` = ``hid_central.py``, and ``/hidc_fast.py`` = the same
+ with ``FAST = True`` (a 7.5-15 ms interval);
+* ``hid_script.mpy`` in both boards' ``/lib``.
+
+Then::
+
+ python.exe p4_hid_gate.py --p4 COM4 --dev COM17 --mode default --runs 10 --log hid.log
+
+Each run soft-resets the device and starts it, hard-resets the P4 (BLE can't
+start twice after a soft reset there, micropython-pydevices#13), then starts
+the host at the friendly prompt, where ESP-IDF's log still reaches the UART:
+every ``alloc_acl_from_ll failed`` (a packet ESP-Hosted dropped) is counted.
+``--mode plant`` runs the planted device, which must FAIL. A run whose host
+never connects is counted as a connect failure and not as a gate result.
+
+``--mode connect`` measures connects instead: the device runs
+``connect_loop_server.py`` and the P4 ``connect_loop_client.py`` (both at
+``/``), ten connects per hard reset. ``--wifi-off`` turns the P4's Wi-Fi
+station off after each boot, for a board whose boot.py joins a network.
+"""
+import re
+import sys
+import time
+
+from uart_rig import Board, wait
+
+
+WIFI_OFF = "--wifi-off" in sys.argv # turn the P4's station off after boot.py joins it
+
+
+def arg(name, default=None):
+ return sys.argv[sys.argv.index(name) + 1] if name in sys.argv else default
+
+
+def connects(p4, dev):
+ r = {"mode": "connect"}
+ if not dev.soft_reset():
+ r["error"] = "device: no prompt after soft reset"
+ return r
+ dev.mark()
+ dev.line("import connect_loop_server")
+ time.sleep(2)
+ if not p4.hard_reset():
+ r["error"] = "P4: no prompt after hard reset"
+ return r
+ if WIFI_OFF:
+ p4.say("import network; network.WLAN(network.STA_IF).active(False)")
+ p4.mark()
+ p4.line("import connect_loop_client")
+ wait([p4, dev], lambda: (b"CONNECTS" in p4.buf or b"Traceback" in p4.buf) and p4.buf.rstrip().endswith(b">>>"), 600)
+ host = p4.text()
+ r["drops"] = host.count("alloc_acl_from_ll failed")
+ m = re.search(r"CONNECTS (\d+) ok (\d+) timeouts (\d+) other (\d+) not found (\d+) times (.*)", host)
+ if m:
+ r.update(attempts=int(m.group(1)), ok=int(m.group(2)), timeouts=int(m.group(3)), other=int(m.group(4)),
+ missing=int(m.group(5)), times=m.group(6).strip())
+ else:
+ r["error"] = "no CONNECTS line"
+ r["accepted"] = dev.text().count("accepted")
+ return r
+
+
+def one(p4, dev, mode):
+ if mode == "connect":
+ return connects(p4, dev)
+ r = {"mode": mode}
+ if not dev.soft_reset():
+ r["error"] = "device: no prompt after soft reset"
+ return r
+ dev.mark()
+ dev.line("import hidp_plant" if mode == "plant" else "import hidp")
+ if not wait([dev], lambda: b"advertising as" in dev.buf, 20):
+ r["error"] = "device never advertised"
+ return r
+ if not p4.hard_reset():
+ r["error"] = "P4: no prompt after hard reset"
+ return r
+ if WIFI_OFF:
+ p4.say("import network; network.WLAN(network.STA_IF).active(False)")
+ p4.mark()
+ p4.line("import hidc_fast" if mode == "fast" else "import hidc")
+ finished = lambda: (b"RESULT " in p4.buf or b"Traceback" in p4.buf) and p4.buf.rstrip().endswith(b">>>")
+ ok = wait([p4, dev], finished, 180)
+ wait([p4, dev], lambda: b"DONE" in dev.buf, 10 if b"RESULT" in p4.buf else 0)
+ host, device = p4.text(), dev.text()
+ r["drops"] = host.count("alloc_acl_from_ll failed")
+ m = re.search(r"connected and started in (\d+) ms", host)
+ r["connect_ms"] = int(m.group(1)) if m else None
+ m = re.search(r"events (\d+) want (\d+)", host)
+ r["events"] = "{}/{}".format(m.group(1), m.group(2)) if m else None
+ m = re.search(r"MISMATCH (.*)", host)
+ r["mismatch"] = m.group(1)[:100] if m else None
+ m = re.search(r"RESULT (\w+)", host)
+ r["result"] = m.group(1) if m else None
+ m = re.search(r"latency phase stalled after (\d+)", host)
+ r["stalled_after"] = int(m.group(1)) if m else None
+ m = re.search(r"round trip, press to the host's LED write: (.*)", device)
+ r["rtt"] = m.group(1).strip() if m else None
+ if "Traceback" in host:
+ tb = host[host.index("Traceback"):].strip().split("\n")
+ r["exception"] = tb[-2] if tb[-1].startswith(">>>") else tb[-1]
+ r["connect_timeout"] = r["connect_ms"] is None and "Timeout" in r["exception"] and "no device matching" not in r["exception"]
+ if not ok:
+ r["error"] = "host didn't finish in 180 s"
+ return r
+
+
+def main():
+ mode = arg("--mode", "default")
+ runs = int(arg("--runs", "1"))
+ log = open(arg("--log", "p4_hid_gate.log"), "a", encoding="utf-8")
+ p4 = Board(arg("--p4", "COM4"), "P4", log)
+ dev = Board(arg("--dev", "COM17"), "DEV", log)
+ results = []
+ for i in range(runs):
+ log.write("==== run {} mode {} {}\n".format(i + 1, mode, time.strftime("%H:%M:%S")))
+ r = one(p4, dev, mode)
+ results.append(r)
+ print("run", i + 1, r, flush=True)
+ dev.prompt()
+ p4.hard_reset()
+ if mode == "connect":
+ print("SUMMARY connect sessions", len(results), "attempts", sum(r.get("attempts", 0) for r in results),
+ "ok", sum(r.get("ok", 0) for r in results), "timeouts", sum(r.get("timeouts", 0) for r in results),
+ "other", sum(r.get("other", 0) for r in results), "not found", sum(r.get("missing", 0) for r in results),
+ "drops", sum(r.get("drops") or 0 for r in results))
+ return
+ done = [r for r in results if r.get("result")]
+ print("SUMMARY mode", mode, "runs", len(results), "completed", len(done),
+ "PASS", sum(r["result"] == "PASS" for r in done),
+ "with drops", sum(bool(r.get("drops")) for r in results),
+ "drops total", sum(r.get("drops") or 0 for r in results),
+ "connect timeouts", sum(bool(r.get("connect_timeout")) for r in results))
+
+
+main()
diff --git a/tests/bledev_board/passkey_gate.py b/tests/bledev_board/passkey_gate.py
new file mode 100644
index 00000000..1afa68e7
--- /dev/null
+++ b/tests/bledev_board/passkey_gate.py
@@ -0,0 +1,153 @@
+# SPDX-FileCopyrightText: 2026 Brad Barnett
+#
+# SPDX-License-Identifier: MIT
+"""The pairing gate, board to board with a passkey: one board shows it, the other types it.
+
+Runs on Windows Python with pyserial (``uart_rig.py``). The serving board has
+a display and ``/pair_server.py``; the typing board has
+``/files_pair_client.py``. Both have bledev in ``/lib``::
+
+ python.exe passkey_gate.py --server COM17 --client COM4 right wrong --log gate.log
+
+The passkey travels the way a person would carry it. The serving board draws
+it on its screen and prints it on its console (``bledev: Bluetooth passkey
+123 456``); this script reads that line off the serving board's UART. The
+typing board asks with ``PASSKEY?`` on its own console, and this script types
+the digits into that board's UART, where ``input()`` reads them. Nothing else
+passes between the two.
+
+Steps, each starting from hard resets and no bonds on either board:
+
+``right`` type what the screen shows. The link must be encrypted,
+ authenticated and bonded on both boards, 20 KB must cross both ways
+ byte for byte (the serving board's copy hashed too), and a second
+ connection must encrypt from the bond without asking again. After
+ hard resets both boards must still hold one bond.
+``wrong`` type the passkey one off. Pairing must be refused, and after hard
+ resets neither board may hold a bond.
+
+``--plant justworks`` starts the server with pair_server's ``justworks``
+plant (it says passkey but pairs without one); ``wrong`` must then FAIL.
+Ends with ``GATE PASS`` or ``GATE FAIL``.
+"""
+import re
+import sys
+import time
+
+from uart_rig import Board, wait
+
+BONDS = "from bledev import security; security.use_store(); print('BONDS', security.bonds())"
+
+
+def arg(name, default=None):
+ return sys.argv[sys.argv.index(name) + 1] if name in sys.argv else default
+
+
+def bonds(board):
+ out = board.say(BONDS)
+ m = re.search(r"BONDS (\d+)", out)
+ return int(m.group(1)) if m else None
+
+
+def fresh(server, client):
+ for b in (server, client):
+ b.hard_reset()
+ b.say("from bledev import security; security.use_store(); security.forget()")
+ server.say("import os; [os.remove(f) for f in os.listdir('/') if f == 'pair_20k.bin']")
+
+
+def step(server, client, kind, plant, say):
+ failures = []
+
+ def check(ok, what):
+ say(("PASS " if ok else "FAIL ") + what)
+ if not ok:
+ failures.append(what)
+
+ fresh(server, client)
+ check(bonds(server) == 0 and bonds(client) == 0, "both boards start with no bond")
+ for b in (server, client):
+ b.hard_reset()
+ started = server.say("import pair_server; pair_server.start('passkey'{})".format(
+ ", plant='justworks'" if plant else ""), 20)
+ say("server:", started)
+ client.mark()
+ server.mark()
+ client.line("import files_pair_client as c; c.run(passkey='console')")
+ t0 = time.monotonic()
+ shown = asked = None
+ finished = lambda: b"RESULT" in client.buf and client.buf.rstrip().endswith(b">>>")
+
+ def ready():
+ nonlocal shown, asked
+ m = re.search(rb"Bluetooth passkey (\d{3}) (\d{3})", server.buf)
+ if m and shown is None:
+ shown = int(m.group(1) + m.group(2))
+ say("the server's console showed {:06d} after {:.2f} s".format(shown, time.monotonic() - t0))
+ if b"PASSKEY?" in client.buf and asked is None:
+ asked = time.monotonic() - t0
+ say("the client asked for it after {:.2f} s".format(asked))
+ return (shown is not None and asked is not None) or finished()
+
+ wait([server, client], ready, 60)
+ if shown is not None and asked is not None:
+ typed = shown if kind == "right" else (shown + 1) % 1000000
+ say("typing {:06d} into the client's console ({})".format(typed, "as shown" if typed == shown else "one off"))
+ client.write(b"%06d\r" % typed)
+ else:
+ say("no passkey exchange: shown", shown, "asked", asked)
+ wait([server, client], finished, 120)
+ out = client.text()
+ for line in out.split("\n"):
+ if line.startswith(("PASS", "FAIL", "RESULT", "connected", "write", "reconnected", "sha256", "passkey asked",
+ "bonds after", "store")):
+ say("client:", line.strip())
+ status = server.say("pair_server.status()")
+ say("server:", status)
+ result = re.search(r"RESULT (.*)", out)
+ result = result.group(1).strip() if result else None
+ if kind == "right":
+ check(result == "PASS", "the client's checks pass (paired, authenticated, 20 KB, reconnect from the bond)")
+ check(shown is not None and asked is not None, "the passkey was shown and asked for")
+ sha = re.search(r"sha256 ([0-9a-f]{64})", out)
+ theirs = server.say("import hashlib, binascii; print(binascii.hexlify(hashlib.sha256("
+ "open('/pair_20k.bin','rb').read()).digest()).decode())")
+ check(bool(sha) and sha.group(1) in theirs, "the serving board's copy has the same SHA-256")
+ m = re.search(r"links \{(.*?)\}", status)
+ say("server links:", m.group(1) if m else None)
+ else:
+ check(result is not None and result.startswith("FAIL") and "connect and pair" in result,
+ "pairing is refused")
+ for b in (server, client):
+ b.hard_reset()
+ n_server, n_client = bonds(server), bonds(client)
+ say("after hard resets: server bonds", n_server, "client bonds", n_client)
+ want = 1 if kind == "right" else 0
+ check(n_server == want and n_client == want, "each board holds {} bond(s) after resets".format(want))
+ return failures
+
+
+def main():
+ log = open(arg("--log", "passkey_gate.log"), "a", encoding="utf-8")
+ server = Board(arg("--server", "COM17"), "SERVER", log)
+ client = Board(arg("--client", "COM4"), "CLIENT", log)
+ plant = arg("--plant")
+ steps = [a for a in sys.argv[1:] if a in ("right", "wrong")]
+
+ def say(*parts):
+ line = " ".join(str(p) for p in parts)
+ print(line, flush=True)
+ log.write("#### " + line + "\n")
+ log.flush()
+
+ failed = []
+ for kind in steps:
+ say("==== step", kind, "plant", plant, time.strftime("%H:%M:%S"))
+ failed += ["{}: {}".format(kind, f) for f in step(server, client, kind, plant, say)]
+ for b in (server, client):
+ b.say("from bledev import security; security.use_store(); security.forget()")
+ b.hard_reset()
+ say("GATE", "FAIL " + "; ".join(failed) if failed else "PASS")
+
+
+main()
diff --git a/tests/bledev_board/uart_rig.py b/tests/bledev_board/uart_rig.py
new file mode 100644
index 00000000..b81fdfa6
--- /dev/null
+++ b/tests/bledev_board/uart_rig.py
@@ -0,0 +1,103 @@
+# SPDX-FileCopyrightText: 2026 Brad Barnett
+#
+# SPDX-License-Identifier: MIT
+"""Drive boards' friendly REPLs over their UARTs, several at once, from Windows Python.
+
+The board-to-board gates that need what mpftp can't give use this: a run
+started at the friendly prompt keeps ESP-IDF's log on the console (the raw
+REPL mutes it, so the P4's ``alloc_acl_from_ll failed`` only shows here), a
+hard reset between runs (the P4 can't turn BLE on twice after a soft reset,
+micropython-pydevices#13), and one board's console read while another board
+is typed at, which is how ``passkey_gate.py`` carries a passkey from one
+screen to another board's keyboard.
+
+Ports open with pyserial's defaults (DTR and RTS asserted), as mpremote opens
+them: on the LCD-7's CH343, opening with both released resets the board.
+Everything each board prints goes to the log, prefixed with its name.
+"""
+import time
+
+import serial
+
+
+class Board:
+ def __init__(self, port, name, log=None):
+ self.s = serial.Serial(port, 115200, timeout=0)
+ self.name = name
+ self.log = log
+ self.buf = b"" # since the last mark()
+ self._partial = b""
+
+ def pump(self):
+ data = self.s.read(8192)
+ if data:
+ self.buf += data
+ if self.log is not None:
+ self._partial += data
+ *lines, self._partial = self._partial.split(b"\n")
+ for line in lines:
+ self.log.write("{:9.3f} {} | {}\n".format(
+ time.monotonic(), self.name, line.decode("utf-8", "replace").rstrip("\r")))
+ self.log.flush()
+ return data
+
+ def mark(self):
+ self.pump()
+ self.buf = b""
+
+ def text(self):
+ return self.buf.decode("utf-8", "replace").replace("\r", "")
+
+ def write(self, data):
+ self.s.write(data)
+
+ def line(self, text):
+ self.s.write(text.encode() + b"\r")
+
+ def prompt(self, timeout=5):
+ """Stop whatever runs and get to the friendly prompt."""
+ self.mark()
+ self.write(b"\x03\x03")
+ time.sleep(0.1)
+ self.write(b"\r")
+ return wait([self], lambda: self.buf.rstrip().endswith(b">>>"), timeout)
+
+ def soft_reset(self, timeout=20):
+ self.prompt()
+ self.mark()
+ self.write(b"\x04")
+ return wait([self], lambda: b"soft reboot" in self.buf and self.buf.rstrip().endswith(b">>>"), timeout)
+
+ def hard_reset(self, timeout=40):
+ """machine.reset() from the prompt, then wait for the next prompt."""
+ self.prompt()
+ self.mark()
+ self.line("import machine; machine.reset()")
+ return wait([self], lambda: b"MicroPython v" in self.buf and self.buf.rstrip().endswith(b">>>"), timeout)
+
+ def say(self, text, timeout=10):
+ """Type one line at the prompt and return what it printed."""
+ self.mark()
+ self.line(text)
+ wait([self], lambda: b"\n" in self.buf and self.buf.split(b"\n", 1)[1].rstrip().endswith(b">>>"), timeout)
+ body = self.text().split("\n", 1)[1] if "\n" in self.text() else ""
+ return body.rstrip().removesuffix(">>>").rstrip()
+
+ def close(self):
+ self.s.close()
+
+
+def wait(boards, done, timeout):
+ """Pump every board until ``done()`` or the timeout. True if done."""
+ end = time.monotonic() + timeout
+ while True:
+ got = False
+ for b in boards:
+ if b.pump():
+ got = True
+ if done():
+ return True
+ if time.monotonic() > end:
+ return False
+ if not got:
+ time.sleep(0.01)
diff --git a/tests/bledev_web/gate.py b/tests/bledev_web/gate.py
new file mode 100644
index 00000000..127ec38d
--- /dev/null
+++ b/tests/bledev_web/gate.py
@@ -0,0 +1,181 @@
+# SPDX-FileCopyrightText: 2026 Brad Barnett
+#
+# SPDX-License-Identifier: MIT
+"""The browser gate: 16 KB up, down and echoed over nus through bledev.webble.
+
+The same three phases as ``tests/bledev_board/nus_client.py``, against
+``nus_gate_server.py`` on a board. It runs unchanged on PyScript's Pyodide,
+PyScript's MicroPython, and the direct MicroPython WebAssembly build; the page
+around it says which. Tapping the page's button starts it, because the
+browser's device chooser opens only from a tap.
+
+Every byte is checked against the pattern and every length must match. The
+page's ``plant=1`` flips one bit at offset 5000 of the UP stream, which the
+board must catch; the board's ``PLANT`` does the same to DOWN and ECHO, which
+this side must catch. Lines go to the page and to the test server's log, and
+the last one is ``RESULT PASS`` or ``RESULT FAIL``.
+"""
+import asyncio
+import json
+import sys
+import time
+
+import js
+
+import bledev.nus as nus
+from bledev import webble
+
+N = 16384
+
+
+def params():
+ return json.loads(str(js.JSON.stringify(js.gateParams)))
+
+
+def log(*parts):
+ line = " ".join(str(p) for p in parts)
+ print(line)
+ js.gateLog(line)
+
+
+def now_ms():
+ if hasattr(time, "ticks_ms"):
+ return time.ticks_ms()
+ return int(time.monotonic() * 1000)
+
+
+def since(t0):
+ if hasattr(time, "ticks_diff"):
+ return time.ticks_diff(time.ticks_ms(), t0)
+ return now_ms() - t0
+
+
+def pattern(n, seed):
+ return bytes(((i * 7 + (i >> 8) * 13 + seed) & 0xFF) for i in range(n))
+
+
+def compare(got, want):
+ bad = 0
+ first = -1
+ for i in range(min(len(got), len(want))):
+ if got[i] != want[i]:
+ bad += 1
+ if first < 0:
+ first = i
+ return bad, first
+
+
+def kbps(n, ms):
+ return "{:.2f} KB/s".format(n / 1024 / (ms / 1000)) if ms else "inf"
+
+
+async def gate():
+ p = params()
+ plant = bool(p.get("plant"))
+ mtu = int(p.get("mtu") or 23)
+ name = p.get("name") or "bledev-web"
+ if name == "any":
+ name = None # filter on the Nordic UART service alone
+ log("GATE start runtime", p.get("runtime"), sys.implementation.name, sys.version.split()[0],
+ "plant", plant, "mtu", mtu)
+ ble = webble.WebBLE(mtu=mtu)
+ t0 = now_ms()
+ link = await nus.connect(ble, name=name)
+ log("connected in", since(t0), "ms (chooser included), page mtu", link.connection.mtu,
+ "device", link.connection.device, "connect retries", ble.connect_retries)
+ ok = True
+ try:
+ await link.write(b"MTU\n")
+ reply = (await asyncio.wait_for(link.readline(), 10)).decode().split()
+ log("board sees mtu", reply[1])
+
+ # UP: we send, the board checks.
+ data = pattern(N, 1)
+ if plant:
+ data = bytearray(data)
+ data[5000] ^= 0x01
+ data = bytes(data)
+ await link.write("UP {}\n".format(N).encode())
+ t0 = now_ms()
+ await link.write(data)
+ sent_ms = since(t0)
+ reply = (await asyncio.wait_for(link.readline(), 60)).decode().split()
+ ms = int(reply[2])
+ log("UP", reply[1], N, "bytes: board received in", ms, "ms =", kbps(N, ms),
+ "(our send", sent_ms, "ms) mismatches", reply[3], "first", reply[4])
+ ok = ok and reply[1] == "OK"
+
+ # DOWN: the board sends, we check.
+ await link.write("DOWN {}\n".format(N).encode())
+ t0 = now_ms()
+ got = await asyncio.wait_for(link.readexactly(N), 120)
+ ms = since(t0)
+ bad, first = compare(got, pattern(N, 2))
+ verdict = "OK" if bad == 0 and len(got) == N else "BAD"
+ log("DOWN", verdict, len(got), "bytes in", ms, "ms =", kbps(N, ms), "mismatches", bad, "first", first)
+ ok = ok and verdict == "OK"
+
+ # ECHO: through the board and back.
+ await link.write("ECHO {}\n".format(N).encode())
+ data = pattern(N, 3)
+ t0 = now_ms()
+ failed = []
+
+ async def send():
+ try:
+ await link.write(data)
+ except Exception as e:
+ failed.append(e)
+ log("ECHO send failed after", link.bytes_out, "bytes out:", type(e).__name__, e)
+ await link.close()
+
+ sender = asyncio.create_task(send())
+ try:
+ got = await asyncio.wait_for(link.readexactly(N), 180)
+ except asyncio.TimeoutError:
+ log("ECHO stalled: sent", link.bytes_out, "bytes in all, received", link.bytes_in,
+ "(", len(link._buf), "of", N, "echo bytes buffered )")
+ raise
+ ms = since(t0)
+ await sender
+ if failed:
+ raise failed[0]
+ bad, first = compare(got, data)
+ verdict = "OK" if bad == 0 and len(got) == N else "BAD"
+ log("ECHO", verdict, len(got), "bytes round trip in", ms, "ms =", kbps(N, ms),
+ "each way; mismatches", bad, "first", first)
+ ok = ok and verdict == "OK"
+
+ await link.write(b"BYE\n")
+ await asyncio.sleep(0.3)
+ finally:
+ await link.close()
+ log("RESULT", "PASS" if ok else "FAIL")
+
+
+_running = False
+
+
+async def guarded():
+ global _running
+ try:
+ await gate()
+ except BaseException as e:
+ log("EXCEPTION", type(e).__name__, e)
+ log("RESULT", "FAIL")
+ finally:
+ _running = False
+
+
+def on_click(*args):
+ global _running
+ if _running:
+ return
+ _running = True
+ # Starting the task from the click keeps the browser's user activation
+ # alive for requestDevice(): it lasts about five seconds.
+ asyncio.get_event_loop().create_task(guarded())
+
+
+js.document.getElementById("go").addEventListener("click", webble._proxy(on_click))
+js.gateReady()
diff --git a/tests/bledev_web/mock_bluetooth.js b/tests/bledev_web/mock_bluetooth.js
new file mode 100644
index 00000000..c8be6462
--- /dev/null
+++ b/tests/bledev_web/mock_bluetooth.js
@@ -0,0 +1,154 @@
+// A stand-in for navigator.bluetooth, loaded when the page has ?mock=1.
+//
+// It plays nus_gate_server.py's part in JavaScript: one device named
+// "bledev-web" with the Nordic UART service, answering MTU, UP, DOWN, ECHO
+// and BYE. It lets the gate page run headless in any browser, so the Python to
+// JavaScript plumbing is checked before a radio is involved. Like Android, it
+// cuts a write without response to the link's payload (mock_mtu - 3) without
+// saying so, which is why webble sizes writes to the MTU it's told.
+// ?mockplant=1 flips the bit at offset 5000 of DOWN and ECHO.
+(() => {
+ const q = new URLSearchParams(location.search);
+ if (q.get("mock") !== "1") return;
+ const MTU = parseInt(q.get("mock_mtu") || "23", 10);
+ const PLANT = q.get("mockplant") === "1";
+ const NUS = "6e400001-b5a3-f393-e0a9-e50e24dcca9e";
+ const RX = "6e400002-b5a3-f393-e0a9-e50e24dcca9e";
+ const TX = "6e400003-b5a3-f393-e0a9-e50e24dcca9e";
+ const later = (f, ms = 0) => setTimeout(f, ms);
+ const pattern = (n, seed) => {
+ const a = new Uint8Array(n);
+ for (let i = 0; i < n; i++) a[i] = (i * 7 + (i >> 8) * 13 + seed) & 255;
+ return a;
+ };
+ const err = (name, message) => Object.assign(new Error(message), { name });
+
+ class Char extends EventTarget {
+ constructor(service, uuid, props) {
+ super();
+ this.service = service;
+ this.uuid = uuid;
+ this.properties = props;
+ this.value = null;
+ this.notifying = false;
+ }
+ async writeValueWithoutResponse(v) {
+ if (!device.gatt.connected) throw err("NetworkError", "GATT Server is disconnected.");
+ const bytes = new Uint8Array(v.buffer ? v.buffer.slice(v.byteOffset, v.byteOffset + v.byteLength) : v);
+ server.receive(bytes.slice(0, MTU - 3)); // silent truncation, as Android does
+ }
+ async writeValueWithResponse(v) { return this.writeValueWithoutResponse(v); }
+ async readValue() { return new DataView(new Uint8Array(0).buffer); }
+ async startNotifications() { this.notifying = true; return this; }
+ async stopNotifications() { this.notifying = false; return this; }
+ }
+
+ // The server side: nus_gate_server.py's protocol.
+ const server = {
+ buf: [],
+ mode: null,
+ need: 0,
+ got: [],
+ sent: 0,
+ queue: [],
+ sending: false,
+ send(bytes) {
+ for (let i = 0; i < bytes.length; i += MTU - 3) this.queue.push(bytes.slice(i, i + MTU - 3));
+ this.pump();
+ },
+ pump() {
+ if (this.sending) return;
+ this.sending = true;
+ const step = () => {
+ const chunk = this.queue.shift();
+ if (!chunk || !device.gatt.connected) { this.sending = false; return; }
+ if (tx.notifying) {
+ tx.value = new DataView(chunk.buffer, chunk.byteOffset, chunk.byteLength);
+ tx.dispatchEvent(new Event("characteristicvaluechanged"));
+ }
+ later(step);
+ };
+ later(step);
+ },
+ line(text) { this.send(new TextEncoder().encode(text + "\n")); },
+ flip(chunk, sent) {
+ if (PLANT && sent <= 5000 && 5000 < sent + chunk.length) chunk[5000 - sent] ^= 1;
+ return chunk;
+ },
+ receive(bytes) {
+ if (this.mode === "UP") {
+ this.got.push(...bytes);
+ if (this.got.length >= this.need) {
+ const want = pattern(this.need, 1);
+ let bad = 0, first = -1;
+ for (let i = 0; i < this.need; i++) if (this.got[i] !== want[i]) { bad++; if (first < 0) first = i; }
+ const ok = bad === 0 && this.got.length === this.need;
+ this.line(`UP ${ok ? "OK" : "BAD"} ${Date.now() - this.t0} ${bad} ${first}`);
+ this.mode = null;
+ }
+ return;
+ }
+ if (this.mode === "ECHO") {
+ const chunk = this.flip(Uint8Array.from(bytes), this.sent);
+ this.sent += chunk.length;
+ this.send(chunk);
+ if (this.sent >= this.need) this.mode = null;
+ return;
+ }
+ this.buf.push(...bytes);
+ const nl = this.buf.indexOf(10);
+ if (nl < 0) return;
+ const cmd = new TextDecoder().decode(Uint8Array.from(this.buf.slice(0, nl))).trim();
+ const rest = this.buf.slice(nl + 1);
+ this.buf = [];
+ const [op, n] = cmd.split(" ");
+ if (op === "MTU") this.line(`MTU ${MTU}`);
+ else if (op === "UP") { this.mode = "UP"; this.need = +n; this.got = []; this.t0 = Date.now(); }
+ else if (op === "DOWN") this.send(this.flip(pattern(+n, 2), 0));
+ else if (op === "ECHO") { this.mode = "ECHO"; this.need = +n; this.sent = 0; }
+ else if (op === "BYE") later(() => device.gatt.disconnect(), 50);
+ if (rest.length) this.receive(Uint8Array.from(rest));
+ },
+ };
+
+ const service = { uuid: NUS, isPrimary: true };
+ const rx = new Char(service, RX, { read: false, write: true, writeWithoutResponse: true, notify: false, indicate: false });
+ const tx = new Char(service, TX, { read: false, write: false, writeWithoutResponse: false, notify: true, indicate: false });
+ service.getCharacteristics = async (uuid) => {
+ const all = [rx, tx].filter((c) => !uuid || c.uuid === uuid);
+ if (!all.length) throw err("NotFoundError", "No Characteristics matching UUID found in Service.");
+ return all;
+ };
+
+ const device = new EventTarget();
+ device.id = "mock-device-id";
+ device.name = "bledev-web";
+ device.gatt = {
+ connected: false,
+ device,
+ async connect() { this.connected = true; return this; },
+ disconnect() {
+ if (!this.connected) return;
+ this.connected = false;
+ later(() => device.dispatchEvent(new Event("gattserverdisconnected")));
+ },
+ async getPrimaryServices(uuid) {
+ if (uuid && uuid !== NUS) throw err("NotFoundError", "No Services matching UUID found in Device.");
+ return [service];
+ },
+ };
+
+ const bluetooth = {
+ async getAvailability() { return true; },
+ async requestDevice(options) {
+ if (!navigator.userActivation || !navigator.userActivation.isActive) {
+ throw err("SecurityError", "Must be handling a user gesture to show a permission request.");
+ }
+ const f = (options.filters || [])[0] || {};
+ if (f.name && f.name !== device.name) throw err("NotFoundError", "User cancelled the requestDevice() chooser.");
+ return device;
+ },
+ };
+ Object.defineProperty(navigator, "bluetooth", { value: bluetooth, configurable: true });
+ console.log("mock navigator.bluetooth installed, mtu", MTU, "plant", PLANT);
+})();
diff --git a/tests/bledev_web/mpy.html b/tests/bledev_web/mpy.html
new file mode 100644
index 00000000..b6ae1f49
--- /dev/null
+++ b/tests/bledev_web/mpy.html
@@ -0,0 +1,18 @@
+
+
+
+
+
+ bledev.webble gate (mpy)
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/tests/bledev_web/mpy.json b/tests/bledev_web/mpy.json
new file mode 100644
index 00000000..fd846230
--- /dev/null
+++ b/tests/bledev_web/mpy.json
@@ -0,0 +1,10 @@
+{
+ "files": {
+ "../../lib/bledev/__init__.py": "./bledev/__init__.py",
+ "../../lib/bledev/auto.py": "./bledev/auto.py",
+ "../../lib/bledev/nus.py": "./bledev/nus.py",
+ "../../lib/bledev/webble.py": "./bledev/webble.py",
+ "./gate.py": "./gate.py"
+ },
+ "interpreter": "/pyscript/micropython/micropython.mjs"
+}
\ No newline at end of file
diff --git a/tests/bledev_web/nus_gate_server.py b/tests/bledev_web/nus_gate_server.py
new file mode 100644
index 00000000..39293be0
--- /dev/null
+++ b/tests/bledev_web/nus_gate_server.py
@@ -0,0 +1,106 @@
+# SPDX-FileCopyrightText: 2026 Brad Barnett
+#
+# SPDX-License-Identifier: MIT
+"""The board side of the browser gate: serve Nordic UART as ``bledev-web``.
+
+The same protocol as ``tests/bledev_board/nus_server.py`` (UP, DOWN, ECHO,
+BYE), but it keeps serving, one link after another, so a browser can run the
+gate again without restarting the board. Each link's MTU is logged, because
+the browser can't report it. ``PLANT = True`` flips one bit at offset 5000 of
+everything this side sends, and the page must then report FAIL. Logs to
+/gate_server.log. Stop it with Ctrl-C (any mpftp command does).
+"""
+import asyncio, time
+import bledev.mpble
+import bledev.nus as nus
+
+PLANT = False # the planted run sets this: flip one bit at offset 5000 of what we send
+NAME = "bledev-web" # not bledev-gate: tests/bledev_board uses that name on other boards
+LOG = "/gate_server.log"
+
+
+def log(*parts):
+ line = " ".join(str(p) for p in parts)
+ print(line)
+ with open(LOG, "a") as f:
+ f.write(line + "\n")
+
+
+def pattern(n, seed):
+ return bytes(((i * 7 + (i >> 8) * 13 + seed) & 0xFF) for i in range(n))
+
+
+def compare(got, want):
+ bad = 0
+ first = -1
+ for i in range(min(len(got), len(want))):
+ if got[i] != want[i]:
+ bad += 1
+ if first < 0:
+ first = i
+ return bad, first
+
+
+def flip(chunk, sent):
+ # Flip the bit at stream offset 5000 if it falls inside this chunk.
+ if PLANT and sent <= 5000 < sent + len(chunk):
+ chunk = bytearray(chunk)
+ chunk[5000 - sent] ^= 0x01
+ chunk = bytes(chunk)
+ return chunk
+
+
+async def session(link):
+ log("connected, mtu", link.connection.mtu)
+ while True:
+ line = await link.readline()
+ cmd = line.decode().strip()
+ if not cmd or cmd == "BYE":
+ break
+ if cmd == "MTU":
+ await link.write("MTU {}\n".format(link.connection.mtu).encode())
+ continue
+ op, n = cmd.split()
+ n = int(n)
+ if op == "UP":
+ t0 = time.ticks_ms()
+ got = await link.readexactly(n)
+ ms = time.ticks_diff(time.ticks_ms(), t0)
+ bad, first = compare(got, pattern(n, 1))
+ verdict = "OK" if bad == 0 and len(got) == n else "BAD"
+ log("UP", verdict, n, "bytes", ms, "ms mismatches", bad, "first", first, "mtu", link.connection.mtu)
+ await link.write("UP {} {} {} {}\n".format(verdict, ms, bad, first).encode())
+ elif op == "DOWN":
+ data = flip(pattern(n, 2), 0)
+ t0 = time.ticks_ms()
+ await link.write(data)
+ log("DOWN sent", n, "bytes in", time.ticks_diff(time.ticks_ms(), t0), "ms")
+ elif op == "ECHO":
+ left = n
+ sent = 0
+ while left:
+ chunk = await link.read(left)
+ if not chunk:
+ break
+ await link.write(flip(chunk, sent))
+ sent += len(chunk)
+ left -= len(chunk)
+ log("ECHO echoed", sent, "of", n)
+ log("done; bytes_in", link.bytes_in, "bytes_out", link.bytes_out)
+
+
+async def main():
+ open(LOG, "w").close()
+ ble = bledev.mpble.get()
+ log("serving as", NAME, "plant", PLANT)
+ while True:
+ link = await nus.serve(ble, name=NAME)
+ try:
+ await session(link)
+ except Exception as e:
+ log("session ended:", repr(e))
+ if link.connection.is_connected():
+ await link.close()
+
+
+asyncio.run(main())
diff --git a/tests/bledev_web/page.js b/tests/bledev_web/page.js
new file mode 100644
index 00000000..f5634adb
--- /dev/null
+++ b/tests/bledev_web/page.js
@@ -0,0 +1,33 @@
+// The part of the gate page every runtime shares: the query parameters, the
+// log (on the page and POSTed to serve.py), and the button.
+const q = new URLSearchParams(location.search);
+window.gateParams = {
+ runtime: document.documentElement.dataset.runtime,
+ plant: q.get("plant") === "1" ? 1 : 0,
+ mtu: parseInt(q.get("mtu") || "23", 10),
+ name: q.get("name") || "bledev-web",
+ mock: q.get("mock") === "1" ? 1 : 0,
+};
+
+const out = () => document.getElementById("log");
+
+window.gateLog = (line) => {
+ line = String(line);
+ out().textContent += line + "\n";
+ fetch("/log", { method: "POST", body: `[${window.gateParams.runtime}] ${line}` }).catch(() => {});
+ if (line.startsWith("RESULT ")) {
+ const go = document.getElementById("go");
+ go.textContent = line;
+ go.className = line.endsWith("PASS") ? "pass" : "fail";
+ }
+};
+
+window.gateReady = () => {
+ const go = document.getElementById("go");
+ go.disabled = false;
+ go.textContent = "Connect and run the gate";
+ window.gateLog("READY " + navigator.userAgent);
+};
+
+window.addEventListener("error", (e) => window.gateLog("PAGE ERROR " + e.message));
+window.addEventListener("unhandledrejection", (e) => window.gateLog("PAGE REJECTION " + (e.reason && e.reason.stack || e.reason)));
diff --git a/tests/bledev_web/pyodide.html b/tests/bledev_web/pyodide.html
new file mode 100644
index 00000000..8d119b68
--- /dev/null
+++ b/tests/bledev_web/pyodide.html
@@ -0,0 +1,18 @@
+
+
+
+
+
+ bledev.webble gate (pyodide)
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/tests/bledev_web/pyodide.json b/tests/bledev_web/pyodide.json
new file mode 100644
index 00000000..ddb4b976
--- /dev/null
+++ b/tests/bledev_web/pyodide.json
@@ -0,0 +1,10 @@
+{
+ "files": {
+ "../../lib/bledev/__init__.py": "./bledev/__init__.py",
+ "../../lib/bledev/auto.py": "./bledev/auto.py",
+ "../../lib/bledev/nus.py": "./bledev/nus.py",
+ "../../lib/bledev/webble.py": "./bledev/webble.py",
+ "./gate.py": "./gate.py"
+ },
+ "interpreter": "/pyscript/pyodide/pyodide.mjs"
+}
\ No newline at end of file
diff --git a/tests/bledev_web/serve.py b/tests/bledev_web/serve.py
new file mode 100644
index 00000000..d170306b
--- /dev/null
+++ b/tests/bledev_web/serve.py
@@ -0,0 +1,105 @@
+#!/usr/bin/env python3
+# SPDX-FileCopyrightText: 2026 Brad Barnett
+#
+# SPDX-License-Identifier: MIT
+"""Serve the webble gate pages, and collect what they log.
+
+ python tests/bledev_web/serve.py [--port 8765] [--log gate.log]
+
+Then open ``http://localhost:8765/tests/bledev_web/pyodide.html`` (or
+``mpy.html``, ``wasm.html``) in Chrome or Edge. Web Bluetooth needs a secure
+context, and ``localhost`` is one; to reach it from an Android phone, run
+``adb reverse tcp:8765 tcp:8765`` and open the same URL there.
+
+The repository root is served as it is. ``/pyscript/`` comes from the
+pyscript-template checkout's vendored PyScript (Pyodide and MicroPython,
+offline), and ``/wasm/`` from the workspace's ``bin/`` (the direct MicroPython
+WebAssembly build), both found by walking up from this file. Every line a
+page logs is POSTed to ``/log``, printed, and appended to ``--log``.
+"""
+
+import argparse
+import functools
+import http.server
+import os
+import sys
+from pathlib import Path
+
+HERE = Path(__file__).resolve().parent
+ROOT = HERE.parent.parent
+
+
+def _find(relative):
+ for parent in HERE.parents:
+ candidate = parent / relative
+ if candidate.exists():
+ return candidate
+ return None
+
+
+class Handler(http.server.SimpleHTTPRequestHandler):
+ extensions_map = dict(
+ http.server.SimpleHTTPRequestHandler.extensions_map,
+ **{".mjs": "text/javascript", ".js": "text/javascript", ".wasm": "application/wasm",
+ ".json": "application/json", ".py": "text/plain"},
+ )
+ mounts = {}
+ log_path = None
+
+ def translate_path(self, path):
+ clean = path.split("?", 1)[0].split("#", 1)[0]
+ for prefix, directory in self.mounts.items():
+ if clean.startswith(prefix):
+ return str(Path(directory) / clean[len(prefix):])
+ return super().translate_path(path)
+
+ def end_headers(self):
+ self.send_header("Cache-Control", "no-store")
+ super().end_headers()
+
+ def do_POST(self):
+ length = int(self.headers.get("Content-Length") or 0)
+ body = self.rfile.read(length).decode("utf-8", "replace")
+ if self.path.startswith("/log"):
+ print(body, flush=True)
+ if self.log_path:
+ with open(self.log_path, "a") as f:
+ f.write(body + "\n")
+ self.send_response(204)
+ self.end_headers()
+
+ def log_message(self, fmt, *args):
+ if "--quiet" not in sys.argv:
+ super().log_message(fmt, *args)
+
+
+def main():
+ parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
+ parser.add_argument("--port", type=int, default=8765)
+ parser.add_argument("--bind", default="127.0.0.1")
+ parser.add_argument("--log", default=None, help="append every logged line here")
+ parser.add_argument("--quiet", action="store_true", help="don't log requests")
+ args = parser.parse_args()
+
+ pyscript = _find("pyscript-template/vendor/pyscript")
+ wasm = _find("bin/micropython.mjs")
+ Handler.mounts = {}
+ if pyscript:
+ Handler.mounts["/pyscript/"] = pyscript
+ else:
+ print("no pyscript-template/vendor/pyscript found: the PyScript pages won't load", file=sys.stderr)
+ if wasm:
+ Handler.mounts["/wasm/"] = wasm.parent
+ else:
+ print("no bin/micropython.mjs found: wasm.html won't load", file=sys.stderr)
+ Handler.log_path = args.log
+
+ handler = functools.partial(Handler, directory=str(ROOT))
+ server = http.server.ThreadingHTTPServer((args.bind, args.port), handler)
+ print("serving {} on http://localhost:{}/tests/bledev_web/ (pyscript {}, wasm {})".format(
+ ROOT, args.port, pyscript, wasm and wasm.parent), flush=True)
+ server.serve_forever()
+
+
+if __name__ == "__main__":
+ main()
diff --git a/tests/bledev_web/style.css b/tests/bledev_web/style.css
new file mode 100644
index 00000000..f87f3598
--- /dev/null
+++ b/tests/bledev_web/style.css
@@ -0,0 +1,6 @@
+body { font: 16px system-ui, sans-serif; margin: 0; background: #111; color: #ddd; }
+#go { display: block; width: 100%; height: 40vh; font-size: 2em; border: 0; background: #235; color: #fff; }
+#go:disabled { background: #333; color: #888; }
+#go.pass { background: #174; }
+#go.fail { background: #822; }
+#log { white-space: pre-wrap; word-break: break-all; font: 13px monospace; padding: 8px; }
diff --git a/tests/bledev_web/wasm.html b/tests/bledev_web/wasm.html
new file mode 100644
index 00000000..8c56b326
--- /dev/null
+++ b/tests/bledev_web/wasm.html
@@ -0,0 +1,16 @@
+
+
+
+
+
+ bledev.webble gate (wasm)
+
+
+
+
+
+
+
+
+
+
diff --git a/tests/bledev_web/wasm.js b/tests/bledev_web/wasm.js
new file mode 100644
index 00000000..f47be78f
--- /dev/null
+++ b/tests/bledev_web/wasm.js
@@ -0,0 +1,20 @@
+// The direct MicroPython WebAssembly runtime, as Workbench and the PyDevices
+// gallery load it: loadMicroPython() from micropython.mjs, files written into
+// its filesystem, Python run with runPythonAsync(). serve.py maps /wasm/ to
+// the workspace's bin/ directory.
+const FILES = ["__init__", "auto", "nus", "webble"].map((m) => [`../../lib/bledev/${m}.py`, `/lib/bledev/${m}.py`]);
+FILES.push(["./gate.py", "/lib/gate.py"]);
+
+const { loadMicroPython } = await import("/wasm/micropython.mjs");
+const mp = await loadMicroPython({ stdout: (line) => console.log(line), linebuffer: true, heapsize: 16 * 1024 * 1024 });
+mp.FS.mkdirTree("/lib/bledev");
+for (const [url, path] of FILES) {
+ const response = await fetch(url, { cache: "no-store" });
+ if (!response.ok) throw new Error(`${url}: ${response.status}`);
+ mp.FS.writeFile(path, await response.text());
+}
+try {
+ await mp.runPythonAsync("import sys\nif '/lib' not in sys.path: sys.path.append('/lib')\nimport gate");
+} catch (e) {
+ window.gateLog("PYTHON ERROR " + e);
+}
diff --git a/tests/test_bledev_web.py b/tests/test_bledev_web.py
new file mode 100644
index 00000000..cc1225f9
--- /dev/null
+++ b/tests/test_bledev_web.py
@@ -0,0 +1,147 @@
+# SPDX-FileCopyrightText: 2026 Brad Barnett
+#
+# SPDX-License-Identifier: MIT
+"""bledev.webble in a real browser, against the mock Web Bluetooth device.
+
+Serves ``tests/bledev_web/`` with its own ``serve.py`` and drives the gate page
+in headless Chromium through Playwright, on all three runtimes: PyScript's
+Pyodide, PyScript's MicroPython, and the direct MicroPython WebAssembly build.
+``mock_bluetooth.js`` stands in for the radio, so this checks the Python to
+JavaScript plumbing, not a link; the same page runs against a board with
+``nus_gate_server.py`` (see ``docs/bledev-internals.md``).
+
+Two planted faults must fail: one flipped bit each way, and a page told a
+larger MTU than the link has, whose writes the mock cuts short without an
+error, as Android does.
+
+Needs Playwright with Chromium, the pyscript-template checkout's vendored
+PyScript and the workspace's ``bin/micropython.mjs``. Without them the skip
+is loud, and ``PYDEVICES_REQUIRE_BROWSER=1`` turns it into a failure.
+"""
+
+from pathlib import Path
+import os
+import socket
+import subprocess
+import sys
+import time
+import unittest
+import urllib.request
+
+HERE = Path(__file__).resolve().parent / "bledev_web"
+
+
+def _find(relative):
+ for parent in HERE.parents:
+ if (parent / relative).exists():
+ return parent / relative
+ return None
+
+
+def _missing():
+ try:
+ import playwright.sync_api # noqa: F401
+ except ImportError:
+ return "playwright is not installed"
+ if _find("pyscript-template/vendor/pyscript") is None:
+ return "no pyscript-template/vendor/pyscript beside this checkout"
+ if _find("bin/micropython.mjs") is None:
+ return "no bin/micropython.mjs in the workspace"
+ return None
+
+
+def _free_port():
+ with socket.socket() as s:
+ s.bind(("127.0.0.1", 0))
+ return s.getsockname()[1]
+
+
+class WebBLEGate(unittest.TestCase):
+ @classmethod
+ def setUpClass(cls):
+ reason = _missing()
+ if reason:
+ message = "bledev.webble browser gate did NOT run: " + reason
+ if os.environ.get("PYDEVICES_REQUIRE_BROWSER") == "1":
+ raise AssertionError(message)
+ sys.stderr.write("\n*** " + message + " ***\n")
+ raise unittest.SkipTest(message)
+ from playwright.sync_api import sync_playwright
+
+ cls.port = _free_port()
+ cls.server = subprocess.Popen(
+ [sys.executable, str(HERE / "serve.py"), "--port", str(cls.port), "--quiet"],
+ stdout=subprocess.DEVNULL,
+ stderr=subprocess.DEVNULL,
+ )
+ base = "http://127.0.0.1:{}/tests/bledev_web/".format(cls.port)
+ for _ in range(100):
+ try:
+ urllib.request.urlopen(base + "page.js")
+ break
+ except OSError:
+ time.sleep(0.1)
+ cls.base = base
+ cls.pw = sync_playwright().start()
+ try:
+ cls.browser = cls.pw.chromium.launch()
+ except Exception as e:
+ cls.pw.stop()
+ cls.server.terminate()
+ raise unittest.SkipTest("no Chromium for Playwright: {}".format(e))
+
+ @classmethod
+ def tearDownClass(cls):
+ cls.browser.close()
+ cls.pw.stop()
+ cls.server.terminate()
+ cls.server.wait()
+
+ def gate(self, runtime, query):
+ page = self.browser.new_page()
+ try:
+ page.goto(self.base + "{}.html?mock=1&{}".format(runtime, query))
+ page.wait_for_function(
+ "!document.getElementById('go').disabled || "
+ "document.getElementById('log').textContent.includes('ERROR')",
+ timeout=180000,
+ )
+ page.click("#go") # a real click: requestDevice needs the user activation
+ page.wait_for_function(
+ "document.getElementById('log').textContent.includes('RESULT')", timeout=240000
+ )
+ return page.inner_text("#log")
+ finally:
+ page.close()
+
+ def test_every_runtime_passes(self):
+ for runtime in ("pyodide", "mpy", "wasm"):
+ with self.subTest(runtime=runtime):
+ log = self.gate(runtime, "mtu=247&mock_mtu=247")
+ self.assertIn("RESULT PASS", log, log)
+ for phase in ("UP OK", "DOWN OK", "ECHO OK"):
+ self.assertIn(phase, log, log)
+
+ def test_default_mtu_is_safe_on_a_small_link(self):
+ log = self.gate("pyodide", "mock_mtu=23")
+ self.assertIn("RESULT PASS", log, log)
+ self.assertIn("page mtu 23", log, log)
+
+ def test_planted_bit_flips_fail_every_phase(self):
+ for runtime in ("pyodide", "wasm"):
+ with self.subTest(runtime=runtime):
+ log = self.gate(runtime, "mtu=247&mock_mtu=247&plant=1&mockplant=1")
+ self.assertIn("RESULT FAIL", log, log)
+ for phase in ("UP BAD", "DOWN BAD", "ECHO BAD"):
+ self.assertIn(phase, log, log)
+ self.assertEqual(log.count("mismatches 1 first 5000"), 3, log)
+
+ def test_an_overstated_mtu_fails(self):
+ # The page believes 247, the link has 23: every write is cut short.
+ log = self.gate("pyodide", "mtu=247&mock_mtu=23")
+ self.assertIn("RESULT FAIL", log, log)
+ self.assertNotIn("UP OK", log, log)
+
+
+if __name__ == "__main__":
+ unittest.main()