From f9e9271b29dc4ff19e9547db0702cbe42141899c Mon Sep 17 00:00:00 2001 From: Jean-Pierre Martineau Date: Mon, 21 Sep 2026 03:26:13 -0400 Subject: [PATCH] The connect screen keeps looking, instead of looking once It scanned on the way in and never again, so a radio plugged in ten seconds late was invisible until MeshTerm was restarted. The worst version of that is the first run: the screen says no companion devices detected and offers a network address typed by hand, or quitting. Going to fetch the cable is the obvious thing to do and it was the one thing that did not work. The three transports want three different answers, and the reason is cost. Enumerating serial ports is a SetupAPI walk -- not free, but cheap enough to repeat every couple of seconds off the event loop without anyone noticing, and it covers the case this is really about. Bluetooth is a bounded listen, so "keep scanning" there means keeping the radio listening for as long as the splash is open, on the handhelds this program exists for, on a battery, on the screen somebody leaves up while they walk away. It gets a window every fifteen seconds instead, eight of them, and the count is ignored while the list is still empty -- an empty splash is precisely where somebody is waiting, and another listen is the cost of being useful. TCP is not scanned at all, because it never was: a network companion does not advertise, and its rows are read from the registry and the profiles, which change only when the reader changes them. Only a change redraws. A poll that finds the same ports says nothing, which matters for somebody part-way through arrowing down the list: a redraw they did not ask for, on a list that has not moved, is the screen shifting under their hands for no reason. When it does redraw, `replace_items` keeps their place by value rather than by index, so a device appearing above the highlight does not carry the selection with it. `select_startup` grew a `live` argument to make this possible -- work to do while the list is on screen, handed a `redraw(items)` rather than the screen itself, since a caller holding a screen could not repaint it anyway. It is cancelled when the screen resolves, so nothing outlives the list it was redrawing and no repaint lands on the PIN prompt or the smoke test that opens over it. Fixes: #13 Co-Authored-By: Claude Opus 5 (1M context) --- meshterm/ui/device_picker.py | 84 ++++++++++++++++++-- meshterm/ui/tui/session.py | 28 ++++++- tests/test_tui.py | 148 +++++++++++++++++++++++++++++++++++ 3 files changed, 250 insertions(+), 10 deletions(-) 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)" + )