# SPDX-License-Identifier: Apache-2.0 """The live TX-optimization screen: select the link, arm the sweep, and watch the levels arrive. This is the interactive face of the ``tx-optimize`` tool. The scripted CLI keeps its one-shot table and HTML chart. The screen is built again on the armed-but-idle pattern of the trace screen. One dialog with steps in the tool selects the *link*. First it asks for the admin node whose transmit power the sweep tunes, then for the target whose reception the sweep optimizes. The screen then opens idle over the main menu. The route, the sweep parameters, and the login state are across the top. A list of actions controls everything, and nothing transmits until the user commits Sweep. * **Route** opens the path composer of the trace screen, which goes hop by hop and is pinned at the tuned node. The user composes how the measurement reaches that node, and each step is suggested from the links that MeshTerm observed. The code adds the target hop automatically. The default is the direct shot. * **Range / Step / Samples** float small dialogs that adjust the sweep: the TX range, the spacing of the coarse grid, and the traces that are measured for each level. The Sweep row always shows the resulting worst-case number of transmissions. Thus the cost of a commit is on the screen before it occurs (each run is paced, like the sampling of the trace tool). * **Sweep** first logs in, if it must. The password prompt floats only if no password is remembered, and MeshTerm remembers a password that works for the next time. Then it runs the optimizer under a floating in-flight dialog that has Abort. The phases are coarse, refine, and verify. Each measured level goes into the bar chart behind the dialog, and the current best level has a star. If the user aborts in the middle of the sweep, the code restores the original power of the node. * **Apply winner** is added to the actions when a sweep has found a winner (the highlight moves to it). It is the same offer as the dialog after the sweep, for the case that the user first declined it. Each sweep opens one ``runs`` row and stores its levels and traces in the same way as a scripted run records them. A finished sweep leaves the screen armed. The user can adjust the range and sweep again. A new sweep is a new measurement, so the chart clears when it starts. """ from __future__ import annotations import asyncio import re from collections.abc import Callable from typing import TYPE_CHECKING, Any from rich.console import Group, RenderableType from rich.table import Table from rich.text import Text from ..core.connection import REMOTE_TX_MAX, REMOTE_TX_MIN from ..core.models import Contact, LoginResult, TraceResult, TxLevelResult, TxOptResult from ..services import trace_runner, tx_optimizer from ..services.topology import build_topology, render_custom_spec from .menus import icon_lane, icon_mark from .pathline import SELF_GLYPH, PathHop, PathLine, cut_to, path_line from .theme import name_style, snr_style from .trace_screen import TracingDialog, collapse_trace_width, snr_bar from .tui.render import render_lines, render_to_ansi from .tui.screen import Screen from .tui.spinner import Spinner, spinner_interval from .widgets import NodeResolver if TYPE_CHECKING: from ..context import AppContext #: The words for the user for each optimizer phase that is reported through ``on_phase``. _PHASE_LABELS = { "coarse": "coarse sweep", "refine": "refining around the leader", "verify": "verifying the leader", } #: The coarse grid spacings that the Step dialog offers. 1 measures each level, so no refine #: pass is necessary. A wider grid needs fewer transmissions, but it needs a refine pass #: later. STEP_CHOICES = (1, 2, 3, 4, 5) #: The numbers of samples for each level that the Samples dialog offers (the ladder of the #: trace tool). SAMPLE_CHOICES = (1, 2, 3, 5, 8) #: Each mark that the action rows of the sweep start with. The list is there to *measure* #: the icon column, so that each label starts in the same place. Also, a platform that #: draws no icon lane (refer to :func:`~meshterm.ui.menus.command_icon`) makes the column #: zero cells wide. _ACTION_ICONS = ("✎", "⚙", "#", "▶", "★") def _icon_lane() -> int: """The icon column of the action rows, in cells (the marks of this screen, measured once).""" return icon_lane(_ACTION_ICONS) def _icon(text: Text, icon: str, style: str) -> None: """Append the mark of an action row in the icon column, with its trailing space.""" text.append_text(icon_mark(icon, style, _icon_lane())) class TxSweepScreen(Screen): """A full-screen session of a TX-power sweep. It is armed, but idle until the user starts it. ↑/↓ move the highlight over the action rows, and Enter commits the selected row. The highlight opens on Sweep, so a plain Enter sweeps. PgUp/PgDn/Home/End scroll the body. Esc leaves the screen and cancels any sweep that is in flight (the unwind of the optimizer restores the original power of the node). The screen renders the state and sends the keys to the correct code. The owning session (refer to :func:`open_tx_optimize`) gives it the flows: the route composer, the parameter dialogs, and the sweep runner. The session sends the measurements back through :meth:`on_phase`, :meth:`on_level`, :meth:`complete`, and :meth:`fail`. """ floating = False def __init__( self, *, admin_label: str, target_label: str, device_label: str, device_hash: str | None, resolve: NodeResolver, session: Any, tx_min: int, tx_max: int, admin_key: str | None = None, target_key: str | None = None, step: int = 3, samples: int = 3, run_sweep: Callable[[], None] = lambda: None, apply_winner: Callable[[], None] = lambda: None, compose_route: Callable[[], None] = lambda: None, pick_range: Callable[[], None] = lambda: None, pick_step: Callable[[], None] = lambda: None, pick_samples: Callable[[], None] = lambda: None, ) -> None: """Create the sweep screen. Nothing transmits until the user commits Sweep. Args: admin_label: The display name of the node that the sweep tunes. target_label: The display name of the node at which the SNR is measured. device_label: The name of our node, which starts the route line. device_hash: The public key of our node, so that the ends of the route have a hash. resolve: Maps the raw hash of a hop to the name of a contact, when it is known. admin_key: The key or hash of the tuned node, so that its name has its own hue. target_key: The key or hash of the target, for the same reason. session: The running TUI session (for paints). tx_min: The initial low end of the sweep range. tx_max: The initial high end of the sweep range. step: The initial spacing of the coarse grid. samples: The initial number of traces that are measured for each level. run_sweep: Starts one sweep. The owner guards it, and it does nothing while a sweep is in flight. apply_winner: Offers the apply dialog again for the winner of a completed sweep. compose_route: Opens the route composer flow over this screen. pick_range: Floats the TX-range dialog. pick_step: Floats the grid-spacing dialog. pick_samples: Floats the dialog for the number of samples for each level. """ super().__init__() self.title = f"TX optimize — {admin_label} → {target_label}" self._admin_label = admin_label self._target_label = target_label self._admin_key = admin_key self._target_key = target_key self._device_label = device_label self._device_hash = device_hash self._resolve = resolve self._session = session self._run_sweep = run_sweep self._apply_winner = apply_winner self._compose_route = compose_route self._pick_range = pick_range self._pick_step = pick_step self._pick_samples = pick_samples #: The sweep parameters. The dialogs of the owner change them between sweeps. self.tx_min = tx_min self.tx_max = tx_max self.step = step self.samples = samples #: The hops that the measurement crosses *before* the tuned node (output of the #: composer, at the spec width of the session). Empty means the direct shot. self.route_hops: list[str] = [] #: One line that describes the login state (remembered, will ask, or logged in). The #: owner maintains it, and the header shows it. self.login_text = "" self.running = False self._dialog_open = False self._phase: str | None = None self._done = 0 self._total = 0 self._levels: dict[int, TxLevelResult] = {} self._best_tx: int | None = None self._result: TxOptResult | None = None self._applied = False self._error: str | None = None self._worker: asyncio.Task | None = None self._index = self._actions.index("sweep") self._pin_cursor = False # pin the view only while the user uses ↑/↓ # --- state ------------------------------------------------------------------- @property def _actions(self) -> tuple[str, ...]: """The action rows in display order. Apply is added when there is a winner.""" rows = ["route", "range", "step", "samples", "sweep"] if self._can_apply(): rows.append("apply") return tuple(rows) @property def footer_hint(self) -> str: # type: ignore[override] """The footer keys. They change when a sweep is in flight.""" if self.running: return "sweeping… · PgUp/PgDn scroll · Esc back" return "↑↓ actions · Enter run · PgUp/PgDn scroll · Esc back" def _can_apply(self) -> bool: """Whether Apply is offered: the sweep finished, it has a winner, and it is not applied.""" return self._result is not None and self._result.best_snr is not None and not self._applied def estimated_traces(self) -> int: """The worst-case number of transmissions that one commit of Sweep can run. The number is the levels of the coarse grid, plus up to ``2 · (step − 1)`` integer fill-ins of the refine pass around the winner, plus the verify measurement again. Each is measured with the chosen number of samples for each level. The real number is usually lower, because the refine pass stops at the ends of the range and skips levels that are already measured. """ coarse = len(tx_optimizer.coarse_levels(self.tx_min, self.tx_max, self.step)) refine = max(0, 2 * (self.step - 1)) return (coarse + refine + 1) * self.samples # --- owner feed ---------------------------------------------------------------- def sweep_started(self, worker: asyncio.Task) -> None: """Take a sweep that just started. It is a new measurement, so clear the old evidence.""" self.running = True self._worker = worker self._phase = None self._done = self._total = 0 self._levels.clear() self._best_tx = None self._result = None self._applied = False self._error = None self._session.invalidate() def sweep_finished(self) -> None: """Mark that the sweep is not in flight any more, however it ended.""" self.running = False self._worker = None self._session.invalidate() def on_phase(self, phase: str) -> None: """Record that the optimizer enters a search phase (refer to ``tx_optimizer.PHASES``).""" self._phase = phase self._session.invalidate() def on_level(self, done: int, total: int, level: TxLevelResult) -> None: """Record one measured level. It replaces an 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: """Take the selection of the finished sweep and move the highlight to Apply.""" self._result = result self._best_tx = result.best_tx if self._can_apply(): self._index = self._actions.index("apply") self._session.invalidate() def fail(self, error: str) -> None: """Mark the sweep as failed. The levels that already arrived stay on the screen.""" self._error = error self._session.invalidate() def mark_applied(self) -> None: """Record that the code wrote the winner to the admin node.""" self._applied = True self._index = min(self._index, len(self._actions) - 1) self._session.invalidate() def phase_label(self) -> str: """The status line of the in-flight dialog: the phase and the progress of the levels.""" label = _PHASE_LABELS.get(self._phase or "", "starting…") if self._total: return f"{label} · level {self._done}/{self._total}" return label def cancel(self) -> None: """Cancel any sweep that is in flight. The unwind of the optimizer restores the power.""" if self._worker is not None and not self._worker.done(): self._worker.cancel() # --- input ------------------------------------------------------------------- def handle(self, action: str, data: str = "") -> None: """Move the highlight of the actions, commit the selected action, scroll, or leave.""" if action == "enter": self._commit_action() elif action == "up": # Both ends clamp and do not wrap. This is the rule of the whole app for the # highlight of a row. A highlight that jumps from one end to the other takes # the page of results with it. self._index = max(0, self._index - 1) self._pin_cursor = True elif action == "down": self._index = min(len(self._actions) - 1, self._index + 1) self._pin_cursor = True elif action == "pageup": self._pin_cursor = False self.scroll_pages(-1) elif action in ("pagedown", "space"): self._pin_cursor = False self.scroll_pages(1) elif action in ("home", "ctrl_home"): self._pin_cursor = False self.scroll_to_top() elif action in ("end", "ctrl_end"): self._pin_cursor = False self.scroll_to_bottom() elif action == "escape": self.cancel() self.resolve(None) def _commit_action(self) -> None: """Run the action row under the highlight. The flows guard against a second entry.""" actions = self._actions key = actions[min(self._index, len(actions) - 1)] if key == "sweep": if not self.running: self._run_sweep() elif key == "apply": if self._can_apply(): self._apply_winner() elif key == "route": self._open_flow(self._compose_route) elif key == "range": self._open_flow(self._pick_range) elif key == "step": self._open_flow(self._pick_step) elif key == "samples": self._open_flow(self._pick_samples) def _open_flow(self, flow: Callable[[], None]) -> None: """Float a parameter flow over the screen. Only one at a time, and never during a sweep.""" if self._dialog_open or self.running: return self._dialog_open = True async def run() -> None: try: await flow() # type: ignore[misc] # the flows of the owner are async closures finally: self._dialog_open = False self._session.invalidate() asyncio.ensure_future(run()) # --- rendering ----------------------------------------------------------------- def render_body(self, width: int) -> list[str]: """Render the header, the list of actions, the chart of levels, and the outcome.""" lines = self._header_lines(width) lines.append("") self._cursor: int | None = None actions = self._actions self._index = min(self._index, len(actions) - 1) for i, key in enumerate(actions): selected = i == self._index text = self._action_text(key, selected) text.no_wrap = True # The route row has a path line, so the code cuts it and does not truncate it. # A chip that goes past the end of the row cracks. Each other row keeps the # ellipsis. text = cut_to(text, width) text.no_wrap = True if selected: self._cursor = len(lines) lines.append(render_to_ansi(text, width)) if key == "samples": lines.append("") # separate the sweep group from the parameters tail: list[RenderableType] = [] if self.running or self._levels: tail += [Text(), Text("Levels", style="accent"), self._levels_table()] outcome = self._outcome() if outcome is not None: tail += [Text(), outcome] lines.extend(render_lines(Group(*tail), width)) self._scroll_total = max(1, len(lines)) return lines def cursor_line(self) -> int | None: """The line of the highlighted action row while ↑/↓ are in use. Else the page scrolls.""" return getattr(self, "_cursor", None) if self._pin_cursor else None def _action_text(self, key: str, selected: bool) -> Text: """One action row: the pointer, the glyph, and the label with the current value.""" text = Text("❯ " if selected else " ", style="cursor" if selected else "") if key == "route": _icon(text, "✎", "brand") if self.route_hops: # Only the relays that the user composed. The sweep always starts from us and # always arrives at the repeater that it tunes, and the user did not select # either of them here as a hop. Thus both ends are drawn open. The chevrons # say that the route continues past them. This is exactly what ``via`` says # in the word before them. text.append("Route — via ") text.append_text( path_line( list(self.route_hops), self._resolve, from_origin=False, to_destination=False, ).text() ) else: text.append(f"Route — direct to {self._admin_label}") elif key == "range": _icon(text, "⚙", "accent") text.append(f"Range — TX {self.tx_min}–{self.tx_max}") elif key == "step": _icon(text, "⚙", "accent") text.append( f"Step — every {_nth(self.step)} level, then refine" if self.step > 1 else "Step — every level (no refine needed)" ) elif key == "samples": _icon(text, "#", "accent") text.append( f"Samples — {self.samples} trace{'s' if self.samples != 1 else ''} per level" ) elif key == "sweep": _icon(text, "▶", "ok") text.append(f"Sweep — up to {self.estimated_traces()} paced transmissions") else: # apply: it is there only when there is a winner to set _icon(text, "★", "ok") best = self._result.best_tx if self._result is not None else "?" text.append(f"Apply winner — set TX {best} on {self._admin_label}") if selected: text.style = "cursor" return text def _header_lines(self, width: int) -> list[str]: """The header lanes: tuned link, measured route, sweep range, and login state. The route renders through the only path widget. It wraps at the boundaries of hops, under its own value column (the hanging-indent rule), and does not fold back to column zero. The tuned link above it has the same key hues as the route. """ tuning = Text.assemble( ("tuning ", "muted"), (self._admin_label, name_style(self._admin_label, self._admin_key)), (" → ", "muted"), (self._target_label, name_style(self._target_label, self._target_key)), ) lines = [render_to_ansi(tuning, width, no_wrap=True)] route = self._route_line().wrapped(width, indent=9) first = Text("route ", style="muted") first.append_text(route[0]) lines.append(render_to_ansi(first, width, no_wrap=True)) lines.extend(render_to_ansi(cont, width, no_wrap=True) for cont in route[1:]) rest = Text("sweep ", style="muted") rest.append( f"TX {self.tx_min}–{self.tx_max} · step {self.step} · " f"{self.samples} trace{'s' if self.samples != 1 else ''}/level" ) if self.login_text: rest.append("\n") rest.append("login ", style="muted") rest.append(self.login_text) if self._result is not None and self._result.original_tx is not None: rest.append("\n") rest.append("was ", style="muted") rest.append(f"TX {self._result.original_tx}") lines.extend(render_lines(rest, width)) return lines def _route_line(self) -> PathLine: """The walk of one measurement: out through the tuned link, and back as a mirror image. The outbound leg is us, the composed hops, the tuned node, and the target. The code draws it in full colour (each name in its own key hue). The return is the outbound leg mirrored back. It is the trace boomerang that the optimizer really sends. The code dims it, and it means "the user does not compose this". Both our ends use the ``★`` of the whole app and not our name. Each measurement that this screen makes leaves us and comes back to us. If the line spelled out our name two times, the lane would lose cells that the hops being tuned need. The trace route lane and the trophy card make the same choice. """ hops: list[PathHop] = [PathHop(SELF_GLYPH, you=True)] for hop in self.route_hops: hops.append(PathHop(self._hop_name(hop), key=hop)) hops.append(PathHop(self._admin_label, key=self._admin_key)) hops.append(PathHop(self._target_label, key=self._target_key)) hops.append(PathHop(self._admin_label, key=self._admin_key, dim=True)) for hop in reversed(self.route_hops): hops.append(PathHop(self._hop_name(hop), key=hop, dim=True)) hops.append(PathHop(SELF_GLYPH, you=True, dim=True)) return PathLine(hops) def _hop_name(self, hop: str) -> str: """The name of a route hop when it is known, otherwise its raw hash.""" named = self._resolve(hop) return named if named else hop def _levels_table(self) -> Table: """Each measured level as a bar chart, in ascending order of TX. The best has a star.""" 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) -> Text | None: """The verdict and the apply status of the completed sweep, or the error of the sweep.""" if self._error is not None: return Text(f"✗ sweep failed: {self._error}", style="err") 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 route, then sweep again.", ""), ) outcome = Text.assemble( ("best ", "muted"), (f"TX {result.best_tx}", "brand"), (" · ", "muted"), ) outcome.append(f"{result.best_snr:+.1f} dB", style=snr_style(result.best_snr)) 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" — the node keeps TX {result.original_tx}" if result.original_tx is not None else "" ) outcome.append(f"not applied{restored}", style="muted") return outcome def _nth(n: int) -> str: """Give the ordinal for the Step action row: ``2 → "2nd"``, ``3 → "3rd"``, and so on.""" suffix = {1: "st", 2: "nd", 3: "rd"}.get(n if n < 20 else n % 10, "th") return f"{n}{suffix}" def parse_tx_range(text: str) -> tuple[int, int] | None: """Parse a typed TX range, such as ``"12-28"`` or ``"12 28"``, into ``(low, high)``. The function limits the values to the range that the remote firmware can represent. It returns ``None`` if the text does not have exactly two numbers in order. """ numbers = re.findall(r"\d+", text) if len(numbers) != 2: return None lo, hi = int(numbers[0]), int(numbers[1]) if lo > hi: return None return max(REMOTE_TX_MIN, lo), min(REMOTE_TX_MAX, hi) async def open_tx_optimize( ctx: AppContext, *, admin_node: Contact, target_label: str, target_hash: str, ) -> dict[str, Any]: """Run the live TX-optimization session for one tuned link. The dialog of the tool has selected the link and is already gone. This function connects the armed-idle screen to the radio, to the topology evidence, and to the database. It runs the screen until the user leaves it, and Esc goes to the main menu, because the sweep is the whole visit. The login occurs in the first commit of Sweep. If a password is remembered, the login is silent. Otherwise, one floating prompt asks for the password. MeshTerm remembers the password if the login works, and forgets it if the node rejects it (the convention of the trace composer). Each sweep opens its own ``runs`` row and records each level and trace under it, in the same way as a scripted run does. Args: ctx: The shared application context. It must run the interactive menu. admin_node: The contact that the sweep tunes (the hop before the target). target_label: The display name of the node at which the SNR is measured. target_hash: The hex hash of the target (the full key, or the prefix that the user typed). Returns: The recorded summary of the last sweep (empty if the user left before a sweep ran). Raises: RuntimeError: If the caller calls it outside the interactive menu (there is no full-screen session). """ from .path_composer import AUTO_SPEC, PathComposerScreen from .surface import TuiUi from .tui import CANCEL, Choice, SelectScreen if not isinstance(ctx.ui, TuiUi): # pragma: no cover - the caller is only in the menu raise RuntimeError("the live TX sweep is only available in the menu") session = ctx.ui.session device = await ctx.device() # The contacts, the self-info, and the routing width come from the session cache (refer # to DeviceState). The code still keeps the live ``device`` for the transmissions of # the sweep itself below. contacts = await ctx.devstate.contacts() resolve = trace_runner.make_node_resolver(contacts) self_info = await ctx.devstate.self_info() device_label = str(self_info.get("name") or "this node") device_hash = str(self_info.get("public_key") or "") or None try: width_bytes = collapse_trace_width(int(await ctx.devstate.path_hash_mode())) except Exception: # noqa: BLE001 - an optional read. The default of 1 byte always works width_bytes = 1 admin_hash = (admin_node.public_key or admin_node.key_prefix).lower().removeprefix("0x") target_hex = target_hash.lower().removeprefix("0x") logged_in = False summary: dict[str, Any] = {} def login_text() -> str: if logged_in: return f"✓ logged in to {admin_node.name}" if ctx.admin_store.get(admin_node) is not None: return "password remembered — logs in at the first sweep" return "no saved password — asks at the first sweep" screen = TxSweepScreen( admin_label=admin_node.name, target_label=target_label, device_label=device_label, device_hash=device_hash, resolve=resolve, session=session, tx_min=ctx.preferences.tx_opt_min, tx_max=ctx.preferences.tx_opt_max, admin_key=admin_hash, target_key=target_hex, ) screen.login_text = login_text() def spec() -> str: """The one-way wire spec of the next sweep: composed hops, tuned node, and target.""" return render_custom_spec((*screen.route_hops, admin_hash, target_hex), width_bytes) # --- parameter flows (floated over the screen) --------------------------------- async def compose_route() -> None: """Compose how the measurement reaches the tuned node, which is pinned as the target.""" topo = build_topology( self_id=device_hash or "local", contacts=contacts, trace_paths=ctx.repo.trace_paths(), packet_paths=ctx.repo.packet_paths(), neighbour_links=ctx.repo.neighbour_links(), ) admin_id = topo.canonical(admin_hash) or admin_hash[:12] composer = PathComposerScreen( device_label=device_label, device_hash=device_hash, topology=topo, width_bytes=width_bytes, target_id=admin_id, target_hash=admin_hash, target_label=admin_node.name, hops=[topo.canonical(h) or h for h in screen.route_hops], resolve=resolve, # name a hop that the topology left ambiguous, as we do ) result = await session.run_screen(composer) if result is CANCEL or not isinstance(result, str): return if result == AUTO_SPEC: screen.route_hops = [] # routing by the device has no meaning here: direct shot return # The composer gives the symmetric boomerang around the tuned node. Keep only the # outbound hops before it. The code adds the target leg. tokens = [t for t in result.split(",") if t] if tokens and len(tokens) % 2 == 1 and tokens == tokens[::-1]: screen.route_hops = tokens[: len(tokens) // 2] else: screen.route_hops = tokens[:-1] if tokens else [] async def pick_range() -> None: """Select the TX range of the sweep: the default, the full range, or a typed range.""" preferred_lo, preferred_hi = ctx.preferences.tx_opt_min, ctx.preferences.tx_opt_max items = [ Choice( title=f"TX {preferred_lo}–{preferred_hi} — your configured default", value=(preferred_lo, preferred_hi), ), Choice( title=f"TX {REMOTE_TX_MIN}–{REMOTE_TX_MAX} — the full remote range", value=(REMOTE_TX_MIN, REMOTE_TX_MAX), ), Choice(title="Custom… — type a window", value="custom"), ] picked = await session.run_screen( SelectScreen( "Sweep range", items, prompt="The TX window the sweep explores:", default=(screen.tx_min, screen.tx_max), footer_hint="↑↓ move · Enter set · Esc keep", filterable=False, ) ) if picked is CANCEL or picked is None: return if picked == "custom": typed = await session.text( "Sweep range", prompt=f"Low–high, within {REMOTE_TX_MIN}–{REMOTE_TX_MAX}:", default=f"{screen.tx_min}-{screen.tx_max}", validate=lambda t: ( parse_tx_range(t) is not None or "Enter two numbers, low then high (e.g. 14-24)." ), ) if not typed: return parsed = parse_tx_range(typed) if parsed is None: return picked = parsed screen.tx_min, screen.tx_max = picked async def pick_step() -> None: """Select the coarse grid spacing. 1 measures each level, so there is nothing to refine.""" items = [ Choice( title=f"{n}" + (" — every level up front" if n == 1 else f" — every {_nth(n)} level"), value=n, ) for n in STEP_CHOICES ] picked = await session.run_screen( SelectScreen( "Coarse step", items, prompt="The coarse sweep measures the window at this spacing, then refines.", default=screen.step, footer_hint="↑↓ move · Enter set · Esc keep", filterable=False, ) ) if picked is not CANCEL and picked is not None: screen.step = int(picked) async def pick_samples() -> None: """Select the number of traces for each level. They are paced, like the trace tool.""" pace = ctx.preferences.trace_cooldown_s items = [ Choice( title=f"{n} trace{'s' if n > 1 else ' '}" + (" — fast but noisy" if n == 1 else ""), value=n, ) for n in SAMPLE_CHOICES ] picked = await session.run_screen( SelectScreen( "Samples per level", items, prompt=f"Each TX level is measured with this many traces, {pace:g} s apart.", default=screen.samples, footer_hint="↑↓ move · Enter set · Esc keep", filterable=False, ) ) if picked is not CANCEL and picked is not None: screen.samples = int(picked) # --- login, sweep, apply -------------------------------------------------------- async def ensure_login() -> bool: """Log in to the tuned node one time for each session. Ask for a password only if necessary. This runs *before* the in-flight dialog opens. Thus the password prompt floats over the idle screen, and the login command itself is behind the skeleton card. This is the same pattern as each other device call before a transmission. """ nonlocal logged_in if logged_in: return True password = ctx.admin_store.get(admin_node) if password is None: password = await session.text( f"Admin password for {admin_node.name}", prompt="The repeater ignores tuning commands without an admin login.", password=True, ) if not password: return False async with ctx.ui.busy_overlay(): outcome = await device.admin_login(admin_node, password) ctx.admin_store.record(admin_node, password, outcome) if outcome is LoginResult.REFUSED: screen.login_text = login_text() await session.message_dialog( Text( f"{admin_node.name!r} rejected the admin login (wrong password?). " "The saved password was cleared; sweep again to enter a new one.", style="err", ), title="Admin login", ) return False if not outcome: # No reply is not a denial. The password stays for the next attempt. screen.login_text = login_text() await session.message_dialog( Text( f"No reply from {admin_node.name} — it may be out of reach, asleep, " "or busy. The saved password was kept; sweep again when it answers.", style="warn", ), title="Admin login", ) return False logged_in = True screen.login_text = login_text() return True async def apply_dialog() -> None: """Offer the winner in a floating dialog. If the user selects Apply, write it.""" 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}", ""), ) # The convention of platform dialogs: the safe way out is on the left, and the # committing action is on the right and is the default. Thus Enter applies, and Esc # backs out. choice = await session.button_dialog( prompt, [("Cancel", "cancel"), ("Apply", "apply")], title="Apply winner", default=1, # The default hint of the dialog: Enter commits the button that is highlighted, # and ←→ moves the highlight (refer to the quit confirm in ui/menu.py). ) if choice != "apply": return async with ctx.ui.busy_overlay(): 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 async def sweep() -> None: """One commit of Sweep: log in if it is necessary, then run the optimizer under a dialog.""" run_id: int | None = None try: if not await ensure_login(): return except asyncio.CancelledError: raise except Exception as exc: # noqa: BLE001 - the screen shows the error. It is not a crash screen.fail(str(exc)) return finally: if not logged_in: screen.sweep_finished() spinner = Spinner() flight = TracingDialog( f"Sweeping — {admin_node.name} → {target_label}", spinner=spinner, on_abort=screen.cancel, ) session.push(flight) ticker = asyncio.ensure_future(_animate(session, spinner, flight, screen)) try: path = spec() run_id = ctx.repo.start_run( "tx-optimize", { "path": path, "samples": screen.samples, "step": screen.step, "tx_min": screen.tx_min, "tx_max": screen.tx_max, }, ctx.profile_name, ) def on_trace(trace: TraceResult) -> None: flight.last = trace assert run_id is not None ctx.repo.record_trace(run_id, trace) result = await tx_optimizer.optimize_tx_power( device, target_label, admin_node, path, tx_min=screen.tx_min, tx_max=screen.tx_max, coarse_step=screen.step, samples_per_level=screen.samples, apply=False, # the dialog makes the decision, after the evidence is complete cooldown_s=ctx.preferences.trace_cooldown_s, snr_tolerance=ctx.preferences.tx_snr_tolerance_db, on_level=screen.on_level, on_phase=screen.on_phase, persist_level=lambda lv: ctx.repo.record_tx_sample(run_id, lv), persist_trace=on_trace, ) except asyncio.CancelledError: if run_id is not None: ctx.repo.finish_run(run_id, "error", {"error": "cancelled"}) raise except Exception as exc: # noqa: BLE001 - the screen shows the error. It is not a crash screen.fail(str(exc)) if run_id is not None: 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 short fault of the spinner must never break the sweep pass session.pop(flight) screen.sweep_finished() summary.clear() 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() def run_sweep() -> None: if screen.running: return screen.sweep_started(asyncio.ensure_future(sweep())) def apply_winner() -> None: asyncio.ensure_future(apply_dialog()) screen._run_sweep = run_sweep screen._apply_winner = apply_winner screen._compose_route = compose_route screen._pick_range = pick_range screen._pick_step = pick_step screen._pick_samples = pick_samples try: await session.run_screen(screen) finally: worker = screen._worker if worker is not None and not worker.done(): # Esc in the middle of a sweep: stop the search and let the unwind of the # optimizer restore the original power of the node. The skeleton card 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, so there is nothing to show pass return summary async def _animate( session: Any, spinner: Spinner, flight: TracingDialog, screen: TxSweepScreen ) -> None: """Advance the spinner and the status of the in-flight dialog at a steady rate.""" while True: await asyncio.sleep(spinner_interval()) spinner.tick() flight.status = screen.phase_label() session.invalidate()