diff --git a/src/glider/gui/widgets/schema_form.py b/src/glider/gui/widgets/schema_form.py index 888a2bce..1736712a 100644 --- a/src/glider/gui/widgets/schema_form.py +++ b/src/glider/gui/widgets/schema_form.py @@ -167,14 +167,16 @@ async def _scan(): combo.clear() if not results: combo.addItem("(no devices found)", None) - for name, address in results: - # Show the advertised name; fall back to the address for - # unnamed peripherals so they stay distinguishable. The - # address is the item data (and tooltip) and is what gets - # saved. - label = name if name and name != "(unknown)" else address - combo.addItem(label, address) - combo.setItemData(combo.count() - 1, address, Qt.ItemDataRole.ToolTipRole) + for peripheral in results: + # An unnamed peripheral shows as its address and signal + # strength, with its advertised services in the tooltip -- + # which is how you tell which bare MAC is the stimulator + # when its name did not survive the scan response. + combo.addItem(peripheral.label, peripheral.address) + detail = peripheral.address + if peripheral.service_uuids: + detail += chr(10) + "services: " + ", ".join(peripheral.service_uuids) + combo.setItemData(combo.count() - 1, detail, Qt.ItemDataRole.ToolTipRole) except ImportError: QMessageBox.critical(parent, "Scan failed", "bleak is not installed.") except Exception as e: # noqa: BLE001 - surfaced to the user diff --git a/src/glider/hal/boards/ble_board.py b/src/glider/hal/boards/ble_board.py index 3d7ea2a5..fe0ddda8 100644 --- a/src/glider/hal/boards/ble_board.py +++ b/src/glider/hal/boards/ble_board.py @@ -15,6 +15,7 @@ """ import logging +from dataclasses import dataclass from glider.hal.base_board import ( BaseBoard, @@ -27,6 +28,31 @@ logger = logging.getLogger(__name__) +@dataclass(frozen=True) +class DiscoveredPeripheral: + """One peripheral seen by a scan. + + ``name`` is empty when the peripheral advertised none, which is common and + not a defect: a Zephyr device puts its name in the scan response, and + Windows drops that often enough that a working device routinely appears + nameless. + """ + + address: str + name: str = "" + rssi: int | None = None + service_uuids: tuple[str, ...] = () + + @property + def label(self) -> str: + """What to show a human picking one out of a list.""" + base = self.name or self.address + return f"{base} ({self.rssi} dBm)" if self.rssi is not None else base + + def advertises(self, service_uuid: str) -> bool: + return str(service_uuid).strip().lower() in self.service_uuids + + class BLEBoard(BaseBoard): """Host Bluetooth LE adapter. Peripherals connect per-device via bleak.""" @@ -68,17 +94,23 @@ async def disconnect(self) -> None: logger.info("BLEBoard: adapter released") @staticmethod - async def scan(timeout: float = 8.0) -> list[tuple[str, str]]: + async def scan(timeout: float = 8.0) -> list[DiscoveredPeripheral]: """Discover nearby BLE peripherals. - Returns a list of ``(name, address)`` tuples; ``name`` falls back to - ``"(unknown)"`` when the peripheral advertises none. - Reads the name from the advertisement data (``local_name``) rather than ``device.name``: many peripherals (e.g. Zephyr devices) send their name in the SCAN RESPONSE, which an active scan captures into ``local_name`` even when ``device.name`` comes back empty on Windows. Static so callers can scan the host adapter without needing a board instance. + + Keeps the signal strength and the advertised service UUIDs, because a + peripheral whose name did not come through is otherwise a bare MAC in a + list of bare MACs. The services say *what* it is and the RSSI says which + one is on the bench in front of you -- and a device that knows its own + service UUID can be matched without a name at all. + + Sorted strongest-first, so the peripheral you are holding is near the + top rather than wherever the adapter happened to enumerate it. """ from bleak import BleakScanner @@ -87,8 +119,22 @@ async def scan(timeout: float = 8.0) -> list[tuple[str, str]]: results = [] for dev, adv in discovered.values(): name = (getattr(adv, "local_name", None) or getattr(dev, "name", None) or "").strip() - results.append((name or "(unknown)", dev.address)) - logger.info("BLEBoard: scan found %d peripheral(s)", len(results)) + results.append( + DiscoveredPeripheral( + name=name, + address=dev.address, + rssi=getattr(adv, "rssi", None), + service_uuids=tuple( + str(u).lower() for u in (getattr(adv, "service_uuids", None) or ()) + ), + ) + ) + results.sort(key=lambda p: (p.rssi if p.rssi is not None else -999), reverse=True) + logger.info( + "BLEBoard: scan found %d peripheral(s): %s", + len(results), + ", ".join(p.label for p in results) or "none", + ) return results # --- pin operations are not applicable to BLE --- diff --git a/src/glider/hal/devices/ble_device.py b/src/glider/hal/devices/ble_device.py index 8c52c2c3..8b3ea49f 100644 --- a/src/glider/hal/devices/ble_device.py +++ b/src/glider/hal/devices/ble_device.py @@ -239,6 +239,41 @@ async def _resolve_address(self) -> str: return self._resolved_address return await self._find_by_name() + async def _find_by_service(self) -> str | None: + """Address of a peripheral advertising ``service_uuid``, if exactly one is. + + The identifier of last resort, and the sturdiest available: a service + UUID neither rotates the way a private address does nor depends on a + scan response surviving the trip -- which is how a Zephyr device ends up + nameless on Windows while a phone app sees it fine. + + Returns None rather than guessing when several peripherals advertise the + service. With six identical stimulators on a bench that is the normal + case, and picking one would connect to the wrong animal's. + """ + if not self._service_uuid: + return None + from glider.hal.boards.ble_board import BLEBoard + + found = await BLEBoard.scan(timeout=RESOLVE_SCAN_S) + matches = [p for p in found if p.advertises(self._service_uuid)] + if not matches: + return None + if len(matches) > 1: + logger.warning( + "BLE %s: %d peripherals advertise service %s (%s); set an address " + "or an advertised name to say which one", + self._name, + len(matches), + self._service_uuid, + ", ".join(p.label for p in matches), + ) + return None + logger.info( + "BLE %s: matched service %s -> %s", self._name, self._service_uuid, matches[0].address + ) + return matches[0].address + async def _find_by_name(self) -> str: """Scan for ``name`` and return the address it is advertising *now*.""" from bleak import BleakScanner @@ -280,7 +315,7 @@ async def _ensure_connected(self) -> None: # is worth a rescan before giving up. The Scan button fills the # address and the operator often fills the name too, which makes # this the common case rather than an exotic one. - if not self._adv_name: + if not self._adv_name and not self._service_uuid: raise logger.info( "BLE %s: no answer at %s (%s); re-resolving by name %r", @@ -290,8 +325,13 @@ async def _ensure_connected(self) -> None: self._adv_name, ) self._resolved_address = None - fresh = await self._find_by_name() - if fresh == address: + # Name first when there is one -- it names *this* unit. The service + # UUID is shared by every device of the type, so it can only help + # when exactly one is in range. + fresh = await self._find_by_name() if self._adv_name else None + if fresh is None: + fresh = await self._find_by_service() + if fresh is None or fresh == address: # The name resolved to the address that just failed, so this is # not a rotation. Report the original failure rather than a # second identical one. diff --git a/tests/unit/hal/test_ble_board.py b/tests/unit/hal/test_ble_board.py index e239932e..2229bf33 100644 --- a/tests/unit/hal/test_ble_board.py +++ b/tests/unit/hal/test_ble_board.py @@ -21,9 +21,10 @@ def __init__(self, name, address): class _FakeAdv: - def __init__(self, local_name): + def __init__(self, local_name, service_uuids=None, rssi=None): self.local_name = local_name - self.service_uuids = [] + self.service_uuids = list(service_uuids or []) + self.rssi = rssi @pytest.fixture @@ -54,13 +55,86 @@ async def test_connect_marks_ready(fake_bleak): assert board.is_connected -async def test_scan_returns_name_address_pairs(fake_bleak): +async def test_scan_reports_each_peripheral(fake_bleak): board = BLEBoard() await board.connect() + + results = await board.scan(timeout=0.1) + + by_address = {p.address: p for p in results} + # "Opto-A" advertises its name only in the scan response. + assert by_address["AA:BB"].name == "Opto-A" + # A peripheral that advertised no name keeps an empty one rather than a + # placeholder, so callers can tell "nameless" from "named '(unknown)'". + assert by_address["CC:DD"].name == "" + + +async def test_an_unnamed_peripheral_is_labelled_by_address(fake_bleak): + """A Zephyr device whose scan response was dropped is a bare MAC in a list + of bare MACs; the label is what a human has to pick from.""" + board = BLEBoard() + await board.connect() + results = await board.scan(timeout=0.1) - assert ("Opto-A", "AA:BB") in results - # Unnamed peripheral falls back to "(unknown)". - assert ("(unknown)", "CC:DD") in results + + assert next(p for p in results if p.address == "CC:DD").label == "CC:DD" + assert next(p for p in results if p.address == "AA:BB").label == "Opto-A" + + +async def test_the_label_carries_signal_strength_when_known(monkeypatch, fake_bleak): + """Which of several identical peripherals is the one on the bench in front + of you is answered by RSSI and nothing else.""" + + class _Scanner: + @staticmethod + async def discover(timeout=5.0, return_adv=False, **kwargs): + return {"AA:BB": (_FakeBLEDevice(None, "AA:BB"), _FakeAdv("Stim", rssi=-42))} + + fake_bleak.BleakScanner = _Scanner + board = BLEBoard() + await board.connect() + + assert (await board.scan(timeout=0.1))[0].label == "Stim (-42 dBm)" + + +async def test_results_are_sorted_strongest_first(monkeypatch, fake_bleak): + class _Scanner: + @staticmethod + async def discover(timeout=5.0, return_adv=False, **kwargs): + return { + "FAR": (_FakeBLEDevice(None, "FAR"), _FakeAdv("far", rssi=-90)), + "NEAR": (_FakeBLEDevice(None, "NEAR"), _FakeAdv("near", rssi=-30)), + } + + fake_bleak.BleakScanner = _Scanner + board = BLEBoard() + await board.connect() + + assert [p.address for p in await board.scan(timeout=0.1)] == ["NEAR", "FAR"] + + +async def test_advertised_services_are_kept_and_matchable(monkeypatch, fake_bleak): + """The sturdiest identifier a nameless peripheral has.""" + service = "12345678-1234-5678-1234-56789ABCDEF0" + + class _Scanner: + @staticmethod + async def discover(timeout=5.0, return_adv=False, **kwargs): + return { + "AA:BB": (_FakeBLEDevice(None, "AA:BB"), _FakeAdv(None, service_uuids=[service])) + } + + fake_bleak.BleakScanner = _Scanner + board = BLEBoard() + await board.connect() + + found = (await board.scan(timeout=0.1))[0] + # Case-insensitive: advertisements and configuration disagree on case + # routinely, and a UUID that matched only sometimes would be worse than one + # that never did. + assert found.advertises(service.lower()) + assert found.advertises(service.upper()) + assert not found.advertises("00000000-0000-0000-0000-000000000000") async def test_pin_operations_raise(fake_bleak): diff --git a/tests/unit/hal/test_ble_device_full.py b/tests/unit/hal/test_ble_device_full.py index 54d9128b..ac65d44a 100644 --- a/tests/unit/hal/test_ble_device_full.py +++ b/tests/unit/hal/test_ble_device_full.py @@ -487,3 +487,109 @@ async def test_a_name_that_resolves_to_nothing_says_why(rotating_bleak): with pytest.raises(RuntimeError, match="connected to something else"): await device.initialize() + + +async def test_a_nameless_peripheral_is_found_by_its_service(monkeypatch): + """The case from the bench: the address rotated, the name never made it + into the advertisement, and a phone app saw the device the whole time. + The service UUID is the one identifier that survives both.""" + from unittest.mock import MagicMock + + from glider.hal.boards.ble_board import DiscoveredPeripheral + + service = "12345678-1234-5678-1234-56789abcdef0" + attempts: list[str] = [] + + def _client(address, *a, **k): + attempts.append(address) + client = _FakeClient(address) + if address == "AA:STALE": + + async def _refuse(): + raise RuntimeError("Device with address AA:STALE was not found.") + + client.connect = _refuse + return client + + module = MagicMock(name="bleak") + module.BleakClient = _client + monkeypatch.setitem(sys.modules, "bleak", module) + + async def _scan(timeout=8.0): + return [ + DiscoveredPeripheral(address="OTHER", name="lab chair", rssi=-70), + DiscoveredPeripheral(address="BB:LIVE", name="", rssi=-40, service_uuids=(service,)), + ] + + monkeypatch.setattr("glider.hal.boards.ble_board.BLEBoard.scan", staticmethod(_scan)) + + device = _make_device( + settings={"address": "AA:STALE", "service_uuid": service, "write_char_uuid": "c"} + ) + await device.initialize() + + assert attempts == ["AA:STALE", "BB:LIVE"] + assert device.is_initialized + + +async def test_several_peripherals_advertising_the_service_is_refused(monkeypatch): + """Six identical stimulators on a bench is the normal case, and connecting + to whichever answered first would be the wrong animal's.""" + from unittest.mock import MagicMock + + from glider.hal.boards.ble_board import DiscoveredPeripheral + + service = "12345678-1234-5678-1234-56789abcdef0" + + def _client(address, *a, **k): + client = _FakeClient(address) + + async def _refuse(): + raise RuntimeError("Device with address AA:STALE was not found.") + + client.connect = _refuse + return client + + module = MagicMock(name="bleak") + module.BleakClient = _client + monkeypatch.setitem(sys.modules, "bleak", module) + + async def _scan(timeout=8.0): + return [ + DiscoveredPeripheral(address="ONE", service_uuids=(service,)), + DiscoveredPeripheral(address="TWO", service_uuids=(service,)), + ] + + monkeypatch.setattr("glider.hal.boards.ble_board.BLEBoard.scan", staticmethod(_scan)) + + device = _make_device( + settings={"address": "AA:STALE", "service_uuid": service, "write_char_uuid": "c"} + ) + + with pytest.raises(RuntimeError, match="was not found"): + await device.initialize() + + +async def test_the_name_is_preferred_over_the_service(rotating_bleak, monkeypatch): + """A name identifies *this* unit; a service UUID is shared by every device + of the type. When both are configured the specific one has to win.""" + scanned = [] + + async def _scan(timeout=8.0): + scanned.append(True) + return [] + + monkeypatch.setattr("glider.hal.boards.ble_board.BLEBoard.scan", staticmethod(_scan)) + + device = _make_device( + settings={ + "address": "AA:OLD", + "name": "maimu_ezurio", + "service_uuid": "12345678-1234-5678-1234-56789abcdef0", + "write_char_uuid": "c", + } + ) + await device.initialize() + + assert rotating_bleak.attempts == ["AA:OLD", "BB:NEW"] + assert scanned == [], "the service scan ran even though the name resolved"