# SPDX-License-Identifier: Apache-2.0 """The repeater-admin screens: edit the settings of a remote node over the mesh, in an editor. These screens are the interactive side of the ``repeater-admin`` tool. First the user selects a node (in the shared node picker, which shows the nodes with credentials first) and logs in (with a remembered password, or else with a one-time prompt). Then an editor opens. Its shape is the same as the Device config screen, on purpose: the same setting, value, and description lanes, staged ``current → new`` values, and Apply. But it uses the text CLI of the repeater (refer to :mod:`meshterm.core.remote_config`) instead of the binary protocol of the companion. Thus it has the settings that only a repeater has, which the local editor never had: TX delay, Direct TX delay, duty cycle, flood caps, and the bridge. Each read is one paced round trip over the mesh. Thus the editor does not read all the values when it opens. Each value shows what the last read (or the last applied ``set``) said, from the cache for each node (:class:`~meshterm.core.remote_store.RemoteStore`). The "Read settings" action refreshes all of them under a progress dialog that the user can abort, one paced read at a time. ``^R`` (``Read`` on the F-key lane) reads only the highlighted row again. A setting that the firmware of the node does not have shows as ``n/a``. That is different from ``?``, which means never read. Apply sends the staged values in the same way, and puts each confirmed value immediately back into the cache. Other than the settings, the action rows are for the node itself: advert, clock sync, change of the admin password, and reboot, each behind its own floating confirm dialog. **Regions** opens the region editor (:mod:`meshterm.ui.region_editor`), which sets the floods that the node relays, region by region. **Command line** opens the readline-style remote CLI (:mod:`meshterm.ui.remote_cli`) for all that the catalog does not include. That command line is also where the page gets more rows. Third-party builds have settings that the catalog does not know. MeshTerm cannot know which build a node runs unless it asks the node. Each question is one paced round trip, and each guessed row is a useless ``n/a`` on the page of each other node. Thus MeshTerm probes nothing on speculation. When the user runs a ``get`` or ``set`` here, and this node answers it, the setting key gets a row on the page of this node, under **Extra** (:func:`learn_from_cli`). The user used that round trip anyway, the catalog does not grow, and the page of no other node changes. Only the user can remove a row, by hand (``Del``). A setting key that stops answering shows ``n/a``, like any other setting key. If it did not, a misread reply would delete the one thing about this node that nobody else can restore. """ from __future__ import annotations import asyncio import re from collections.abc import Iterable from dataclasses import dataclass from typing import TYPE_CHECKING, Any from rich.cells import cell_len from rich.text import Text from ..core.models import Contact from ..core.remote_config import ( DISCOVERED_CATEGORY, NEVER_STORED, REPEATER_SETTINGS, RemoteSetting, composite_members, composite_reads, discovered_setting, get_setting, learnable_from_get, normalize_value, parse_reply_value, parse_setting_command, range_hint, read_plan, reply_is_error, settings_by_category, validate_value, write_plan, ) from .menus import ( PICK_LOCATION_HELP, PICK_LOCATION_LABEL, confirm_discard, exit_rows, fit_cells, icon_lane, lane_header, lane_row, marked_label, menu_rows, section_heading, ) from .trace_screen import TracingDialog from .tui import Choice, Separator from .tui.select import DeleteRequest, SelectScreen, splice_hint from .tui.spinner import Spinner, spinner_interval if TYPE_CHECKING: from ..context import AppContext from ..core.connection import Device #: The time in seconds to wait for one remote reply. Repeaters answer over the mesh, and a #: route of more than one hop takes seconds. Thus this value is longer on purpose than the #: value for a local command. _REPLY_TIMEOUT_S = 10.0 # The sentinels of the menu actions. They are different from the setting keys (CLI # parameter names). _CLI = "__cli__" _LOCATION = "__location__" _READ = "__read__" _ADVERT = "__advert__" _CLOCK = "__clock__" _PASSWORD = "__password__" _REBOOT = "__reboot__" _REGIONS = "__regions__" _APPLY = "__apply__" _CANCEL = "__cancel__" #: The footer atom that names the chord that reads the highlighted setting again. READ_ONE_HINT = "^R read" #: The footer atom for the keyboard key that removes a discovered row (only on those rows). FORGET_HINT = "Del forget" #: The minimum width of the value lane for a discovered setting. Thus a page whose catalog #: rows are all ``?`` still shows a part of what the node answered. _EXTRA_VALUE_MIN = 14 def _extra_value(text: str, width: int) -> str: """One discovered value, fitted to the lane: whitespace collapsed, and a long reply cut. The value of a discovered setting is whatever the node said. A build can answer with a full diagnostic line (``desired=off effective=off supported=yes …``). Without this cut, that line would set the width of the value lane (one width for all the rows on the page), and push the description of each setting off the edge. Thus the function cuts the row here, and the page keeps its shape. The user reads the full reply on the command line. """ flat = " ".join(text.split()) return fit_cells(flat, width) if cell_len(flat) > width else flat def _spec_for(key: str, cache: dict) -> RemoteSetting | None: """The setting that ``key`` names on this node: from the catalog, or discovered there.""" spec = get_setting(key) if spec is not None: return spec cached = cache.get(key) return discovered_setting(key) if cached is not None and cached.discovered else None def _discovered_specs(cache: dict) -> list[RemoteSetting]: """The discovered settings of this node, sorted by setting key: rows the catalog lacks.""" return [discovered_setting(key) for key, cached in sorted(cache.items()) if cached.discovered] def _all_specs(cache: dict) -> list[RemoteSetting]: """All the settings on the page of this node: the catalog, and what the node told us.""" return [*REPEATER_SETTINGS, *_discovered_specs(cache)] def _extra_keys(cache: dict) -> frozenset[str]: """The setting keys on the page of this node that came from the node, not the catalog.""" return frozenset(key for key, cached in cache.items() if cached.discovered) @dataclass(frozen=True, slots=True) class ReadOne: """The result of the admin menu when ``^R`` asks for the value of one setting again. Attributes: key: The catalog key of the highlighted setting. """ key: str class AdminMenu(SelectScreen): """The admin editor's list, with a keyboard key that reads the highlighted setting again. A full read is one paced round trip for each setting. On a node with all the sections, that takes minutes. Thus, to check whether one value was accepted, or to refresh the one row that timed out, the user must not have to read all of them. The keyboard key is ``^R``, the chord that the app already uses for "retry" (the chat sends an unacknowledged message again with it). Here too, it asks the mesh the same question again, for the one highlighted row. It must be a chord, because this list filters as you type, and each bare letter already has a use. The footer names this keyboard key, and the F-key lane draws its ``Read`` chip live, only on a setting row that can be read. They never do this on an action row, where there is nothing to read. It is a full-screen page, not a floating dialog. The full catalog of a node is a place where the user works for some time, like the Device config editor that it copies. A box that fitted its content, over the node picker, left most of the width of the frame to the backdrop. The node picker still floats, and the value prompts, the confirm dialogs, and the progress dialogs float over this page. """ floating = False #: The setting keys on the page of this node that are not in the catalog. They are #: refreshed with the rows. extra_keys: frozenset[str] = frozenset() def __init__(self, title: str, items: list, **kwargs: Any) -> None: """Build the list of the page. Its rows end at the edge, and ←→ does not scroll them. This is the same as on the Device config page, on purpose. The Actions rows pin a head block (:func:`~meshterm.ui.menus.menu_rows`). That alone turns on the ←→ scroll for the full list. Then a setting row, which pins nothing, moved its label off the screen with its description, and the footer got a ``←→ scroll`` atom. ``hscroll=False`` keeps each row complete and cut with the ellipsis, like each other editor lane. """ kwargs.setdefault("hscroll", False) super().__init__(title, items, **kwargs) def _row_key(self) -> str | None: """The highlighted row's setting key, or ``None`` on an action row.""" current = self._current_choice() if current is None or not isinstance(current.value, str): return None return current.value def _readable_key(self) -> str | None: """The setting key of the highlighted row, if the node can be asked for that setting.""" key = self._row_key() if key is None: return None if key in self.extra_keys: return key # a discovered row is a string setting, and can always be read again spec = get_setting(key) return spec.key if spec is not None and spec.readable else None def _forgettable_key(self) -> str | None: """The setting key of the highlighted row, if it is a discovered row that Del removes.""" key = self._row_key() return key if key is not None and key in self.extra_keys else None @property def footer_hint(self) -> str: # type: ignore[override] """The list's own hint, plus the read atom while the highlight is on a readable setting.""" base = super().footer_hint return splice_hint(base, READ_ONE_HINT) if self._readable_key() else base @property def picocalc_lyra_lane(self): """The list's lane, with ``Read`` on F3 and ``Forget`` behind it: dim where inactive. F1/F2 are the section jumps of this grouped list, and F4/F5 are the pager. F3 is the slot that a select list uses for its own verb. Read and Forget are the two halves of the same slot, because they have the same subject (this row). They are never both live: a catalog row can be read but not forgotten, and only a discovered row can be removed. On an action row, they are dim, not empty: the two actions exist on this screen, but not for the highlighted row. """ from .tui.fkeys import FPair lane = list(super().picocalc_lyra_lane) lane[2] = FPair( "Read", "retry", opp_label="Forget", opp_action="delete", enabled=self._readable_key() is not None, opp_enabled=self._forgettable_key() is not None, ) return lane def handle(self, action: str, data: str = "") -> None: """Ask for the value of the highlighted setting, or act as each select list does.""" if action == "retry": key = self._readable_key() if key is not None: self.resolve(ReadOne(key)) return super().handle(action, data) async def open_repeater_admin(ctx: AppContext) -> dict[str, Any] | None: """Run the repeater-admin flow: select a node, log in, and administer it. The node picker is a dialog: a question before the user goes in, which closes when it is answered. Thus the admin page opens over the main menu, and Esc on the page goes back there: the page is the full visit. If the node refuses the login (or never answers), the picker asks again, with the same node highlighted. Thus a mistyped password costs one Enter and a retry, not a new start of the flow. The list once stayed pushed below the page as a hub. That looked like two places where there is only one. Args: ctx: The shared application context (it must run the interactive TUI surface). Returns: A summary of the administered node (for the log of the tool), or ``None`` if the user left the picker and never started a session. Raises: RuntimeError: If it is called outside the interactive menu (no full-screen session). """ from .admin_picker import pick_admin_node from .surface import TuiUi if not isinstance(ctx.ui, TuiUi): # pragma: no cover - the menu-only caller prevents this raise RuntimeError("repeater admin is only available in the menu") device = await ctx.device() # Through the session cache: when this screen opens, it must not read the (slow) # contacts table again if another screen already read it (refer to # :class:`~meshterm.services.device_state.DeviceState`). The device handle above is # still necessary for the admin login and for the CLI session that come after. contacts = await ctx.devstate.contacts() node: Contact | None = None while True: node = await pick_admin_node( ctx, contacts, title="Repeater admin — node to manage", prompt="The remote node to set up (you need its admin password):", default=node, ) if node is None: return None if await _login(ctx, device, node): return await _admin_session(ctx, device, node) async def _login(ctx: AppContext, device: Device, node: Contact) -> bool: """Log in to ``node``: with a remembered password silently, or else one floating prompt. MeshTerm remembers a password that works. It forgets a rejected password, so that the next try asks for a new one (the convention for remote admin in all the app). But it forgets only a rejected password. A node that never answered said nothing about the password, so when there is no answer, MeshTerm keeps the password. Before, if the user administered a repeater that was down at that time, MeshTerm erased its credential. """ from ..core.models import LoginResult session = ctx.ui.session password = ctx.admin_store.get(node) prompted = password is None if password is None: password = await session.text( f"Admin password for {node.name}", prompt="The node ignores admin commands without a login.", password=True, # The picker dialog is closed now, so this prompt is the only screen on the stack. # The flag keeps it a box over a blank base, not a full-frame prompt. floating=True, ) if not password: return False async with ctx.ui.busy_overlay(): outcome = await device.admin_login(node, password) ctx.admin_store.record(node, password, outcome) if outcome is LoginResult.REFUSED: await session.message_dialog( Text( f"{node.name!r} rejected the admin login (wrong password?). " + ("" if prompted else "The saved password was cleared; ") + "try again to enter a new one.", style="err", ), title="Admin login", ) return False if not outcome: await session.message_dialog( Text( f"No reply from {node.name} — it may be out of reach, asleep, or busy. " + ("" if prompted else "The saved password was kept; ") + "try again when it answers.", style="warn", ), title="Admin login", ) return False return True async def _admin_session(ctx: AppContext, device: Device, node: Contact) -> dict[str, Any]: """Run the editor loop for one node that the user is logged in to. One screen for the full session. Its rows refresh in place after each action: they hold the ``current → new`` values of the cache, and the title counts the staged values. Thus the content changes below a highlight that stays where the user put it, and the typed filter stays too. Each sub-prompt floats over the screen, and Esc leaves the node. """ from .tui import CANCEL session = ctx.ui.session pending: dict[str, str] = {} # setting key -> staged new value applied = 0 cache = ctx.remote_store.settings(node) title, items = _menu_items(node, cache, pending) menu = AdminMenu( title, items, footer_hint="↑↓ move · type to filter · Enter select · Esc back", delete_hint=FORGET_HINT, ) menu.extra_keys = _extra_keys(cache) async with session.stay(menu) as visit: while True: choice = await visit.result() if choice is CANCEL: choice = _CANCEL if choice in (None, _CANCEL): if pending and not await confirm_discard(ctx, len(pending), verb="sending"): continue return {"node": node.name, "applied": applied} if isinstance(choice, ReadOne): spec = _spec_for(choice.key, cache) if spec is not None: await read_settings(ctx, device, node, [spec]) elif isinstance(choice, DeleteRequest): await _forget_discovered(ctx, node, str(choice.value), pending) elif choice == _APPLY: applied += await _apply(ctx, device, node, pending) elif choice == _READ: # The page of this node, not the catalog: the same sweep refreshes a # discovered row and all the other rows, and the row shows n/a if the node # stops answering. await read_settings(ctx, device, node, _all_specs(cache)) elif choice == _LOCATION: await _stage_location(ctx, cache, pending) elif choice == _CLI: await _command_line(ctx, device, node) elif choice == _REGIONS: from .region_editor import open_region_editor await open_region_editor(ctx, device, node) elif choice == _ADVERT: await _simple_action( ctx, device, node, "advert", title="Send advert", prompt=f"Ask {node.name} to announce itself (flood) now?", commit="Send", ) elif choice == _CLOCK: await _simple_action( ctx, device, node, "clock sync", title="Sync clock", prompt=f"Set {node.name}'s clock from this companion's time?", commit="Sync", ) elif choice == _PASSWORD: await _change_password(ctx, device, node) elif choice == _REBOOT: await _simple_action( ctx, device, node, "reboot", title="Reboot node", prompt=f"Reboot {node.name} now? It drops off the mesh while booting.", commit="Reboot", danger=True, ) else: # a setting key spec = _spec_for(str(choice), cache) if spec is not None and not spec.writable: # A read-only fact has nothing to stage. Enter can only ask for it again. await read_settings(ctx, device, node, [spec]) else: await _stage_setting(ctx, node, str(choice), cache, pending) # A read or an apply writes the cache that the rows come from again. Read the # cache again, so that the values are the values that the action produced. cache = ctx.remote_store.settings(node) title, items = _menu_items(node, cache, pending) menu.extra_keys = _extra_keys(cache) menu.replace_items(items, title=title) # --- the menu ------------------------------------------------------------------------ def _menu_items(node: Contact, cache: dict, pending: dict[str, str]) -> tuple[str, list]: """Build the title and the rows of the editor menu from the cache and the staged values. The lane layout is the same as on the Device config editor: setting, value (with any staged ``→ new``), and description, below one header line. Thus the administration of a remote node looks exactly like the edit of the settings of the local one. """ sections: list[tuple[str, list[tuple[str, Text, str, Any]]]] = [] for category, specs in settings_by_category(): rows: list[tuple[str, Text, str, Any]] = [] for spec in specs: # In the words of the node that the page is open on: on a room server, the # guest password is the room password (RemoteSetting.for_node). spec = spec.for_node(node.node_type) if spec.key == "lat": # A position selected on the map sets the two coordinates at the same time. # Thus its row is first, above the pair that it fills: the same row, at the # same place, as on the Device config page. rows.append((PICK_LOCATION_LABEL, Text(), PICK_LOCATION_HELP, _LOCATION)) rows.append((spec.label, _value_text(spec, cache, pending), spec.help, spec.key)) sections.append((category, rows)) # What this node told us itself, at the end and in its own section. The section is not # there on a node that told us nothing, which is each node until a user asks it # something. Its values fit in the lane that the catalog rows already need. The value # lane has one width for the full page. Thus a reply with no limit here would push the # description of each setting to the right, off the edge of a 72-cell screen. It would # also change the shape of a page that the user opened to read the catalog. extra = _discovered_specs(cache) if extra: budget = max( _EXTRA_VALUE_MIN, max((cell_len(value.plain) for _, rows in sections for _, value, _, _ in rows)), ) sections.append( ( DISCOVERED_CATEGORY, [(s.label, _value_text(s, cache, pending, budget), s.help, s.key) for s in extra], ) ) label_w = max(cell_len(label) for _, rows in sections for label, _, _, _ in rows) value_w = max(cell_len(value.plain) for _, rows in sections for _, value, _, _ in rows) # The same pinned header, which fits itself, as the header above the lanes of the local # editor. It stays on the screen with the category headings for the full list. When a # long cached value leaves no space for the last label, the header abbreviates, and it # does not wrap (menus.lane_header). items: list = [Separator(lambda w: lane_header(label_w, value_w, w), pinned=True)] for category, rows in sections: items.append(section_heading(category)) for label, value, help_text, key in rows: items.append( Choice( title=lane_row(label, value, help_text, label_w, value_w), value=key, # Only a discovered row can be removed. A catalog row that goes away # would come back at the next paint, because the catalog still names it. deletable=category == DISCOVERED_CATEGORY, ) ) # ↻ and ⌨ are one cell wide, but 📡 🕒 🔐 🔄 are two. Thus the icon lane is measured one # time, and each icon is padded to that width. If not, the labels of Read settings and # Command line would start one cell to the left of the rows below them. actions = [ ("↻", "Read settings", "Fetch every value from the node, one paced read", _READ), ("⌨", "Command line…", "Talk to the node's CLI directly", _CLI), ("🔖", "Regions…", "Which scoped floods it relays", _REGIONS), ("📡", "Send advert…", "Have the node announce itself now", _ADVERT), ("🕒", "Sync clock…", "Set the node's clock over the mesh", _CLOCK), ("🔐", "Admin password…", "Change the node's admin password", _PASSWORD), ("🔄", "Reboot node…", "Restart it remotely", _REBOOT), ] lane = icon_lane(icon for icon, _, _, _ in actions) items.append(section_heading("Actions")) items.extend( menu_rows( (marked_label(icon, label, "", lane=lane), help_text, value) for icon, label, help_text, value in actions ) ) staged = len(pending) items.extend(exit_rows(staged, apply_value=_APPLY, back_value=_CANCEL)) title = f"Repeater admin — {node.name}" + (f" · {staged} staged" if staged else "") return title, items def _value_text( spec: RemoteSetting, cache: dict, pending: dict[str, str], width: int = _EXTRA_VALUE_MIN ) -> Text: """The VALUE lane of one setting: what the node said last, and any staged arrow. There are five states, each with its own word: - The value. - ``empty``, for a string that the node holds blank. - ``n/a``, when the node answered with something that cannot be the value of this setting (its firmware has no such setting). - ``?``, when the setting was never read (or the node never answered). - ``write-only``, for a setting that the node cannot be asked for. No age: the lane is the value, and a time stamp beside it only made the lane too full. """ cached = cache.get(spec.key) if not spec.readable: value = Text("write-only", style="faint") elif cached is None: value = Text("?", style="muted") elif not cached.supported: value = Text("n/a", style="muted") elif cached.value == "": value = Text("empty", style="muted") elif spec.discovered: value = Text(_extra_value(cached.value, width)) else: value = Text(spec.display(cached.value)) if spec.key in pending: staged = pending[spec.key] shown = _extra_value(staged, width) if spec.discovered else spec.display(staged) value.append(f" → {shown}", style="warn") return value # --- staging and applying --------------------------------------------------------- async def _stage_setting( ctx: AppContext, node: Contact, key: str, cache: dict, pending: dict[str, str] ) -> None: """Prompt for the new value of one setting, and stage it (MeshTerm sends nothing yet). The prompt names the setting as its row does, in the words of the node that it is for. """ spec = _spec_for(key, cache) if spec is None or not spec.writable: # pragma: no cover - the menu offers only real settings return spec = spec.for_node(node.node_type) cached = cache.get(key) known = cached.value if cached is not None and cached.supported else None current = pending.get(key, known or "") if spec.kind == "bool": picked = await ctx.ui.dialog( spec.help, [("Off", "off"), ("On", "on")], title=spec.label, default=1 if current == "on" else 0, keys={"0": "off", "1": "on", "n": "off", "y": "on"}, ) if picked is None: return value = str(picked) elif spec.kind == "enum": items = [ Choice( title=option.label + (f" — {option.help}" if option.help else "") + (" (current)" if option.value == current else ""), value=option.value, ) for option in spec.options ] picked = await ctx.ui.select( spec.label, items, prompt=spec.help, default=current if any(o.value == current for o in spec.options) else None, ) if picked is None: return value = str(picked) else: raw = await ctx.ui.text( spec.label, prompt=spec.help, default=str(current), validate=lambda t: validate_value(spec, t), help_text=range_hint(spec), ) if raw is None: return value = normalize_value(spec, raw) if value == known: pending.pop(key, None) # the same as what the node said last, so nothing to send else: pending[key] = value async def _forget_discovered( ctx: AppContext, node: Contact, key: str, pending: dict[str, str] ) -> None: """Remove one discovered row from the page of this node, after a confirm dialog. This is the only way to remove a discovered row. A read could remove one, because a setting key that the node no longer answers can be real (after a reused identity moves to a different board). But then one misread reply would silently delete the row: a truncated line, a node that answers during a reboot, or a stray reply that MeshTerm matched to the wrong command. When that occurs on a catalog row, the cost is an ``n/a``, and nothing is lost, because the catalog still names the setting key. But only this page knows the setting key of a discovered row, and the user may not remember it. Thus a read can only show the same ``n/a``, and a removal is a decision. It is a red confirm dialog, like each other delete of one record, but nothing changes on the node. What goes is what MeshTerm remembers. If the user asks for the setting key again, the row comes back. """ from .tui import CANCEL choice = await ctx.ui.dialog( f"Stop showing {key} on {node.name}? Nothing on the node changes — " "ask it for the setting again and the row comes back.", [("Cancel", CANCEL), ("Forget", "forget")], title="Forget setting", default=1, destructive=True, ) if choice != "forget": return pending.pop(key, None) ctx.remote_store.forget_setting(node, key) async def _stage_location(ctx: AppContext, cache: dict, pending: dict[str, str]) -> None: """Select the advertised location of the node on the map, and stage the two coordinates. This is the row of the Device config page, but it uses the CLI of this node. The map opens on the staged position (or on the position that the node said last), or on the mesh if there is no position. Each selected coordinate is put in the same format as a read of it. Thus, if the user selects the position that the node already holds, the staged value is removed, and no ``set`` that does nothing goes in the queue. MeshTerm sends nothing until Apply. """ from .map_screen import coords_or_none, pick_location def known(key: str) -> str | None: cached = cache.get(key) return cached.value if cached is not None and cached.supported else None lat = pending.get("lat", known("lat")) lon = pending.get("lon", known("lon")) picked = await pick_location(ctx, initial=coords_or_none(lat, lon)) if picked is None: return for key, value in zip(("lat", "lon"), picked, strict=True): spec = get_setting(key) if spec is None: # pragma: no cover - both keys are in the catalog continue # Six decimals ≈ 0.1 m: more than the precision of the map, and enough for an advert. text = normalize_value(spec, f"{value:.6f}") if text == known(key): pending.pop(key, None) # what the node already says, so nothing to send else: pending[key] = text def _known_values(ctx: AppContext, node: Contact) -> dict[str, str]: """The last values read from the node, to state the unstaged fields of a composite again.""" return { key: cached.value for key, cached in ctx.remote_store.settings(node).items() if cached.supported } def _remark(reply: str, value: str) -> str: """What the reply to a successful write adds after ``OK`` (``reboot to apply``), or ``""``.""" remark = re.sub(r"^ok\b[\s,:-]*", "", reply.strip(), flags=re.IGNORECASE) return "" if remark.lower() in ("", value.lower()) else remark async def _apply(ctx: AppContext, device: Device, node: Contact, pending: dict[str, str]) -> int: """Send all the staged values, paced, under a progress dialog that the user can abort. MeshTerm sends the staged values in the groups of :func:`~meshterm.core.remote_config.write_plan`: one command for each setting, but the four fields of the radio go together as one ``set radio``. That command states each field again. Thus, if a radio field is staged and its sibling fields were never read, MeshTerm reads them first (one paced ``get radio``), and does not guess a frequency. Each confirmed value goes immediately into the cache for the node, and MeshTerm forgets each other spelling of the same firmware value. A write that the node rejects or does not answer stays staged, so that the user can try it again (or unstage it). It is not silently removed. The function stores one ``runs`` row for the batch. Returns: The number of settings that the node accepted. """ session = ctx.ui.session run_id = ctx.repo.start_run( "repeater-admin", {"node": node.name, "mode": "apply", "settings": sorted(pending)}, ctx.profile_name, ) outcomes: list[Text] = [] applied = 0 async def work(dialog: TracingDialog) -> None: nonlocal applied first = True async def send(command: str, status: str) -> str | None: nonlocal first if not first: # each transmission after the first waits for the cooldown await asyncio.sleep(ctx.preferences.trace_cooldown_s) first = False dialog.status = status session.invalidate() return await device.send_remote_command(node, command, timeout=_REPLY_TIMEOUT_S) for command in composite_reads(pending, _known_values(ctx, node)): fills = [s for s in REPEATER_SETTINGS if s.get_command == command] reply = await send(command, command) if reply is not None: remember_reply(ctx, node, fills, reply) writes = write_plan(pending, _known_values(ctx, node)) # The catalog, plus the discovered rows of this node. A staged setting key that only # this node has is still a row, with a label to name in the outcome and a write to # count. specs = _all_specs(ctx.remote_store.settings(node)) for i, write in enumerate(writes, start=1): staged = [s for s in specs if s.key in write.values and s.key in pending] names = ", ".join(f"{s.label} = {s.display(write.values[s.key])}" for s in staged) if write.missing: unread = ", ".join(s.label for s in map(get_setting, write.missing) if s) outcomes.append( Text.assemble( ("✗ ", "err"), f"{names} — {unread} couldn't be read (still staged)" ) ) continue verb = " ".join(write.command.split()[:2]) # never the value of a secret on screen reply = await send(write.command, f"{verb} · {i}/{len(writes)}") if reply is not None and not reply_is_error(reply): for key, value in write.values.items(): ctx.remote_store.remember_setting(node, key, value) spec = get_setting(key) for other in spec.overlaps if spec is not None else (): ctx.remote_store.forget_setting(node, other) for spec in staged: pending.pop(spec.key, None) applied += len(staged) remark = _remark(reply, write.values[staged[0].key]) if staged else "" outcomes.append( Text.assemble(("✓ ", "ok"), names, (f" — {remark}" if remark else "", "muted")) ) elif reply is None: outcomes.append(Text.assemble(("? ", "warn"), f"{names} — no reply (still staged)")) else: outcomes.append( Text.assemble(("✗ ", "err"), f"{names} — {reply.strip()} (still staged)") ) aborted = await _run_under_dialog(ctx, f"Applying — {node.name}", work) ctx.repo.finish_run( run_id, "error" if aborted else "ok", {"applied": applied, "staged_left": len(pending)}, ) summary = Text.assemble( (f"{applied}", "brand"), f" of {applied + len(pending)} settings applied" ) if aborted: summary.append(" (aborted — the rest stay staged)", style="warn") await ctx.ui.session.message_dialog( Text("\n").join([summary, Text(), *outcomes]) if outcomes else summary, title=f"Apply — {node.name}", ) return applied def remember_reply(ctx: AppContext, node: Contact, fills: list[RemoteSetting], reply: str) -> int: """Put the reply of one read into the cache, for each setting that it answers. A reply that cannot be the value of the setting means that the node has no such setting. MeshTerm remembers it as that (the row shows ``n/a``). This applies to an error (``??: key``, ``Error: unsupported``) and also to an answer to a different question. The second type is how older firmware says it. ``get`` matches its setting keys by prefix. Thus, on v1.15, a setting key that the firmware does not have goes into the branch of a shorter sibling key: ``get radio.fem.rxgain`` answers with the ``> 910.525,62.5,7,5`` of ``get radio``. Also, ``gps`` on a board with no GPS receiver answers ``Can't find GPS``. Before, these rows stayed on ``?``, which means never asked. Now only a read that got no reply at all (which this function does not get) keeps ``?``. Returns: The number of settings that got a value. """ got = 0 for spec in fills: value = parse_reply_value(spec, reply) # None for an error reply too if value is None: ctx.remote_store.remember_unsupported(node, spec.key) else: ctx.remote_store.remember_setting(node, spec.key, value) got += 1 return got def learn_from_cli(ctx: AppContext, node: Contact, command: str, reply: str) -> None: """Put what one exchange on the command line proved about ``node`` into its page. The command line is the only place where the user can reach a setting outside the catalog. Thus it is also the only place where MeshTerm can find one. A ``get`` or ``set`` that this node answered proves that the setting key exists. The row that the setting key gets costs no round trip that the user did not ask for. The catalog does not grow: a setting key learned here belongs only to this node (refer to :func:`~meshterm.core.remote_config.discovered_setting`), and the page of each other node stays as it was. There are three cases, in the order of the tests: * The **own key of a composite** (``get radio``) fills all four of its fields, exactly as a read of any one of them does. * A **catalog key** goes into the cache in the same way as from the sweep or an Apply. That is also why ``get tx`` on the command line no longer leaves the TX power row out of date. * Each other setting key is **discovered**, and the two verbs are not equally good evidence. ``handleSetCmd`` matches a setting key with its trailing space, and refuses an unknown setting key immediately. Thus an accepted write is proof. ``handleGetCmd`` matches on a bare prefix, and can answer a setting key that it does not have from the branch of a shorter setting key. Thus a read is proof only when the reply cannot be the reply for that shorter setting key (:func:`~meshterm.core.remote_config.learnable_from_get`). MeshTerm never stores ``prv.key``, however the user asked for it (:data:`~meshterm.core.remote_config.NEVER_STORED`). """ parsed = parse_setting_command(command) if parsed is None or parsed.key in NEVER_STORED: return members = composite_members(parsed.key) spec = _spec_for(parsed.key, ctx.remote_store.settings(node)) if parsed.verb == "get": if members: remember_reply(ctx, node, members, reply) elif spec is not None: remember_reply(ctx, node, [spec], reply) elif not reply_is_error(reply) and learnable_from_get( parsed.key, reply, _known_values(ctx, node) ): found = discovered_setting(parsed.key) value = parse_reply_value(found, reply) if value is not None: ctx.remote_store.remember_discovered(node, found.key, value) return if reply_is_error(reply): return # a refused write tells nothing about the setting key or its value if members: for member in members: value = parse_reply_value(member, parsed.value) if value is not None: ctx.remote_store.remember_setting(node, member.key, value) return if spec is not None and not spec.discovered: value = parse_reply_value(spec, parsed.value) if value is None: return # the catalog cannot read back these words, so keep the row as it was ctx.remote_store.remember_setting(node, spec.key, value) for other in spec.overlaps: ctx.remote_store.forget_setting(node, other) return ctx.remote_store.remember_discovered(node, parsed.key, parsed.value) async def read_settings( ctx: AppContext, device: Device, node: Contact, specs: Iterable[RemoteSetting] ) -> None: """Refresh ``specs`` from the node, one paced read at a time. The four radio fields share one ``get radio`` (refer to :func:`~meshterm.core.remote_config.read_plan`), so a full read asks each question one time. Values that parse go into the cache (and onto the screen). A setting that the firmware does not have shows ``n/a``. An abort keeps all the values that were already read. After the reads, a dialog lists the reads that the node never answered. Their rows continue to show what they said last, and without the list, they would look new. """ session = ctx.ui.session plan = read_plan(specs) run_id = ctx.repo.start_run( "repeater-admin", {"node": node.name, "mode": "read"}, ctx.profile_name ) read = 0 unanswered: list[str] = [] async def work(dialog: TracingDialog) -> None: nonlocal read for i, (command, fills) in enumerate(plan, start=1): dialog.status = f"{command} · {i}/{len(plan)}" session.invalidate() reply = await device.send_remote_command(node, command, timeout=_REPLY_TIMEOUT_S) if reply is None: unanswered.extend(spec.label for spec in fills) else: read += remember_reply(ctx, node, fills, reply) if i < len(plan): await asyncio.sleep(ctx.preferences.trace_cooldown_s) aborted = await _run_under_dialog(ctx, f"Reading — {node.name}", work) asked = sum(len(fills) for _command, fills in plan) ctx.repo.finish_run( run_id, "error" if aborted else "ok", {"read": read, "asked": asked, "unanswered": len(unanswered)}, ) if unanswered and not aborted: await session.message_dialog( Text( f"No reply from {node.name} for: {', '.join(unanswered)}. " "Those rows still show what they last said.", style="warn", ), title=f"Read — {node.name}", ) async def _run_under_dialog(ctx: AppContext, title: str, work) -> bool: """Run an async remote batch under a floating spinner dialog with Abort. Args: ctx: The shared application context. title: The heading of the dialog. work: ``async work(dialog)``, which does the batch and updates ``dialog.status``. Returns: ``True`` if the user aborted, ``False`` if the batch ran to the end. """ session = ctx.ui.session spinner = Spinner() dialog = TracingDialog(title, spinner=spinner, on_abort=lambda: None) dialog.show_last = False task = asyncio.ensure_future(work(dialog)) dialog.on_abort = task.cancel async def animate() -> None: while True: await asyncio.sleep(spinner_interval()) spinner.tick() session.invalidate() ticker = asyncio.ensure_future(animate()) session.push(dialog) aborted = False try: await task except asyncio.CancelledError: aborted = True except Exception as exc: # noqa: BLE001 - show it in a dialog, and keep the screen await session.message_dialog(Text(str(exc), style="err"), title=title) finally: ticker.cancel() try: await ticker except asyncio.CancelledError: pass except Exception: # noqa: BLE001 - a small spinner fault must never stop the batch pass session.pop(dialog) return aborted # --- one-shot actions ------------------------------------------------------------ async def _simple_action( ctx: AppContext, device: Device, node: Contact, command: str, *, title: str, prompt: str, commit: str, danger: bool = False, ) -> None: """Confirm and send one fixed CLI command, then show the reply of the node.""" choice = await ctx.ui.dialog( prompt, [("Cancel", None), (commit, "go")], title=title, default=1, danger=danger, ) if choice != "go": return async with ctx.ui.busy_overlay(): reply = await device.send_remote_command(node, command, timeout=_REPLY_TIMEOUT_S) if reply is None: body = Text("no reply — the command may still have landed", style="warn") else: body = Text(reply.strip(), style="err" if reply_is_error(reply) else "ok") await ctx.ui.session.message_dialog(body, title=title) async def _change_password(ctx: AppContext, device: Device, node: Contact) -> None: """Change the admin password of the node (and remember the new password if it succeeds).""" new = await ctx.ui.text( f"New admin password for {node.name}", prompt="Sent over the mesh; the node applies it immediately.", password=True, ) if not new: return confirmed = await ctx.ui.dialog( f"Change {node.name}'s admin password now? You will need the new one everywhere.", [("Cancel", None), ("Change", "go")], title="Admin password", default=1, danger=True, ) if confirmed != "go": return async with ctx.ui.busy_overlay(): reply = await device.send_remote_command(node, f"password {new}", timeout=_REPLY_TIMEOUT_S) if reply is not None and not reply_is_error(reply): ctx.admin_store.remember(node, new) # the password that works is now the new one body = Text("✓ password changed and remembered", style="ok") elif reply is None: body = Text( "no reply — the change may or may not have landed; the old password " "stays remembered until a login proves otherwise", style="warn", ) else: body = Text(reply.strip(), style="err") await ctx.ui.session.message_dialog(body, title="Admin password") # --- the command line --------------------------------------------------------------- async def _command_line(ctx: AppContext, device: Device, node: Contact) -> None: """Open the readline-style remote CLI for ``node`` (the history is stored for each node).""" from .remote_cli import RemoteCliScreen session = ctx.ui.session worker: asyncio.Task | None = None ticker: asyncio.Task | None = None def send(command: str) -> None: nonlocal worker if worker is not None and not worker.done(): return # one command at a time, because each send is a transmission screen.sent(command) ctx.remote_store.append_history(node, command) worker = asyncio.ensure_future(roundtrip(command)) async def roundtrip(command: str) -> None: try: reply = await device.send_remote_command(node, command, timeout=_REPLY_TIMEOUT_S) except asyncio.CancelledError: screen.failed("cancelled", error=False) raise except Exception as exc: # noqa: BLE001 - shown inline, and the screen stays open screen.failed(str(exc), error=True) return if reply is None: screen.failed(f"no reply within {_REPLY_TIMEOUT_S:.0f} s") else: learn_from_cli(ctx, node, command, reply) screen.reply(reply) screen = RemoteCliScreen( node_label=node.name, history=ctx.remote_store.history(node), send=send, session=session, extra_keys=_extra_keys(ctx.remote_store.settings(node)), ) async def animate() -> None: while True: await asyncio.sleep(spinner_interval()) if screen.busy: screen.tick() session.invalidate() ticker = asyncio.ensure_future(animate()) try: await session.run_screen(screen) finally: for task in (worker, ticker): if task is not None and not task.done(): task.cancel() for task in (worker, ticker): if task is not None: try: await task except asyncio.CancelledError: pass except Exception: # noqa: BLE001 - the screen is closed, so nothing to show pass