diff --git a/meshterm/ui/device_picker.py b/meshterm/ui/device_picker.py index 21d331d..bb89c50 100644 --- a/meshterm/ui/device_picker.py +++ b/meshterm/ui/device_picker.py @@ -28,6 +28,7 @@ is connected to again. from __future__ import annotations +import asyncio from collections.abc import Awaitable, Callable, Mapping from typing import TYPE_CHECKING @@ -40,6 +41,8 @@ from ..core.device_store import DeviceStore, RememberedDevice from ..core.discovery import ( DEFAULT_TCP_PORT, DiscoveredDevice, + discover_ble_devices, + discover_devices, parse_tcp_endpoint, serial_device, tcp_device, @@ -298,6 +301,27 @@ async def _smoke_test( return info +#: How often the splash re-enumerates serial ports. `comports()` is a SetupAPI/sysfs +#: walk rather than a free read, so it runs off the event loop -- but it is cheap enough +#: to repeat at this rate without anyone noticing, and it covers the common case: a USB +#: radio plugged in after the splash appeared. +_POLL_S = 2.0 + +#: Bluetooth is not cheap in the same way. Each scan is a bounded listen, so keeping one +#: running means keeping the radio listening for as long as this screen is open -- and +#: this screen is the one a reader leaves up while they go and find a cable, on a handheld, +#: on a battery. So it gets a duty cycle rather than a loop: a window this often, and only +#: while it is still plausible somebody is waiting on one. +_BLE_EVERY_S = 15.0 +_BLE_WINDOW_S = 4.0 + +#: How many windows to run before giving up on finding a companion nobody has turned on. +#: Once something *is* listed the reader has what they came for; while the list is still +#: empty the count is ignored, because an empty splash is exactly where somebody is +#: waiting and the cost of another listen is the cost of being useful. +_BLE_WINDOWS = 8 + + async def prompt_device( ui: Ui, devices: list[DiscoveredDevice], @@ -327,11 +351,24 @@ async def prompt_device( leave the picker without selecting one — by pressing Esc or choosing the Quit row — which the caller treats as a request to exit. """ - scanned = list(devices) + # Serial and Bluetooth are kept apart because they are refreshed on different clocks: + # a poll replaces everything attached, a Bluetooth window replaces everything in range, + # and one must not wipe the other's findings. + wired = [d for d in devices if not d.is_ble] + wireless = [d for d in devices if d.is_ble] #: The row the *next* redraw should open on, when the pass just finished moved the list #: under the reader (see :func:`_after_hiding`). Cleared as soon as it is spent. focus: DiscoveredDevice | None = None - while True: + + def assemble() -> tuple[ + list[DiscoveredDevice], list, RememberedDevice | None, dict[str, RememberedDevice] + ]: + """The list as it stands this instant: what is there, less what is hidden. + + Built in one place because two callers need it now -- the loop below, and the + rescan that redraws the rows under the reader while the splash is open. + """ + scanned = wired + wireless remembered = store.load() # The full registry (not just the single last device) so *every* confirmed companion # can be named, highlighted, and sorted to the top — keyed by stable_id. Reloaded each @@ -353,7 +390,42 @@ async def prompt_device( # only feedback hiding needs. hidden = store.hidden_ids() # In display order, so a shortcut acting on a row can say what the row *after* it is. - ordered = _order([d for d in listed if d.stable_id not in hidden], registry) + order = _order([d for d in listed if d.stable_id not in hidden], registry) + items = _build_items(order, remembered, registry, hidden=len(hidden)) + return order, items, remembered, registry + + async def keep_looking(redraw: Callable[[list], None]) -> None: + """Re-enumerate while the splash is up, and redraw when the answer changes. + + The screen used to scan once on the way in and never again, so a radio plugged in + ten seconds late was invisible until MeshTerm was restarted -- and the screen that + said so offered only a network address typed by hand, or quitting. Fetching the + cable is the obvious thing to do and it was the one thing that did not work. + + Only a change redraws. A poll that finds the same ports is silence, which matters + because the reader may be part-way through arrowing down the list. + """ + windows = 0 + waited = 0.0 + while True: + await asyncio.sleep(_POLL_S) + before = {d.stable_id for d in wired + wireless} + + # ``comports()`` blocks; off the event loop so the splash stays live. + wired[:] = await asyncio.to_thread(discover_devices) + + waited += _POLL_S + if waited >= _BLE_EVERY_S and (windows < _BLE_WINDOWS or not wired + wireless): + waited = 0.0 + windows += 1 + wireless[:] = await discover_ble_devices(_BLE_WINDOW_S) + + if {d.stable_id for d in wired + wireless} != before: + _, rows, _, _ = assemble() + redraw(rows) + + while True: + ordered, items, remembered, registry = assemble() # Preselect the remembered "last known good" device when it is currently attached/in # range — unless the last pass asked for a particular row, which a hide does so the # highlight lands where the vanished row was rather than jumping back to the default. @@ -362,16 +434,14 @@ async def prompt_device( default = focus if focus in ordered else default focus = None - # Build the rows once so the same list can be redrawn as the backdrop behind a - # removal confirm (so it floats over the picker rather than replacing it). - items = _build_items(ordered, remembered, registry, hidden=len(hidden)) chosen = await ui.select_startup( "Select a companion device", items, default=default, banner=load_logo(), keys=_SHORTCUTS, - key_hint=_shortcut_hint(len(hidden)), + key_hint=_shortcut_hint(len(store.hidden_ids())), + live=keep_looking, ) # Esc (``None``) and the Quit row both mean "leave the picker" — surface that to the # caller as ``None`` so it can exit the program instead of continuing device-less. diff --git a/meshterm/ui/tui/session.py b/meshterm/ui/tui/session.py index c1e2f84..adc9d7a 100644 --- a/meshterm/ui/tui/session.py +++ b/meshterm/ui/tui/session.py @@ -15,8 +15,8 @@ from __future__ import annotations import asyncio import os import sys -from collections.abc import AsyncIterator, Callable, Mapping, Sequence -from contextlib import asynccontextmanager, nullcontext +from collections.abc import AsyncIterator, Awaitable, Callable, Mapping, Sequence +from contextlib import asynccontextmanager, nullcontext, suppress from typing import Any from prompt_toolkit.application import Application @@ -917,6 +917,7 @@ class TuiSession: footer_hint: str = "↑↓ move · Enter select · Esc bye", keys: Mapping[str, Any] | None = None, key_hint: Callable[[Any], str] | None = None, + live: Callable[[Callable[[list], None]], Awaitable[None]] | None = None, ) -> Any: """Show a chromeless select splash (banner above a content-sized box). @@ -931,6 +932,15 @@ class TuiSession: removal key advertises itself exactly where it acts (see :attr:`~meshterm.ui.tui.select.SelectScreen.footer_hint`). + ``live`` is work to do *while* the list is up — the device picker's rescan, which + keeps looking for a companion the whole time the splash is open. It is handed one + function, ``redraw(items)``, which swaps the rows under the reader and repaints; + the screen itself stays in here, since a caller holding one could not repaint it + anyway (``invalidate`` is the session's, not the screen's). It runs as a task for + exactly as long as the screen does and is cancelled when the screen resolves, so + nothing outlives the list it was redrawing and no repaint lands on a dialog that + has since opened over it. + ``keys`` hands the list bare-key shortcuts — which a splash can afford precisely because it does not filter, so every letter is free (see :class:`SelectScreen`) — and ``key_hint`` says what they are called on the row the highlight is standing on, @@ -959,7 +969,19 @@ class TuiSession: # ⇧H only appears at all once something is hidden, which is exactly the state where # a reader needs to be told how to undo it, and by then they have already found h. screen.spare_hint_atoms = ("←→ scroll", "h hide") - result = await self.run_screen(screen) + def redraw(new_items: list) -> None: + """Swap the rows and repaint. The reader keeps their place, by value.""" + screen.replace_items(new_items) + self.invalidate() + + worker = asyncio.ensure_future(live(redraw)) if live else None + try: + result = await self.run_screen(screen) + finally: + if worker is not None: + worker.cancel() + with suppress(asyncio.CancelledError): + await worker return None if result is CANCEL else result async def confirm_startup( diff --git a/tests/test_tui.py b/tests/test_tui.py index 137bd31..29d6498 100644 --- a/tests/test_tui.py +++ b/tests/test_tui.py @@ -3425,3 +3425,151 @@ def test_a_shortcut_is_only_honoured_where_a_letter_is_free() -> None: filtering.handle("text", "h") assert len(resolved) == 1 # nothing new resolved assert filtering._filter == "h" # the letter went to the query, where it belongs + + +# -- the splash keeps looking --------------------------------------------------------- + + +def _drive_picker(tmp_path, monkeypatch, *, serial_rounds, ble_rounds=None, hide=None): + """Run the picker's rescan against a scripted sequence of scans. + + The fake UI stands in for the splash being open: it is handed the ``live`` callable + and runs it until the script is exhausted, recording every redraw. Discovery is + replaced rather than mocked at the port level, because what is under test is the + rescan's behaviour -- when it redraws, when it keeps quiet -- and not pyserial. + """ + import asyncio + + from meshterm.core.device_store import DeviceStore + from meshterm.ui import device_picker + + monkeypatch.setattr(device_picker, "_POLL_S", 0.001) + monkeypatch.setattr(device_picker, "_BLE_EVERY_S", 0.0025) + monkeypatch.setattr(device_picker, "_BLE_WINDOW_S", 0.0) + + serial = list(serial_rounds) + ble = list(ble_rounds or [[]]) + calls = {"serial": 0, "ble": 0} + + def _serial(): + calls["serial"] += 1 + return list(serial[min(calls["serial"] - 1, len(serial) - 1)]) + + async def _ble(_timeout): + calls["ble"] += 1 + return list(ble[min(calls["ble"] - 1, len(ble) - 1)]) + + monkeypatch.setattr(device_picker, "discover_devices", _serial) + monkeypatch.setattr(device_picker, "discover_ble_devices", _ble) + + store = DeviceStore(tmp_path / "devices.json") + for stable in hide or []: + store.hide(stable) + + redraws: list = [] + + class _Ui: + async def select_startup( # noqa: ANN001, ANN201, ANN003 + self, title, items, *, default=None, banner=None, footnote=None, live=None, **_kw + ): + task = asyncio.ensure_future(live(redraws.append)) + # Long enough for the script to run several times over at this cadence. + await asyncio.sleep(0.05) + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + return None + + async def _never(_device): + raise AssertionError("verify should not run when selection is skipped") + + asyncio.run(device_picker.prompt_device(_Ui(), list(serial[0]), store, _never)) + return redraws, calls + + +def _labels(items) -> list[str]: + return [ + (it.title.plain if hasattr(it.title, "plain") else str(it.title)) + for it in items + if isinstance(it, Choice) + ] + + +def test_splash_redraws_when_a_device_is_plugged_in(tmp_path, monkeypatch) -> None: + """A radio attached after the splash opened appears without restarting MeshTerm.""" + from meshterm.core.discovery import DiscoveredDevice + + first = DiscoveredDevice(port="COM5", product="Wio SX1262", vid=0x2886) + later = DiscoveredDevice(port="COM9", product="RAK4631", vid=0x239A) + + redraws, _ = _drive_picker(tmp_path, monkeypatch, serial_rounds=[[first], [first, later]]) + + assert redraws, "the splash never redrew after the device appeared" + assert any("COM9" in label for label in _labels(redraws[-1])) + + +def test_splash_stays_quiet_when_nothing_changed(tmp_path, monkeypatch) -> None: + """A poll that finds the same ports redraws nothing. + + This is the half that matters for somebody mid-way through arrowing down the list: + a redraw they did not ask for, on a list that has not changed, is the screen moving + under their hands for no reason. + """ + from meshterm.core.discovery import DiscoveredDevice + + same = [DiscoveredDevice(port="COM5", product="Wio SX1262", vid=0x2886)] + redraws, calls = _drive_picker(tmp_path, monkeypatch, serial_rounds=[same]) + + assert calls["serial"] > 1, "the rescan never ran" + assert redraws == [] + + +def test_splash_notices_a_device_going_away(tmp_path, monkeypatch) -> None: + """Unplugging removes the row, which is the same question asked backwards.""" + from meshterm.core.discovery import DiscoveredDevice + + one = DiscoveredDevice(port="COM5", product="Wio SX1262", vid=0x2886) + redraws, _ = _drive_picker(tmp_path, monkeypatch, serial_rounds=[[one], []]) + + assert redraws, "the splash never redrew after the device went away" + assert not any("COM5" in label for label in _labels(redraws[-1])) + + +def test_a_rescan_does_not_unhide_what_the_reader_hid(tmp_path, monkeypatch) -> None: + """A refresh is not ⇧H. Whatever `h` hid stays hidden when the list is rebuilt.""" + from meshterm.core.discovery import DiscoveredDevice + + kept = DiscoveredDevice(port="COM5", product="Wio SX1262", vid=0x2886) + buried = DiscoveredDevice(port="COM7", product="FT232R USB UART", vid=0x0403) + later = DiscoveredDevice(port="COM9", product="RAK4631", vid=0x239A) + + redraws, _ = _drive_picker( + tmp_path, + monkeypatch, + serial_rounds=[[kept, buried], [kept, buried, later]], + hide=[buried.stable_id], + ) + + assert redraws, "the splash never redrew" + labels = _labels(redraws[-1]) + assert any("COM9" in label for label in labels) + assert not any("COM7" in label for label in labels) + + +def test_bluetooth_is_not_scanned_on_every_poll(tmp_path, monkeypatch) -> None: + """Serial is cheap and polled; Bluetooth is a listen and gets a duty cycle. + + Each BLE scan keeps the radio listening, and this is the screen somebody leaves open + on a battery-powered handheld while they go and find a cable -- so it must not be in + the poll loop. + """ + from meshterm.core.discovery import DiscoveredDevice + + same = [DiscoveredDevice(port="COM5", product="Wio SX1262", vid=0x2886)] + _, calls = _drive_picker(tmp_path, monkeypatch, serial_rounds=[same]) + + assert calls["serial"] > calls["ble"], ( + f"bluetooth scanned {calls['ble']} time(s) against {calls['serial']} serial poll(s)" + )