diff --git a/meshterm/services/tx_optimizer.py b/meshterm/services/tx_optimizer.py index 1b77bd9..2e46491 100644 --- a/meshterm/services/tx_optimizer.py +++ b/meshterm/services/tx_optimizer.py @@ -32,6 +32,11 @@ from . import trace_runner LevelCallback = Callable[[int, int, TxLevelResult], None] PersistLevel = Callable[[TxLevelResult], None] +PhaseCallback = Callable[[str], None] + +#: The search phases, in order, as reported to ``on_phase``: the coarse grid sweep, the +#: integer fill-in around the coarse winner, and the extra-samples re-measure of the leader. +PHASES = ("coarse", "refine", "verify") #: On a near-tie in target SNR (within this many dB of the best), the lower TX power wins. DEFAULT_SNR_TOLERANCE_DB = 1.0 @@ -142,6 +147,7 @@ async def optimize_tx_power( cooldown_s: float = 1.0, snr_tolerance: float = DEFAULT_SNR_TOLERANCE_DB, on_level: Optional[LevelCallback] = None, + on_phase: Optional[PhaseCallback] = None, persist_level: Optional[PersistLevel] = None, persist_trace: Optional[Callable] = None, ) -> TxOptResult: @@ -168,6 +174,8 @@ async def optimize_tx_power( cooldown_s: Delay between individual traces (duty-cycle safety). snr_tolerance: dB band for the lower-power tie-break (see :func:`select_best`). on_level: Optional progress callback ``(completed, total, level_result)``. + on_phase: Optional callback announcing each search phase as it begins (one of + :data:`PHASES`), so a live view can say *what kind* of measuring is happening. persist_level: Optional callback to store each level's aggregated result. persist_trace: Optional callback to store each individual trace. @@ -221,12 +229,16 @@ async def optimize_tx_power( return level try: + if on_phase is not None: + on_phase("coarse") for tx in coarse: await measure_level(tx) best = select_best(list(levels.values()), snr_tolerance=snr_tolerance) if refine: + if on_phase is not None: + on_phase("refine") lo = max(tx_min, best.tx_power - coarse_step + 1) hi = min(tx_max, best.tx_power + coarse_step - 1) for tx in range(lo, hi + 1): @@ -237,6 +249,8 @@ async def optimize_tx_power( if verify: # Re-measure the leader with extra samples; if it was an outlier the larger # sample will pull it back and a steadier neighbor can take over. + if on_phase is not None: + on_phase("verify") await measure_level(best.tx_power) best = select_best(list(levels.values()), snr_tolerance=snr_tolerance) diff --git a/meshterm/tools/tx_optimize.py b/meshterm/tools/tx_optimize.py index 9749d01..e3321b3 100644 --- a/meshterm/tools/tx_optimize.py +++ b/meshterm/tools/tx_optimize.py @@ -3,8 +3,13 @@ You force a path (just like ``trace``) ending at the **target** node where SNR is measured. The node one hop *before* the target — one you hold admin rights on — is the node whose transmit power gets swept and tuned. Admin passwords are remembered between -runs, progress streams live like a trace, and the winning power is written back to the -node when the sweep finishes. +runs. + +In the interactive menu one prompt picks the path and the live sweep screen +(:mod:`meshterm.ui.tx_screen`) takes it from there: levels land in a bar chart as they +are measured, and the apply decision is made *after* the sweep, over the evidence. On the +CLI it stays a scriptable one-shot with the full flag set (range, step, samples, +``--apply``, the HTML chart), and progress streams like a trace. """ from __future__ import annotations @@ -37,65 +42,121 @@ class TxOptimizeTool(Tool): order = 15 # right after Trace, its measurement sibling async def prompt_params(self, ctx: AppContext) -> Optional[dict[str, Any]]: - """Interactively gather the path, sampling, range, and apply choices. + """Pick the path for the live sweep (everything else defaults, tuned in-screen). Args: ctx: Shared application context. Returns: - A parameter dict, or ``None`` if the user cancelled. + ``{"live": True, "path": spec}``, or ``None`` if the user cancelled. """ device = await ctx.device() contacts = await device.get_contacts() choices = [c.name for c in contacts] - path_prompt = ( - "Path to the target (comma-separated contacts/hex, ending at the target;\n" - "the node just before the target is the one we'll tune):" + prompt = ( + "Comma-separated contacts/hex, ending at the target — " + "the node just before it is tuned" ) validate_path = lambda v: _validate_link_path(v, contacts) # noqa: E731 if choices: - path_spec = await ctx.ui.autocomplete(path_prompt, choices, validate=validate_path) + path_spec = await ctx.ui.autocomplete( + "Path to the target", choices, prompt=prompt, validate=validate_path + ) else: - path_spec = await ctx.ui.text(path_prompt, validate=validate_path) + path_spec = await ctx.ui.text( + "Path to the target", prompt=prompt, validate=validate_path + ) if not path_spec: return None + return {"live": True, "path": path_spec.strip()} - samples = await ctx.ui.text( - f"Traces per TX level? (1-{MAX_SAMPLES})", default="3", validate=_is_valid_sample_count - ) - if samples is None: - return None - tx_min = await ctx.ui.text( - "Lowest TX power to try:", default=str(ctx.settings.tx_opt_min), validate=_is_int - ) - if tx_min is None: - return None - tx_max = await ctx.ui.text( - "Highest TX power to try:", default=str(ctx.settings.tx_opt_max), validate=_is_int - ) - if tx_max is None: - return None - step = await ctx.ui.text("Coarse step:", default="3", validate=_is_valid_sample_count) - if step is None: - return None - apply = await ctx.ui.confirm( - "Set the winning TX power on the node afterward?", default=True - ) - if apply is None: - return None + async def execute(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult: + """Run — without the outer run row on the live path. - return { - "path": path_spec.strip(), - "samples": int(samples), - "tx_min": int(tx_min), - "tx_max": int(tx_max), - "step": int(step), - "apply": bool(apply), - "viz": True, - } + The live screen opens its own ``runs`` row for the sweep (matching what a + scripted invocation records), so wrapping the screen session in another row + would double-log it. Scripted runs keep the base class's logging. + + Args: + ctx: Shared application context. + params: Parameters for this invocation. + + Returns: + The :class:`ToolResult` from :meth:`run`. + """ + if params.get("live"): + return await self.run(ctx, params) + return await super().execute(ctx, params) async def run(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult: + """Open the live sweep (menu) or run the scripted optimization (CLI). + + Args: + ctx: Shared application context. + params: ``live`` + ``path`` from the menu prompt; or ``path``, ``samples``, + ``tx_min``, ``tx_max``, ``step``, ``apply``, optional ``password``/ + ``viz``, and the injected ``_run_id`` from the CLI. + + Returns: + A :class:`ToolResult` with the optimum and any chart path. + """ + if params.get("live"): + return await self._run_live(ctx, str(params["path"])) + return await self._run_cli(ctx, params) + + # -- interactive (menu) --------------------------------------------------------- + + async def _run_live(self, ctx: AppContext, path_spec: str) -> ToolResult: + """Resolve the link, log in, and hand off to the live sweep screen. + + Args: + ctx: Shared application context. + path_spec: The user's comma-separated path entry (names and/or hex). + + Returns: + A :class:`ToolResult` echoing the sweep's recorded summary. + """ + from ..ui.tx_screen import open_tx_optimize + + device = await ctx.device() + contacts = await device.get_contacts() + path = trace_runner.parse_trace_path(path_spec, contacts) + admin_node, target_label = _resolve_link(path, contacts) + await self._login(ctx, admin_node, {}) + + summary = await open_tx_optimize( + ctx, admin_node=admin_node, target_label=target_label, path=path + ) + return ToolResult(summary=summary) + + async def _login( + self, ctx: AppContext, admin_node: Contact, params: dict[str, Any] + ) -> None: + """Authenticate against the admin node, remembering a working password. + + Args: + ctx: Shared application context. + admin_node: The node we're about to tune. + params: Tool params (may carry an explicit ``password`` on the CLI). + + Raises: + DeviceCommandError: If the login is rejected (the stored password, now + known bad, is forgotten so the next run asks fresh). + """ + device = await ctx.device() + password = await self._resolve_password(ctx, admin_node, params) + if not await device.admin_login(admin_node, password): + ctx.admin_store.forget(admin_node) # bad password: don't keep reusing it + raise DeviceCommandError( + f"admin login to {admin_node.name!r} failed (wrong password?). " + "The saved password was cleared; re-run to enter a new one." + ) + ctx.admin_store.remember(admin_node, password) + + # -- scripted (CLI) --------------------------------------------------------------- + + async def _run_cli(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult: """Resolve the link, log in, sweep TX power, render results, and apply the winner. Args: @@ -119,15 +180,8 @@ class TxOptimizeTool(Tool): if int(params.get("samples", 3)) > MAX_SAMPLES: ctx.ui.note(f"[warn]capping at {MAX_SAMPLES} traces per level[/warn]") - password = await self._resolve_password(ctx, admin_node, params) ctx.ui.note(f"[muted]logging in to[/muted] [brand]{admin_node.name}[/brand] [muted]…[/muted]") - if not await device.admin_login(admin_node, password): - ctx.admin_store.forget(admin_node) # bad password: don't keep reusing it - raise DeviceCommandError( - f"admin login to {admin_node.name!r} failed (wrong password?). " - "The saved password was cleared; re-run to enter a new one." - ) - ctx.admin_store.remember(admin_node, password) + await self._login(ctx, admin_node, params) ctx.ui.note( f"[muted]tuning[/muted] [brand]{admin_node.name}[/brand] " @@ -349,42 +403,6 @@ def _validate_link_path(value: str, contacts: list[Contact]) -> bool | str: return True -def _is_valid_sample_count(value: str) -> bool | str: - """Validate a positive count in ``1..MAX_SAMPLES`` for a questionary text answer. - - Args: - value: Raw input. - - Returns: - ``True`` if valid, otherwise an error message string. - """ - try: - count = int(value) - except ValueError: - return "Enter a whole number." - if count < 1: - return "Enter a number greater than zero." - if count > MAX_SAMPLES: - return f"Maximum {MAX_SAMPLES}." - return True - - -def _is_int(value: str) -> bool | str: - """Validate that a questionary answer is a whole number. - - Args: - value: Raw input. - - Returns: - ``True`` if valid, otherwise an error message string. - """ - try: - int(value) - return True - except ValueError: - return "Enter a whole number." - - def _fmt_snr(snr: Optional[float]) -> str: """Format an optional SNR for the progress description. diff --git a/meshterm/ui/tx_screen.py b/meshterm/ui/tx_screen.py new file mode 100644 index 0000000..52ee964 --- /dev/null +++ b/meshterm/ui/tx_screen.py @@ -0,0 +1,463 @@ +"""The live TX-optimization screen: watch a remote node's power sweep land level by level. + +This is the interactive face of the ``tx-optimize`` tool (the scripted CLI keeps its +one-shot table and HTML chart). Where the old flow front-loaded six prompts — including +"apply the winner?" before anything had been measured — this screen follows the app's +living-screen pattern: one prompt picks the path, the sweep starts immediately, and every +measured level lands in a bar chart as it completes, phase by phase (coarse → refine → +verify), with the current best starred. The apply decision moves to *after* the evidence +is in: when the sweep finishes, a floating dialog offers the winner (Cancel left, Apply +right, Enter commits — the platform-dialog convention), and ``a`` re-offers it later if +the user first backs out of the dialog. + +Backing out mid-sweep is safe: the optimizer's unwind restores the admin node's original +transmit power, and every level already measured is persisted under the sweep's ``runs`` +row exactly as a scripted run records it. +""" + +from __future__ import annotations + +import asyncio +from typing import TYPE_CHECKING, Any, Callable, Optional + +from rich.console import Group, RenderableType +from rich.table import Table +from rich.text import Text + +from ..core.models import Contact, TxLevelResult, TxOptResult +from ..services import tx_optimizer +from .theme import snr_style +from .trace_screen import snr_bar +from .tui.render import render_lines +from .tui.screen import Screen +from .tui.spinner import Spinner + +if TYPE_CHECKING: + from ..context import AppContext + +#: Seconds between spinner frames while the sweep is measuring. +_SPINNER_INTERVAL = 0.12 + +#: Human phrasing for each optimizer phase reported through ``on_phase``. +_PHASE_LABELS = { + "coarse": "coarse sweep", + "refine": "refining around the leader", + "verify": "verifying the leader", +} + + +class TxSweepScreen(Screen): + """A full-screen live view of one remote-admin TX-power sweep. + + The screen is fed by the optimizer's callbacks (each measured level, each phase + change) and by the controller marking completion/failure/apply; it renders state and + handles keys only. Esc backs out at any time — mid-sweep that cancels the search and + restores the node's original power; after completion, ``a`` (re-)offers the apply + dialog. + """ + + floating = False + + def __init__( + self, + *, + admin_label: str, + target_label: str, + path: str, + tx_min: int, + tx_max: int, + step: int, + samples: int, + session: Any, + offer_apply: Callable[[], None], + ) -> None: + """Create the sweep screen (the controller starts the sweep alongside it). + + Args: + admin_label: Display name of the node being tuned. + target_label: Display name of the node the SNR is measured at. + path: The forced one-way path (comma-separated hashes) out to the target. + tx_min: Lowest TX power the sweep explores. + tx_max: Highest TX power the sweep explores. + step: Coarse-sweep grid spacing. + samples: Traces measured per level. + session: The running TUI session (for repaints). + offer_apply: Re-opens the apply dialog (bound by the controller); wired to + the ``a`` key once a completed sweep has a winner that wasn't applied. + """ + super().__init__() + self.title = f"tx optimize · {admin_label} → {target_label}" + self._admin_label = admin_label + self._target_label = target_label + self._path = path + self._tx_min = tx_min + self._tx_max = tx_max + self._step = step + self._samples = samples + self._session = session + self._offer_apply = offer_apply + self._spinner = Spinner() + self._phase: Optional[str] = None + self._done = 0 + self._total = 0 + self._levels: dict[int, TxLevelResult] = {} + self._best_tx: Optional[int] = None + self._result: Optional[TxOptResult] = None + self._applied = False + self._error: Optional[str] = None + + # --- controller feed -------------------------------------------------------- + + @property + def running(self) -> bool: + """Whether the sweep is still measuring.""" + return self._result is None and self._error is None + + @property + def footer_hint(self) -> str: # type: ignore[override] + """The footer keys, tracking sweep state.""" + if self.running: + return "measuring… · ↑↓ scroll · Esc stop & restore" + if self._can_apply(): + return "a apply winner · ↑↓ scroll · Esc back" + return "↑↓ scroll · Esc back" + + def tick(self) -> None: + """Advance the measuring spinner (driven by the controller's animation timer).""" + self._spinner.tick() + + def on_phase(self, phase: str) -> None: + """Record the optimizer entering a search phase (see ``tx_optimizer.PHASES``).""" + self._phase = phase + self._session.invalidate() + + def on_level(self, done: int, total: int, level: TxLevelResult) -> None: + """Record one measured level (replacing any earlier pass at the same power).""" + self._done, self._total = done, total + self._levels[level.tx_power] = level + self._best_tx = tx_optimizer.select_best(list(self._levels.values())).tx_power + self._session.invalidate() + + def complete(self, result: TxOptResult) -> None: + """Mark the sweep finished and adopt its final selection.""" + self._result = result + self._best_tx = result.best_tx + self._session.invalidate() + + def fail(self, error: str) -> None: + """Mark the sweep failed, keeping whatever levels already landed on screen.""" + self._error = error + self._session.invalidate() + + def mark_applied(self) -> None: + """Record that the winner was written to the admin node.""" + self._applied = True + self._session.invalidate() + + def _can_apply(self) -> bool: + """Whether ``a`` should offer the winner (finished, got a result, not yet set).""" + return ( + self._result is not None + and self._result.best_snr is not None + and not self._applied + ) + + # --- input -------------------------------------------------------------------- + + def handle(self, action: str, data: str = "") -> None: + """Offer the apply dialog, scroll, or dismiss.""" + if action == "text" and data.lower() == "a": + if self._can_apply(): + self._offer_apply() + elif action == "up": + self.scroll_lines(-1) + elif action == "down": + self.scroll_lines(1) + elif action == "pageup": + self.scroll_pages(-1) + elif action in ("pagedown", "space"): + self.scroll_pages(1) + elif action in ("home", "ctrl_home"): + self.scroll_to_top() + elif action in ("end", "ctrl_end"): + self.scroll_to_bottom() + elif action == "escape": + self.resolve(None) + + # --- rendering ------------------------------------------------------------------ + + def render_body(self, width: int) -> list[str]: + """Render the link header, phase line, level chart, and outcome.""" + lines = render_lines(Group(*self._sections()), width) + self._scroll_total = max(1, len(lines)) + return lines + + def _sections(self) -> list[RenderableType]: + """Assemble the screen's stacked sections for the current state.""" + sections: list[RenderableType] = [self._header(), Text(), self._phase_line()] + if self._levels: + sections += [Text(), self._levels_table()] + outcome = self._outcome() + if outcome is not None: + sections += [Text(), outcome] + return sections + + def _header(self) -> Text: + """The tuned link and the sweep's fixed parameters, label-aligned.""" + header = Text.assemble( + ("tuning ", "muted"), (self._admin_label, "brand"), + (" → ", "muted"), (self._target_label, "brand"), ("\n", ""), + ("via ", "muted"), (self._path, ""), ("\n", ""), + ("sweep ", "muted"), + (f"TX {self._tx_min}–{self._tx_max} · step {self._step} · " + f"{self._samples} trace{'s' if self._samples != 1 else ''}/level", ""), + ) + if self._result is not None and self._result.original_tx is not None: + header.append("\n") + header.append("was ", style="muted") + header.append(f"TX {self._result.original_tx}") + return header + + def _phase_line(self) -> Text: + """The live phase/progress line, or the terminal state once the sweep ends.""" + if self._error is not None: + return Text(f"✗ sweep failed: {self._error}", style="err") + if self._result is not None: + return Text("✓ sweep complete", style="ok") + line = self._spinner.text() + label = _PHASE_LABELS.get(self._phase or "", "starting…") + line.append(f" {label}", style="muted") + if self._total: + line.append(f" · level {self._done}/{self._total}", style="muted") + return line + + def _levels_table(self) -> Table: + """Every measured level as a bar chart, ascending by TX, the best starred.""" + table = Table(box=None, padding=(0, 1, 0, 0), expand=False, header_style="muted") + table.add_column("TX", justify="right", min_width=4) + table.add_column("SNR AT TARGET") + table.add_column("", justify="right") # numeric SNR + table.add_column("RELIABILITY", justify="right") + table.add_column("TRACES", justify="right", style="muted") + for level in sorted(self._levels.values(), key=lambda lv: lv.tx_power): + is_best = level.tx_power == self._best_tx + tx_cell = Text() + tx_cell.append("★ " if is_best else " ", style="ok") + tx_cell.append(str(level.tx_power), style="brand" if is_best else "") + snr = level.target_snr + if snr is None: + bar_cell: Text = Text("✗ no reply", style="err") + snr_cell = Text("—", style="muted") + else: + bar_cell = snr_bar(snr) + snr_cell = Text(f"{snr:+.1f} dB", style=snr_style(snr)) + rate = level.success_rate + rate_style = "ok" if rate >= 1.0 else ("warn" if rate > 0 else "err") + table.add_row( + tx_cell, + bar_cell, + snr_cell, + Text(f"{rate:.0%}", style=rate_style), + f"{level.successes}/{level.samples}", + ) + return table + + def _outcome(self) -> Optional[Text]: + """The completed sweep's verdict and apply status, or ``None`` while measuring.""" + result = self._result + if result is None: + return None + if result.best_snr is None: + return Text.assemble( + ("! ", "warn"), + ("no traces reached ", ""), + (self._target_label, "brand"), + (" at any level — check the path ends at the target and is reachable.", ""), + ) + snr_text = Text(f"{result.best_snr:+.1f} dB", style=snr_style(result.best_snr)) + outcome = Text.assemble( + ("best ", "muted"), (f"TX {result.best_tx}", "brand"), (" · ", "muted"), + ) + outcome.append_text(snr_text) + outcome.append(f" at {self._target_label} · ", style="muted") + outcome.append(f"{result.best_success_rate:.0%} reliable") + outcome.append("\n") + outcome.append("status ", style="muted") + if self._applied: + outcome.append(f"✓ TX {result.best_tx} set on {self._admin_label}", style="ok") + else: + restored = ( + f" (node left at TX {result.original_tx})" + if result.original_tx is not None + else "" + ) + outcome.append(f"not applied{restored} — press a to apply", style="muted") + return outcome + + +async def open_tx_optimize( + ctx: "AppContext", + *, + admin_node: Contact, + target_label: str, + path: str, +) -> dict[str, Any]: + """Run one live TX-power sweep for an already-authenticated admin node. + + The caller (the ``tx-optimize`` tool) has resolved the path and logged in to + ``admin_node``; this opens the sweep screen, drives the optimizer against it with + ``apply=False`` (the node is restored to its original power when the search ends or + is cancelled), persists every level and trace under one ``runs`` row, and raises the + apply dialog once the evidence is in. Cancelling mid-sweep (Esc) stops the search; + the optimizer's unwind restores the original power before the screen returns. + + Args: + ctx: The shared application context (must be running the interactive TUI surface). + admin_node: The contact being tuned (already logged in). + target_label: Display name of the node the SNR is measured at. + path: The forced one-way path (comma-separated hashes) ending at the target. + + Returns: + The summary dict recorded on the sweep's run row (empty if cancelled before any + level landed). + + Raises: + RuntimeError: If called outside the interactive menu (no full-screen session). + """ + from .surface import TuiUi + + if not isinstance(ctx.ui, TuiUi): # pragma: no cover - guarded by the menu-only caller + raise RuntimeError("the live TX sweep is only available in the menu") + session = ctx.ui.session + device = await ctx.device() + + tx_min = ctx.settings.tx_opt_min + tx_max = ctx.settings.tx_opt_max + step = 3 + samples = 3 + summary: dict[str, Any] = {} + + def offer_apply() -> None: + asyncio.ensure_future(apply_dialog()) + + screen = TxSweepScreen( + admin_label=admin_node.name, + target_label=target_label, + path=path, + tx_min=tx_min, + tx_max=tx_max, + step=step, + samples=samples, + session=session, + offer_apply=offer_apply, + ) + + run_id = ctx.repo.start_run( + "tx-optimize", + {"path": path, "samples": samples, "step": step, "tx_min": tx_min, "tx_max": tx_max}, + ctx.profile_name, + ) + + async def apply_dialog() -> None: + """Offer the winner in a floating dialog; write it to the node on Apply.""" + result = screen._result + if result is None or result.best_snr is None or screen._applied: + return + was = f" (currently {result.original_tx})" if result.original_tx is not None else "" + prompt = Text.assemble( + ("Set TX ", ""), (str(result.best_tx), "brand"), + (f" on {admin_node.name}?{was}", ""), + ) + # Platform-dialog convention: the safe way out left, the committing action right + # and default, so Enter applies and Esc backs out. + choice = await session.button_dialog( + prompt, + [("Cancel", "cancel"), ("Apply", "apply")], + title="Apply winner", + default=1, + footer_hint="Esc cancel · Enter apply", + ) + if choice != "apply": + return + await device.set_remote_tx_power(admin_node, result.best_tx) + ctx.log.info("set TX power %s on %s", result.best_tx, admin_node.name) + screen.mark_applied() + summary["applied"] = True + ctx.repo.finish_run(run_id, "ok", summary) + + async def sweep() -> None: + """Drive the optimizer to completion, then raise the apply dialog.""" + ticker = asyncio.ensure_future(_animate(session, screen)) + try: + result = await tx_optimizer.optimize_tx_power( + device, + target_label, + admin_node, + path, + tx_min=tx_min, + tx_max=tx_max, + coarse_step=step, + samples_per_level=samples, + apply=False, # the decision moves to the dialog, after the evidence is in + cooldown_s=ctx.settings.trace_cooldown_s, + on_level=screen.on_level, + on_phase=screen.on_phase, + persist_level=lambda lv: ctx.repo.record_tx_sample(run_id, lv), + persist_trace=lambda t: ctx.repo.record_trace(run_id, t), + ) + except asyncio.CancelledError: + ctx.repo.finish_run(run_id, "error", {"error": "cancelled"}) + raise + except Exception as exc: # noqa: BLE001 - shown on the screen, not crashed through + screen.fail(str(exc)) + ctx.repo.finish_run(run_id, "error", {"error": str(exc)}) + return + finally: + ticker.cancel() + try: + await ticker + except asyncio.CancelledError: + pass + except Exception: # noqa: BLE001 - a spinner hiccup must never break the sweep + pass + summary.update( + { + "target": result.target, + "admin_node": result.admin_node, + "path": result.path, + "best_tx": result.best_tx, + "best_snr": result.best_snr, + "best_success_rate": round(result.best_success_rate, 3), + "original_tx": result.original_tx, + "applied": False, + "levels_measured": len(result.levels), + } + ) + ctx.repo.finish_run(run_id, "ok", summary) + screen.complete(result) + await apply_dialog() + + worker = asyncio.ensure_future(sweep()) + try: + await session.run_screen(screen) + finally: + if not worker.done(): + # Esc mid-sweep: stop the search and let the optimizer's unwind restore the + # node's original power. The ring spinner covers that last device command. + worker.cancel() + async with ctx.ui.busy_overlay(): + try: + await worker + except asyncio.CancelledError: + pass + except Exception: # noqa: BLE001 - the sweep is over; nothing to surface + pass + return summary + + +async def _animate(session: Any, screen: TxSweepScreen) -> None: + """Advance the measuring spinner and repaint on a steady cadence, until cancelled.""" + while True: + await asyncio.sleep(_SPINNER_INTERVAL) + screen.tick() + session.invalidate() diff --git a/tests/test_live_screens.py b/tests/test_live_screens.py new file mode 100644 index 0000000..cbe06fa --- /dev/null +++ b/tests/test_live_screens.py @@ -0,0 +1,313 @@ +"""Unit tests for the live Trace and TX-optimize screens. + +These drive the two full-screen tools' pure logic — burst/sweep state machines, key +handling, and rendering — against fake sessions and injected runners, so they run fast +and headless (the same approach as ``test_tui``). +""" + +from __future__ import annotations + +import asyncio +from datetime import timedelta + +import pytest + +from meshterm.core.models import Hop, TraceResult, TraceStats, TxLevelResult, TxOptResult, utcnow +from meshterm.ui.trace_screen import TraceScreen, snr_bar +from meshterm.ui.tx_screen import TxSweepScreen + + +class _FakeSession: + """The one session capability the screens use directly: requesting a repaint.""" + + def __init__(self) -> None: + self.repaints = 0 + + def invalidate(self) -> None: + self.repaints += 1 + + +def _trace(*snrs: float, success: bool = True, target: str = "Alice") -> TraceResult: + """Build a trace with one hop per SNR reading.""" + hops = [Hop(index=i, node=f"{i:02x}{i:02x}", snr=snr) for i, snr in enumerate(snrs)] + return TraceResult( + target=target, success=success, hops=hops, round_trip_ms=200.0, path_hash_bytes=2 + ) + + +def _plain(lines: list[str]) -> str: + """Join rendered ANSI lines and strip nothing — assertions use substring checks.""" + return "\n".join(lines) + + +# --- snr_bar ------------------------------------------------------------------- + + +def test_snr_bar_scales_with_signal_quality() -> None: + """A stronger signal fills more of the track; None renders an empty muted track.""" + weak = snr_bar(-12.0).plain + strong = snr_bar(8.0).plain + assert weak.count("▆") < strong.count("▆") + assert len(weak) == len(strong) # the track width is constant + assert "▆" not in snr_bar(None).plain + + +def test_snr_bar_clamps_out_of_range_readings() -> None: + """Readings beyond the display range clamp to the ends instead of over/underflowing.""" + assert snr_bar(99.0).plain.count("▆") == snr_bar(10.0).plain.count("▆") + assert snr_bar(-99.0).plain.count("▆") == 1 # a heard hop always shows something + + +# --- TraceScreen ---------------------------------------------------------------- + + +def _trace_screen(burst=None, edit_path=None, previous=None) -> tuple[TraceScreen, _FakeSession]: + session = _FakeSession() + + async def default_burst(samples, path_spec, on_trace): # noqa: ANN001 + for _ in range(samples): + on_trace(_trace(5.0, 2.0)) + + async def default_edit(current): # noqa: ANN001 + return current + + screen = TraceScreen( + "Alice", + device_label="Us", + device_hash="aabb" + "00" * 30, + resolve=lambda label: label, + session=session, + burst=burst or default_burst, + edit_path=edit_path or default_edit, + previous=previous, + ) + return screen, session + + +async def test_trace_screen_burst_streams_results_into_the_log() -> None: + """A burst appends each landed trace; the log and aggregates render them.""" + screen, _ = _trace_screen() + screen.start_burst() + await asyncio.sleep(0) # let the worker run + await screen._worker + body = _plain(screen.render_body(100)) + assert "success rate" in body and "3/3" in body + assert "#3" in body # newest-first numbering + assert "Per-hop medians" in body + + +async def test_trace_screen_seeds_route_from_previous_trace() -> None: + """Before any fresh reply, the stored route shows, marked as previous.""" + old = _trace(4.0) + old.timestamp = utcnow() - timedelta(hours=3) + screen, _ = _trace_screen(previous=old) + body = _plain(screen.render_body(100)) + assert "(previous" in body + # A fresh success replaces the seeded route and drops the marker. + screen._on_trace(_trace(6.0)) + body = _plain(screen.render_body(100)) + assert "(previous" not in body + + +async def test_trace_screen_only_one_burst_at_a_time() -> None: + """Enter during a burst is a no-op; the running flag gates re-entry.""" + started = 0 + release = asyncio.Event() + + async def burst(samples, path_spec, on_trace): # noqa: ANN001 + nonlocal started + started += 1 + await release.wait() + + screen, _ = _trace_screen(burst=burst) + screen.start_burst() + assert screen._running + screen.handle("enter") # ignored while running + release.set() + await screen._worker + assert started == 1 + assert not screen._running + + +async def test_trace_screen_burst_failure_reads_inline() -> None: + """A failed burst reports its error in the log area instead of crashing the screen.""" + + async def burst(samples, path_spec, on_trace): # noqa: ANN001 + raise RuntimeError("no route") + + screen, _ = _trace_screen(burst=burst) + screen.start_burst() + await screen._worker + assert "trace failed: no route" in _plain(screen.render_body(100)) + + +async def test_trace_screen_samples_cycle_only_when_idle() -> None: + """`s` cycles the burst size through the odd sample counts, but never mid-burst.""" + screen, _ = _trace_screen() + assert screen._samples == 3 + screen.handle("text", "s") + assert screen._samples == 5 + screen._running = True + screen.handle("text", "s") + assert screen._samples == 5 # locked while a burst is in flight + + +async def test_trace_screen_path_dialog_updates_the_spec() -> None: + """`p` opens the injected path prompt; its result becomes the next burst's path.""" + asked: list[str] = [] + + async def edit(current: str): # noqa: ANN001 + asked.append(current) + return "3d,f2" + + screen, _ = _trace_screen(edit_path=edit) + screen.handle("text", "p") + await asyncio.sleep(0) + assert asked == [""] + assert screen._path_spec == "3d,f2" + assert "3d,f2" in _plain(screen.render_body(100)) + + +async def test_trace_screen_escape_cancels_the_inflight_burst() -> None: + """Esc resolves the screen and cancels a burst that is still measuring.""" + release = asyncio.Event() + + async def burst(samples, path_spec, on_trace): # noqa: ANN001 + await release.wait() + + screen, _ = _trace_screen(burst=burst) + screen.start_burst() + screen.future = asyncio.get_running_loop().create_future() + screen.handle("escape") + assert screen.future.result() is None + with pytest.raises(asyncio.CancelledError): + await screen._worker + + +# --- TxSweepScreen ----------------------------------------------------------------- + + +def _level(tx: int, snr: float | None, successes: int = 2, samples: int = 2) -> TxLevelResult: + return TxLevelResult( + tx_power=tx, + samples=samples, + successes=successes, + target_snr=snr, + score=snr if snr is not None else float("-inf"), + stats=TraceStats.from_traces("Alice", []), + ) + + +def _sweep_screen(offer_apply=None) -> tuple[TxSweepScreen, _FakeSession]: + session = _FakeSession() + screen = TxSweepScreen( + admin_label="Repeater", + target_label="Alice", + path="a1b2,d4e5", + tx_min=12, + tx_max=28, + step=3, + samples=3, + session=session, + offer_apply=offer_apply or (lambda: None), + ) + return screen, session + + +def _result(best_tx: int = 19, best_snr: float | None = 8.8) -> TxOptResult: + return TxOptResult( + target="Alice", + admin_node="Repeater", + path="a1b2,d4e5", + original_tx=20, + best_tx=best_tx, + best_snr=best_snr, + best_success_rate=1.0, + applied=False, + levels=[_level(best_tx, best_snr)], + ) + + +def test_sweep_screen_stars_the_running_best() -> None: + """Each landed level renders ascending by TX with the current best starred.""" + screen, _ = _sweep_screen() + screen.on_phase("coarse") + screen.on_level(1, 7, _level(12, -2.0)) + screen.on_level(2, 7, _level(18, 7.5)) + body = _plain(screen.render_body(100)) + assert "coarse sweep" in body and "level 2/7" in body + starred = next(line for line in body.splitlines() if "★" in line) + assert "18" in starred + + +def test_sweep_screen_shows_failed_levels_distinctly() -> None: + """A level nothing got through at reads as a no-reply row, not an empty bar.""" + screen, _ = _sweep_screen() + screen.on_level(1, 7, _level(28, None, successes=0)) + assert "✗ no reply" in _plain(screen.render_body(100)) + + +def test_sweep_screen_completion_offers_the_winner() -> None: + """After completion the outcome shows and `a` re-offers the apply dialog.""" + offered: list[bool] = [] + screen, _ = _sweep_screen(offer_apply=lambda: offered.append(True)) + screen.handle("text", "a") # mid-sweep: nothing to offer yet + assert offered == [] + screen.complete(_result()) + body = _plain(screen.render_body(100)) + assert "sweep complete" in body and "TX 19" in body and "press a to apply" in body + assert "a apply winner" in screen.footer_hint + screen.handle("text", "a") + assert offered == [True] + + +def test_sweep_screen_apply_updates_status_and_retires_the_key() -> None: + """Marking the winner applied flips the status line and stops offering `a`.""" + offered: list[bool] = [] + screen, _ = _sweep_screen(offer_apply=lambda: offered.append(True)) + screen.complete(_result()) + screen.mark_applied() + assert "✓ TX 19 set on Repeater" in _plain(screen.render_body(100)) + assert "a apply winner" not in screen.footer_hint + screen.handle("text", "a") + assert offered == [] + + +def test_sweep_screen_no_result_never_offers_apply() -> None: + """A sweep where nothing got through warns and keeps the apply key retired.""" + screen, _ = _sweep_screen() + screen.complete(_result(best_snr=None)) + body = _plain(screen.render_body(100)) + assert "no traces reached" in body + assert "a apply winner" not in screen.footer_hint + + +def test_sweep_screen_failure_keeps_measured_levels_on_screen() -> None: + """A mid-sweep error is reported while the levels already measured stay visible.""" + screen, _ = _sweep_screen() + screen.on_level(1, 7, _level(12, -2.0)) + screen.fail("link lost") + body = _plain(screen.render_body(100)) + assert "sweep failed: link lost" in body + assert "-2.0" in body + + +def test_sweep_screen_escape_resolves() -> None: + """Esc resolves the screen (the controller then cancels and restores).""" + screen, _ = _sweep_screen() + + class _Fut: + def __init__(self) -> None: + self.value = None + self._done = False + + def done(self) -> bool: + return self._done + + def set_result(self, value) -> None: # noqa: ANN001 + self.value = value + self._done = True + + screen.future = _Fut() + screen.handle("escape") + assert screen.future.done() diff --git a/tests/test_skeleton.py b/tests/test_skeleton.py index 0577e1f..c053ef9 100644 --- a/tests/test_skeleton.py +++ b/tests/test_skeleton.py @@ -342,6 +342,29 @@ async def test_tx_optimizer_no_apply_restores_original(tmp_path: Path) -> None: assert await device.get_remote_tx_power(admin) == 15 # restored, not the winner +async def test_tx_optimizer_reports_phases_in_order() -> None: + """on_phase announces each enabled search stage, in coarse → refine → verify order.""" + from meshterm.services import tx_optimizer + + device, admin, target, path = await _setup_link() + phases: list[str] = [] + + await tx_optimizer.optimize_tx_power( + device, target.name, admin, path, + samples_per_level=2, coarse_step=6, apply=False, cooldown_s=0, + on_phase=phases.append, + ) + assert phases == list(tx_optimizer.PHASES) + + phases.clear() + await tx_optimizer.optimize_tx_power( + device, target.name, admin, path, + samples_per_level=2, coarse_step=6, refine=False, verify=False, + apply=False, cooldown_s=0, on_phase=phases.append, + ) + assert phases == ["coarse"] # disabled stages are never announced + + def test_select_best_prefers_reliability_then_lowest_power() -> None: """A 100%-reliable level beats a flakier higher-SNR one; ties go to lower TX.""" from meshterm.core.models import TraceStats, TxLevelResult