From e7913702b558886df0ee24170f21d2e31557e32e Mon Sep 17 00:00:00 2001 From: David Lechner Date: Sat, 17 Oct 2026 23:52:20 -0500 Subject: [PATCH 1/7] pb-service: the byte stream a hub talks over is not USB's alone The EV3 and NXT speak over an RFCOMM channel exactly what the other hubs speak over USB: the same COBS variant, the same envelope from Pybricks Profile v1.6.0, and the same rule of one request at a time. So everything that was about the byte stream rather than about serial ports moves to hub/stream: cobs and wire as they were, the link that was _UsbLink as StreamLink, and SerialPort as BytePipe -- the two methods of a stream the link reads, which an RFCOMM channel can offer as readily as a serial port can. Nothing changes in behaviour. The one visible difference is wording: a characteristic with no equivalent on a stream says so, rather than naming USB. --- .../pb-service/src/pb_service/hub/protocol.py | 2 +- .../src/pb_service/hub/stream/__init__.py | 10 + .../pb_service/hub/{usb => stream}/cobs.py | 9 +- .../src/pb_service/hub/stream/link.py | 249 ++++++++++++++++++ .../src/pb_service/hub/stream/pipe.py | 45 ++++ .../pb_service/hub/{usb => stream}/wire.py | 6 +- .../pb-service/src/pb_service/hub/usb/fake.py | 9 +- .../pb-service/src/pb_service/hub/usb/port.py | 32 +-- .../src/pb_service/hub/usb/transport.py | 228 +--------------- packages/pb-service/tests/test_cobs.py | 2 +- packages/pb-service/tests/test_usb_link.py | 10 +- 11 files changed, 333 insertions(+), 269 deletions(-) create mode 100644 packages/pb-service/src/pb_service/hub/stream/__init__.py rename packages/pb-service/src/pb_service/hub/{usb => stream}/cobs.py (93%) create mode 100644 packages/pb-service/src/pb_service/hub/stream/link.py create mode 100644 packages/pb-service/src/pb_service/hub/stream/pipe.py rename packages/pb-service/src/pb_service/hub/{usb => stream}/wire.py (96%) diff --git a/packages/pb-service/src/pb_service/hub/protocol.py b/packages/pb-service/src/pb_service/hub/protocol.py index 6b1b12f..b9b05f5 100644 --- a/packages/pb-service/src/pb_service/hub/protocol.py +++ b/packages/pb-service/src/pb_service/hub/protocol.py @@ -97,7 +97,7 @@ def _standard_uuid(short: int) -> str: The firmware header declares v1.6.0, whose addition was a USB interface: the same commands and events inside a message envelope a byte stream needs, which -belongs to :mod:`pb_service.hub.usb.wire` rather than here. v1.7.0 is the +belongs to :mod:`pb_service.hub.stream.wire` rather than here. v1.7.0 is the telemetry pair below, where the two upstreams disagree -- the firmware carries it as unreleased, Pybricks Code already calls it v1.7.0 -- and the higher number is the safer of the two to claim, since claiming it only ever admits diff --git a/packages/pb-service/src/pb_service/hub/stream/__init__.py b/packages/pb-service/src/pb_service/hub/stream/__init__.py new file mode 100644 index 0000000..08c11b7 --- /dev/null +++ b/packages/pb-service/src/pb_service/hub/stream/__init__.py @@ -0,0 +1,10 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 The Pybricks Authors + +"""What every hub reached over a byte stream needs, whatever carries it. + +USB was the first such transport and RFCOMM is the second, but nothing in here +knows about either: a pipe with no message boundaries needs framing, and an +envelope to say which characteristic a message is about, and both are the same +whether the bytes came off a cable or a radio. +""" diff --git a/packages/pb-service/src/pb_service/hub/usb/cobs.py b/packages/pb-service/src/pb_service/hub/stream/cobs.py similarity index 93% rename from packages/pb-service/src/pb_service/hub/usb/cobs.py rename to packages/pb-service/src/pb_service/hub/stream/cobs.py index 872a7b4..55cb862 100644 --- a/packages/pb-service/src/pb_service/hub/usb/cobs.py +++ b/packages/pb-service/src/pb_service/hub/stream/cobs.py @@ -3,10 +3,11 @@ """Consistent Overhead Byte Stuffing, the SPIKE Prime variant. -A CDC pipe is a raw byte stream with no message boundaries, so each direction -frames its messages. This is a port of ``lib/pbio/src/cobs.c``, and it is -deliberately not classic COBS -- close enough to look familiar and different -enough that assuming the familiar version produces a hub that never answers: +A byte stream -- a CDC pipe over USB, an RFCOMM channel over Bluetooth -- has +no message boundaries, so each direction frames its messages. This is a port of +``lib/pbio/src/cobs.c``, and it is deliberately not classic COBS -- close enough +to look familiar and different enough that assuming the familiar version +produces a hub that never answers: * the frame delimiter is ``0x02``, not ``0x00``; * three values are escaped rather than one, so that ``0x01`` and ``0x02`` are diff --git a/packages/pb-service/src/pb_service/hub/stream/link.py b/packages/pb-service/src/pb_service/hub/stream/link.py new file mode 100644 index 0000000..54ebf49 --- /dev/null +++ b/packages/pb-service/src/pb_service/hub/stream/link.py @@ -0,0 +1,249 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 The Pybricks Authors + +"""One hub's connection, over whatever carries the bytes. + +The hub protocol above this file is written in terms of reading and writing +characteristics. A byte stream supplies none of that by itself, so this turns +those calls into framed messages and matches each reply to the request that +caused it. Nothing here knows whether the bytes came off a cable or a radio. +""" + +from __future__ import annotations + +import asyncio +import logging +from collections.abc import Callable +from typing import Final, final + +import anyio +from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream + +from pb_service.hub import protocol as p +from pb_service.hub.stream import cobs, wire +from pb_service.hub.stream.pipe import BytePipe +from pb_service.hub.transport import ( + HubCommandError, + HubLink, + NotificationHandler, +) + +logger = logging.getLogger(__name__) + +__all__ = ["REPLY_PATIENCE_SECONDS", "StreamLink"] + +REPLY_PATIENCE_SECONDS: Final = 5.0 +"""How long any one request waits before the hub is treated as unresponsive.""" + +_PENDING_REPLIES: Final = 8 +"""Frames the reader may hold for a requester before dropping the oldest news.""" + + +@final +class StreamLink(HubLink): + """An open connection to one hub over a byte stream. + + Requests are serialised, one at a time. That is not caution: a hub answers + one read at a time, and issuing several without waiting loses all but the + last -- measured, not documented anywhere. With at most one request + outstanding, correlating a reply needs no bookkeeping beyond checking that + it is the one that was asked for. + """ + + def __init__(self, pipe: BytePipe, device_id: str) -> None: + self._pipe = pipe + self._device_id = device_id + self._lock = anyio.Lock() + self._handler: NotificationHandler | None = None + self._connected = True + self._tag = 0 + + send, receive = anyio.create_memory_object_stream[cobs.Frame](_PENDING_REPLIES) + self._replies_in: MemoryObjectSendStream[cobs.Frame] = send + self._replies_out: MemoryObjectReceiveStream[cobs.Frame] = receive + + @property + def is_connected(self) -> bool: + return self._connected + + async def read_characteristic(self, uuid: str) -> bytes: + selector = wire.READ_SELECTORS.get(uuid) + + if selector is None: + raise ValueError(f"no stream equivalent for characteristic {uuid}") + + async with self._lock: + self._discard_stale() + await self._write(wire.encode_read(selector)) + + with anyio.fail_after(REPLY_PATIENCE_SECONDS): + while True: + frame = await self._next_reply() + + if frame.type != wire.InMessage.READ_REPLY: + continue + + reply = wire.parse_read_reply(frame.payload) + + if reply is not None and reply[0] == selector: + return reply[1] + + async def write_characteristic( + self, uuid: str, data: bytes, *, response: bool = True + ) -> None: + if uuid != p.PYBRICKS_COMMAND_EVENT_UUID: + raise ValueError(f"characteristic {uuid} is not writable over a stream") + + self._tag = (self._tag + 1) % 256 + tag = self._tag + + async with self._lock: + self._discard_stale() + await self._write(wire.encode_command(tag, data)) + + # The hub answers every command, so the answer is collected even + # when the caller does not want one. Leaving it in the stream would + # make it the next request's problem. + with anyio.fail_after(REPLY_PATIENCE_SECONDS): + status = await self._await_response(tag) + + if status and response: + raise _rejection(status) + + async def start_notify(self, uuid: str, handler: NotificationHandler) -> None: + if uuid != p.PYBRICKS_COMMAND_EVENT_UUID: + raise ValueError(f"characteristic {uuid} does not notify over a stream") + + self._handler = handler + + async with self._lock: + await self._write(wire.encode_subscribe(subscribed=True)) + + async def _await_response(self, tag: int) -> int: + """Wait for the response carrying ``tag``. + + Args: + tag: What was sent with the command. + + Returns: + The firmware's status code, zero for success. + """ + while True: + frame = await self._next_reply() + + if frame.type != wire.InMessage.RESPONSE: + continue + + answer = wire.parse_response(frame.payload) + + if answer is not None and answer.tag == tag: + return answer.status + + async def _next_reply(self) -> cobs.Frame: + """Take the next frame addressed to whichever request is waiting. + + Returns: + The frame. + + Raises: + ConnectionError: If the link ended while waiting. The stream closing + under a request is how a hub going away arrives here, and + reporting it as a disconnect is what makes it the same failure a + write reports rather than an anyio internal nobody handles. + """ + try: + return await self._replies_out.receive() + except (anyio.EndOfStream, anyio.ClosedResourceError) as exc: + self._connected = False + raise ConnectionError(f"the hub on {self._device_id} went away") from exc + + async def _write(self, framed: bytes) -> None: + """Send one framed message, reporting a dead link as a disconnect.""" + try: + await self._pipe.write(framed) + except (OSError, ConnectionError) as exc: + self._connected = False + raise ConnectionError(f"the hub on {self._device_id} went away") from exc + + def _discard_stale(self) -> None: + """Drop anything left over from a request that gave up. + + A reply to a request nobody is waiting for any more would otherwise be + read as the answer to the next one. + """ + while True: + try: + self._replies_out.receive_nowait() + except (anyio.WouldBlock, anyio.EndOfStream, anyio.ClosedResourceError): + return + + async def run(self, on_lost: Callable[[str], None] | None) -> None: + """Read frames for as long as the link lasts. + + Events go straight to the notification handler and everything else to + whichever request is waiting. This is the only task reading the port, so + nothing here needs a lock. + + Args: + on_lost: Called with the hub's id if the link ends *by itself*, and + deliberately not when this task is cancelled. Disconnecting on + request unwinds through the same code, and reporting that as a + hub going away would tell somebody to go and rescue a hub that + is sitting there perfectly happily. + """ + try: + while True: + body = await self._pipe.readuntil(bytes((cobs.DELIMITER,))) + frame = cobs.decode(body[:-1]) + + if frame is None: + logger.debug("%s: discarding an undecodable frame", self._device_id) + continue + + if frame.type == wire.InMessage.EVENT: + if self._handler is not None: + self._handler(frame.payload) + + continue + + try: + self._replies_in.send_nowait(frame) + except anyio.WouldBlock: + logger.debug( + "%s: no one is waiting for %r", self._device_id, frame.type + ) + except ( + OSError, + ConnectionError, + asyncio.IncompleteReadError, + asyncio.LimitOverrunError, + ) as exc: + logger.info("%s: the link ended: %s", self._device_id, exc) + + if on_lost is not None: + on_lost(self._device_id) + finally: + self._connected = False + # Both ends: closing only the sending one leaves the receiving one + # to be collected instead, which the suite reports as a + # ResourceWarning, and a port opened and closed many times in a run + # should accumulate nothing at all. + self._replies_in.close() + self._replies_out.close() + + +def _rejection(status: int) -> HubCommandError: + """Turn a firmware status code into the hub's own refusal. + + Args: + status: What the response carried. + + Returns: + The equivalent error, naming the code where it is one we know. + """ + try: + error = p.HubError(status) + except ValueError: + return HubCommandError(status, f"hub refused the command (error {status:#04x})") + + return HubCommandError(error, f"hub refused the command: {error.name}") diff --git a/packages/pb-service/src/pb_service/hub/stream/pipe.py b/packages/pb-service/src/pb_service/hub/stream/pipe.py new file mode 100644 index 0000000..bb21f9b --- /dev/null +++ b/packages/pb-service/src/pb_service/hub/stream/pipe.py @@ -0,0 +1,45 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 The Pybricks Authors + +"""The little of an open byte stream a hub link uses. + +Two methods, because that is all the framing needs. Everything else about a +particular carrier -- the baud rate and modem pins of a cable, the address and +channel of a radio -- belongs to opening one, and so lives with that transport +rather than here. +""" + +from __future__ import annotations + +from collections.abc import Callable +from contextlib import AbstractAsyncContextManager +from typing import Protocol + +__all__ = ["BytePipe", "PipeOpener"] + + +class BytePipe(Protocol): + """An open byte stream, readable and writable.""" + + async def readuntil(self, separator: bytes = b"\n") -> bytes: + """Read up to and including the separator. + + Args: + separator: The byte sequence to stop after. + + Returns: + Everything up to and including it. + """ + ... + + async def write(self, data: bytes | bytearray | memoryview) -> None: + """Send bytes, waiting for them to leave the application buffer. + + Args: + data: What to send. + """ + ... + + +PipeOpener = Callable[[str], AbstractAsyncContextManager[BytePipe]] +"""How a pipe is opened, given the name of what is on the other end.""" diff --git a/packages/pb-service/src/pb_service/hub/usb/wire.py b/packages/pb-service/src/pb_service/hub/stream/wire.py similarity index 96% rename from packages/pb-service/src/pb_service/hub/usb/wire.py rename to packages/pb-service/src/pb_service/hub/stream/wire.py index d4c96c2..8b579b9 100644 --- a/packages/pb-service/src/pb_service/hub/usb/wire.py +++ b/packages/pb-service/src/pb_service/hub/stream/wire.py @@ -1,9 +1,9 @@ # SPDX-License-Identifier: MIT # Copyright (c) 2026 The Pybricks Authors -"""The messages a hub exchanges over USB. +"""The messages a hub exchanges over a byte stream. -The commands themselves are the ones the Bluetooth transport already sends -- +The commands themselves are the ones the GATT transport already sends -- :mod:`pb_service.hub.protocol` builds them and nothing here duplicates that. What this module adds is the envelope a byte stream needs and a radio does not: a message type, a tag so a response can be matched to the command that caused @@ -27,7 +27,7 @@ from typing import Final, NamedTuple from pb_service.hub import protocol as p -from pb_service.hub.usb import cobs +from pb_service.hub.stream import cobs __all__ = [ "MIN_PROFILE", diff --git a/packages/pb-service/src/pb_service/hub/usb/fake.py b/packages/pb-service/src/pb_service/hub/usb/fake.py index 3d9f624..75f8628 100644 --- a/packages/pb-service/src/pb_service/hub/usb/fake.py +++ b/packages/pb-service/src/pb_service/hub/usb/fake.py @@ -10,7 +10,7 @@ unreachable with hardware. Here they are one method call. **Replies are built by hand rather than with the encoders under test.** The -framing is shared, because :mod:`~pb_service.hub.usb.cobs` is already checked +framing is shared, because :mod:`~pb_service.hub.stream.cobs` is already checked against frames captured from a real hub and reusing a verified codec proves nothing wrong; but the envelope is packed here with :mod:`struct`, so a mistake in ``parse_response`` or ``parse_read_reply`` shows up as a failure instead of @@ -33,8 +33,9 @@ import anyio from pb_service.hub import protocol as p -from pb_service.hub.usb import cobs, wire -from pb_service.hub.usb.port import PortInfo, PortLister, PortOpener, SerialPort +from pb_service.hub.stream import cobs, wire +from pb_service.hub.stream.pipe import BytePipe +from pb_service.hub.usb.port import PortInfo, PortLister, PortOpener __all__ = ["CAPABILITIES", "FakeHubPort", "lister", "opener"] @@ -295,7 +296,7 @@ def opener(*ports: FakeHubPort) -> PortOpener: by_device = {port.info.device: port for port in ports} @asynccontextmanager - async def open_port(device: str) -> AsyncGenerator[SerialPort, None]: + async def open_port(device: str) -> AsyncGenerator[BytePipe, None]: found = by_device.get(device) if found is None: diff --git a/packages/pb-service/src/pb_service/hub/usb/port.py b/packages/pb-service/src/pb_service/hub/usb/port.py index 91449c6..d722218 100644 --- a/packages/pb-service/src/pb_service/hub/usb/port.py +++ b/packages/pb-service/src/pb_service/hub/usb/port.py @@ -21,10 +21,12 @@ import logging from collections.abc import AsyncGenerator, Callable, Iterable from contextlib import AbstractAsyncContextManager, asynccontextmanager -from typing import Final, NamedTuple, Protocol +from typing import Final, NamedTuple import serialx +from pb_service.hub.stream.pipe import BytePipe + logger = logging.getLogger(__name__) __all__ = [ @@ -32,7 +34,6 @@ "PortInfo", "PortLister", "PortOpener", - "SerialPort", "list_hub_ports", "open_hub_port", ] @@ -71,33 +72,10 @@ class PortInfo(NamedTuple): """ -class SerialPort(Protocol): - """The little of an open serial port this transport uses.""" - - async def readuntil(self, separator: bytes = b"\n") -> bytes: - """Read up to and including the separator. - - Args: - separator: The byte sequence to stop after. - - Returns: - Everything up to and including it. - """ - ... - - async def write(self, data: bytes | bytearray | memoryview) -> None: - """Send bytes, waiting for them to leave the application buffer. - - Args: - data: What to send. - """ - ... - - PortLister = Callable[[], Iterable[PortInfo]] """Where the list of ports that might be hubs comes from.""" -PortOpener = Callable[[str], AbstractAsyncContextManager[SerialPort]] +PortOpener = Callable[[str], AbstractAsyncContextManager[BytePipe]] """How a port is opened, given its device name.""" @@ -119,7 +97,7 @@ def list_hub_ports() -> Iterable[PortInfo]: @asynccontextmanager -async def open_hub_port(device: str) -> AsyncGenerator[SerialPort, None]: +async def open_hub_port(device: str) -> AsyncGenerator[BytePipe, None]: """Open a serial port ready to talk to a hub. Asserting DTR is part of opening rather than something the caller does, diff --git a/packages/pb-service/src/pb_service/hub/usb/transport.py b/packages/pb-service/src/pb_service/hub/usb/transport.py index 0864e36..d8a6822 100644 --- a/packages/pb-service/src/pb_service/hub/usb/transport.py +++ b/packages/pb-service/src/pb_service/hub/usb/transport.py @@ -23,30 +23,26 @@ from __future__ import annotations -import asyncio import logging from collections.abc import AsyncGenerator, Callable from contextlib import asynccontextmanager, suppress from typing import Final, final import anyio -from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream from pb_service.hub import protocol as p +from pb_service.hub.stream import wire +from pb_service.hub.stream.link import StreamLink from pb_service.hub.transport import ( DeviceInfo, - HubCommandError, HubLink, - NotificationHandler, ScanFilter, UnavailableDevice, ) -from pb_service.hub.usb import cobs, wire from pb_service.hub.usb.port import ( PortInfo, PortLister, PortOpener, - SerialPort, list_hub_ports, open_hub_port, ) @@ -57,9 +53,6 @@ SCHEME: Final = "usb" -REPLY_PATIENCE_SECONDS: Final = 5.0 -"""How long any one request waits before the hub is treated as unresponsive.""" - PROBE_PATIENCE_SECONDS: Final = 2.0 """How long discovery spends naming one hub before giving up on it.""" @@ -76,9 +69,6 @@ is small. It also runs only while somebody is scanning. """ -_PENDING_REPLIES: Final = 8 -"""Frames the reader may hold for a requester before dropping the oldest news.""" - def device_id_for(port: PortInfo) -> str: """Return the service-facing id for a port. @@ -96,216 +86,6 @@ def device_id_for(port: PortInfo) -> str: return f"{SCHEME}:{port.serial_number or port.device}" -@final -class _UsbLink(HubLink): - """An open connection to one hub over its serial port. - - Requests are serialised, one at a time. That is not caution: a hub answers - one read at a time, and issuing several without waiting loses all but the - last -- measured, not documented anywhere. With at most one request - outstanding, correlating a reply needs no bookkeeping beyond checking that - it is the one that was asked for. - """ - - def __init__(self, serial: SerialPort, device_id: str) -> None: - self._serial = serial - self._device_id = device_id - self._lock = anyio.Lock() - self._handler: NotificationHandler | None = None - self._connected = True - self._tag = 0 - - send, receive = anyio.create_memory_object_stream[cobs.Frame](_PENDING_REPLIES) - self._replies_in: MemoryObjectSendStream[cobs.Frame] = send - self._replies_out: MemoryObjectReceiveStream[cobs.Frame] = receive - - @property - def is_connected(self) -> bool: - return self._connected - - async def read_characteristic(self, uuid: str) -> bytes: - selector = wire.READ_SELECTORS.get(uuid) - - if selector is None: - raise ValueError(f"no USB equivalent for characteristic {uuid}") - - async with self._lock: - self._discard_stale() - await self._write(wire.encode_read(selector)) - - with anyio.fail_after(REPLY_PATIENCE_SECONDS): - while True: - frame = await self._next_reply() - - if frame.type != wire.InMessage.READ_REPLY: - continue - - reply = wire.parse_read_reply(frame.payload) - - if reply is not None and reply[0] == selector: - return reply[1] - - async def write_characteristic( - self, uuid: str, data: bytes, *, response: bool = True - ) -> None: - if uuid != p.PYBRICKS_COMMAND_EVENT_UUID: - raise ValueError(f"characteristic {uuid} is not writable over USB") - - self._tag = (self._tag + 1) % 256 - tag = self._tag - - async with self._lock: - self._discard_stale() - await self._write(wire.encode_command(tag, data)) - - # The hub answers every command, so the answer is collected even - # when the caller does not want one. Leaving it in the stream would - # make it the next request's problem. - with anyio.fail_after(REPLY_PATIENCE_SECONDS): - status = await self._await_response(tag) - - if status and response: - raise _rejection(status) - - async def start_notify(self, uuid: str, handler: NotificationHandler) -> None: - if uuid != p.PYBRICKS_COMMAND_EVENT_UUID: - raise ValueError(f"characteristic {uuid} does not notify over USB") - - self._handler = handler - - async with self._lock: - await self._write(wire.encode_subscribe(subscribed=True)) - - async def _await_response(self, tag: int) -> int: - """Wait for the response carrying ``tag``. - - Args: - tag: What was sent with the command. - - Returns: - The firmware's status code, zero for success. - """ - while True: - frame = await self._next_reply() - - if frame.type != wire.InMessage.RESPONSE: - continue - - answer = wire.parse_response(frame.payload) - - if answer is not None and answer.tag == tag: - return answer.status - - async def _next_reply(self) -> cobs.Frame: - """Take the next frame addressed to whichever request is waiting. - - Returns: - The frame. - - Raises: - ConnectionError: If the link ended while waiting. The stream closing - under a request is how a hub going away arrives here, and - reporting it as a disconnect is what makes it the same failure a - write reports rather than an anyio internal nobody handles. - """ - try: - return await self._replies_out.receive() - except (anyio.EndOfStream, anyio.ClosedResourceError) as exc: - self._connected = False - raise ConnectionError(f"the hub on {self._device_id} went away") from exc - - async def _write(self, framed: bytes) -> None: - """Send one framed message, reporting a dead link as a disconnect.""" - try: - await self._serial.write(framed) - except (OSError, ConnectionError) as exc: - self._connected = False - raise ConnectionError(f"the hub on {self._device_id} went away") from exc - - def _discard_stale(self) -> None: - """Drop anything left over from a request that gave up. - - A reply to a request nobody is waiting for any more would otherwise be - read as the answer to the next one. - """ - while True: - try: - self._replies_out.receive_nowait() - except (anyio.WouldBlock, anyio.EndOfStream, anyio.ClosedResourceError): - return - - async def run(self, on_lost: Callable[[str], None] | None) -> None: - """Read frames for as long as the link lasts. - - Events go straight to the notification handler and everything else to - whichever request is waiting. This is the only task reading the port, so - nothing here needs a lock. - - Args: - on_lost: Called with the hub's id if the link ends *by itself*, and - deliberately not when this task is cancelled. Disconnecting on - request unwinds through the same code, and reporting that as a - hub going away would tell somebody to go and rescue a hub that - is sitting there perfectly happily. - """ - try: - while True: - body = await self._serial.readuntil(bytes((cobs.DELIMITER,))) - frame = cobs.decode(body[:-1]) - - if frame is None: - logger.debug("%s: discarding an undecodable frame", self._device_id) - continue - - if frame.type == wire.InMessage.EVENT: - if self._handler is not None: - self._handler(frame.payload) - - continue - - try: - self._replies_in.send_nowait(frame) - except anyio.WouldBlock: - logger.debug( - "%s: no one is waiting for %r", self._device_id, frame.type - ) - except ( - OSError, - ConnectionError, - asyncio.IncompleteReadError, - asyncio.LimitOverrunError, - ) as exc: - logger.info("%s: the link ended: %s", self._device_id, exc) - - if on_lost is not None: - on_lost(self._device_id) - finally: - self._connected = False - # Both ends: closing only the sending one leaves the receiving one - # to be collected instead, which the suite reports as a - # ResourceWarning, and a port opened and closed many times in a run - # should accumulate nothing at all. - self._replies_in.close() - self._replies_out.close() - - -def _rejection(status: int) -> HubCommandError: - """Turn a firmware status code into the hub's own refusal. - - Args: - status: What the response carried. - - Returns: - The equivalent error, naming the code where it is one we know. - """ - try: - error = p.HubError(status) - except ValueError: - return HubCommandError(status, f"hub refused the command (error {status:#04x})") - - return HubCommandError(error, f"hub refused the command: {error.name}") - - @final class UsbHubTransport: """Finds hubs on the serial ports and opens connections to them.""" @@ -430,7 +210,7 @@ async def _probe( try: with anyio.fail_after(PROBE_PATIENCE_SECONDS): async with self._open_port(port.device) as serial: - link = _UsbLink(serial, device_id) + link = StreamLink(serial, device_id) async with anyio.create_task_group() as tg: tg.start_soon(link.run, None) @@ -492,7 +272,7 @@ async def open(self, device_id: str) -> AsyncGenerator[HubLink, None]: raise ConnectionError(f"no hub is plugged in as {device_id}") async with self._open_port(port.device) as serial: - link = _UsbLink(serial, device_id) + link = StreamLink(serial, device_id) async with anyio.create_task_group() as tg: tg.start_soon(link.run, self._on_lost) diff --git a/packages/pb-service/tests/test_cobs.py b/packages/pb-service/tests/test_cobs.py index 4c934f9..b3888d2 100644 --- a/packages/pb-service/tests/test_cobs.py +++ b/packages/pb-service/tests/test_cobs.py @@ -14,7 +14,7 @@ import pytest -from pb_service.hub.usb import cobs +from pb_service.hub.stream import cobs SUBSCRIBE = 1 READ = 3 diff --git a/packages/pb-service/tests/test_usb_link.py b/packages/pb-service/tests/test_usb_link.py index faba1fd..49ce826 100644 --- a/packages/pb-service/tests/test_usb_link.py +++ b/packages/pb-service/tests/test_usb_link.py @@ -22,6 +22,8 @@ import pytest from pb_service.hub import protocol as p +from pb_service.hub.stream import wire +from pb_service.hub.stream.pipe import BytePipe from pb_service.hub.transport import ( HubCommandError, HubLink, @@ -29,9 +31,7 @@ UnavailableDevice, ) from pb_service.hub.usb import transport as usb_transport -from pb_service.hub.usb import wire from pb_service.hub.usb.fake import FakeHubPort, lister, opener -from pb_service.hub.usb.port import SerialPort from pb_service.hub.usb.transport import UsbHubTransport pytestmark = pytest.mark.anyio @@ -287,7 +287,7 @@ async def test_a_late_reply_is_not_the_next_requests_answer() -> None: async def test_a_characteristic_with_no_usb_equivalent_is_refused( link: HubLink, ) -> None: - with pytest.raises(ValueError, match="no USB equivalent"): + with pytest.raises(ValueError, match="no stream equivalent"): await link.read_characteristic("0000face-0000-1000-8000-00805f9b34fb") @@ -354,7 +354,7 @@ async def test_two_hubs_are_told_apart_by_their_serial_numbers() -> None: @asynccontextmanager -async def _refused(device: str) -> AsyncGenerator[SerialPort, None]: +async def _refused(device: str) -> AsyncGenerator[BytePipe, None]: """Refuse to open a port, the way the operating system refuses one. Args: @@ -421,7 +421,7 @@ async def test_a_port_that_opens_on_a_later_turn_stops_being_reported( refusals = 0 @asynccontextmanager - async def open_port(device: str) -> AsyncGenerator[SerialPort, None]: + async def open_port(device: str) -> AsyncGenerator[BytePipe, None]: nonlocal refusals if refusals == 0: From a770a862f713fc7e4a86c09e17a388df3a077f9a Mon Sep 17 00:00:00 2001 From: David Lechner Date: Sat, 17 Oct 2026 23:52:42 -0500 Subject: [PATCH 2/7] pb-service: EV3 and NXT hubs over Bluetooth Classic The EV3 and NXT have no BLE, so they are reached over RFCOMM, through aio-rfcomm, which gives one asyncio API over BlueZ, IOBluetooth and WinRT. What crosses the channel is the framed stream USB already carries, so the transport is mostly a matter of finding hubs and handing the channel to StreamLink. Finding them is not a scan and cannot be one. Bluetooth Classic pairs in the operating system, so what is listed is the paired set: a hub appears as soon as the computer knows it and stays listed whether it is switched on or not. Only opening it says which, so a hub that is off arrives as ConnectionError from open() -- the routine way to fail here, not an edge case. A hub is recognised by its service class. The EV3 has its own, from the 0x01xx block of the Pybricks UUIDs, which is not GATT attributes. The NXT publishes only plain Serial Port, which every paired headset offers too, so one of those counts only when its name ends in the "(MINDSTORMS NXT)" the firmware appends. The same suffix gives the type for the scan's own column, and is taken off the name shown. Bluetooth being off, or this program not being allowed it, is reported as the transport's one unavailable entry rather than raised, so the other transports still answer. Discovery follows Bluetooth's availability as well as polling the paired set every 2 s, so that entry appears and clears as soon as the switch does. aio-rfcomm comes from PyPI. It adds rubicon-objc on macOS and the WinRT RFCOMM projection on Windows. On Linux it needs only dbus-fast, which bleak already brings. --- packages/pb-service/pyproject.toml | 1 + .../pb-service/src/pb_service/hub/protocol.py | 24 ++ .../src/pb_service/hub/rfcomm/__init__.py | 10 + .../src/pb_service/hub/rfcomm/fake.py | 216 ++++++++++++++++ .../src/pb_service/hub/rfcomm/port.py | 194 +++++++++++++++ .../src/pb_service/hub/rfcomm/transport.py | 228 +++++++++++++++++ .../pb-service/tests/test_rfcomm_naming.py | 61 +++++ packages/pb-service/tests/test_rfcomm_pipe.py | 109 +++++++++ .../pb-service/tests/test_rfcomm_transport.py | 230 ++++++++++++++++++ uv.lock | 55 +++++ 10 files changed, 1128 insertions(+) create mode 100644 packages/pb-service/src/pb_service/hub/rfcomm/__init__.py create mode 100644 packages/pb-service/src/pb_service/hub/rfcomm/fake.py create mode 100644 packages/pb-service/src/pb_service/hub/rfcomm/port.py create mode 100644 packages/pb-service/src/pb_service/hub/rfcomm/transport.py create mode 100644 packages/pb-service/tests/test_rfcomm_naming.py create mode 100644 packages/pb-service/tests/test_rfcomm_pipe.py create mode 100644 packages/pb-service/tests/test_rfcomm_transport.py diff --git a/packages/pb-service/pyproject.toml b/packages/pb-service/pyproject.toml index 82c6826..037bd77 100644 --- a/packages/pb-service/pyproject.toml +++ b/packages/pb-service/pyproject.toml @@ -23,6 +23,7 @@ dependencies = [ "bleak>=1.1.0", "packaging>=22", "serialx>=1.9", + "aio-rfcomm>=1.0", "semver>=3", "typing-extensions>=4.15", ] diff --git a/packages/pb-service/src/pb_service/hub/protocol.py b/packages/pb-service/src/pb_service/hub/protocol.py index b9b05f5..cf95c67 100644 --- a/packages/pb-service/src/pb_service/hub/protocol.py +++ b/packages/pb-service/src/pb_service/hub/protocol.py @@ -29,6 +29,7 @@ __all__ = [ "PYBRICKS_COMMAND_EVENT_UUID", "PYBRICKS_HUB_CAPABILITIES_UUID", + "PYBRICKS_RFCOMM_SERVICE_UUID", "PYBRICKS_SERVICE_UUID", "Capabilities", "Command", @@ -48,6 +49,7 @@ "encode_write_stdin", "encode_write_user_program_meta", "encode_write_user_ram", + "hub_kind_name", "hub_type_name", "parse_status_report", "parse_version", @@ -73,6 +75,14 @@ def _standard_uuid(short: int) -> str: PYBRICKS_COMMAND_EVENT_UUID: Final = _pybricks_uuid(0x0002) PYBRICKS_HUB_CAPABILITIES_UUID: Final = _pybricks_uuid(0x0003) +PYBRICKS_RFCOMM_SERVICE_UUID: Final = _pybricks_uuid(0x0101) +"""Service class of the hub's Bluetooth Classic serial port. + +From the 0x01xx block, which is not GATT attributes. A dedicated class rather +than plain Serial Port, which would match every paired headset and GPS, or the +GATT service UUID, which would match every BLE-only hub. +""" + DEVICE_INFORMATION_SERVICE_UUID: Final = _standard_uuid(0x180A) DEVICE_NAME_UUID: Final = _standard_uuid(0x2A00) FIRMWARE_REVISION_UUID: Final = _standard_uuid(0x2A26) @@ -617,6 +627,20 @@ class VendorIdSource(IntEnum): """ +def hub_kind_name(kind: HubType) -> str | None: + """Name a kind of hub that is known without a PnP ID. + + Args: + kind: Which hub. + + Returns: + Its name, or ``None`` where the name depends on the product version + and so cannot be told from the kind alone, as for the large and small + Technic hubs. + """ + return _HUB_TYPE_NAMES.get((kind, None)) + + def hub_type_name(pnp: PnpId | None) -> str | None: """Name the kind of hub a PnP ID describes. diff --git a/packages/pb-service/src/pb_service/hub/rfcomm/__init__.py b/packages/pb-service/src/pb_service/hub/rfcomm/__init__.py new file mode 100644 index 0000000..1e2c5cb --- /dev/null +++ b/packages/pb-service/src/pb_service/hub/rfcomm/__init__.py @@ -0,0 +1,10 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 The Pybricks Authors + +"""RFCOMM transport: a hub reached over Bluetooth Classic serial.""" + +from __future__ import annotations + +from pb_service.hub.rfcomm.transport import RfcommHubTransport, device_id_for + +__all__ = ["RfcommHubTransport", "device_id_for"] diff --git a/packages/pb-service/src/pb_service/hub/rfcomm/fake.py b/packages/pb-service/src/pb_service/hub/rfcomm/fake.py new file mode 100644 index 0000000..d92c678 --- /dev/null +++ b/packages/pb-service/src/pb_service/hub/rfcomm/fake.py @@ -0,0 +1,216 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 The Pybricks Authors + +"""Bluetooth with simulated hubs paired to it. + +What this exists for is the half of the RFCOMM transport a real hub cannot +help with. A hub that is switched on answers every time, so the paths that +matter when something goes wrong -- a hub paired but switched off, a radio the +user has not allowed, a paired device that is not a hub at all, a hub that +stops answering mid-request -- are unreachable with hardware. Here each is one +argument. + +The hub on the far end of a channel is :class:`~pb_service.hub.usb.fake. +FakeHubPort`, which answers the same framed protocol whatever carries it. That +it lives under ``usb`` is a leftover of being written there first; it belongs +beside the rest of the shared stream code, and moving it is a tidy-up this +does not wait for. +""" + +from __future__ import annotations + +from collections.abc import AsyncGenerator, Collection, Sequence +from contextlib import asynccontextmanager +from typing import Any, Self, cast, final +from uuid import UUID + +import aio_rfcomm +import anyio +from aio_rfcomm.errors import ( + BluetoothUnavailableError, + DeviceNotFoundError, + ServiceNotFoundError, +) + +from pb_service.hub.rfcomm.port import PYBRICKS_RFCOMM_SERVICE, split_hub_name +from pb_service.hub.stream import cobs +from pb_service.hub.usb.fake import FakeHubPort + +__all__ = ["FakePairedHub", "radio", "switch"] + +_DELIMITER = bytes((cobs.DELIMITER,)) + + +@final +class FakePairedHub: + """A device this computer is paired with.""" + + def __init__( + self, + *, + name: str = "Bumblebee (MINDSTORMS EV3)", + address: str = "00:16:53:4A:75:15", + service: UUID = PYBRICKS_RFCOMM_SERVICE, + reachable: bool = True, + port: FakeHubPort | None = None, + ) -> None: + """Pair a device with the radio. + + Args: + name: What the operating system calls it, suffix and all. + address: Its Bluetooth address. + service: The service class it answers on. Plain Serial Port makes + it a device that is not necessarily a hub. + reachable: Whether opening it succeeds. False is the common case + this transport has to get right: a hub the computer is paired + with but which is switched off. + port: The hub on the far end, or None for a stock one. + """ + self.name = name + self.address = address + self.service = service + self.reachable = reachable + self._port = port + + @property + def port(self) -> FakeHubPort: + """The hub on the far end, made only once something opens it. + + Lazily, because a hub holds streams that have to be closed again, and + most of what this file is for never opens a channel at all. + """ + if self._port is None: + bare, _ = split_hub_name(self.name) + self._port = FakeHubPort(name=bare or self.name) + + return self._port + + @property + def info(self) -> aio_rfcomm.RfcommDeviceInfo: + """How the radio describes it.""" + return aio_rfcomm.RfcommDeviceInfo(self.address, self.name) + + +@final +class _FakeChannel: + """One frame at a time, as a radio hands them over.""" + + def __init__(self, port: FakeHubPort) -> None: + self._port = port + + async def receive_until(self, delimiter: bytes, *, max_bytes: int) -> bytes: + return (await self._port.readuntil(delimiter))[: -len(delimiter)] + + async def send(self, data: bytes) -> None: + await self._port.write(data) + + +@final +class _FakeWatcher: + """Availability now, then each change, as the real watcher gives it.""" + + def __init__(self, bluetooth: _FakeBluetooth) -> None: + self._bluetooth = bluetooth + self._last: aio_rfcomm.Availability | None = None + + def __aiter__(self) -> Self: + return self + + async def __anext__(self) -> aio_rfcomm.Availability: + while True: + changed = self._bluetooth.changed + current = self._bluetooth.availability + + if current is not self._last: + self._last = current + return current + + await changed.wait() + + +@final +class _FakeBluetooth: + def __init__(self, hubs: Sequence[FakePairedHub], unavailable: str | None) -> None: + self._hubs = hubs + self._unavailable = unavailable + self.changed = anyio.Event() + + @property + def availability(self) -> aio_rfcomm.Availability: + if self._unavailable is None: + return aio_rfcomm.Availability.USABLE + + return aio_rfcomm.Availability.SWITCHED_OFF + + def switch(self, unavailable: str | None) -> None: + self._unavailable = unavailable + self.changed.set() + self.changed = anyio.Event() + + def watch_availability(self) -> _FakeWatcher: + return _FakeWatcher(self) + + def _check(self) -> None: + if self._unavailable is not None: + raise BluetoothUnavailableError( + aio_rfcomm.Availability.SWITCHED_OFF, self._unavailable + ) + + async def list_known_devices( + self, *, service: UUID | Collection[UUID] | None = None + ) -> list[aio_rfcomm.RfcommDeviceInfo]: + self._check() + wanted = {service} if isinstance(service, UUID) else service + return [ + hub.info for hub in self._hubs if wanted is None or hub.service in wanted + ] + + @asynccontextmanager + async def open_service( + self, device: aio_rfcomm.RfcommDeviceInfo, service: UUID + ) -> AsyncGenerator[Any, None]: + self._check() + hub = next((h for h in self._hubs if h.address == device.address), None) + + if hub is None or not hub.reachable: + raise DeviceNotFoundError(device.address) + + if service != hub.service: + raise ServiceNotFoundError(f"{device.address} does not offer {service}") + + port = hub.port + + try: + yield _FakeChannel(port) + finally: + port.close() + + +def radio(*hubs: FakePairedHub, unavailable: str | None = None) -> aio_rfcomm.Bluetooth: + """Build Bluetooth with these devices paired to it. + + Args: + hubs: What the computer is paired with. + unavailable: Why Bluetooth cannot be used, or None for usable. This is + the switched-off or not-allowed case, which the transport has to + report without stopping the other transports. Opening Bluetooth + succeeds regardless, so it is every call that fails. + + Returns: + Something to hand :class:`~pb_service.hub.rfcomm.transport. + RfcommHubTransport`. Cast, because the real one is a class rather + than a protocol and nothing structural can stand in for it. + """ + return cast("aio_rfcomm.Bluetooth", _FakeBluetooth(hubs, unavailable)) + + +def switch(bluetooth: aio_rfcomm.Bluetooth, unavailable: str | None) -> None: + """Make Bluetooth usable or unusable while it is in use. + + Args: + bluetooth: What :func:`radio` built. + unavailable: Why Bluetooth cannot be used from now on, or None for + usable again. Watchers hear of it at once, as they would of the + user flipping the switch. + """ + cast("_FakeBluetooth", bluetooth).switch(unavailable) diff --git a/packages/pb-service/src/pb_service/hub/rfcomm/port.py b/packages/pb-service/src/pb_service/hub/rfcomm/port.py new file mode 100644 index 0000000..bc2dc34 --- /dev/null +++ b/packages/pb-service/src/pb_service/hub/rfcomm/port.py @@ -0,0 +1,194 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 The Pybricks Authors + +"""Where an RFCOMM channel to a hub comes from. + +Discovery here is not a scan. Bluetooth Classic devices are paired in the +operating system first, and what this can offer is the paired set -- so a hub +appears as soon as it is known and stays there whether it is switched on or +not, which is a different promise from the one a radio scan makes. +""" + +from __future__ import annotations + +import logging +from collections.abc import AsyncGenerator, Mapping +from contextlib import asynccontextmanager +from typing import Final +from uuid import UUID + +import aio_rfcomm +from aio_rfcomm.errors import ( + BluetoothUnavailableError, + ChannelClosedError, + DelimiterNotFoundError, +) + +from pb_service.hub import protocol as p +from pb_service.hub.protocol import HubType +from pb_service.hub.stream.pipe import BytePipe + +logger = logging.getLogger(__name__) + +__all__ = [ + "HUB_SERVICES", + "HUB_TYPE_SUFFIXES", + "MAX_FRAME", + "PYBRICKS_RFCOMM_SERVICE", + "SERIAL_PORT_SERVICE", + "ChannelPipe", + "list_hub_devices", + "open_hub_channel", + "split_hub_name", +] + +PYBRICKS_RFCOMM_SERVICE: Final = UUID(p.PYBRICKS_RFCOMM_SERVICE_UUID) +"""The EV3's own service class.""" + +SERIAL_PORT_SERVICE: Final = UUID("00001101-0000-1000-8000-00805f9b34fb") +"""Plain Serial Port, which the NXT publishes because it has nothing else yet. + +Specific to neither Pybricks nor LEGO: every paired headset and GPS offers it, +so what it finds are candidates rather than hubs. +""" + +HUB_SERVICES: Final = (PYBRICKS_RFCOMM_SERVICE, SERIAL_PORT_SERVICE) +"""The service classes a hub may answer on, most specific first.""" + +MAX_FRAME: Final = 2**16 +"""The longest frame read before giving up on the hub, delimiter included. + +The same bound USB reads have: serialx reads through an asyncio.StreamReader +left at its default limit. +""" + +HUB_TYPE_SUFFIXES: Final[Mapping[str, HubType]] = { + "MINDSTORMS EV3": HubType.EV3, + "MINDSTORMS NXT": HubType.NXT, +} +"""Which brick each name suffix means. + +The firmware appends "()" to the hub name, from PBSYS_CONFIG_HUB_TYPE_STR +in its platform config, so the name carries the type where neither the service +class nor anything else does. These are presentation strings rather than +protocol: a rewording upstream stops NXTs being recognised, while EV3s keep +being found by their own service class. +""" + + +def split_hub_name(name: str | None) -> tuple[str | None, HubType | None]: + """Separate a hub's own name from the type the firmware appended to it. + + Args: + name: What the operating system calls the device. + + Returns: + The name without the suffix, and the kind it named. The kind is + ``None`` for anything identifying no hub -- which is every other + serial device this computer happens to be paired with -- and the name + is then returned unchanged, since the brackets are then the owner's. + """ + if not name or not name.endswith(")"): + return name, None + + # The last bracketed part, because the firmware appends its own: a hub + # whose user-set name also ends in brackets still answers correctly. + start = name.rfind("(") + + if start < 0: + return name, None + + kind = HUB_TYPE_SUFFIXES.get(name[start + 1 : -1].strip()) + + if kind is None: + return name, None + + return name[:start].strip() or None, kind + + +class ChannelPipe: + """An open RFCOMM channel, as the byte stream the link expects.""" + + def __init__(self, channel: aio_rfcomm.RfcommChannel) -> None: + self._channel = channel + + async def readuntil(self, separator: bytes = b"\n") -> bytes: + """Read up to and including the separator. + + Args: + separator: The byte sequence to stop after. + + Returns: + Everything up to and including it. + + Raises: + ConnectionError: The hub hung up before one arrived, Bluetooth + stopped being usable, or :data:`MAX_FRAME` bytes arrived + without one. Reported the way a cable being pulled is, so the + link above treats a radio going quiet and a plug coming out as + the same thing. + """ + try: + body = await self._channel.receive_until(separator, max_bytes=MAX_FRAME) + except (ChannelClosedError, BluetoothUnavailableError) as exc: + raise ConnectionError(f"the channel to the hub ended: {exc}") from exc + except DelimiterNotFoundError as exc: + raise ConnectionError( + f"the hub sent a frame too long to read: {exc}" + ) from exc + + # The link strips the separator itself. + return body + separator + + async def write(self, data: bytes | bytearray | memoryview) -> None: + """Send bytes to the hub. + + Args: + data: What to send. + + Raises: + ConnectionError: The channel or Bluetooth stopped working. + """ + try: + await self._channel.send(bytes(data)) + except (ChannelClosedError, BluetoothUnavailableError) as exc: + raise ConnectionError(f"the channel to the hub ended: {exc}") from exc + + +async def list_hub_devices( + bluetooth: aio_rfcomm.Bluetooth, service: UUID +) -> list[aio_rfcomm.RfcommDeviceInfo]: + """List the paired devices offering one service. + + One service at a time rather than all at once, because which one answered + is what names the hub. + + Args: + bluetooth: Bluetooth, open. + service: The service class to match. + + Returns: + What the OS knows about each one. + """ + return await bluetooth.list_known_devices(service=service) + + +@asynccontextmanager +async def open_hub_channel( + bluetooth: aio_rfcomm.Bluetooth, + device: aio_rfcomm.RfcommDeviceInfo, + service: UUID, +) -> AsyncGenerator[BytePipe, None]: + """Open a hub's serial service. + + Args: + bluetooth: Bluetooth, open. + device: The hub, as discovery listed it. + service: The service class discovery matched it on. + + Yields: + The open channel, as a byte stream. + """ + async with bluetooth.open_service(device, service) as channel: + logger.debug("opened the hub serial service on %s", device.address) + yield ChannelPipe(channel) diff --git a/packages/pb-service/src/pb_service/hub/rfcomm/transport.py b/packages/pb-service/src/pb_service/hub/rfcomm/transport.py new file mode 100644 index 0000000..d3f3e7b --- /dev/null +++ b/packages/pb-service/src/pb_service/hub/rfcomm/transport.py @@ -0,0 +1,228 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 The Pybricks Authors + +"""RFCOMM transport: a hub reached over Bluetooth Classic serial. + +What travels over the channel is the same framed protocol USB carries, so +everything above :mod:`pb_service.hub.stream` is shared and only getting the +bytes differs. + +Discovery is not a scan and cannot be one. Bluetooth Classic pairs in the +operating system, and what is offered here is the paired set -- so a hub shows +up as soon as it is known to the computer and stays whether it is switched on +or not. A listing therefore means "hubs this computer has been introduced to", +not "hubs that are here", and only opening one says which. +""" + +from __future__ import annotations + +import logging +from collections.abc import AsyncGenerator, Callable +from contextlib import AsyncExitStack, asynccontextmanager +from typing import Final, final +from uuid import UUID + +import aio_rfcomm +import anyio +from aio_rfcomm.errors import BluetoothUnavailableError, RfcommError + +from pb_service.hub import protocol as p +from pb_service.hub.rfcomm.port import ( + HUB_SERVICES, + SERIAL_PORT_SERVICE, + list_hub_devices, + open_hub_channel, + split_hub_name, +) +from pb_service.hub.stream.link import StreamLink +from pb_service.hub.transport import ( + DeviceInfo, + HubLink, + ScanFilter, + UnavailableDevice, +) + +logger = logging.getLogger(__name__) + +__all__ = ["RfcommHubTransport", "device_id_for"] + +SCHEME: Final = "rfcomm" + +POLL_SECONDS: Final = 2.0 +"""How often the paired set is re-read while someone is listening. + +Slower than the USB poll because the answer almost never changes: pairing is +something a person does in system settings, not something that happens while a +scan is running. It is here at all so that a hub paired mid-scan still appears. +""" + + +def device_id_for(address: str) -> str: + """Return the service-facing id for a Bluetooth address. + + Args: + address: The hub's Bluetooth address. + + Returns: + The prefixed identifier. + """ + return f"{SCHEME}:{address}" + + +@final +class RfcommHubTransport: + """Finds paired hubs and opens RFCOMM channels to them.""" + + def __init__(self, bluetooth: aio_rfcomm.Bluetooth) -> None: + """Create a transport. + + Args: + bluetooth: Bluetooth, open for at least as long as this is used. + :func:`~pb_service.backend.open_hub_transport` opens it. + """ + self._bluetooth = bluetooth + self.on_disconnect: Callable[[str], None] | None = None + self._unavailable: dict[str, str] = {} + # The device as listed and the service it answered on, so that opening + # it uses the one that found it rather than guessing. + self._found: dict[str, tuple[aio_rfcomm.RfcommDeviceInfo, UUID]] = {} + + @property + def scheme(self) -> str: + return SCHEME + + @property + def unavailable(self) -> tuple[UnavailableDevice, ...]: + """What the radio would not let us look at. + + A refused permission or a switched-off adapter, rather than a hub: the + paired set cannot be read at all in either case, so the one entry here + stands for the adapter. + """ + return tuple( + UnavailableDevice(address, reason) + for address, reason in self._unavailable.items() + ) + + async def discover( + self, scan_filter: ScanFilter + ) -> AsyncGenerator[DeviceInfo, None]: + """Report paired hubs until the caller stops listening. + + Args: + scan_filter: Which hubs to report. + + Yields: + Each paired hub offering the service, re-reported on every pass. + """ + if scan_filter.transport is not None and scan_filter.transport != SCHEME: + return + + # Followed as well as polled, so that Bluetooth being switched off or + # on is reported at once rather than on the next pass. Only a wake-up: + # each pass still learns what is wrong from reading the paired set, + # whose error carries what the platform said, such as how to allow + # this program. + changes = self._bluetooth.watch_availability() + # The state now, which the first pass is about to read anyway. + await anext(changes) + + while True: + self._unavailable.clear() + + try: + seen: set[str] = set() + + for service in HUB_SERVICES: + for device in await list_hub_devices(self._bluetooth, service): + if device.address in seen: + continue + + bare, kind = split_hub_name(device.name) + + # Plain Serial Port is not ours: every paired + # headset offers it too, so only a name that + # identifies a brick makes one of those a hub. + if kind is None and service == SERIAL_PORT_SERVICE: + continue + + seen.add(device.address) + device_id = device_id_for(device.address) + self._found[device_id] = (device, service) + + if device_id in scan_filter.exclude: + continue + + # Without the type suffix: the listing has its + # own column for that. + name = bare or device.address + + if scan_filter.name is not None and name != scan_filter.name: + continue + + yield DeviceInfo( + id=device_id, + name=name, + address=device.address, + transport=SCHEME, + hub_type=(p.hub_kind_name(kind) if kind else None), + ) + except BluetoothUnavailableError as exc: + # The library's own words, kept rather than paraphrased: they + # are what tells "turn Bluetooth on" from "allow this program". + self._unavailable["bluetooth"] = str(exc) + logger.debug("cannot read the paired set: %s", exc) + + with anyio.move_on_after(POLL_SECONDS): + await anext(changes) + + @asynccontextmanager + async def open(self, device_id: str) -> AsyncGenerator[HubLink, None]: + """Open a channel to a hub. + + Args: + device_id: An id previously reported by :meth:`discover`. + + Yields: + The open link, which disconnects on exit. + """ + address = device_id.removeprefix(f"{SCHEME}:") + + if address == device_id: + raise ConnectionError(f"{device_id} is not an RFCOMM hub") + + found = self._found.get(device_id) + + if found is None: + raise ConnectionError( + f"{device_id} has not been discovered, so there is no knowing " + f"which service to open" + ) + + device, service = found + + # An exit stack rather than "async with", so that entering is inside + # the try while the body is not: with an async context manager the + # work happens in __aenter__. + async with AsyncExitStack() as stack: + try: + pipe = await stack.enter_async_context( + open_hub_channel(self._bluetooth, device, service) + ) + except RfcommError as exc: + raise ConnectionError(f"cannot reach {device_id}: {exc}") from exc + + link = StreamLink(pipe, device_id) + + async with anyio.create_task_group() as tg: + tg.start_soon(link.run, self._on_lost) + + try: + yield link + finally: + tg.cancel_scope.cancel() + + def _on_lost(self, device_id: str) -> None: + """Report a hub that went away by itself.""" + if self.on_disconnect is not None: + self.on_disconnect(device_id) diff --git a/packages/pb-service/tests/test_rfcomm_naming.py b/packages/pb-service/tests/test_rfcomm_naming.py new file mode 100644 index 0000000..14051e0 --- /dev/null +++ b/packages/pb-service/tests/test_rfcomm_naming.py @@ -0,0 +1,61 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 The Pybricks Authors + +"""Telling a brick from everything else the computer is paired with. + +Bluetooth Classic offers no PnP ID to a scan, and plain Serial Port is what +every headset publishes, so the name the firmware builds is what identifies an +NXT. These pin the shape that depends on, because it is a presentation string +upstream and nothing there knows this reads it. +""" + +from __future__ import annotations + +import pytest + +from pb_service.hub.protocol import HubType +from pb_service.hub.rfcomm.port import split_hub_name + + +@pytest.mark.parametrize( + ("full", "bare", "expected"), + [ + ("ev3 (MINDSTORMS EV3)", "ev3", HubType.EV3), + ("Pybricks (MINDSTORMS NXT)", "Pybricks", HubType.NXT), + # The whole name can be the suffix if the hub was never named. + ("(MINDSTORMS EV3)", None, HubType.EV3), + ], +) +def test_the_firmware_suffix_names_the_brick( + full: str, bare: str | None, expected: HubType +) -> None: + assert split_hub_name(full) == (bare, expected) + + +@pytest.mark.parametrize( + "name", + [ + None, + "", + # The reason any of this exists: a headset offers Serial Port too. + "Bose NC 700 Head", + # A hub type this build has never heard of says nothing rather than + # guessing, the same way an unknown PnP ID does. + "something (Another Hub)", + # Brackets alone are not a suffix. + "a hub (with a parenthetical)", + ], +) +def test_anything_else_identifies_no_brick(name: str | None) -> None: + # The name comes back untouched: brackets this does not recognise belong + # to whoever named the device. + assert split_hub_name(name) == (name, None) + + +def test_the_last_suffix_wins() -> None: + # The user sets the prefix and the firmware appends to it, so somebody who + # names their NXT after an EV3 still gets an NXT. + assert split_hub_name("x (MINDSTORMS EV3) (MINDSTORMS NXT)") == ( + "x (MINDSTORMS EV3)", + HubType.NXT, + ) diff --git a/packages/pb-service/tests/test_rfcomm_pipe.py b/packages/pb-service/tests/test_rfcomm_pipe.py new file mode 100644 index 0000000..e4e6286 --- /dev/null +++ b/packages/pb-service/tests/test_rfcomm_pipe.py @@ -0,0 +1,109 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 The Pybricks Authors + +"""Turning an RFCOMM channel into the byte stream the link reads. + +Rejoining frames that arrive in pieces is aio-rfcomm's job, in receive_until. +What is left here is the shape the link expects, and every way the channel can +end arriving as ConnectionError, which is what the link treats as a hub gone. +""" + +from __future__ import annotations + +from typing import cast + +import aio_rfcomm +import pytest +from aio_rfcomm.errors import ( + BluetoothUnavailableError, + ChannelClosedError, + DelimiterNotFoundError, +) + +from pb_service.hub.rfcomm.port import MAX_FRAME, ChannelPipe +from pb_service.hub.stream.pipe import BytePipe + +pytestmark = pytest.mark.anyio + +DELIM = b"\x02" + + +@pytest.fixture(scope="session") +def anyio_backend() -> str: + return "asyncio" + + +class FakeChannel: + """Answers each read with the next result, or raises it.""" + + def __init__(self, *results: bytes | Exception) -> None: + self.results = list(results) + self.limits: list[int] = [] + self.sent: list[bytes] = [] + + async def receive_until(self, delimiter: bytes, *, max_bytes: int) -> bytes: + self.limits.append(max_bytes) + result = self.results.pop(0) + + if isinstance(result, Exception): + raise result + + return result + + async def send(self, data: bytes) -> None: + result = self.results.pop(0) if self.results else None + + if isinstance(result, Exception): + raise result + + self.sent.append(data) + + +def _pipe(channel: FakeChannel) -> ChannelPipe: + return ChannelPipe(cast("aio_rfcomm.RfcommChannel", channel)) + + +def test_it_is_the_stream_the_link_asks_for() -> None: + # Checked by assignment rather than isinstance: BytePipe is a plain + # Protocol, so this is pyright's job, and the test records the intent. + pipe: BytePipe = _pipe(FakeChannel()) + assert pipe is not None + + +async def test_the_delimiter_is_included_because_the_link_strips_it() -> None: + # The link does body[:-1], so handing back a frame without its delimiter + # would silently eat the last byte of every message. + pipe = _pipe(FakeChannel(b"abc")) + assert await pipe.readuntil(DELIM) == b"abc" + DELIM + + +async def test_a_frame_is_read_with_a_bound() -> None: + channel = FakeChannel(b"abc") + await _pipe(channel).readuntil(DELIM) + assert channel.limits == [MAX_FRAME] + + +@pytest.mark.parametrize( + "error", + [ + ChannelClosedError(), + BluetoothUnavailableError(aio_rfcomm.Availability.SWITCHED_OFF), + DelimiterNotFoundError("no delimiter"), + ], +) +async def test_a_read_that_cannot_finish_reports_a_disconnect(error: Exception) -> None: + pipe = _pipe(FakeChannel(error)) + with pytest.raises(ConnectionError): + await pipe.readuntil(DELIM) + + +async def test_writing_reaches_the_channel() -> None: + channel = FakeChannel() + await _pipe(channel).write(bytearray(b"hello")) + assert channel.sent == [b"hello"] + + +async def test_writing_to_a_closed_channel_reports_a_disconnect() -> None: + pipe = _pipe(FakeChannel(ChannelClosedError())) + with pytest.raises(ConnectionError): + await pipe.write(b"hello") diff --git a/packages/pb-service/tests/test_rfcomm_transport.py b/packages/pb-service/tests/test_rfcomm_transport.py new file mode 100644 index 0000000..d3e9a66 --- /dev/null +++ b/packages/pb-service/tests/test_rfcomm_transport.py @@ -0,0 +1,230 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 The Pybricks Authors + +"""Finding hubs over Bluetooth Classic, and failing to reach them. + +The failing half is the point. A paired hub that is switched off is listed +exactly like one that is on -- the listing is the paired set, not a scan -- so +connecting to something that is not there is routine here rather than an edge +case, and has to arrive as a refusal rather than as a crash. +""" + +from __future__ import annotations + +from collections.abc import Callable +from contextlib import aclosing + +import anyio +import pytest + +from pb_service.hub import protocol as p +from pb_service.hub.rfcomm.fake import FakePairedHub, radio, switch +from pb_service.hub.rfcomm.port import PYBRICKS_RFCOMM_SERVICE, SERIAL_PORT_SERVICE +from pb_service.hub.rfcomm.transport import POLL_SECONDS, RfcommHubTransport +from pb_service.hub.transport import DeviceInfo, ScanFilter +from pb_service.hub.usb.fake import FakeHubPort + +pytestmark = pytest.mark.anyio + +EV3 = "00:16:53:4A:75:15" +NXT = "00:16:53:0D:5E:AF" + + +@pytest.fixture(scope="session") +def anyio_backend() -> str: + return "asyncio" + + +async def found( + transport: RfcommHubTransport, scan_filter: ScanFilter | None = None +) -> list[DeviceInfo]: + """Everything one pass of discovery reports.""" + seen: list[DeviceInfo] = [] + + # One pass yields straight away and then waits out the poll, so a short + # window collects exactly that pass. + with anyio.move_on_after(0.1): + async with aclosing(transport.discover(scan_filter or ScanFilter())) as hubs: + async for hub in hubs: + seen.append(hub) + + return seen + + +async def test_a_paired_hub_is_reported_with_its_type() -> None: + transport = RfcommHubTransport(radio(FakePairedHub(address=EV3))) + + (hub,) = await found(transport) + + assert hub.id == f"rfcomm:{EV3}" + assert hub.address == EV3 + assert hub.transport == "rfcomm" + assert hub.hub_type == "EV3 brick" + + +async def test_the_type_suffix_is_not_part_of_the_name() -> None: + # The listing has a column for the type, and the name has to match what + # the hub answers with once connected, since a choice is stored by name. + transport = RfcommHubTransport( + radio(FakePairedHub(name="Bumblebee (MINDSTORMS EV3)", address=EV3)) + ) + + (hub,) = await found(transport) + + assert hub.name == "Bumblebee" + + +async def test_a_serial_device_that_is_not_a_hub_is_left_out() -> None: + # Every paired headset offers Serial Port, so the name is what separates a + # brick from the rest of somebody's Bluetooth devices. + transport = RfcommHubTransport( + radio( + FakePairedHub( + name="Bose NC 700 Head", + address="4C:87:5D:29:FA:5C", + service=SERIAL_PORT_SERVICE, + ) + ) + ) + + assert await found(transport) == [] + + +async def test_an_nxt_on_plain_serial_port_is_still_a_hub() -> None: + transport = RfcommHubTransport( + radio( + FakePairedHub( + name="Pybricks (MINDSTORMS NXT)", + address=NXT, + service=SERIAL_PORT_SERVICE, + ) + ) + ) + + (hub,) = await found(transport) + + assert hub.name == "Pybricks" + assert hub.hub_type == "NXT brick" + + +async def test_a_hub_answering_on_both_services_is_reported_once() -> None: + # A real EV3 does: its record declares Serial Port as well as the Pybricks + # class, so both listings return it. + both = [ + FakePairedHub(address=EV3, service=PYBRICKS_RFCOMM_SERVICE), + FakePairedHub(address=EV3, service=SERIAL_PORT_SERVICE), + ] + transport = RfcommHubTransport(radio(*both)) + + assert len(await found(transport)) == 1 + + +async def test_asking_for_another_transport_reports_nothing() -> None: + transport = RfcommHubTransport(radio(FakePairedHub(address=EV3))) + + assert await found(transport, ScanFilter(transport="usb")) == [] + + +async def test_an_excluded_hub_is_not_reported() -> None: + transport = RfcommHubTransport(radio(FakePairedHub(address=EV3))) + + excluded = ScanFilter(exclude=frozenset({f"rfcomm:{EV3}"})) + + assert await found(transport, excluded) == [] + + +async def test_a_name_filter_matches_the_trimmed_name() -> None: + transport = RfcommHubTransport( + radio(FakePairedHub(name="Bumblebee (MINDSTORMS EV3)", address=EV3)) + ) + + assert len(await found(transport, ScanFilter(name="Bumblebee"))) == 1 + assert await found(transport, ScanFilter(name="Bumblebee (MINDSTORMS EV3)")) == [] + + +async def test_bluetooth_that_cannot_be_used_is_reported_rather_than_raised() -> None: + # One transport's problem must not stop the others answering. + transport = RfcommHubTransport(radio(unavailable="Bluetooth is turned off")) + + assert await found(transport) == [] + assert [u.reason for u in transport.unavailable] == ["Bluetooth is turned off"] + + +async def test_bluetooth_switched_off_and_on_mid_scan_is_noticed_at_once() -> None: + # Well within the poll: a scan that ends just after the switch must + # already say why, and stop saying it, and list the hub again, as soon as + # that is untrue. + bluetooth = radio(FakePairedHub(address=EV3)) + transport = RfcommHubTransport(bluetooth) + seen: list[DeviceInfo] = [] + + async def scan() -> None: + # Pulling continuously, as the manager does: discovery only moves on + # while someone is listening. + async with aclosing(transport.discover(ScanFilter())) as hubs: + async for hub in hubs: + seen.append(hub) + + async def until(condition: Callable[[], object]) -> None: + with anyio.fail_after(POLL_SECONDS / 4): + # A poll rather than an event: what changes is the transport's own + # state, which has nothing to signal with. + while not condition(): # noqa: ASYNC110 + await anyio.sleep(0.005) + + async with anyio.create_task_group() as tg: + tg.start_soon(scan) + await until(lambda: seen) + + switch(bluetooth, "Bluetooth is turned off") + await until(lambda: transport.unavailable) + assert [u.reason for u in transport.unavailable] == ["Bluetooth is turned off"] + + seen.clear() + switch(bluetooth, None) + await until(lambda: seen) + assert transport.unavailable == () + + tg.cancel_scope.cancel() + + +async def test_a_hub_that_is_switched_off_refuses_rather_than_crashing() -> None: + # The whole reason this file exists. A paired hub is listed whether or not + # it is switched on, so this is the ordinary way connecting fails, and it + # has to arrive as ConnectionError like every other transport reports. + transport = RfcommHubTransport(radio(FakePairedHub(address=EV3, reachable=False))) + + (hub,) = await found(transport) + + with pytest.raises(ConnectionError): + async with transport.open(hub.id): + pass + + +async def test_a_hub_that_was_never_discovered_refuses() -> None: + transport = RfcommHubTransport(radio()) + + with pytest.raises(ConnectionError): + async with transport.open("rfcomm:00:00:00:00:00:00"): + pass + + +async def test_an_id_from_another_transport_refuses() -> None: + transport = RfcommHubTransport(radio()) + + with pytest.raises(ConnectionError): + async with transport.open("usb:12345"): + pass + + +async def test_an_open_hub_answers_the_protocol() -> None: + port = FakeHubPort(name="Bumblebee") + transport = RfcommHubTransport(radio(FakePairedHub(address=EV3, port=port))) + + (hub,) = await found(transport) + + async with transport.open(hub.id) as link: + with anyio.fail_after(5): + name = await link.read_characteristic(p.DEVICE_NAME_UUID) + + assert name == b"Bumblebee" diff --git a/uv.lock b/uv.lock index 72c0a61..e7c79b3 100644 --- a/uv.lock +++ b/uv.lock @@ -42,6 +42,27 @@ test = [ ] typecheck = [{ name = "pyright", specifier = ">=1.1.400" }] +[[package]] +name = "aio-rfcomm" +version = "1.0.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "dbus-fast", marker = "sys_platform == 'linux'" }, + { name = "rubicon-objc", marker = "sys_platform == 'darwin'" }, + { name = "typing-extensions" }, + { name = "winrt-windows-devices-bluetooth", marker = "sys_platform == 'win32'" }, + { name = "winrt-windows-devices-bluetooth-rfcomm", marker = "sys_platform == 'win32'" }, + { name = "winrt-windows-devices-enumeration", marker = "sys_platform == 'win32'" }, + { name = "winrt-windows-devices-radios", marker = "sys_platform == 'win32'" }, + { name = "winrt-windows-foundation", marker = "sys_platform == 'win32'" }, + { name = "winrt-windows-foundation-collections", marker = "sys_platform == 'win32'" }, + { name = "winrt-windows-storage-streams", marker = "sys_platform == 'win32'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/ff/f9/d0123d34731da8b544376ff29bef60842ef0c587223b240ff11fc276ad68/aio_rfcomm-1.0.0.tar.gz", hash = "sha256:3d4458519ecbe437967623b6ad72815294bbff7213a6d18152087cb98f233651", size = 108909, upload-time = "2026-10-18T04:38:40.416Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/6a/28/85ff0a652fa0609f894c9afbbe17f02a168e40e9e403ac574b3ab3bcd565/aio_rfcomm-1.0.0-py3-none-any.whl", hash = "sha256:88e6a12f3023cc6e46b74a306d0b501185f6eefc6faa53844a2d62acac5f7f95", size = 138379, upload-time = "2026-10-18T04:38:38.862Z" }, +] + [[package]] name = "anyio" version = "4.14.2" @@ -938,6 +959,7 @@ name = "pybricks-service" version = "0.1.0" source = { editable = "packages/pb-service" } dependencies = [ + { name = "aio-rfcomm" }, { name = "anyio" }, { name = "bleak" }, { name = "packaging" }, @@ -949,6 +971,7 @@ dependencies = [ [package.metadata] requires-dist = [ + { name = "aio-rfcomm", specifier = ">=1.0" }, { name = "anyio", specifier = ">=4.14" }, { name = "bleak", specifier = ">=1.1.0" }, { name = "packaging", specifier = ">=22" }, @@ -1238,6 +1261,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/2a/68/1fc93dd759605b5d00fc98b50200739e41ed32bd22d6ba35ca6c3932371b/rich_rst-2.1.0-py3-none-any.whl", hash = "sha256:7ecd1343ee12c879d0e7ae74c3eb6d263b023d2929c6d114212eb1fd91057255", size = 272987, upload-time = "2026-07-05T02:59:42.792Z" }, ] +[[package]] +name = "rubicon-objc" +version = "0.5.7" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/46/7a/b90db65d32583ecace5164420ec93a5321a0b90e7951f6b93aebd3e70eee/rubicon_objc-0.5.7.tar.gz", hash = "sha256:ed2a56284a18ceb6370276069e8aab685b5465bbb457a5f13e09e37d0618703d", size = 197088, upload-time = "2026-09-30T04:12:18.303Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/09/68/f2548a31901e437c49b676a395bb7bef9f879d4df496e0a18e17160816ad/rubicon_objc-0.5.7-py3-none-any.whl", hash = "sha256:3dc76bc29f300ee6b9a4f408748099f70f8472ca59fe70cca42dfa1f2f1bfe3f", size = 64666, upload-time = "2026-09-30T04:12:16.966Z" }, +] + [[package]] name = "ruff" version = "0.16.5" @@ -1594,6 +1626,29 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/78/08/51c53ac3c704cd92da5ed7e7b9b57159052f6e46744e4f7e447ed708aa22/winrt_windows_devices_bluetooth_genericattributeprofile-3.2.1-cp314-cp314-win_arm64.whl", hash = "sha256:5140682da2860f6a55eb6faf9e980724dc457c2e4b4b35a10e1cebd8fc97d892", size = 194836, upload-time = "2025-09-20T07:10:00.87Z" }, ] +[[package]] +name = "winrt-windows-devices-bluetooth-rfcomm" +version = "3.2.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "winrt-runtime" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/2a/b4/c182d73c9562f10fc345d1c5cf97ab0be5e2d6adbc83d52792e41ad730f8/winrt_windows_devices_bluetooth_rfcomm-3.2.1.tar.gz", hash = "sha256:98eb180b95539c0831cc9878984192063875bd806f8d054afc3fabf7945634ce", size = 9324, upload-time = "2025-06-06T14:41:23.667Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e2/37/e48883277129f45e2f445d202051a709b828e192163585e98a286cbf31bc/winrt_windows_devices_bluetooth_rfcomm-3.2.1-cp311-cp311-win32.whl", hash = "sha256:b49b5d36adf2584579d58f44049a552c078d6ede94ca80003845e9d21b42a0a1", size = 56339, upload-time = "2025-06-06T07:01:04.774Z" }, + { url = "https://files.pythonhosted.org/packages/21/a6/d68c821104f899d54b328b8cb56af54f8c0f3e03c07044c6032b82a919b0/winrt_windows_devices_bluetooth_rfcomm-3.2.1-cp311-cp311-win_amd64.whl", hash = "sha256:d846786c5f6af6e684161d40ec29fee4408422b87870725a9c2e844a1d97ff1b", size = 59860, upload-time = "2025-06-06T07:01:05.578Z" }, + { url = "https://files.pythonhosted.org/packages/35/97/a148ef1c8bf627d3d7fbcc3d4a262d80f7c932e1b87295dea08f4472e956/winrt_windows_devices_bluetooth_rfcomm-3.2.1-cp311-cp311-win_arm64.whl", hash = "sha256:5b7813c6e4fc213d2174d80843d15d096704050f3f03e81a83fe0adc6dd214bb", size = 54862, upload-time = "2025-06-06T07:01:06.367Z" }, + { url = "https://files.pythonhosted.org/packages/8c/8f/73ae88689f34c4ece5c995c2c276bc18ecaf83a0edf4dfc5524e5cfaaefe/winrt_windows_devices_bluetooth_rfcomm-3.2.1-cp312-cp312-win32.whl", hash = "sha256:daf332a1508687318139ef710a229f2e2e24dc34ff5e6c97bb57aa1134bb5b55", size = 56409, upload-time = "2025-06-06T07:01:07.488Z" }, + { url = "https://files.pythonhosted.org/packages/39/1d/9be78401783d89165642b4bd33ddfaf4a8dff76e08fc863984be5f55cce4/winrt_windows_devices_bluetooth_rfcomm-3.2.1-cp312-cp312-win_amd64.whl", hash = "sha256:2193818b705c64312b650b81c675a476feb9f09a6450f5b3e1fe47725c4ac39d", size = 60073, upload-time = "2025-06-06T07:01:08.666Z" }, + { url = "https://files.pythonhosted.org/packages/2e/92/6e336e5806fe407903c91ef8f6ec94d3bd8fb0462fa7d5c45e842bb643e8/winrt_windows_devices_bluetooth_rfcomm-3.2.1-cp312-cp312-win_arm64.whl", hash = "sha256:2eff3b3f7b09957cd83134bdd39d33b344aa98cf15269767e5d6a9a9eb6ceb58", size = 54838, upload-time = "2025-06-06T07:01:09.447Z" }, + { url = "https://files.pythonhosted.org/packages/a0/95/99c18c537f9abe54318122e0729067b23c8a6067016b7f8985bb8aaaf1b3/winrt_windows_devices_bluetooth_rfcomm-3.2.1-cp313-cp313-win32.whl", hash = "sha256:c751fcd5bc4dfa61bc7c8c52c143bd148c56122a34deb7c79d5065696ded1cc4", size = 56419, upload-time = "2025-06-06T07:01:10.974Z" }, + { url = "https://files.pythonhosted.org/packages/71/0c/c821f538857a90cc04ba759e4425c22488c93facbeab0d84713dad58a754/winrt_windows_devices_bluetooth_rfcomm-3.2.1-cp313-cp313-win_amd64.whl", hash = "sha256:a800ae9fbea506142eb4284a5a4e390896402eab4eb5150608478571eaec80d3", size = 60102, upload-time = "2025-06-06T07:01:11.766Z" }, + { url = "https://files.pythonhosted.org/packages/46/15/ae05619969114256f83ab33374dd1c1e53a2b0b459b55e3358cfeca0f57c/winrt_windows_devices_bluetooth_rfcomm-3.2.1-cp313-cp313-win_arm64.whl", hash = "sha256:0ace6a12b6f051bb3a09c29956fd4ba58744ed5cd8584b12b1ef6eaab311f456", size = 54851, upload-time = "2025-06-06T07:01:12.544Z" }, + { url = "https://files.pythonhosted.org/packages/b2/00/d0931934b5b98ecfb68c2e455fe2318f3912c52910a8e8f674b2c0b6e98f/winrt_windows_devices_bluetooth_rfcomm-3.2.1-cp314-cp314-win32.whl", hash = "sha256:ae79b8bb7178e07f5edf4bae66cfce763f8dbe31bd5c1207d40fb61a7a99e206", size = 57861, upload-time = "2025-09-20T07:10:01.711Z" }, + { url = "https://files.pythonhosted.org/packages/b5/5d/d1148f216f5481467177b7f9fe9596a21b514e7c640025862ac587eaf9d0/winrt_windows_devices_bluetooth_rfcomm-3.2.1-cp314-cp314-win_amd64.whl", hash = "sha256:356b9a3e79da980ccdcf4858103bb922664a985e1bfe552d4317cc6e855b5bc9", size = 61547, upload-time = "2025-09-20T07:10:02.478Z" }, + { url = "https://files.pythonhosted.org/packages/fa/8e/a90c322f6380f3527666ba0f0337000cd19acc0aa6b69f52b6ff464a3660/winrt_windows_devices_bluetooth_rfcomm-3.2.1-cp314-cp314-win_arm64.whl", hash = "sha256:3eb710f6f0308baf4aa3572963bf136139c8249f1f52adc9507eb0b661435fe8", size = 56593, upload-time = "2025-09-20T07:10:03.258Z" }, +] + [[package]] name = "winrt-windows-devices-enumeration" version = "3.2.1" From aa2c192c56d769a81367fee157665fa721af1d75 Mon Sep 17 00:00:00 2001 From: David Lechner Date: Sat, 17 Oct 2026 23:52:59 -0500 Subject: [PATCH 3/7] pb-service: RFCOMM hubs are found alongside the others The service now opens Bluetooth Classic itself and hands it to the RFCOMM transport, next to USB and BLE. aio-rfcomm allows it open once per process, and everything opened through it stops when its block ends, so it is opened around the hub manager and outlives every hub. PB_HUB_BACKEND=rfcomm uses it alone, and is then the one case where failing to open it is an error; with every transport in use, RFCOMM is left out with a warning and the rest work. On macOS the service asks for Bluetooth under the name "Pybricks". A service started from a command line has no application to speak for it, so without a name it is refused rather than asked, and macOS remembers the answer under the name, so it must not change. RFCOMM ranks after BLE when a hub is reachable more than one way, because its listing includes hubs that are switched off, where hearing a hub advertise says it is there. In practice they rarely compete: a hub speaks one radio or the other. Connecting to a hub that cannot be reached now fails as HUB_NOT_CONNECTED with the transport's reason in the log, rather than as an internal error. It was reachable on USB by unplugging between the scan and the connect, and on RFCOMM it is the ordinary result for a paired hub that is switched off. --- packages/pb-service/src/pb_service/backend.py | 88 +++++++++++++++++-- .../src/pb_service/hub/transport.py | 7 +- .../pb-service/src/pb_service/methods/hub.py | 9 ++ packages/pb-service/src/pb_service/service.py | 10 ++- .../tests/test_transport_preference.py | 8 +- 5 files changed, 108 insertions(+), 14 deletions(-) diff --git a/packages/pb-service/src/pb_service/backend.py b/packages/pb-service/src/pb_service/backend.py index e0e7a4c..c913036 100644 --- a/packages/pb-service/src/pb_service/backend.py +++ b/packages/pb-service/src/pb_service/backend.py @@ -11,19 +11,25 @@ import logging import os +from collections.abc import AsyncGenerator +from contextlib import AsyncExitStack, asynccontextmanager +from typing import TYPE_CHECKING from pb_service.hub.transport import HubTransport +if TYPE_CHECKING: + from aio_rfcomm import Bluetooth + logger = logging.getLogger(__name__) -__all__ = ["HUB_BACKEND_ENV_VAR", "get_hub_transport"] +__all__ = ["HUB_BACKEND_ENV_VAR", "get_hub_transport", "open_hub_transport"] HUB_BACKEND_ENV_VAR = "PB_HUB_BACKEND" """Selects the transport. -Left alone, the service uses USB and Bluetooth together, which is what lets a -hub be found whichever way it happens to be reachable. Set it to ``usb`` or -``ble`` to use only one, or to ``fake`` for a simulated hub. +Left alone, the service uses every transport together, which is what lets a hub +be found whichever way it happens to be reachable. Set it to ``usb``, ``ble`` or +``rfcomm`` to use only one, or to ``fake`` for a simulated hub. """ @@ -39,7 +45,20 @@ def _ble() -> HubTransport: return BleHubTransport() -def get_hub_transport() -> HubTransport: +def _rfcomm(bluetooth: Bluetooth | None) -> tuple[HubTransport, ...]: + if bluetooth is None: + return () + + from pb_service.hub.rfcomm import RfcommHubTransport + + return (RfcommHubTransport(bluetooth),) + + +def _backend() -> str: + return os.environ.get(HUB_BACKEND_ENV_VAR, "").lower() + + +def get_hub_transport(bluetooth: Bluetooth | None = None) -> HubTransport: """Return the transport this service should use. Every import here is inside the branch that needs it, so a service running @@ -55,10 +74,14 @@ def get_hub_transport() -> HubTransport: in exactly one place rather than once per backend. The simulated backend is what lets the whole service be exercised on a CI runner with no radio and nothing plugged in. + + Args: + bluetooth: Bluetooth Classic, already open, or None to leave RFCOMM + out. :func:`open_hub_transport` opens it when it is wanted. """ from pb_service.hub.combined import CombinedTransport - backend = os.environ.get(HUB_BACKEND_ENV_VAR, "").lower() + backend = _backend() if backend == "fake": from pb_service.hub.fake import FakeHub, FakeHubTransport @@ -77,6 +100,9 @@ def get_hub_transport() -> HubTransport: if backend == "ble": return CombinedTransport(_ble()) + if backend == "rfcomm": + return CombinedTransport(*_rfcomm(bluetooth)) + if backend: logger.warning( "unknown %s=%r; using every transport", HUB_BACKEND_ENV_VAR, backend @@ -84,4 +110,52 @@ def get_hub_transport() -> HubTransport: # In TRANSPORT_PREFERENCE order, which is what makes USB the one chosen when # a hub is reachable both ways. A test holds the two together. - return CombinedTransport(_usb(), _ble()) + return CombinedTransport(_usb(), _ble(), *_rfcomm(bluetooth)) + + +@asynccontextmanager +async def open_hub_transport() -> AsyncGenerator[HubTransport, None]: + """Open the transport this service should use, for as long as the block lasts. + + Bluetooth Classic is opened here when RFCOMM is in use, and nowhere else: + aio-rfcomm allows it open once per process, and everything opened through + it stops working when its block ends, so it has to outlive every hub. + + Yields: + What :func:`get_hub_transport` returns. + + Raises: + RfcommError: ``PB_HUB_BACKEND`` asks for RFCOMM alone, and Bluetooth + could not be opened. With every transport in use, RFCOMM is left + out instead, so the others still work. + """ + backend = _backend() + + if backend in ("fake", "usb", "ble"): + yield get_hub_transport() + return + + import aio_rfcomm + import aio_rfcomm.macos + from aio_rfcomm.errors import RfcommError + + # macOS asks the responsible process for Bluetooth permission, and a + # service started from a command line has no application to speak for it, + # so without this it is refused rather than asked. The answer is + # remembered under the name, so changing it asks the user again. + aio_rfcomm.macos.prompt_under_name( + name="Pybricks", reason="to connect to LEGO hubs over Bluetooth" + ) + + async with AsyncExitStack() as stack: + bluetooth: Bluetooth | None = None + + try: + bluetooth = await stack.enter_async_context(aio_rfcomm.open_bluetooth()) + except RfcommError as exc: + if backend == "rfcomm": + raise + + logger.warning("RFCOMM hubs will not be found: %s", exc) + + yield get_hub_transport(bluetooth) diff --git a/packages/pb-service/src/pb_service/hub/transport.py b/packages/pb-service/src/pb_service/hub/transport.py index 79a9a98..99c81b6 100644 --- a/packages/pb-service/src/pb_service/hub/transport.py +++ b/packages/pb-service/src/pb_service/hub/transport.py @@ -144,9 +144,14 @@ class UnavailableDevice: """ -TRANSPORT_PREFERENCE: Final = ("usb", "ble", "fake") +TRANSPORT_PREFERENCE: Final = ("usb", "ble", "rfcomm", "fake") """How transports rank when a hub is reachable more than one way, best first. +RFCOMM comes after BLE: its listing is the paired set, which includes hubs +that are switched off, where hearing a hub advertise says it is actually there. +The two rarely compete anyway, since a hub speaks one radio protocol or the +other. + USB comes first on three counts, each measured rather than argued: enumerating serial ports answers in milliseconds where a Bluetooth scan has to wait to hear something; a bulk endpoint has no connection interval to wait for; and a hub diff --git a/packages/pb-service/src/pb_service/methods/hub.py b/packages/pb-service/src/pb_service/methods/hub.py index da030ef..fdb35fb 100644 --- a/packages/pb-service/src/pb_service/methods/hub.py +++ b/packages/pb-service/src/pb_service/methods/hub.py @@ -375,6 +375,15 @@ async def hub_connect(session: Session, params: dict[str, Any]) -> dict[str, Any raise JsonRpcError( ErrorCode.HUB_NOT_CONNECTED, f"timed out connecting to {device.name}" ) from None + except ConnectionError as exc: + # How every transport reports a hub it cannot reach. Routine on + # RFCOMM, where the listing is the paired set and so names hubs that + # are switched off, and reachable on USB too by unplugging between + # the scan and the connect. + logger.info("could not connect to %s: %s", device.name, exc) + raise JsonRpcError( + ErrorCode.HUB_NOT_CONNECTED, f"could not connect to {device.name}" + ) from exc except (HubNotFoundError, HubAlreadyConnectedError, HubDisconnectedError) as exc: raise as_wire_error(exc) from exc diff --git a/packages/pb-service/src/pb_service/service.py b/packages/pb-service/src/pb_service/service.py index 6f80cd0..7859b57 100644 --- a/packages/pb-service/src/pb_service/service.py +++ b/packages/pb-service/src/pb_service/service.py @@ -19,7 +19,7 @@ import logging import time from collections.abc import AsyncGenerator -from contextlib import asynccontextmanager +from contextlib import AbstractAsyncContextManager, asynccontextmanager, nullcontext from typing import TYPE_CHECKING, Final import anyio @@ -187,16 +187,20 @@ async def run_service( Raises: AddressInUseError: If another service already holds the endpoint. """ - from pb_service.backend import get_hub_transport + from pb_service.backend import open_hub_transport from pb_service.session import serve_session token = auth.read_or_create_token() + opened: AbstractAsyncContextManager[HubTransport] = ( + open_hub_transport() if transport is None else nullcontext(transport) + ) + async with anyio.create_task_group() as root: # Entered before the listener and left after it. Hub connections # therefore live in a sibling scope of the client sessions, so a client # going away cancels only its own tasks and never a hub. - async with run_hub_manager(transport or get_hub_transport()) as hubs: + async with opened as hub_transport, run_hub_manager(hub_transport) as hubs: service = Service(endpoint, token, idle_timeout, hubs) root.start_soon(service._watchdog) # pyright: ignore[reportPrivateUsage] diff --git a/packages/pb-service/tests/test_transport_preference.py b/packages/pb-service/tests/test_transport_preference.py index 421975f..0dbf412 100644 --- a/packages/pb-service/tests/test_transport_preference.py +++ b/packages/pb-service/tests/test_transport_preference.py @@ -18,7 +18,7 @@ def test_usb_is_preferred_over_bluetooth() -> None: def test_a_transport_this_build_does_not_know_loses_every_tie() -> None: # It can still be used; it just is not chosen over something known. - assert preference_rank("rfcomm") > preference_rank("ble") + assert preference_rank("unknown") > preference_rank("ble") assert preference_rank("") > preference_rank("fake") @@ -34,12 +34,14 @@ def test_the_service_builds_its_transports_in_preference_order( # decision, and nothing else would notice them drifting apart: connecting # would simply start choosing the wrong route. from pb_service.backend import HUB_BACKEND_ENV_VAR, get_hub_transport + from pb_service.hub.rfcomm.fake import radio # The real transports, not whatever the suite is otherwise pointed at. - # Building them touches no hardware. + # Building them touches no hardware, and RFCOMM's needs Bluetooth open, so + # it gets a simulated one. monkeypatch.delenv(HUB_BACKEND_ENV_VAR, raising=False) - schemes = get_hub_transport().scheme.split("+") + schemes = get_hub_transport(radio()).scheme.split("+") assert schemes == sorted(schemes, key=preference_rank) assert schemes[0] == "usb", "USB is meant to be the preferred route" From 21f834874482edd6098107ee9290e24369c1cc79 Mon Sep 17 00:00:00 2001 From: David Lechner Date: Sat, 17 Oct 2026 23:53:11 -0500 Subject: [PATCH 4/7] pb-cli: rfcomm is a transport to ask for, and to explain pb scan and pb connect pass --transport through to the service unchecked, so --transport rfcomm worked as soon as the service knew it. What the CLI says around it had not caught up: - The help named two transports, and now names three. - A hub found over Bluetooth Classic had an empty signal meter, which reads as a hub at the edge of range. Nothing was heard at all: the listing is the paired set, and the hub may be switched off. Its row now says "paired". - pb status showed the service's own word, "rfcomm". It is now "Bluetooth Classic", Bluetooth's name for the older radio, which keeps it apart from the BLE hubs shown as "Bluetooth". - Finding nothing over RFCOMM means nothing is paired, so pb scan and pb connect --transport rfcomm now ask about pairing rather than suggesting the hub be switched on or its button pressed. --- packages/pb-cli/src/pb_cli/commands/hub.py | 25 +++++++++++++------ packages/pb-cli/src/pb_cli/ui.py | 16 +++++++++---- packages/pb-cli/tests/test_ui.py | 28 +++++++++++++++++++--- 3 files changed, 55 insertions(+), 14 deletions(-) diff --git a/packages/pb-cli/src/pb_cli/commands/hub.py b/packages/pb-cli/src/pb_cli/commands/hub.py index a981f1e..5211b98 100644 --- a/packages/pb-cli/src/pb_cli/commands/hub.py +++ b/packages/pb-cli/src/pb_cli/commands/hub.py @@ -61,6 +61,10 @@ "hub that is already connected, or running a program, does not " "advertise." ), + "rfcomm": ( + "Pair the hub with this computer in its Bluetooth settings first. Only " + "the EV3 and NXT are reached this way." + ), } """What to suggest when a hub to connect to could not be found. @@ -117,7 +121,8 @@ async def scan_cmd( Args: timeout: How long to scan, in seconds. name: Only show hubs with this name. - transport: Only look on this transport, usb or ble. Both by default. + transport: Only look on this transport: usb, ble or rfcomm. All of + them by default. verbose: Increase log detail. """ configure_logging(verbose) @@ -157,10 +162,15 @@ async def _scan(seconds: float, name: str | None, transport: str | None) -> int: seen[device.id] = device if not seen: - # "In range" is Bluetooth's word for it and means nothing to somebody - # who asked about a cable. - where = "plugged in" if transport == "usb" else "in range" - err.print(f"[yellow]No hubs found.[/] Is the hub switched on and {where}?") + if transport == "rfcomm": + # The listing is the paired set, so an empty one means nothing is + # paired. Switching the hub on would not change it. + err.print("[yellow]No hubs found.[/] Is the hub paired with this computer?") + else: + # "In range" is Bluetooth's word for it and means nothing to + # somebody who asked about a cable. + where = "plugged in" if transport == "usb" else "in range" + err.print(f"[yellow]No hubs found.[/] Is the hub switched on and {where}?") return EXIT_OK @@ -184,8 +194,9 @@ async def connect_cmd( scan: Watch for hubs and choose one from the list. There is no time limit, so the hub can be switched on whenever you like, and Ctrl-C gives up without connecting to anything. - transport: Reach the hub this way, usb or ble. By default whichever - works, preferring USB where a hub is reachable both ways. + transport: Reach the hub this way: usb, ble or rfcomm. By default + whichever works, preferring USB where a hub is reachable more than + one way. verbose: Increase log detail. """ diff --git a/packages/pb-cli/src/pb_cli/ui.py b/packages/pb-cli/src/pb_cli/ui.py index e1e7eab..3206bf4 100644 --- a/packages/pb-cli/src/pb_cli/ui.py +++ b/packages/pb-cli/src/pb_cli/ui.py @@ -188,13 +188,16 @@ def usage(message: str, hint: str | None = None) -> int: _TRANSPORT_NAMES: Final[Mapping[str, str]] = { "ble": "Bluetooth", + "rfcomm": "Bluetooth Classic", "usb": "USB", "fake": "simulated", } """What to call each way of reaching a hub. -The service's words are ``ble`` and ``usb``, which are the right names on the -wire and the wrong ones on a screen: nobody calls it BLE except us. +The service's words are ``ble``, ``rfcomm`` and ``usb``, which are the right +names on the wire and the wrong ones on a screen: nobody calls it BLE or RFCOMM +except us. Classic is Bluetooth's own name for the older radio the EV3 and NXT +use, and it keeps the two apart where both are in use. """ @@ -316,11 +319,13 @@ def signal_cell(device: pb_client.DeviceInfo, connected: Collection[str] = ()) - connected: The ids of hubs the service is holding open. Returns: - A meter for a hub that was heard over Bluetooth, and a word for the two + A meter for a hub that was heard over Bluetooth, and a word for the cases where a meter would be a lie. A connected hub is not advertising, so there is no signal to draw; a hub on a cable has no signal at all, and drawing an empty meter for one would read as a hub that can barely - be heard when it is in fact the best-connected hub on the list. + be heard when it is in fact the best-connected hub on the list. A hub + listed over Bluetooth Classic was not heard at all: it is there because + it is paired, which says nothing about whether it is switched on. """ if device.id in connected: return "[green]connected[/]" @@ -328,6 +333,9 @@ def signal_cell(device: pb_client.DeviceInfo, connected: Collection[str] = ()) - if device.transport == "usb": return "[green]USB[/]" + if device.transport == "rfcomm": + return "[dim]paired[/]" + return signal_bars(device.rssi) diff --git a/packages/pb-cli/tests/test_ui.py b/packages/pb-cli/tests/test_ui.py index 16f27d6..941dcb5 100644 --- a/packages/pb-cli/tests/test_ui.py +++ b/packages/pb-cli/tests/test_ui.py @@ -114,6 +114,27 @@ def test_a_hub_that_does_not_say_what_it_is_still_gets_a_row() -> None: assert "Pybricks Hub" in console.export_text() +def test_a_paired_hub_is_not_drawn_as_a_faint_one() -> None: + # Bluetooth Classic lists the paired set, so nothing was heard. An empty + # meter would read as a hub at the edge of range rather than as one that + # may simply be switched off. + paired = pb_client.DeviceInfo( + id="rfcomm:00:16:53:4A:75:15", + name="Ostrich", + address="00:16:53:4A:75:15", + rssi=None, + transport="rfcomm", + hub_type="EV3 brick", + ) + assert ui.signal_cell(paired) == "[dim]paired[/]" + + console = _terminal() + console.print(ui.scan_table([paired])) + drawn = console.export_text() + assert "paired" in drawn + assert ui.SIGNAL_EMPTY not in drawn + + def test_the_chooser_marks_the_row_it_is_on() -> None: console = _terminal() console.print(ui.chooser_table([_device("First"), _device("Second")], 1)) @@ -199,6 +220,7 @@ def test_a_version_in_a_message_is_drawn_in_one_piece(err: bool) -> None: # "BLE" is our word, not a user's; nobody with a hub on a desk calls # the radio that. ("ble", "Bluetooth"), + ("rfcomm", "Bluetooth Classic"), ("usb", "USB"), ("fake", "simulated"), ], @@ -214,6 +236,6 @@ def test_a_transport_the_service_did_not_report_names_nothing() -> None: def test_an_unrecognised_transport_is_passed_through() -> None: - # RFCOMM is planned and not built. A name we do not have yet is still a - # better answer than none, and it cannot be wrong the way a guess could. - assert ui.transport_name("rfcomm") == "rfcomm" + # A newer service's transport. A name we do not have yet is still a better + # answer than none, and it cannot be wrong the way a guess could. + assert ui.transport_name("carrier-pigeon") == "carrier-pigeon" From 45565ad044d68805221fc0b345c7429be8de5a2a Mon Sep 17 00:00:00 2001 From: David Lechner Date: Sun, 18 Oct 2026 10:27:43 -0500 Subject: [PATCH 5/7] pb-service: Bluetooth that cannot be used is not a port that will not open With Bluetooth switched off or not allowed, RFCOMM cannot read the paired set, and it reported the adapter as an unavailable device. Finding nothing then raised SERIAL_PORT_UNAVAILABLE, and the CLI told the user to join dialout or unplug a hub. An unavailable device now says whether it is a port or the adapter. An adapter gives BLUETOOTH_UNAVAILABLE with the system's own words, and the CLI's advice covers Bluetooth Classic too. --- packages/pb-cli/src/pb_cli/errors.py | 8 ++-- packages/pb-cli/tests/test_errors.py | 15 ++++++++ .../src/pb_service/hub/rfcomm/transport.py | 2 +- .../src/pb_service/hub/transport.py | 9 ++++- .../pb-service/src/pb_service/methods/hub.py | 22 ++++++++--- .../pb-service/tests/test_rfcomm_transport.py | 4 +- tests/e2e/test_hub_lifecycle.py | 38 +++++++++++++++++++ 7 files changed, 86 insertions(+), 12 deletions(-) diff --git a/packages/pb-cli/src/pb_cli/errors.py b/packages/pb-cli/src/pb_cli/errors.py index 2a53c56..e455d6f 100644 --- a/packages/pb-cli/src/pb_cli/errors.py +++ b/packages/pb-cli/src/pb_cli/errors.py @@ -201,12 +201,12 @@ def serial_port_advice(platform: str = sys.platform) -> str: message="The hub's button interrupted this.", ), # Transport and lifecycle. + # The service repeats what the system said, which is what tells a radio + # that is switched off from one this program may not use. ErrorCode.BLUETOOTH_UNAVAILABLE: _Rule( - "Connect or enable a Bluetooth Low Energy adapter, and check that this " - "computer has permission to use it.", + "Switch Bluetooth on, or plug in an adapter, and check that this " + "computer lets pb use Bluetooth.", exit_code=EXIT_ERROR, - message="No Bluetooth adapter could be found, or permission to use it " - "has not been granted.", ), ErrorCode.SCAN_TIMEOUT: _Rule( "Switch the hub on and press its button so that its light flashes." diff --git a/packages/pb-cli/tests/test_errors.py b/packages/pb-cli/tests/test_errors.py index 245c048..c08b9a4 100644 --- a/packages/pb-cli/tests/test_errors.py +++ b/packages/pb-cli/tests/test_errors.py @@ -424,6 +424,21 @@ def test_a_refused_serial_port_keeps_the_port_in_the_message() -> None: assert problems[0].hint +def test_unusable_bluetooth_keeps_the_systems_words() -> None: + # They are what tells a radio switched off from one pb may not use, and + # the advice covers Bluetooth Classic as well as Low Energy. + exc = JsonRpcError( + ErrorCode.BLUETOOTH_UNAVAILABLE, + "Bluetooth cannot be used: Bluetooth is switched off", + ) + problems = errors.explain(exc) + + assert problems is not None + assert "switched off" in problems[0].message + assert problems[0].hint + assert "Low Energy" not in problems[0].hint + + @pytest.mark.parametrize( ("platform", "expected"), [ diff --git a/packages/pb-service/src/pb_service/hub/rfcomm/transport.py b/packages/pb-service/src/pb_service/hub/rfcomm/transport.py index d3f3e7b..ea2e1f3 100644 --- a/packages/pb-service/src/pb_service/hub/rfcomm/transport.py +++ b/packages/pb-service/src/pb_service/hub/rfcomm/transport.py @@ -100,7 +100,7 @@ def unavailable(self) -> tuple[UnavailableDevice, ...]: stands for the adapter. """ return tuple( - UnavailableDevice(address, reason) + UnavailableDevice(address, reason, kind="adapter") for address, reason in self._unavailable.items() ) diff --git a/packages/pb-service/src/pb_service/hub/transport.py b/packages/pb-service/src/pb_service/hub/transport.py index 99c81b6..5dd2241 100644 --- a/packages/pb-service/src/pb_service/hub/transport.py +++ b/packages/pb-service/src/pb_service/hub/transport.py @@ -18,7 +18,7 @@ from collections.abc import AsyncGenerator, Callable from contextlib import AbstractAsyncContextManager from dataclasses import dataclass, field -from typing import Final, Protocol, runtime_checkable +from typing import Final, Literal, Protocol, runtime_checkable from pb_service.hub import protocol as p @@ -143,6 +143,13 @@ class UnavailableDevice: problems this is, and the client turns the code into the advice. """ + kind: Literal["port", "adapter"] = "port" + """What was turned away: a hub's own port, or the Bluetooth adapter. + + They need opposite advice. A refused port is a hub that is there but out of + reach, while an adapter that cannot be used hides every hub behind it. + """ + TRANSPORT_PREFERENCE: Final = ("usb", "ble", "rfcomm", "fake") """How transports rank when a hub is reachable more than one way, best first. diff --git a/packages/pb-service/src/pb_service/methods/hub.py b/packages/pb-service/src/pb_service/methods/hub.py index fdb35fb..bfa90b2 100644 --- a/packages/pb-service/src/pb_service/methods/hub.py +++ b/packages/pb-service/src/pb_service/methods/hub.py @@ -57,7 +57,8 @@ def _found_nothing(manager: HubManager) -> JsonRpcError: A hub that is plugged in but whose port will not open is the opposite problem from no hub at all -- the user needs to fix their computer, not go and find their hub -- so it gets a code of its own rather than the wording - being bent to cover both. + being bent to cover both. So does a Bluetooth adapter that cannot be used, + which hides every hub behind it. Args: manager: Whose transports did the looking. @@ -68,14 +69,24 @@ def _found_nothing(manager: HubManager) -> JsonRpcError: neither. """ unavailable = manager.transport.unavailable + ports = [d for d in unavailable if d.kind == "port"] + adapters = [d for d in unavailable if d.kind == "adapter"] - if unavailable: - refused = ", ".join(f"{d.address} ({d.reason})" for d in unavailable) + # A refused port first: it is a hub actually there, while an adapter that + # cannot be used only means hubs behind it could not be looked for. + if ports: + refused = ", ".join(f"{d.address} ({d.reason})" for d in ports) return JsonRpcError( ErrorCode.SERIAL_PORT_UNAVAILABLE, f"found a hub but could not open its port: {refused}", ) + if adapters: + reasons = ", ".join(d.reason for d in adapters) + return JsonRpcError( + ErrorCode.BLUETOOTH_UNAVAILABLE, f"Bluetooth cannot be used: {reasons}" + ) + return JsonRpcError(ErrorCode.HUB_NOT_FOUND, "no hub found") @@ -289,8 +300,9 @@ async def hub_scan(session: Session, params: dict[str, Any]) -> dict[str, Any]: session.peer.notify(HUB_FOUND, _device_json(device)) # A scan that finds nothing is an ordinary answer and not an error, so it - # returns an empty list -- unless a hub was there and would not open, which - # scanning again will not fix and which the user can do something about. + # returns an empty list -- unless a hub was there and would not open, or + # Bluetooth could not be used, which scanning again will not fix and which + # the user can do something about. if not found and manager.transport.unavailable: raise _found_nothing(manager) diff --git a/packages/pb-service/tests/test_rfcomm_transport.py b/packages/pb-service/tests/test_rfcomm_transport.py index d3e9a66..64d64c1 100644 --- a/packages/pb-service/tests/test_rfcomm_transport.py +++ b/packages/pb-service/tests/test_rfcomm_transport.py @@ -147,7 +147,9 @@ async def test_bluetooth_that_cannot_be_used_is_reported_rather_than_raised() -> transport = RfcommHubTransport(radio(unavailable="Bluetooth is turned off")) assert await found(transport) == [] - assert [u.reason for u in transport.unavailable] == ["Bluetooth is turned off"] + assert [(u.reason, u.kind) for u in transport.unavailable] == [ + ("Bluetooth is turned off", "adapter") + ] async def test_bluetooth_switched_off_and_on_mid_scan_is_noticed_at_once() -> None: diff --git a/tests/e2e/test_hub_lifecycle.py b/tests/e2e/test_hub_lifecycle.py index 445d83b..5925ace 100644 --- a/tests/e2e/test_hub_lifecycle.py +++ b/tests/e2e/test_hub_lifecycle.py @@ -164,6 +164,44 @@ async def test_a_scan_that_reports_nothing_still_ends(service_at: Endpoint) -> N assert [device async for device in devices] == [] +@pytest.fixture +async def bluetooth_off_at(endpoint: Endpoint) -> AsyncGenerator[Endpoint, None]: + """A service whose Bluetooth adapter cannot be used. + + What RFCOMM reports with the radio switched off or not allowed: it cannot + read the paired set at all, so it names the adapter rather than a hub. + + Args: + endpoint: Where the service should listen. + + Yields: + That endpoint, with the service running. + """ + transport = FakeHubTransport() + transport.hubs = [] + transport.unavailable = ( + UnavailableDevice("bluetooth", "Bluetooth is switched off", kind="adapter"), + ) + + async with service_module.run_service( + endpoint, idle_timeout=0, transport=transport + ): + yield endpoint + + +async def test_bluetooth_that_cannot_be_used_is_not_reported_as_a_port( + bluetooth_off_at: Endpoint, +) -> None: + # Told a port would not open, somebody on Linux joins dialout and somebody + # on a Mac unplugs a hub, and neither helps. The radio is what to fix. + async with pb_client.connect(bluetooth_off_at, autostart=False) as service: + with pytest.raises(JsonRpcError) as caught: + await service.scan(seconds=0.3) + + assert caught.value.code == ErrorCode.BLUETOOTH_UNAVAILABLE + assert "switched off" in str(caught.value), "the system's words are lost" + + @pytest.fixture async def out_of_reach_at(endpoint: Endpoint) -> AsyncGenerator[Endpoint, None]: """A service whose only hub is behind a port it may not open. From fd243fb1ab92e2f6e7b7af60d1e5828380add896 Mon Sep 17 00:00:00 2001 From: David Lechner Date: Sun, 18 Oct 2026 10:27:43 -0500 Subject: [PATCH 6/7] pb-service: a hub gone mid-handshake is unreachable, not an internal error Both stream transports read the link in a task group beside the caller's block, and a group wraps whatever leaves it. A channel or cable that dropped before the handshake finished came out as an ExceptionGroup, which connecting did not catch and reported as INTERNAL_ERROR rather than HUB_NOT_CONNECTED. StreamLink.running now owns that task group, and a failure in it comes out as itself. --- .../src/pb_service/hub/rfcomm/transport.py | 9 +--- .../src/pb_service/hub/stream/link.py | 42 ++++++++++++++++++- .../src/pb_service/hub/usb/transport.py | 6 +-- .../pb-service/tests/test_rfcomm_transport.py | 17 ++++++++ packages/pb-service/tests/test_usb_link.py | 17 ++++++++ 5 files changed, 78 insertions(+), 13 deletions(-) diff --git a/packages/pb-service/src/pb_service/hub/rfcomm/transport.py b/packages/pb-service/src/pb_service/hub/rfcomm/transport.py index ea2e1f3..c6f58f9 100644 --- a/packages/pb-service/src/pb_service/hub/rfcomm/transport.py +++ b/packages/pb-service/src/pb_service/hub/rfcomm/transport.py @@ -214,13 +214,8 @@ async def open(self, device_id: str) -> AsyncGenerator[HubLink, None]: link = StreamLink(pipe, device_id) - async with anyio.create_task_group() as tg: - tg.start_soon(link.run, self._on_lost) - - try: - yield link - finally: - tg.cancel_scope.cancel() + async with link.running(self._on_lost): + yield link def _on_lost(self, device_id: str) -> None: """Report a hub that went away by itself.""" diff --git a/packages/pb-service/src/pb_service/hub/stream/link.py b/packages/pb-service/src/pb_service/hub/stream/link.py index 54ebf49..fff57a7 100644 --- a/packages/pb-service/src/pb_service/hub/stream/link.py +++ b/packages/pb-service/src/pb_service/hub/stream/link.py @@ -13,7 +13,8 @@ import asyncio import logging -from collections.abc import Callable +from collections.abc import AsyncGenerator, Callable +from contextlib import asynccontextmanager from typing import Final, final import anyio @@ -231,6 +232,45 @@ async def run(self, on_lost: Callable[[str], None] | None) -> None: self._replies_in.close() self._replies_out.close() + @asynccontextmanager + async def running( + self, on_lost: Callable[[str], None] | None + ) -> AsyncGenerator[StreamLink, None]: + """Read the link in the background for as long as the block lasts. + + Args: + on_lost: As for :meth:`run`. + + Yields: + This link. + + Raises: + ConnectionError: The hub went away while the block was using it, + such as part way through the handshake. + """ + try: + async with anyio.create_task_group() as tg: + tg.start_soon(self.run, on_lost) + + try: + yield self + finally: + tg.cancel_scope.cancel() + except ExceptionGroup as group: + # The reader is this link's own business, so what fails while the + # block uses the link comes out as itself, as from any other open. + # A hub that drops mid-handshake is a ConnectionError, which the + # caller handles, not a group it would report as an internal error. + failures = group.exceptions + alone = len(failures) == 1 and not isinstance( + failures[0], BaseExceptionGroup + ) + + if alone or all(isinstance(e, ConnectionError) for e in failures): + raise failures[0] from None + + raise + def _rejection(status: int) -> HubCommandError: """Turn a firmware status code into the hub's own refusal. diff --git a/packages/pb-service/src/pb_service/hub/usb/transport.py b/packages/pb-service/src/pb_service/hub/usb/transport.py index d8a6822..86bf2db 100644 --- a/packages/pb-service/src/pb_service/hub/usb/transport.py +++ b/packages/pb-service/src/pb_service/hub/usb/transport.py @@ -274,9 +274,7 @@ async def open(self, device_id: str) -> AsyncGenerator[HubLink, None]: async with self._open_port(port.device) as serial: link = StreamLink(serial, device_id) - async with anyio.create_task_group() as tg: - tg.start_soon(link.run, self._on_lost) - + async with link.running(self._on_lost): try: yield link finally: @@ -291,8 +289,6 @@ async def open(self, device_id: str) -> AsyncGenerator[HubLink, None]: ): await serial.write(wire.encode_subscribe(subscribed=False)) - tg.cancel_scope.cancel() - def _on_lost(self, device_id: str) -> None: """Report a hub that went away by itself.""" self._named.pop(device_id, None) diff --git a/packages/pb-service/tests/test_rfcomm_transport.py b/packages/pb-service/tests/test_rfcomm_transport.py index 64d64c1..9638479 100644 --- a/packages/pb-service/tests/test_rfcomm_transport.py +++ b/packages/pb-service/tests/test_rfcomm_transport.py @@ -219,6 +219,23 @@ async def test_an_id_from_another_transport_refuses() -> None: pass +async def test_a_hub_gone_part_way_through_a_handshake_refuses_plainly() -> None: + # Review of the port found this: the channel dropping before the handshake + # finished came out of the link's task group as an ExceptionGroup, which + # connecting reported as an internal error rather than an unreachable hub. + port = FakeHubPort(name="Bumblebee") + transport = RfcommHubTransport(radio(FakePairedHub(address=EV3, port=port))) + + (hub,) = await found(transport) + + with pytest.raises(ConnectionError): + async with transport.open(hub.id) as link: + port.vanish() + + with anyio.fail_after(5): + await link.read_characteristic(p.DEVICE_NAME_UUID) + + async def test_an_open_hub_answers_the_protocol() -> None: port = FakeHubPort(name="Bumblebee") transport = RfcommHubTransport(radio(FakePairedHub(address=EV3, port=port))) diff --git a/packages/pb-service/tests/test_usb_link.py b/packages/pb-service/tests/test_usb_link.py index 49ce826..e2465e4 100644 --- a/packages/pb-service/tests/test_usb_link.py +++ b/packages/pb-service/tests/test_usb_link.py @@ -205,6 +205,23 @@ def note(device_id: str) -> None: assert lost == [f"usb:{hub.info.serial_number}"] +async def test_a_hub_gone_part_way_through_a_handshake_refuses_plainly( + hub: FakeHubPort, +) -> None: + # The reader runs in a task group, which wraps whatever leaves the block. + # A hub gone before the handshake finishes has to come out as the + # ConnectionError every caller handles, not as a group, which connecting + # used to report as an internal error. + transport = transport_for(hub) + + with pytest.raises(ConnectionError): + async with transport.open(f"usb:{hub.info.serial_number}") as link: + hub.vanish() + + with anyio.fail_after(5): + await link.read_characteristic(p.DEVICE_NAME_UUID) + + async def test_a_link_closed_on_request_is_not_reported_as_a_hub_going_away( hub: FakeHubPort, ) -> None: From 7ca06f87fe16051efdd63a4864dd75a1c9b87d83 Mon Sep 17 00:00:00 2001 From: David Lechner Date: Sun, 18 Oct 2026 10:38:52 -0500 Subject: [PATCH 7/7] pb-service: open Bluetooth Classic when RFCOMM first needs it The service opened Bluetooth before it started listening, and opening it does not return until the system says whether Bluetooth is usable. On CI's macOS runners that never happened, so the service never started and every end-to-end test that runs one timed out, USB and BLE included. A Mac whose user has not yet answered the permission prompt would do the same. It is now opened the first time a scan or a connection over RFCOMM asks, in a task of the transport's own scope, as bleak does for BLE. A service used only over USB never starts aio-rfcomm's helper. If opening fails, RFCOMM reports Bluetooth as unusable and the other transports carry on. --- packages/pb-service/src/pb_service/backend.py | 41 +++-- .../src/pb_service/hub/rfcomm/lazy.py | 158 ++++++++++++++++++ packages/pb-service/tests/test_rfcomm_lazy.py | 139 +++++++++++++++ tests/e2e/test_service_roundtrip.py | 25 +++ 4 files changed, 341 insertions(+), 22 deletions(-) create mode 100644 packages/pb-service/src/pb_service/hub/rfcomm/lazy.py create mode 100644 packages/pb-service/tests/test_rfcomm_lazy.py diff --git a/packages/pb-service/src/pb_service/backend.py b/packages/pb-service/src/pb_service/backend.py index c913036..6a7dc39 100644 --- a/packages/pb-service/src/pb_service/backend.py +++ b/packages/pb-service/src/pb_service/backend.py @@ -12,8 +12,10 @@ import logging import os from collections.abc import AsyncGenerator -from contextlib import AsyncExitStack, asynccontextmanager -from typing import TYPE_CHECKING +from contextlib import asynccontextmanager +from typing import TYPE_CHECKING, cast + +import anyio from pb_service.hub.transport import HubTransport @@ -117,17 +119,16 @@ def get_hub_transport(bluetooth: Bluetooth | None = None) -> HubTransport: async def open_hub_transport() -> AsyncGenerator[HubTransport, None]: """Open the transport this service should use, for as long as the block lasts. - Bluetooth Classic is opened here when RFCOMM is in use, and nowhere else: - aio-rfcomm allows it open once per process, and everything opened through - it stops working when its block ends, so it has to outlive every hub. + Bluetooth Classic is held open in this scope when RFCOMM is in use, and + nowhere else: aio-rfcomm allows it open once per process, and everything + opened through it stops working when its block ends, so it has to outlive + every hub. It is opened the first time RFCOMM needs it, not here, so the + service never waits on it to start (see + :mod:`pb_service.hub.rfcomm.lazy`). If it cannot be opened, RFCOMM + reports Bluetooth as unusable and the other transports carry on. Yields: What :func:`get_hub_transport` returns. - - Raises: - RfcommError: ``PB_HUB_BACKEND`` asks for RFCOMM alone, and Bluetooth - could not be opened. With every transport in use, RFCOMM is left - out instead, so the others still work. """ backend = _backend() @@ -135,9 +136,9 @@ async def open_hub_transport() -> AsyncGenerator[HubTransport, None]: yield get_hub_transport() return - import aio_rfcomm import aio_rfcomm.macos - from aio_rfcomm.errors import RfcommError + + from pb_service.hub.rfcomm.lazy import LazyBluetooth # macOS asks the responsible process for Bluetooth permission, and a # service started from a command line has no application to speak for it, @@ -147,15 +148,11 @@ async def open_hub_transport() -> AsyncGenerator[HubTransport, None]: name="Pybricks", reason="to connect to LEGO hubs over Bluetooth" ) - async with AsyncExitStack() as stack: - bluetooth: Bluetooth | None = None + async with anyio.create_task_group() as tg: + # Cast: it offers only the calls the RFCOMM transport makes. + bluetooth = cast("Bluetooth", LazyBluetooth(tg)) try: - bluetooth = await stack.enter_async_context(aio_rfcomm.open_bluetooth()) - except RfcommError as exc: - if backend == "rfcomm": - raise - - logger.warning("RFCOMM hubs will not be found: %s", exc) - - yield get_hub_transport(bluetooth) + yield get_hub_transport(bluetooth) + finally: + tg.cancel_scope.cancel() diff --git a/packages/pb-service/src/pb_service/hub/rfcomm/lazy.py b/packages/pb-service/src/pb_service/hub/rfcomm/lazy.py new file mode 100644 index 0000000..fd086de --- /dev/null +++ b/packages/pb-service/src/pb_service/hub/rfcomm/lazy.py @@ -0,0 +1,158 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 The Pybricks Authors + +"""Bluetooth Classic, opened the first time RFCOMM needs it. + +Opening it is not free and may not finish. On macOS it starts aio-rfcomm's +helper, which may ask the user for permission, and it does not return until the +system has said whether Bluetooth is usable -- which, on a Mac where nobody is +there to answer the prompt, is never. Done as the service started, that held up +USB and BLE as well. So it is done in a task of the transport's own scope, the +first time a scan or a connection asks for it, as bleak does for BLE. +""" + +from __future__ import annotations + +import logging +from collections.abc import AsyncGenerator, Awaitable, Callable, Collection +from contextlib import asynccontextmanager +from typing import Self, final +from uuid import UUID + +import aio_rfcomm +import anyio +from aio_rfcomm.errors import BluetoothUnavailableError, RfcommError +from anyio.abc import TaskGroup + +logger = logging.getLogger(__name__) + +__all__ = ["LazyBluetooth"] + + +@final +class LazyBluetooth: + """Stands in for :class:`aio_rfcomm.Bluetooth` until it is open. + + Only the calls the RFCOMM transport makes are offered, each waiting for the + real one first. + """ + + def __init__(self, group: TaskGroup) -> None: + """Make one that will open Bluetooth when first asked. + + Args: + group: Owns the task holding Bluetooth open, so it closes when the + transport's scope does. + """ + self._group = group + self._asked = False + self._settled = anyio.Event() + self._bluetooth: aio_rfcomm.Bluetooth | None = None + self._failure: str | None = None + + async def _get(self) -> aio_rfcomm.Bluetooth: + """Wait for Bluetooth to be open, opening it on the first call. + + Returns: + The open Bluetooth object. + + Raises: + BluetoothUnavailableError: It could not be opened at all. + """ + if not self._asked: + self._asked = True + self._group.start_soon(self._hold) + + await self._settled.wait() + + if self._bluetooth is None: + raise BluetoothUnavailableError( + aio_rfcomm.Availability.UNSUPPORTED, self._failure + ) + + return self._bluetooth + + async def _hold(self) -> None: + """Open Bluetooth and keep it open until the scope ends.""" + try: + async with aio_rfcomm.open_bluetooth() as bluetooth: + self._bluetooth = bluetooth + self._settled.set() + await anyio.sleep_forever() + except RfcommError as exc: + # Reported by each call from now on, as Bluetooth that cannot be + # used, which is what RFCOMM already tells the user about. + logger.warning("RFCOMM hubs will not be found: %s", exc) + self._failure = str(exc) + self._settled.set() + + async def list_known_devices( + self, *, service: UUID | Collection[UUID] | None = None + ) -> list[aio_rfcomm.RfcommDeviceInfo]: + """As :meth:`aio_rfcomm.Bluetooth.list_known_devices`, once it is open. + + Args: + service: The services to filter on, if any. + + Returns: + The devices the system knows. + """ + bluetooth = await self._get() + return await bluetooth.list_known_devices(service=service) + + @asynccontextmanager + async def open_service( + self, device: aio_rfcomm.RfcommDeviceInfo, service: UUID + ) -> AsyncGenerator[aio_rfcomm.RfcommChannel, None]: + """As :meth:`aio_rfcomm.Bluetooth.open_service`, once it is open. + + Args: + device: The device to connect to. + service: The service to open. + + Yields: + The open channel. + """ + bluetooth = await self._get() + + async with bluetooth.open_service(device, service) as channel: + yield channel + + def watch_availability(self) -> _Watcher: + """As :meth:`aio_rfcomm.Bluetooth.watch_availability`, once it is open. + + Returns: + Bluetooth's availability now, and then each change. + """ + return _Watcher(self._get) + + +@final +class _Watcher: + """Availability, from the real watcher once Bluetooth is open.""" + + def __init__(self, get: Callable[[], Awaitable[aio_rfcomm.Bluetooth]]) -> None: + self._get = get + self._watching: aio_rfcomm.AvailabilityWatcher | None = None + self._reported = False + + def __aiter__(self) -> Self: + return self + + async def __anext__(self) -> aio_rfcomm.Availability: + if self._watching is None: + try: + bluetooth = await self._get() + except BluetoothUnavailableError: + # Said once, as a watcher says the state now. It never changes + # after, since nothing will open Bluetooth again. + if not self._reported: + self._reported = True + return aio_rfcomm.Availability.UNSUPPORTED + + await anyio.sleep_forever() + raise AssertionError("sleep_forever returned") from None + + self._watching = bluetooth.watch_availability() + + return await anext(self._watching) diff --git a/packages/pb-service/tests/test_rfcomm_lazy.py b/packages/pb-service/tests/test_rfcomm_lazy.py new file mode 100644 index 0000000..021ec2a --- /dev/null +++ b/packages/pb-service/tests/test_rfcomm_lazy.py @@ -0,0 +1,139 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 The Pybricks Authors + +"""Bluetooth Classic opened the first time RFCOMM needs it, and not before. + +On a Mac where nobody is there to answer the permission prompt, opening it +never finishes. Done as the service started, that stopped the service starting +at all, USB and BLE included, which is how CI's macOS runners found it. +""" + +from __future__ import annotations + +from collections.abc import AsyncGenerator +from contextlib import asynccontextmanager + +import aio_rfcomm +import anyio +import pytest +from aio_rfcomm.errors import BluetoothUnavailableError, RfcommError + +from pb_service.hub.rfcomm.fake import FakePairedHub, radio +from pb_service.hub.rfcomm.lazy import LazyBluetooth + +pytestmark = pytest.mark.anyio + +EV3 = "00:16:53:4A:75:15" + + +@pytest.fixture(scope="session") +def anyio_backend() -> str: + return "asyncio" + + +class _Opener: + """Stands in for ``aio_rfcomm.open_bluetooth``, counting the opens.""" + + def __init__( + self, bluetooth: aio_rfcomm.Bluetooth | None, failure: str | None = None + ) -> None: + self.opens = 0 + self.open = anyio.Event() + self._bluetooth = bluetooth + self._failure = failure + + @asynccontextmanager + async def __call__(self) -> AsyncGenerator[aio_rfcomm.Bluetooth, None]: + self.opens += 1 + + if self._failure is not None: + raise RfcommError(self._failure) + + if self._bluetooth is None: + # What a Mac with an unanswered permission prompt does. + await anyio.sleep_forever() + + assert self._bluetooth is not None + self.open.set() + + try: + yield self._bluetooth + finally: + self.open = anyio.Event() + + +@pytest.fixture +def opener(monkeypatch: pytest.MonkeyPatch) -> _Opener: + opener = _Opener(radio(FakePairedHub(address=EV3))) + monkeypatch.setattr(aio_rfcomm, "open_bluetooth", opener) + return opener + + +async def test_nothing_is_opened_until_asked(opener: _Opener) -> None: + async with anyio.create_task_group() as tg: + LazyBluetooth(tg) + await anyio.sleep(0.05) + tg.cancel_scope.cancel() + + assert opener.opens == 0 + + +async def test_the_first_call_opens_it_and_later_ones_reuse_it( + opener: _Opener, +) -> None: + async with anyio.create_task_group() as tg: + lazy = LazyBluetooth(tg) + + first = await lazy.list_known_devices() + second = await lazy.list_known_devices() + + assert [d.address for d in first] == [EV3] + assert first == second + tg.cancel_scope.cancel() + + assert opener.opens == 1 + + +async def test_bluetooth_closes_with_the_scope(opener: _Opener) -> None: + async with anyio.create_task_group() as tg: + lazy = LazyBluetooth(tg) + await lazy.list_known_devices() + assert opener.open.is_set() + tg.cancel_scope.cancel() + + assert not opener.open.is_set() + + +async def test_availability_is_watched_through_the_real_one(opener: _Opener) -> None: + async with anyio.create_task_group() as tg: + lazy = LazyBluetooth(tg) + + with anyio.fail_after(5): + now = await anext(lazy.watch_availability()) + + assert now == aio_rfcomm.Availability.USABLE + tg.cancel_scope.cancel() + + +async def test_failing_to_open_is_bluetooth_that_cannot_be_used( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # What the RFCOMM transport already reports, rather than an error that + # would end its scan. + monkeypatch.setattr(aio_rfcomm, "open_bluetooth", _Opener(None, "no helper")) + + async with anyio.create_task_group() as tg: + lazy = LazyBluetooth(tg) + + with pytest.raises(BluetoothUnavailableError, match="no helper"): + await lazy.list_known_devices() + + watcher = lazy.watch_availability() + assert await anext(watcher) == aio_rfcomm.Availability.UNSUPPORTED + + # Said once: nothing will open it again, so nothing changes. + with anyio.move_on_after(0.05) as waited: + await anext(watcher) + + assert waited.cancelled_caught + tg.cancel_scope.cancel() diff --git a/tests/e2e/test_service_roundtrip.py b/tests/e2e/test_service_roundtrip.py index f4299b3..4062ef0 100644 --- a/tests/e2e/test_service_roundtrip.py +++ b/tests/e2e/test_service_roundtrip.py @@ -13,8 +13,10 @@ import sys import uuid from collections.abc import AsyncGenerator, Iterator +from contextlib import asynccontextmanager from pathlib import Path +import aio_rfcomm import anyio import pytest @@ -212,3 +214,26 @@ async def test_idle_service_exits_on_its_own(endpoint: Endpoint) -> None: assert "idle" in service.shutdown_reason assert 0.3 <= took < 0.45, f"exited after {took:.3f}s, wanted about 0.3s" + + +@asynccontextmanager +async def _never_opens() -> AsyncGenerator[None, None]: + """What opening Bluetooth does on a Mac with nobody to answer its prompt.""" + await anyio.sleep_forever() + yield # Never reached, but makes this a generator. + + +async def test_a_service_starts_while_bluetooth_never_opens( + monkeypatch: pytest.MonkeyPatch, endpoint: Endpoint +) -> None: + # The CI failure: the service waited on Bluetooth before listening, so + # on a Mac nobody could answer, it never answered anything either. + monkeypatch.setenv("PB_HUB_BACKEND", "") + monkeypatch.setattr(aio_rfcomm, "open_bluetooth", _never_opens) + + with anyio.fail_after(10): + async with ( + service_module.run_service(endpoint, idle_timeout=0), + pb_client.connect(endpoint, autostart=False) as service, + ): + await service.ping()