A setting you use on a node's command line becomes a row on that node's page

Third-party builds carry knobs the catalog has never heard of -- EasySkyMesh's
radio.rxps and reboot.interval among them -- and MeshTerm cannot know which
build a repeater is running without asking it. Asking is the whole problem:
every question is one paced round trip, so probing for a fork's keys would
spend a sweep on every node to benefit a few, and listing them outright would
leave dead n/a rows on everyone else's page.

So nothing is probed speculatively and the catalog does not grow. The command
line is already the only place those settings can be reached, so it is now
also where they are found: a get/set the reader ran, which this node answered,
earns a row on this node's page under Extra. The round trip was one the reader
was spending anyway, and no other node's page changes.

The two verbs are not equal evidence, which the firmware settles. handleSetCmd
compares each key with its trailing space -- "radio " -- so a longer key can
never land in a shorter one's branch, and an unknown one answers "unknown
config:". handleGetCmd compares a bare prefix, so stock firmware answers
"get radio.rxps" out of the get radio branch with a perfectly plausible
910.525,62.5,7,5. A catalog row survives that because its typing rejects it; a
discovered key has no typing, so a read is trusted only when nothing shadows
the key, or when the reply differs from what the shadow last said.

A discovered row's whole claim to exist is that the node answered it, so when
it stops answering the row goes rather than earning the n/a that only a
catalog key can -- which is what tidies up after a board swap onto a reused
identity. Del forgets one without waiting for a read, and asking the key again
brings it back. Its values are fitted to the lane the catalog's rows already
need, since the value lane is one width for the whole page.

prv.key is never stored, whichever way it was asked. The firmware answers it on
the serial console only, and a reply that reached us anyway must not be the
first place a repeater's private key lands in a file.

Two things fall out: get tx at the command line no longer leaves the TX power
row showing last week's value, since catalog keys now fold into the cache the
same way, and a discovered key completes at the prompt where it was learned.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jean-Pierre MartineauandClaude Opus 5 committed 2026-09-17 21:29:26 -04:00
1 parent be7e588c3a
commit a7d201f4da
5 files changed
+628 -23

No files matched your search

+144
View File
@@ -28,6 +28,13 @@ three of its shapes only make sense against that source:
``af`` both read and write ``airtime_factor``, so writing one stales the other
(:attr:`RemoteSetting.overlaps`).
The catalog is not the whole page, though: it is what *any* MeshCore build might expose,
and a node can have settings beyond it (a third-party firmware's own knobs). Those are never
listed here and never probed — they are learned one node at a time from commands the reader
actually ran (:func:`parse_setting_command`, :func:`discovered_setting`), which is why
:attr:`RemoteSetting.discovered` exists and why :func:`learnable_from_get` has to care that
``get`` matches keys by prefix and ``set`` does not.
Replies are parsed *loosely* on purpose: repeater firmware answers tersely and has changed
its phrasing across versions (``"> 20"``, ``"tx: 20"``, ``"OK - repeat is now ON"``), so
values are extracted rather than pattern-matched. Firmware without a given key answers
@@ -118,6 +125,9 @@ class RemoteSetting:
part: This setting's position within its composite's comma-joined value.
overlaps: Keys that are another spelling of the same firmware value, whose cached
reading a write to this one makes stale.
discovered: Whether this setting is not in the catalog at all, but was learned from
a command the reader ran on *one node's* command line (see
:func:`discovered_setting`). Never true of a catalog entry.
"""
key: str
@@ -139,6 +149,7 @@ class RemoteSetting:
composite: str = ""
part: int = 0
overlaps: tuple[str, ...] = ()
discovered: bool = False
@property
def get_command(self) -> str:
@@ -671,6 +682,134 @@ def composite_members(composite: str) -> list[RemoteSetting]:
return sorted((s for s in REPEATER_SETTINGS if s.composite == composite), key=lambda s: s.part)
#: The category a discovered setting sorts under, and the heading its section takes.
DISCOVERED_CATEGORY = "Extra"
#: The description lane's text for a discovered setting. The catalog has no explanation to
#: give — all it knows is that this node answered the key — so the row says exactly that, and
#: says it short: the lane is shared with descriptions written to fit a 72-column screen.
DISCOVERED_HELP = "Found on this node"
#: Keys whose value is never written to disk, however it was asked for. ``prv.key`` is the
#: node's own identity: the firmware answers it on the serial console only (``sender_timestamp
#: == 0`` in ``handleGetCmd``), it is a fact rather than a setting, and a reply that reached
#: us anyway must not be the first place a repeater's private key lands in a file. The
#: command line still runs the command and still prints the answer; only the remembering
#: stops here.
NEVER_STORED = frozenset({"prv.key"})
@dataclass(frozen=True, slots=True)
class SettingCommand:
"""A remote-CLI command recognised as a read or a write of one named setting.
Attributes:
verb: ``"get"`` or ``"set"``.
key: The setting's key, as the reader spelled it.
value: What a ``set`` writes; ``""`` for a ``get``.
"""
verb: str
key: str
value: str = ""
def parse_setting_command(command: str) -> SettingCommand | None:
"""Read one typed CLI command as a setting read/write, or ``None`` if it is neither.
The grammar is the filter, and deliberately so: ``get <key>`` and ``set <key> <value>``
are the firmware's own way of saying *this is a setting*, so recognising them costs no
list of what to exclude. Every top-level verb — ``reboot``, ``advert``, ``clock sync``,
``start ota``, ``powersaving``, ``setperm`` — simply isn't this shape and falls out.
A key is held to the spelling the firmware's own keys use, so a mistyped line can't
invent one.
"""
parts = command.strip().split()
if len(parts) < 2 or parts[0] not in ("get", "set"):
return None
key = parts[1]
if not re.fullmatch(r"[A-Za-z][A-Za-z0-9._]*", key):
return None
if parts[0] == "get":
return SettingCommand("get", key) if len(parts) == 2 else None
value = command.strip()[len("set") :].strip()[len(key) :].strip()
return SettingCommand("set", key, value) if value else None
def get_match_keys() -> list[str]:
"""Every spelling a firmware ``get`` matches on, longest first.
A composite's members share their composite's key (``get radio``); verb-shaped settings
aren't ``get`` keys at all.
"""
keys = {s.composite or s.key for s in REPEATER_SETTINGS if s.readable and not s.verb}
return sorted(keys, key=len, reverse=True)
def shadowing_key(key: str) -> str | None:
"""The catalog ``get`` key that would answer a ``get key`` on firmware lacking ``key``.
``handleGetCmd`` compares with ``memcmp(config, "radio", 5)`` — no trailing space — so a
key it doesn't know falls into the branch of any *shorter* key that prefixes it, and
answers with that one's value. ``get radio.rxps`` on stock firmware comes back
``> 910.525,62.5,7,5``: not an error, and not a value of the key that was asked for.
``handleSetCmd`` matches ``"radio "`` *with* the space, so a write has no such shadow.
"""
return next((k for k in get_match_keys() if key != k and key.startswith(k)), None)
def learnable_from_get(key: str, reply: str, known: Mapping[str, str]) -> bool:
"""Whether a ``get key`` reply really came from ``key`` and not from a shorter sibling.
Only asked of a key the catalog doesn't have, where there is no typing to reject a
wrong-shaped answer with. The reply is trusted when nothing shadows the key
(:func:`shadowing_key`), or when it cannot be the shadow's answer — it doesn't parse as
the shadow's type, or it parses to something other than what the shadow last said. A
shadow whose own value was never read leaves the question unanswerable, and an
unanswerable question is not a discovery.
Args:
key: The key the reader asked for.
reply: What the node answered.
known: The node's last-read values, keyed as the cache keys them.
"""
shadow = shadowing_key(key)
if shadow is None:
return True
members = composite_members(shadow)
if not members:
spec = get_setting(shadow)
members = [spec] if spec is not None else []
if not members: # pragma: no cover - every get key is a catalog key
return True
parsed = {m.key: parse_reply_value(m, reply) for m in members}
if any(value is None for value in parsed.values()):
return True # the reply isn't even shaped like the shadow's answer
if any(known.get(k) is None for k in parsed):
return False # it could be the shadow's answer, and there is nothing to compare
return any(known[k] != value for k, value in parsed.items())
def discovered_setting(key: str) -> RemoteSetting:
"""A catalog entry for a key the catalog hasn't got, which one node said it has.
Everything downstream of the catalog — the read plan, the reply parser, the editor's
lanes, staging, :func:`write_plan` — speaks :class:`RemoteSetting`, so a discovered key
becomes one rather than growing a second path beside it. What it carries is only what
was actually learned: the key as the node spells it, and a string value. No range, no
words, no options — the firmware never said, and guessing is how a row would start
lying about a node whose build we have never seen.
"""
return RemoteSetting(
key=key,
label=key,
help=DISCOVERED_HELP,
category=DISCOVERED_CATEGORY,
kind="str",
discovered=True,
)
def reply_is_error(reply: str) -> bool:
"""Whether a reply text reads as the firmware refusing the command.
@@ -884,6 +1023,11 @@ def write_plan(pending: Mapping[str, str], known: Mapping[str, str]) -> list[Wri
else:
joined = ",".join(str(values[m.key]) for m in members)
writes.append(Write(f"set {spec.composite} {joined}", dict(values))) # type: ignore[arg-type]
# A staged key the catalog hasn't got is a discovered one (nothing else can stage a
# row), and it is written the only way anything knows how: its own plain `set`.
for key, value in pending.items():
if get_setting(key) is None:
writes.append(Write(f"set {key} {value}", {key: value}))
return writes
+22 -2
View File
@@ -35,11 +35,16 @@ class CachedValue:
supported: ``False`` when the node answered the read with an error — its firmware
has no such setting (a board without a front-end module, a build without a
bridge). Kept apart from *never read*, which is no entry at all.
discovered: ``True`` for a setting the catalog hasn't got, learned from a command
the reader ran on *this node's* command line. It is the row's whole claim to
exist, so a node that stops answering the key loses the row rather than
earning an ``n/a`` — the catalog is what says a key *might* be there.
"""
value: str
read_at: datetime | None
supported: bool = True
discovered: bool = False
class RemoteStore:
@@ -84,7 +89,10 @@ class RemoteStore:
except ValueError:
read_at = None
out[key] = CachedValue(
value=str(entry.get("value", "")), read_at=read_at, supported=supported
value=str(entry.get("value", "")),
read_at=read_at,
supported=supported,
discovered=bool(entry.get("discovered", False)),
)
return out
@@ -92,6 +100,10 @@ class RemoteStore:
"""Cache one setting's value for ``node``, stamped now."""
self._put(node, key, {"value": value, "read_at": utcnow().isoformat()})
def remember_discovered(self, node: Contact, key: str, value: str) -> None:
"""Cache a setting the catalog hasn't got, which ``node`` has just proved it has."""
self._put(node, key, {"value": value, "read_at": utcnow().isoformat(), "discovered": True})
def remember_unsupported(self, node: Contact, key: str) -> None:
"""Record that ``node`` answered a read of ``key`` with an error, stamped now."""
self._put(node, key, {"unsupported": True, "read_at": utcnow().isoformat()})
@@ -104,10 +116,18 @@ class RemoteStore:
self._write(records)
def _put(self, node: Contact, key: str, entry: dict) -> None:
"""Store one setting's cache entry for ``node``."""
"""Store one setting's cache entry for ``node``, keeping it discovered if it was.
A discovered row is refreshed by the same reads and writes as any other — through
:meth:`remember_setting` — and that must not quietly demote it to a catalog row it
has no entry for, which would leave a row nothing draws.
"""
records = self._load_all()
record = records.setdefault(admin_key(node), {})
settings = record.setdefault("settings", {})
previous = settings.get(key)
if isinstance(previous, dict) and previous.get("discovered"):
entry = {**entry, "discovered": True}
settings[key] = entry
self._write(records)
+15 -4
View File
@@ -13,8 +13,9 @@ The input line behaves like a shell:
:class:`~meshterm.core.remote_store.RemoteStore`) so last week's incantation is one
keystroke away next session.
* **Tab** completes the current text against the known repeater commands — the fixed
verbs plus every ``get``/``set`` spelling in the settings catalog — and the best
match previews muted, ghost-text style, after the cursor.
verbs, every ``get``/``set`` spelling in the settings catalog, and any key *this node*
taught us here on an earlier visit — and the best match previews muted, ghost-text
style, after the cursor.
* **Enter** sends. The transcript keeps every exchange of the session, newest at the
bottom, and the view follows the prompt as it grows.
@@ -25,7 +26,7 @@ the radio, the history store, and the reply plumbing) and feeds replies back thr
from __future__ import annotations
from collections.abc import Callable
from collections.abc import Callable, Iterable
from rich.console import Group
from rich.text import Text
@@ -70,6 +71,7 @@ class RemoteCliScreen(Screen):
history: list[str],
send: Callable[[str], None],
session: object,
extra_keys: Iterable[str] = (),
) -> None:
"""Create the command line over a node's persisted history.
@@ -78,6 +80,9 @@ class RemoteCliScreen(Screen):
history: The node's prior commands, oldest first (recalled with ↑).
send: Commits one command (owner-guarded; no-op while one is flying).
session: The running TUI session (for repaints).
extra_keys: Settings *this node* has that the catalog hasn't, found here on an
earlier visit. They complete like any other key: the prompt is where they
were learned, so it is where they should be one Tab away.
"""
super().__init__()
self.title = f"Command line — {node_label}"
@@ -89,7 +94,13 @@ class RemoteCliScreen(Screen):
self._history = list(history)
self._recall: int | None = None # index into history while ↑/↓ browse it
self._draft = "" # what was typed before recall began, restored on ↓ past the end
self._completions = known_commands()
self._completions = sorted(
{
*known_commands(),
*(f"get {key}" for key in extra_keys),
*(f"set {key} " for key in extra_keys),
}
)
self.busy = False
self._spinner = Spinner()
self._prompt_line = 0 # body index of the input line (pinned into view)
+247 -17
View File
@@ -21,6 +21,16 @@ Beyond the settings, the action rows cover the box itself — advert, clock sync
admin password, reboot, each behind its own floating confirmation — and the **Command
line** opens the readline-style remote CLI (:mod:`meshterm.ui.remote_cli`) for anything
the catalog doesn't spell.
That command line is also where the page *grows*. Third-party builds carry settings the
catalog has never heard of, and MeshTerm can't know which build a node is running without
asking it — every question being one paced round trip, and every guessed row a dead ``n/a``
on everyone else's page. So nothing is probed speculatively: a ``get``/``set`` the reader
ran here, which *this node* answered, earns a row on *this node's* page under **Extra**
(:func:`learn_from_cli`). The round trip was one the reader was spending anyway, the catalog
doesn't grow, and no other node's page changes. A row the node stops answering goes again —
which is what tidies up after a board swap onto the same identity — and ``Del`` forgets one
without waiting for a read.
"""
from __future__ import annotations
@@ -36,12 +46,18 @@ 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,
@@ -54,6 +70,7 @@ from .menus import (
PICK_LOCATION_LABEL,
confirm_discard,
exit_rows,
fit_cells,
icon_lane,
lane_header,
lane_row,
@@ -63,7 +80,7 @@ from .menus import (
)
from .trace_screen import TracingDialog
from .tui import Choice, Separator
from .tui.select import SelectScreen, splice_hint
from .tui.select import DeleteRequest, SelectScreen, splice_hint
from .tui.spinner import Spinner, spinner_interval
if TYPE_CHECKING:
@@ -88,6 +105,50 @@ _CANCEL = "__cancel__"
#: The footer atom naming the chord that re-reads the highlighted setting.
READ_ONE_HINT = "^R read"
#: The footer atom naming the key that drops a discovered row (surfaced on those rows only).
FORGET_HINT = "Del forget"
#: The narrowest the value lane may be squeezed to for a discovered setting — a floor, so a
#: page whose catalog rows are all ``?`` still shows something of what the node answered.
_EXTRA_VALUE_MIN = 14
def _extra_value(text: str, width: int) -> str:
"""One discovered value, fitted for the lane: whitespace collapsed, a long reply cut.
A discovered setting's value is whatever the node said, and a build that answers a whole
diagnostic line (``desired=off effective=off supported=yes …``) would otherwise set the
value lane's width — one width, for every row on the page — and push each setting's
description off the edge. The row is cut here instead of the page being reshaped around
it; the command line is where the whole reply is read.
"""
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 ``key`` names on this node: the catalog's, or its own discovered one."""
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]:
"""This node's discovered settings, in key order — rows the catalog knows nothing about."""
return [discovered_setting(key) for key, cached in sorted(cache.items()) if cached.discovered]
def _all_specs(cache: dict) -> list[RemoteSetting]:
"""Every setting this node's page draws: the catalog, plus what it taught us itself."""
return [*REPEATER_SETTINGS, *_discovered_specs(cache)]
def _extra_keys(cache: dict) -> frozenset[str]:
"""The keys on this node's page that came from the node rather than from the catalog."""
return frozenset(key for key, cached in cache.items() if cached.discovered)
@dataclass(frozen=True, slots=True)
class ReadOne:
@@ -121,6 +182,9 @@ class AdminMenu(SelectScreen):
floating = False
#: The keys on this node's page that the catalog hasn't got, refreshed with the rows.
extra_keys: frozenset[str] = frozenset()
def __init__(self, title: str, items: list, **kwargs: Any) -> None:
"""Build the page's list, its rows ending at the edge rather than sliding under ←→.
@@ -134,14 +198,28 @@ class AdminMenu(SelectScreen):
kwargs.setdefault("hscroll", False)
super().__init__(title, items, **kwargs)
def _readable_key(self) -> str | None:
"""The highlighted row's setting key, when it is a setting the node can be asked."""
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
spec = get_setting(current.value)
return current.value
def _readable_key(self) -> str | None:
"""The highlighted row's setting key, when it is a setting the node can be asked."""
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 always re-readable
spec = get_setting(key)
return spec.key if spec is not None and spec.readable else None
def _forgettable_key(self) -> str | None:
"""The highlighted row's key, when it is a discovered row Del can drop."""
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 a readable setting is highlighted."""
@@ -150,16 +228,26 @@ class AdminMenu(SelectScreen):
@property
def fkey_lane(self):
"""The list's lane with ``Read`` on F3 — dim on a row there is nothing to read.
"""The list's lane with ``Read`` on F3 and ``Forget`` behind it — dim where inert.
F1/F2 are this grouped list's section jumps and F4/F5 the pager; F3 is the slot a
select list spends on its own verb. Dim rather than empty on an action row: reading
is a thing on this screen, just not for the row the cursor is on.
select list spends on its own verb. Read and Forget are the same slot's two halves
because they are the same subject — *this row* — and they are never both live: a
catalog row can be read and not forgotten, and only a discovered row can be dropped.
Dim rather than empty on an action row: both are things on this screen, just not for
the row the cursor is on.
"""
from .tui.fkeys import FPair
lane = list(super().fkey_lane)
lane[2] = FPair("Read", "retry", enabled=self._readable_key() is not None)
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:
@@ -292,7 +380,9 @@ async def _admin_session(ctx: AppContext, device: Device, node: Contact) -> dict
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()
@@ -304,13 +394,17 @@ async def _admin_session(ctx: AppContext, device: Device, node: Contact) -> dict
continue
return {"node": node.name, "applied": applied}
if isinstance(choice, ReadOne):
spec = get_setting(choice.key)
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:
await read_settings(ctx, device, node, REPEATER_SETTINGS)
# This node's page, not the catalog: a discovered row is refreshed by the
# same sweep as everything else, and drops off it 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:
@@ -349,7 +443,7 @@ async def _admin_session(ctx: AppContext, device: Device, node: Contact) -> dict
danger=True,
)
else: # a setting key
spec = get_setting(str(choice))
spec = _spec_for(str(choice), cache)
if spec is not None and not spec.writable:
# A read-only fact has nothing to stage; asking again is all Enter can do.
await read_settings(ctx, device, node, [spec])
@@ -359,6 +453,7 @@ async def _admin_session(ctx: AppContext, device: Device, node: Contact) -> dict
# the values are the ones the action just 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)
@@ -383,6 +478,25 @@ def _menu_items(node: Contact, cache: dict, pending: dict[str, str]) -> tuple[st
rows.append((spec.label, _value_text(spec, cache, pending), spec.help, spec.key))
sections.append((category, rows))
# What this node taught us itself, last and in its own section — absent entirely on a
# node that has taught us nothing, which is every node until someone asks it something.
# Its values are fitted to the lane the catalog's rows already need: the value lane is
# one width for the whole page, so an unbounded reply here would push every setting's
# description right, off the edge of a 72-column screen, and reshape a page the reader
# 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)
@@ -394,7 +508,13 @@ def _menu_items(node: Contact, cache: dict, pending: dict[str, str]) -> tuple[st
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)
Choice(
title=lane_row(label, value, help_text, label_w, value_w),
value=key,
# Only a discovered row can be dropped: a catalog row that goes away
# would come straight back on the next paint, the catalog still naming it.
deletable=category == DISCOVERED_CATEGORY,
)
)
# ↻ and ⌨ are one cell where 📡 🕒 🔐 🔄 are two, so the column is measured once and
@@ -424,7 +544,9 @@ def _menu_items(node: Contact, cache: dict, pending: dict[str, str]) -> tuple[st
return title, items
def _value_text(spec: RemoteSetting, cache: dict, pending: dict[str, str]) -> Text:
def _value_text(
spec: RemoteSetting, cache: dict, pending: dict[str, str], width: int = _EXTRA_VALUE_MIN
) -> Text:
"""One setting's VALUE lane: what the node last said, and any staged arrow.
Four states, each its own word: the value; ``empty`` for a string the node holds blank;
@@ -441,10 +563,14 @@ def _value_text(spec: RemoteSetting, cache: dict, pending: dict[str, str]) -> Te
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:
value.append(f" → {spec.display(pending[spec.key])}", style="warn")
staged = pending[spec.key]
shown = _extra_value(staged, width) if spec.discovered else spec.display(staged)
value.append(f" → {shown}", style="warn")
return value
@@ -453,7 +579,7 @@ def _value_text(spec: RemoteSetting, cache: dict, pending: dict[str, str]) -> Te
async def _stage_setting(ctx: AppContext, key: str, cache: dict, pending: dict[str, str]) -> None:
"""Prompt for one setting's new value and stage it (nothing is sent yet)."""
spec = get_setting(key)
spec = _spec_for(key, cache)
if spec is None or not spec.writable: # pragma: no cover - menu offers only real keys
return
cached = cache.get(key)
@@ -508,6 +634,33 @@ async def _stage_setting(ctx: AppContext, key: str, cache: dict, pending: dict[s
pending[key] = value
async def _forget_discovered(
ctx: AppContext, node: Contact, key: str, pending: dict[str, str]
) -> None:
"""Drop one discovered row from this node's page, behind a confirm.
The row usually cleans itself up — a node that stops answering the key loses it on the
next read — but that costs a round trip and a node in reach, and neither is a given
after the board it was learned from has been replaced. So Del asks, and forgets. It is
a red confirm like any other single-record delete, though nothing on the node changes:
what goes is what MeshTerm remembers, and asking the key again brings the row 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:
"""Pick the node's advertised location on the map and stage both coordinates it sets.
@@ -598,8 +751,11 @@ async def _apply(ctx: AppContext, device: Device, node: Contact, pending: dict[s
remember_reply(ctx, node, fills, reply)
writes = write_plan(pending, _known_values(ctx, node))
# The catalog plus this node's own discovered rows: a staged 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 REPEATER_SETTINGS if s.key in write.values and s.key in pending]
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)
@@ -668,13 +824,85 @@ def remember_reply(ctx: AppContext, node: Contact, fills: list[RemoteSetting], r
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)
if spec.discovered:
# A discovered row's whole claim to exist is that this node answered the
# key. When it stops, the claim is gone and so is the row — ``n/a`` is for
# a key the *catalog* says might be there, and nothing says that here. This
# is what cleans the page up after a board swap onto the same identity.
ctx.remote_store.forget_setting(node, spec.key)
else:
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:
"""Fold what one command-line exchange proved about ``node`` into its page.
The command line is the only place a setting outside the catalog can be reached, so it
is also the only place one can be *found*: a ``get``/``set`` this node answered is proof
the key is there, and the row it earns costs no round trip nobody asked for. The catalog
doesn't grow — a key learned here belongs to this node alone (see
:func:`~meshterm.core.remote_config.discovered_setting`), and every other node's page is
what it always was.
Three cases, in the order they are tested:
* A **composite's own key** (``get radio``) fills all four of its fields, exactly as a
read of any one of them does.
* A **catalog key** folds into the cache the way the sweep or an Apply would — which is
also how ``get tx`` at the command line stopped leaving the TX power row stale.
* Anything else is **discovered**, and the two verbs are not equally good evidence.
``handleSetCmd`` matches a key with its trailing space and refuses an unknown one
outright, so an accepted write is proof. ``handleGetCmd`` matches on a bare prefix and
will answer a key it hasn't got out of a shorter key's branch, so a read is proof only
when the reply cannot be that shorter key's
(:func:`~meshterm.core.remote_config.learnable_from_get`).
``prv.key`` is never stored, whichever way it was asked
(: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 says nothing about the key, and nothing about 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 # sent in words the catalog can't read back; leave 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:
@@ -870,6 +1098,7 @@ async def _command_line(ctx: AppContext, device: Device, node: Contact) -> None:
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(
@@ -877,6 +1106,7 @@ async def _command_line(ctx: AppContext, device: Device, node: Contact) -> None:
history=ctx.remote_store.history(node),
send=send,
session=session,
extra_keys=_extra_keys(ctx.remote_store.settings(node)),
)
async def animate() -> None:
+200
View File
@@ -23,12 +23,16 @@ from meshterm.core.connection import MockDevice
from meshterm.core.device_store import DeviceStore
from meshterm.core.models import Contact, utcnow
from meshterm.core.remote_config import (
DISCOVERED_HELP,
REPEATER_SETTINGS,
SettingCommand,
composite_reads,
discovered_setting,
get_setting,
known_commands,
normalize_value,
parse_reply_value,
parse_setting_command,
range_hint,
read_plan,
reply_is_error,
@@ -780,3 +784,199 @@ def test_admin_actions_start_every_label_in_the_same_cell() -> None:
starts[label] = cell_len(plain[: plain.index(label)])
assert set(starts) == set(labels)
assert len(set(starts.values())) == 1, f"a label starts a column early: {starts}"
# --- settings discovered on the command line -------------------------------------------
def test_only_get_and_set_are_read_as_settings() -> None:
"""The CLI's own grammar is the filter; every top-level verb falls out of it.
Nothing is excluded by name: ``reboot``, ``advert`` and ``powersaving`` are simply not
``get <key>`` or ``set <key> <value>``, and ``setperm`` only starts with the letters.
"""
assert parse_setting_command("get radio.rxps") == SettingCommand("get", "radio.rxps")
assert parse_setting_command(" set reboot.interval 24 ") == SettingCommand(
"set", "reboot.interval", "24"
)
assert parse_setting_command("set owner.info a b c") == SettingCommand(
"set", "owner.info", "a b c"
)
for command in (
"reboot",
"advert",
"clock sync",
"powersaving on",
"setperm ab12 rw",
"start ota",
"get", # no key
"set tx", # no value
"get ../etc", # not a key's spelling
):
assert parse_setting_command(command) is None, command
def test_a_write_the_node_accepts_is_proof_the_key_is_there(tui_ctx) -> None:
"""``set`` teaches a row: the firmware matches its keys exactly and refuses the rest.
``handleSetCmd`` compares ``"radio "`` *with* the trailing space, so a longer key can
never land in a shorter one's branch, and an unknown one answers ``unknown config:``.
"""
repeater_admin.learn_from_cli(tui_ctx, NODE, "set radio.rxps balanced", "OK")
cached = tui_ctx.remote_store.settings(NODE)["radio.rxps"]
assert (cached.value, cached.discovered) == ("balanced", True)
repeater_admin.learn_from_cli(tui_ctx, NODE, "set nonesuch 1", "unknown config: nonesuch")
assert "nonesuch" not in tui_ctx.remote_store.settings(NODE)
def test_a_read_answered_out_of_a_shorter_keys_branch_is_no_discovery(tui_ctx) -> None:
"""``get`` matches on a bare prefix, so a plausible reply can belong to another key.
Stock firmware answers ``get radio.rxps`` from the ``get radio`` branch. With no typing
to reject it — a discovered key has none — the guard is that the reply is exactly what
the shadowing key last said.
"""
for key, value in (("freq", "910.525"), ("bw", "62.5"), ("sf", "7"), ("cr", "5")):
tui_ctx.remote_store.remember_setting(NODE, key, value)
repeater_admin.learn_from_cli(tui_ctx, NODE, "get radio.rxps", "> 910.525,62.5,7,5")
assert "radio.rxps" not in tui_ctx.remote_store.settings(NODE)
repeater_admin.learn_from_cli(tui_ctx, NODE, "get radio.rxps", "> desired=off level=2")
assert tui_ctx.remote_store.settings(NODE)["radio.rxps"].value == "desired=off level=2"
def test_a_read_nothing_shadows_is_taken_at_its_word(tui_ctx) -> None:
"""No catalog key prefixes ``reboot.interval``, so its reply can only be its own."""
repeater_admin.learn_from_cli(tui_ctx, NODE, "get reboot.interval", "> 24")
assert tui_ctx.remote_store.settings(NODE)["reboot.interval"].value == "24"
repeater_admin.learn_from_cli(tui_ctx, NODE, "get nonesuch", "??: nonesuch")
assert "nonesuch" not in tui_ctx.remote_store.settings(NODE)
def test_an_unread_shadow_leaves_the_question_unanswerable(tui_ctx) -> None:
"""A shadowed key whose shadow was never read is not discovered: we cannot tell."""
repeater_admin.learn_from_cli(tui_ctx, NODE, "get radio.rxps", "> 910.525,62.5,7,5")
assert "radio.rxps" not in tui_ctx.remote_store.settings(NODE)
def test_the_private_key_is_never_written_to_disk(tui_ctx, tmp_path: Path) -> None:
"""``get prv.key`` prints in the transcript and goes no further."""
secret = "ab" * 32
repeater_admin.learn_from_cli(tui_ctx, NODE, "get prv.key", "> " + secret)
assert "prv.key" not in tui_ctx.remote_store.settings(NODE)
written = [p for p in tmp_path.rglob("*.json") if secret in p.read_text(encoding="utf-8")]
assert written == []
def test_a_catalog_key_read_on_the_command_line_refreshes_its_row(tui_ctx) -> None:
"""The cache is the page, so the command line fills it exactly as a read does.
``get tx`` used to leave the TX power row showing what it said last week, and
``get radio`` answered four rows that none of them heard about.
"""
repeater_admin.learn_from_cli(tui_ctx, NODE, "get tx", "> 20")
repeater_admin.learn_from_cli(tui_ctx, NODE, "get radio", "> 869.525,250,10,6")
repeater_admin.learn_from_cli(tui_ctx, NODE, "set txdelay 1.5", "OK")
cache = tui_ctx.remote_store.settings(NODE)
assert cache["tx"].value == "20"
assert (cache["freq"].value, cache["sf"].value) == ("869.525", "10")
assert cache["txdelay"].value == "1.5"
assert not any(cached.discovered for cached in cache.values())
def test_a_write_stales_the_other_spelling_from_the_command_line_too(tui_ctx) -> None:
"""``set dutycycle`` at the prompt drops the cached ``af``, as an Apply would."""
tui_ctx.remote_store.remember_setting(NODE, "af", "1.00")
repeater_admin.learn_from_cli(tui_ctx, NODE, "set dutycycle 40", "OK - 40.0%")
cache = tui_ctx.remote_store.settings(NODE)
assert cache["dutycycle"].value == "40.0" and "af" not in cache
def test_a_discovered_row_survives_a_re_read_and_goes_when_the_key_does(tui_ctx) -> None:
"""The node's answer is the row's whole claim to exist — including after a board swap.
A refreshed value keeps the row discovered (or the page would hold a row the catalog
cannot draw); a key that stops answering loses it, rather than earning the ``n/a`` that
only a catalog key can.
"""
spec = discovered_setting("radio.rxps")
tui_ctx.remote_store.remember_discovered(NODE, "radio.rxps", "off")
repeater_admin.remember_reply(tui_ctx, NODE, [spec], "> balanced")
cached = tui_ctx.remote_store.settings(NODE)["radio.rxps"]
assert (cached.value, cached.discovered) == ("balanced", True)
repeater_admin.remember_reply(tui_ctx, NODE, [spec], "??: radio.rxps")
assert "radio.rxps" not in tui_ctx.remote_store.settings(NODE)
def test_discovered_rows_draw_last_in_their_own_section_and_are_deletable() -> None:
"""A node's own settings sit under ``Extra`` — and only they answer Del."""
from meshterm.ui.tui import Choice
plain = {"txdelay": CachedValue("0.5", utcnow())}
catalog_rows = [
item
for item in repeater_admin._menu_items(NODE, plain, {})[1]
if isinstance(item, Choice) and item.deletable
]
assert catalog_rows == [], "no catalog row may be forgotten"
cache = dict(plain, **{"radio.rxps": CachedValue("balanced", utcnow(), discovered=True)})
items = repeater_admin._menu_items(NODE, cache, {})[1]
deletable = [i for i in items if isinstance(i, Choice) and i.deletable]
assert [i.value for i in deletable] == ["radio.rxps"]
rows = [i for i in items if isinstance(i, Choice) and isinstance(i.value, str)]
settings_rows = [r for r in rows if repeater_admin.get_setting(r.value) or r.deletable]
assert settings_rows[-1].value == "radio.rxps", "the node's own settings come last"
assert DISCOVERED_HELP in settings_rows[-1].title.plain
def test_a_staged_discovered_value_is_written_with_its_own_set() -> None:
"""A key the catalog hasn't got is still written the one way anything can write it."""
writes = write_plan({"radio.rxps": "balanced", "txdelay": "1.5"}, {})
assert [w.command for w in writes] == ["set txdelay 1.5", "set radio.rxps balanced"]
def test_a_long_discovered_value_cannot_widen_the_whole_page() -> None:
"""The value lane is one width for every row, so a node's own setting may not set it.
Some builds answer ``get radio.rxps`` with a whole diagnostic line. Unbounded, that one
row would push every catalog setting's description right and off a 72-column screen —
reshaping a page the reader opened to read the catalog.
"""
from rich.cells import cell_len
from meshterm.ui.tui import Choice
def widest(cache) -> int:
items = repeater_admin._menu_items(NODE, cache, {})[1]
return max(
cell_len(i.title.plain)
for i in items
if isinstance(i, Choice) and hasattr(i.title, "plain")
)
catalog = {
"tx": CachedValue("20", utcnow()),
"owner.info": CachedValue("Jean-Pierre, Montreal", utcnow()),
}
chatty = dict(
catalog,
**{
"radio.rxps": CachedValue(
"desired=off effective=off supported=yes level=0 preamble=16 rx=0 sleep=0",
utcnow(),
discovered=True,
)
},
)
assert widest(chatty) == widest(catalog)
fitted = repeater_admin._extra_value("a value far too long for its lane", 14)
assert cell_len(fitted) == 14
assert repeater_admin._extra_value(" balanced\n", 14) == "balanced"