Move every command onto the seam, and give the plain face its own premise

Forty-nine leaf commands now state a Report instead of printing one, so --json
works on all of them at once rather than on the two that grew a branch for it.
The two hand-rolled JSON paths are gone: devices no longer builds a payload dict
of its own, and its empty scan exits 5 like the plain path always did rather than
0 — a rendering flag has no business moving the report.

The plain face changes premise along with it. It used to be a serialisation
format a person could squint at, and several of its rules existed only to make
splitting safe. Splitting is the machine face's job now, so:

- Names are bare. Alignment is the delimiter; the escaping under the quoting
  stays, because nothing else stops a name a node broadcast from ending its own
  record.
- Times are ages. "Heard recently?" is the question a contacts listing is opened
  to ask, and an ISO instant makes the reader do arithmetic to answer it.
  --absolute puts the instants back; an appointment, the device clock and a live
  capture's own column were never ages and still aren't.
- A route takes arrows and a path keeps its commas. The lexicon has always split
  these and the output never did, so a route line looked pasteable into --path,
  which it is not.
- The listings gain what was cut for parser hygiene: contacts gets the HASH
  --path takes and a LOCATION the model has always carried, preferences gets the
  DESCRIPTION already written in its registry, and the two live streams get
  pinned lanes — which is what let them have columns at all, and with them the
  last call site for the quoting.
- The four written pages wrap at 72 instead of printing 600-cell paragraphs. It
  is the one exception to "no wrapping", and the rule's own reason says why: a
  paragraph has no fields to land under the wrong headings.

info keeps its keys and gains a gloss on the four readings nobody can hold in
their head (93784 (1d 2h)); it is the one key/value block with no set behind it,
so there is no round trip to protect. config show is untouched, deliberately.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jean-Pierre MartineauandClaude Opus 5 committed 2026-09-08 03:43:00 -04:00
1 parent 1c24f9149d
commit 109fe7645d
23 files changed
+2244 -828

No files matched your search

+26 -15
View File
@@ -468,21 +468,36 @@ def platform_command() -> None:
the flavour and the usual reason a Windows session looks plainer than the screenshots
(see :func:`meshterm.ui.termfont.emoji_support`).
"""
from .ui import fields
from .ui.report import Facts
assert _platform_resolution is not None # set by the callback that always runs first
assert _state is not None # ditto
r = _platform_resolution
emoji = emoji_support()
_state.console.print(
script.pairs(
[
("platform", r.platform.name),
("platform_source", r.source),
("flag", r.flag or script.NONE),
("env", r.env or script.NONE),
("device_tree_model", r.detected_model or script.NONE),
("icons", "yes" if emoji.supported else "no"),
("icons_source", emoji.source),
]
renderers.for_format(_state.output, _state.console).render(
(
Facts(
key="platform",
fields=(
fields.word("platform", "platform"),
fields.word("platform_source", "platform_source"),
fields.word("flag", "flag"),
fields.word("env", "env"),
fields.word("device_tree_model", "device_tree_model"),
fields.flag("icons", "icons"),
fields.word("icons_source", "icons_source"),
),
values={
"platform": r.platform.name,
"platform_source": r.source,
"flag": r.flag or None,
"env": r.env or None,
"device_tree_model": r.detected_model or None,
"icons": emoji.supported,
"icons_source": emoji.source,
},
),
)
)
@@ -653,10 +668,6 @@ async def _execute_and_render(tool: Tool, params: dict, ctx: AppContext) -> Tool
"""
result = await tool.execute(ctx, params)
renderers.for_format(ctx.output, ctx.console).render(result.report)
# A produced file is a fact, and its path is the whole of it: bare, one per line, so
# `meshterm config backup out.toml` can be read by the thing that runs it.
for artifact in result.artifacts:
ctx.console.print(artifact, highlight=False)
return result
+87 -3
View File
@@ -26,6 +26,7 @@ from .base import Tool, ToolResult, register
if TYPE_CHECKING:
from ..ui.markdown import MarkdownDoc
from ..ui.report import Facts
class _AboutTool(Tool):
@@ -67,7 +68,7 @@ class _AboutTool(Tool):
return None
async def run(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
"""Print the page — only reachable from the CLI (the menu opens the screen).
"""State the page — only reachable from the CLI (the menu opens the screen).
Without its scannable codes: a QR is a second rendering of a link the page already
prints, drawn for a phone pointed at a screen, and redirected into a file it is a
@@ -80,8 +81,10 @@ class _AboutTool(Tool):
Returns:
A :class:`ToolResult` naming the page that was shown.
"""
ctx.ui.show(self.page().for_script())
return ToolResult(summary={"page": self.name})
return ToolResult(
summary={"page": self.name},
report=(_written_page(self.name, self.title, self.page().for_script()),),
)
@register
@@ -154,3 +157,84 @@ class SupportProjectTool(_AboutTool):
from ..ui.about import support_project
return support_project()
def _written_page(name: str, title: str, doc: MarkdownDoc) -> Facts:
"""One prose page, stated once for both faces.
**Wrapped**, and that is the one deliberate exception to "no line wrapping". The rule
exists so a *record* is never split across two lines with its fields landing under the
wrong headings; a paragraph has no records and no headings, and left unwrapped these
four pages print as single six-hundred-cell lines no terminal can read. So the page is
drawn on a console :data:`~meshterm.ui.script.PAGE_WIDTH` cells wide — a fixed width,
not the terminal's, so the same paragraph reads the same piped, redirected, or shipped
as the document's ``text``.
``links`` is the one machine-actionable thing on a page of prose, and the one thing the
menu bothers to draw a QR of. ``text`` is the page's plain rendering rather than a fold
of the markdown into nested JSON: a structured markdown tree would be a second
rendering, not data, and nothing would consume it.
"""
from .. import __version__
from ..ui import fields
from ..ui.report import BARE, Column, Facts
text = _rendered_page(doc)
return Facts(
key="page",
fields=(
fields.word("page", "page"),
fields.word("title", "title"),
fields.word("version", "version"),
fields.word("text", "text"),
Column(key="links"),
),
values={
"page": name,
"title": title,
# Duplicated from the package on purpose: a page's own facts are part of its
# content, and this is the version its `{version}` placeholder resolved to.
"version": __version__,
"text": text,
"links": _links(text),
},
shape=BARE,
bare="text",
)
def _rendered_page(doc: MarkdownDoc) -> str:
"""The page as plain text, folded to the fixed page width."""
import io
from rich.console import Console
from ..ui import script
buffer = io.StringIO()
Console(
file=buffer,
width=script.PAGE_WIDTH,
color_system=None,
highlight=False,
emoji=False,
markup=False,
).print(doc)
return "\n".join(line.rstrip() for line in buffer.getvalue().splitlines()).strip("\n")
def _links(text: str) -> list[str]:
"""Every URL the page carries, in document order, de-duplicated.
Read off the rendered page rather than the markdown source, so a link written as a
reference and a link written inline both count once and both read as what the reader
sees. Trailing punctuation is not part of a URL and is trimmed.
"""
import re
seen: list[str] = []
for match in re.findall(r"https?://[^\s<>()\[\]]+", text):
url = match.rstrip(".,;:*_")
if url not in seen:
seen.append(url)
return seen
+134 -30
View File
@@ -12,7 +12,7 @@ everything to do with configuring the slots themselves.
from __future__ import annotations
from typing import Any
from typing import TYPE_CHECKING, Any
import typer
from rich.console import Group
@@ -21,7 +21,9 @@ from rich.text import Text
from ..context import AppContext
from ..core import exitcodes
from ..core.channels import (
channel_hash,
derive_secret,
is_public_channel,
normalize_secret,
parse_share_url,
random_secret,
@@ -29,6 +31,9 @@ from ..core.channels import (
)
from .base import Tool, ToolResult, register
if TYPE_CHECKING: # pragma: no cover - typing only
from ..ui.report import Facts, Listing
@register
class ChannelsTool(Tool):
@@ -84,23 +89,15 @@ class ChannelsTool(Tool):
async def _cli_list(self, ctx: AppContext) -> ToolResult:
"""List the configured channel slots."""
from ..ui import script
from ..ui.channels import read_channel_slots
device = await ctx.device()
slots = await read_channel_slots(device)
if not slots:
return ToolResult(summary={"channels": 0}, exit_code=exitcodes.NO_RESULT)
table = script.columns("SLOT", "NAME", "TYPE", "HASH", right=("SLOT",))
for slot in slots:
table.add_row(
str(slot.idx),
script.name(slot.name),
"public" if slot.is_public else "private",
slot.hash,
)
ctx.ui.show(table)
return ToolResult(summary={"channels": len(slots)})
return ToolResult(
summary={"channels": len(slots)},
report=(_channel_listing(slots),),
exit_code=exitcodes.OK if slots else exitcodes.NO_RESULT,
)
async def _cli_add(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
"""Add a public, private-random, or explicitly-keyed channel on a slot."""
@@ -115,8 +112,11 @@ class ChannelsTool(Tool):
else:
secret = normalize_secret(params["secret"]) if params.get("secret") else random_secret()
await write_channel(ctx, device, idx, name, secret)
self._print_share(ctx, name, secret)
return ToolResult(summary={"index": idx, "name": name})
self._show_qr(ctx, share_url(name, secret))
return ToolResult(
summary={"index": idx, "name": name},
report=(_written(idx, name, secret),),
)
async def _cli_join(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
"""Join a channel from a name and an explicit key."""
@@ -127,7 +127,10 @@ class ChannelsTool(Tool):
name = str(params["name"])
secret = normalize_secret(str(params["secret"]))
await write_channel(ctx, device, idx, name, secret)
return ToolResult(summary={"index": idx, "name": name})
return ToolResult(
summary={"index": idx, "name": name},
report=(_written(idx, name, secret, shown=False),),
)
async def _cli_import(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
"""Import a channel from a ``meshcore://channel/add`` link."""
@@ -140,10 +143,18 @@ class ChannelsTool(Tool):
device = await ctx.device()
idx = int(params["index"])
await write_channel(ctx, device, idx, name, secret)
return ToolResult(summary={"index": idx, "name": name})
return ToolResult(
summary={"index": idx, "name": name},
report=(_written(idx, name, secret, shown=False),),
)
async def _cli_share(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
"""Print a channel's share link and QR code."""
"""Print a channel's share link, with its QR code in the menu.
The one command that names a channel's key deliberately, so it is the one place
the secret and the share URL leave the app at all — every other channel document
carries the identity and stops there.
"""
from ..ui.channels import read_channel_slots
device = await ctx.device()
@@ -151,10 +162,15 @@ class ChannelsTool(Tool):
slot = next((s for s in await read_channel_slots(device) if s.idx == idx), None)
if slot is None:
return ToolResult(
summary={"index": idx, "shared": False}, exit_code=exitcodes.NO_RESULT
summary={"index": idx, "shared": False},
report=(_written(idx, None, None, shown=False),),
exit_code=exitcodes.NO_RESULT,
)
self._print_share(ctx, slot.name, slot.secret)
return ToolResult(summary={"index": idx, "shared": True})
self._show_qr(ctx, share_url(slot.name, slot.secret))
return ToolResult(
summary={"index": idx, "shared": True},
report=(_written(idx, slot.name, slot.secret),),
)
async def _cli_clear(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
"""Clear a channel slot, removing the channel from the device.
@@ -171,27 +187,29 @@ class ChannelsTool(Tool):
slot = next((s for s in await read_channel_slots(device) if s.idx == idx), None)
if slot is None:
return ToolResult(
summary={"index": idx, "cleared": False}, exit_code=exitcodes.NO_RESULT
summary={"index": idx, "cleared": False},
report=(_cleared(idx, False),),
exit_code=exitcodes.NO_RESULT,
)
await write_channel(ctx, device, idx, "", None) # empty name => the slot reads as unused
return ToolResult(summary={"index": idx, "cleared": True})
return ToolResult(
summary={"index": idx, "cleared": True},
report=(_cleared(idx, True),),
)
@staticmethod
def _print_share(ctx: AppContext, name: str, secret: bytes) -> None:
"""Render a channel's share link — with its QR code in the menu, without on the CLI.
def _show_qr(ctx: AppContext, url: str) -> None:
"""Draw a channel's share link as a QR code — in the menu, and only there.
The QR is for a phone pointed at the screen. Piped into a file it is a block of
block characters wrapped around the one thing that is actually the answer, so a
scripted run prints the link alone.
scripted run states the link and nothing else.
"""
from ..ui.qr import qr_text
from ..ui.surface import TuiUi
url = share_url(name, secret)
if isinstance(ctx.ui, TuiUi):
ctx.ui.show(Group(qr_text(url), Text(""), Text(url, style="accent")))
else:
ctx.ui.note(url)
def register_cli(self, app: typer.Typer) -> None:
"""Register the ``channels`` subcommand group.
@@ -259,3 +277,89 @@ class ChannelsTool(Tool):
run_tool_command(self, {"cli_action": "clear", "index": index})
app.add_typer(channels_app, name=self.name)
def _channel_listing(slots: list) -> Listing:
"""The configured slots, one record each.
The one listing whose record *is* a shared shape rather than carrying one: a row that
nested its only field under a ``channel`` key would make ``jq '.[].channel.name'``
out of a listing whose every column is already the channel.
"""
from ..ui import fields
from ..ui.report import Listing
return Listing(
key="channels",
columns=(
fields.integer("slot", "SLOT"),
fields.name("name", "NAME"),
fields.word("type", "TYPE"),
fields.hexid("hash", "HASH"),
),
rows=[
{
"slot": slot.idx,
"name": slot.name,
"type": "public" if slot.is_public else "private",
"hash": slot.hash,
}
for slot in slots
],
)
def _written(idx: int, name: str | None, secret: bytes | None, *, shown: bool = True) -> Facts:
"""What a slot now holds, after ``add``, ``join``, ``import`` or ``share``.
All four write or read the same thing and all four know the key they wrote, so all
four state the same shape. The plain face prints the share URL alone where the caller
asked to be given something to pass on (``add``, ``share``) and nothing at all where
it did not (``join``, ``import``): those two were handed the key, and reading it back
to them is not an answer.
Args:
idx: The slot index.
name: The channel name, or ``None`` for a slot that turned out to be empty.
secret: The 16-byte key, or ``None`` for an empty slot.
shown: Whether the plain face prints the URL.
Returns:
The facts block.
"""
from ..ui import fields
from ..ui.fields import ChannelRef
from ..ui.report import BARE, SILENT, Facts
channel = None
if name is not None and secret is not None:
public = is_public_channel(name, secret)
channel = ChannelRef(slot=idx, name=name, public=public, hash=channel_hash(secret))
return Facts(
key="channel",
fields=(
fields.channel("channel"),
fields.hexid("secret", "secret"),
fields.word("url", "url"),
),
values={
"channel": channel,
"secret": secret.hex() if secret else None,
"url": share_url(name, secret) if name is not None and secret is not None else None,
},
shape=BARE if shown else SILENT,
bare="url",
)
def _cleared(idx: int, cleared: bool) -> Facts:
"""What ``channels clear`` did — nothing plain, since the status already said it."""
from ..ui import fields
from ..ui.report import SILENT, Facts
return Facts(
key="cleared",
fields=(fields.integer("slot", "slot"), fields.flag("cleared", "cleared")),
values={"slot": idx, "cleared": cleared},
shape=SILENT,
)
+251 -76
View File
@@ -18,7 +18,6 @@ from collections.abc import Callable
from typing import TYPE_CHECKING, Any
import typer
from rich.table import Table
from rich.text import Text
from ..context import AppContext
@@ -26,19 +25,24 @@ from ..core import exitcodes
from ..core.channels import (
CHANNEL_SLOT_PROBE_CAP,
DEFAULT_PUBLIC_SECRET,
channel_hash,
channel_identity,
is_public_channel,
)
from ..core.connection import Device
from ..core.events import EventKind, MeshEvent
from ..core.models import (
NODE_TYPE_CHAT,
NODE_TYPE_LABELS,
ChatMessage,
Contact,
Conversation,
is_direct_messageable,
)
from ..services.trace_runner import NameKeyResolver, make_name_key_resolver
from ..ui import renderers
from ..ui.chat import _MENTION, _split_channel_sender
from ..ui.fields import ChannelRef, NodeRef
from ..ui.menus import Lane, column_header, fit_cells, section_heading
from ..ui.theme import name_style
from ..ui.tui import Choice, DeleteRequest, Separator
@@ -47,6 +51,7 @@ from .base import Tool, ToolResult, register
if TYPE_CHECKING:
from ..ui.channels import ChannelSlot
from ..ui.report import Facts, Listing
#: How many recent messages ``chat history`` prints by default.
_HISTORY_LIMIT = 50
@@ -261,19 +266,22 @@ class ChatTool(Tool):
already tell the caller: the radio accepted it either way, and whether the peer
answered is a separate fact.
"""
from ..ui import script
device = await ctx.device()
text = str(params["text"])
channel = params.get("channel")
if channel is not None:
await ctx.chat.send_channel(int(channel), text, label=f"#{channel}")
return ToolResult(summary={"channel": channel, "sent": True})
return ToolResult(
summary={"channel": channel, "sent": True},
report=(_sent(channel=int(channel)),),
)
contact = _resolve_contact(await device.get_contacts(), str(params["to"]))
message = await ctx.chat.send_direct(contact, text)
ctx.ui.show(script.pairs([("acked", "yes" if message.acked else "no")]))
return ToolResult(summary={"to": contact.name, "acked": bool(message.acked)})
return ToolResult(
summary={"to": contact.name, "acked": bool(message.acked)},
report=(_sent(contact=contact, acked=bool(message.acked)),),
)
async def _cli_history(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
"""Print a conversation's stored transcript."""
@@ -296,10 +304,9 @@ class ChatTool(Tool):
)
label = contact.name
if messages:
ctx.ui.show(_history_table(label, messages))
return ToolResult(
summary={"conversation": label, "messages": len(messages)},
report=(_transcript(messages),),
exit_code=exitcodes.OK if messages else exitcodes.NO_RESULT,
)
@@ -332,11 +339,6 @@ class ChatTool(Tool):
highlight=False,
)
seen = 0
# One header, then a record per message. Like the monitor's tail this cannot be
# column-aligned (the widths arrive with the data), so the fields are gutter-
# separated and the two that can hold anything — the conversation and the body —
# are quoted and last respectively.
ctx.console.print("TIME CONVERSATION SNR_DB TEXT", highlight=False)
def on_message(event: MeshEvent) -> None:
nonlocal seen
@@ -344,25 +346,35 @@ class ChatTool(Tool):
if message is None:
return
seen += 1
where = f"#{message.channel}" if message.is_channel else (message.sender or "")
fields = [
script.stamp(message.received_at),
script.name(where),
script.number(message.snr, "+.1f"),
script.oneline(message.text),
]
ctx.console.print((" " * script.GUTTER).join(fields), highlight=False)
channel = (
_channel_ref(message.channel, f"#{message.channel}", None)
if message.is_channel
else None
)
emit(
{
"received_at": message.received_at,
"conversation": channel.name if channel else (message.sender or None),
"direction": "in",
"node": None if message.is_channel else NodeRef(hash=message.sender),
"channel": channel,
"snr_db": message.snr,
"text": message.text,
"acked": None,
}
)
unsubscribe = ctx.events.subscribe(on_message, EventKind.MESSAGE)
try:
if seconds:
await asyncio.sleep(seconds)
else:
await asyncio.Event().wait() # until Ctrl-C / cancellation
except (KeyboardInterrupt, asyncio.CancelledError): # pragma: no cover - interactive
pass
finally:
unsubscribe()
with renderers.stream(ctx, _live_messages()) as emit:
unsubscribe = ctx.events.subscribe(on_message, EventKind.MESSAGE)
try:
if seconds:
await asyncio.sleep(seconds)
else:
await asyncio.Event().wait() # until Ctrl-C / cancellation
except (KeyboardInterrupt, asyncio.CancelledError): # pragma: no cover - interactive
pass
finally:
unsubscribe()
return ToolResult(
summary={"messages": seen},
message=f"[muted]stopped — heard {seen} message{'' if seen == 1 else 's'}[/muted]",
@@ -376,30 +388,12 @@ class ChatTool(Tool):
contacts = await device.get_contacts()
lasts = ctx.repo.last_chat_messages()
from ..ui import script
rows = [*channels] + [
Conversation(label=c.name, is_channel=False, contact=c) for c in contacts
]
table = script.columns(
"CONVERSATION", "KIND", "UNREAD", "LAST_TIME", "LAST_TEXT", right=("UNREAD",)
)
for conversation in rows:
last = lasts.get(conversation.key)
table.add_row(
script.name(conversation.label),
"channel" if conversation.is_channel else "direct",
str(ctx.chat.unread(conversation.key)),
script.stamp(last.created_at) if last is not None else script.NONE,
# Not truncated to 40 cells the way the picker's preview is: nothing here
# wraps, so there is no reason to cut a message short — but a body that
# holds a newline would still end the record, so it is folded flat.
script.oneline(last.text) if last is not None else script.NONE,
)
if rows:
ctx.ui.show(table)
return ToolResult(
summary={"conversations": len(rows)},
report=(_conversations(ctx, rows, lasts),),
exit_code=exitcodes.OK if rows else exitcodes.NO_RESULT,
)
@@ -831,37 +825,218 @@ def _ago(when: Any) -> str:
return _format_age(_age_seconds(when))
def _history_table(label: str, messages: list[ChatMessage]) -> Table:
"""A conversation's stored messages, oldest first, one record each.
def _channel_ref(idx: int | None, label: str, secret: bytes | None) -> ChannelRef:
"""One channel as the shared shape, from whatever the surface knows about it.
``DIRECTION`` carries what the transcript wrote into its sender lane as the word
A transcript row and a live message know a slot and a label but not the key, so
``public`` is read off the name where the secret is not in hand — which is what the
leading ``#`` means and how a public channel is written everywhere else in the app.
"""
return ChannelRef(
slot=int(idx or 0),
name=label,
public=is_public_channel(label, secret) if secret else label.startswith("#"),
hash=channel_hash(secret) if secret else None,
)
def _message_columns() -> tuple:
"""One transcript row's columns, shared by ``history`` and the ``listen`` stream.
``DIR`` carries what the menu's transcript writes into its sender lane as the word
"you": which way a message went is a fact of its own, and spending the name lane on it
made our own name a value that lane could not otherwise hold. ``PEER`` is therefore
always the *other* party — the sender of an inbound message, and empty on a channel
broadcast, which has no one party it went to.
always the *other* party.
``TEXT`` is last and unquoted: it is the rest of the line. A message body can hold
absolutely anything, which is why it goes through :func:`~meshterm.ui.script.oneline`
— "the rest of the line" stops being true the moment the body holds a newline, and
the remainder then reads as a second record with an empty ``TIME``.
Args:
label: The conversation's display name (the caller named it; the columns don't).
messages: The messages to show, oldest-first.
Returns:
The scripted table (see :func:`meshterm.ui.script.columns`).
``TEXT`` is last and escaped: it is the rest of the line, and "the rest of the line"
stops being true the moment a body holds a newline — after which the remainder reads
as a second record with an empty time. The document carries the body raw, a JSON string
holding a newline natively.
"""
from ..ui import script
from ..ui import fields
table = script.columns("TIME", "DIRECTION", "PEER", "SNR_DB", "TEXT", right=("SNR_DB",))
return (
fields.when("created_at", "AGE"),
fields.word("direction", "DIR"),
# One column for whichever end the conversation had; the document keeps the two
# apart, because a parser cannot tell a channel from a contact by its label.
fields.plain_only(fields.name("conversation", "PEER")),
fields.node("node", lanes=()),
fields.channel("channel", lanes=()),
fields.snr("snr_db", "SNR_DB"),
fields.free("text", "TEXT"),
fields.hidden("acked"),
)
def _transcript(messages: list[ChatMessage]) -> Listing:
"""A conversation's stored messages, oldest first, one record each."""
from ..ui.report import Listing
rows = []
for message in messages:
peer = message.peer_name or message.peer or ""
table.add_row(
script.stamp(message.created_at),
"out" if message.outbound else "in",
script.name(peer),
script.number(message.snr, "+.1f"),
script.oneline(message.text),
channel = (
_channel_ref(message.channel_idx, message.peer_name or f"#{message.channel_idx}", None)
if message.is_channel
else None
)
return table
rows.append(
{
"created_at": message.created_at,
"direction": "out" if message.outbound else "in",
"conversation": message.peer_name or message.peer or None,
"node": None
if message.is_channel
else NodeRef(name=message.peer_name, hash=message.peer),
"channel": channel,
"snr_db": message.snr,
"text": message.text,
"acked": message.acked,
}
)
return Listing(
key="messages",
columns=_message_columns(),
rows=rows,
order=("AGE", "DIR", "PEER", "SNR_DB", "TEXT"),
)
def _live_messages() -> Listing:
"""The shape ``chat listen`` streams: what arrived, as it arrived.
``TIME`` is absolute here where the transcript's is an age, and for the reason the
whole time rule turns on: every row of a live tail would read ``now``, which is no
information at all. There is no ``DIR`` lane — everything a tail hears came in — and
the body goes last, so the one field with no width can never push a lane.
"""
from dataclasses import replace
from ..ui import fields
from ..ui.report import Listing
def pin(column, width: int): # noqa: ANN001, ANN202 - one column in, one column out
return replace(column, lanes=tuple(replace(lane, width=width) for lane in column.lanes))
return Listing(
key="messages",
columns=(
pin(fields.instant("received_at", "TIME"), 25),
pin(fields.plain_only(fields.name("conversation", "PEER")), 16),
fields.node("node", lanes=()),
fields.channel("channel", lanes=()),
pin(fields.snr("snr_db", "SNR_DB"), 6),
fields.free("text", "TEXT"),
fields.hidden("direction"),
fields.hidden("acked"),
),
order=("TIME", "PEER", "SNR_DB", "TEXT"),
)
def _conversations(ctx: AppContext, rows: list, lasts: dict) -> Listing:
"""Every channel and contact, with the last thing said in each.
``LAST`` is an age rather than an instant: a conversation list is scanned for what is
warm, and ``never`` is a real answer — the conversation exists and nothing has been
said in it.
"""
from ..ui import fields
from ..ui.report import Listing
records = []
for conversation in rows:
last = lasts.get(conversation.key)
channel = (
_channel_ref(conversation.channel_idx, conversation.label, conversation.secret)
if conversation.is_channel
else None
)
contact = getattr(conversation, "contact", None)
records.append(
{
"kind": "channel" if conversation.is_channel else "direct",
"conversation": conversation.label,
"channel": channel,
"node": None
if conversation.is_channel
else NodeRef(
name=conversation.label,
key=(getattr(contact, "public_key", "") or "").lower() or None,
hash=(getattr(contact, "key_prefix", "") or "").lower() or None,
type=NODE_TYPE_LABELS.get(getattr(contact, "node_type", None)),
),
"unread": ctx.chat.unread(conversation.key),
"last_message_at": last.created_at if last is not None else None,
# Not truncated to 40 cells the way the picker's preview is: nothing here
# wraps, so there is no reason to cut a message short — but a body that
# holds a newline would still end the record, so it is folded flat.
"last_text": last.text if last is not None else None,
}
)
return Listing(
key="conversations",
columns=(
fields.word("kind", "KIND"),
fields.plain_only(fields.name("conversation", "CONVERSATION")),
fields.channel("channel", lanes=()),
fields.node("node", lanes=()),
fields.integer("unread", "UNREAD"),
fields.when("last_message_at", "LAST", absent="never"),
fields.free("last_text", "LAST_TEXT"),
),
rows=records,
order=("CONVERSATION", "KIND", "UNREAD", "LAST", "LAST_TEXT"),
)
def _sent(
*,
contact: object | None = None,
channel: int | None = None,
acked: bool | None = None,
) -> Facts:
"""What one ``chat send`` did.
A direct message prints its ``acked`` state, which is the one thing the send does not
already tell the caller: the radio accepted it either way, and whether the peer
answered is a separate fact. A channel broadcast prints nothing, because there *is* no
acknowledgement on a channel — and the document says that in the one way the plain
face never could, with ``null`` rather than ``false``.
"""
from ..ui import fields
from ..ui.report import PAIRS, SILENT, Facts
node = (
NodeRef(
name=contact.name,
key=(getattr(contact, "public_key", "") or "").lower() or None,
hash=(getattr(contact, "key_prefix", "") or "").lower() or None,
type=NODE_TYPE_LABELS.get(getattr(contact, "node_type", None)),
)
if contact is not None
else None
)
return Facts(
key="sent",
fields=(
# `acked` is the only lane: the caller named the recipient and the radio
# accepting the message is what a zero exit already says, so the one fact
# left to report is whether the peer answered.
fields.hidden("kind"),
fields.node("node", lanes=()),
fields.channel("channel", lanes=()),
fields.hidden("sent"),
fields.flag("acked", "acked"),
),
values={
"kind": "channel" if channel is not None else "direct",
"node": node,
"channel": (
_channel_ref(channel, f"#{channel}", None) if channel is not None else None
),
"sent": True,
"acked": acked,
},
shape=SILENT if channel is not None else PAIRS,
)
+346 -61
View File
@@ -12,7 +12,7 @@ management, factory reset) through the same executor.
from __future__ import annotations
from pathlib import Path
from typing import Any
from typing import TYPE_CHECKING, Any
import typer
@@ -30,6 +30,9 @@ from ..core.device_config import (
)
from .base import Tool, ToolResult, register
if TYPE_CHECKING: # pragma: no cover - typing only
from ..ui.report import Block, Facts, Listing, Report
@register
class ConfigTool(Tool):
@@ -71,11 +74,16 @@ class ConfigTool(Tool):
device = await ctx.device()
ops: list[tuple] = list(params.get("ops") or [])
snapshot = await build_snapshot(device)
changes, artifacts = await apply_ops(ctx, device, snapshot, ops)
changes, artifacts, report = await apply_ops(ctx, device, snapshot, ops)
plural = "" if changes == 1 else "s"
message = f"[ok]✓[/ok] applied [brand]{changes}[/brand] change{plural}" if changes else None
return ToolResult(summary={"changes": changes}, message=message, artifacts=artifacts)
return ToolResult(
summary={"changes": changes},
message=message,
artifacts=artifacts,
report=report or None,
)
# -- CLI --------------------------------------------------------------------
@@ -202,7 +210,7 @@ class ConfigTool(Tool):
async def apply_ops(
ctx: AppContext, device: Device, snapshot: dict, ops: list[tuple]
) -> tuple[int, list[str]]:
) -> tuple[int, list[str], Report]:
"""Execute a list of configuration operation tuples against ``device``.
This is the single executor for every config operation, used both by
@@ -218,35 +226,51 @@ async def apply_ops(
ops: Operation tuples (see the module docstring).
Returns:
A ``(changes, artifacts)`` pair: the number of value changes applied and any
file paths produced.
A ``(changes, artifacts, report)`` triple: the number of value changes applied,
any file paths produced, and the blocks stating what happened (empty for the
menu, which shows its own acknowledgements).
"""
changes = 0
artifacts: list[str] = []
report: list[Any] = []
applied: list[dict[str, Any]] = []
for op in ops:
kind = op[0]
if kind == "show":
await _show(ctx, device, snapshot)
block = await _show(ctx, device, snapshot)
if block is not None:
report.append(block)
elif kind == "get":
# The bare value, nothing else: the caller named the key, so repeating it back
# is one more thing for `$(meshterm config get name)` to strip off. The same
# shape `sysctl -n` and `git config --get` print.
# shape `sysctl -n` and `git config --get` print. The document keeps the key,
# the type and an enum's label, none of which the bare form has room for.
spec = get_spec(op[1])
ctx.ui.note(_script_value(spec, spec.getter(snapshot)))
report.append(_one_setting(spec, spec.getter(snapshot)))
elif kind == "set":
changes += await _apply_setting(ctx, device, op[1], op[2], snapshot)
change = await _apply_setting(ctx, device, op[1], op[2], snapshot)
if change is not None:
applied.append(change)
changes += 1
elif kind == "set_custom":
previous = (await device.get_custom_vars()).get(op[1])
await device.set_custom_var(op[1], op[2])
ctx.ui.ack(f"[ok]✓[/ok] custom [brand]{op[1]}[/brand] = {op[2]}")
applied.append({"key": f"custom.{op[1]}", "previous": previous, "value": op[2]})
changes += 1
elif kind == "set_channel":
await device.set_channel(op[1], op[2], op[3])
ctx.ui.ack(f"[ok]✓[/ok] channel {op[1]} = [brand]{op[2]}[/brand]")
changes += 1
report.append(_channel_written(op[1], op[2], op[3]))
elif kind == "backup":
artifacts.append(str(await _backup(device, snapshot, op[1])))
written, counts = await _backup(device, snapshot, op[1])
artifacts.append(str(written))
report.append(_backed_up(written, counts))
elif kind == "restore":
changes += await _restore(ctx, device, snapshot, op[1], op[2])
done, blocks = await _restore(ctx, device, snapshot, op[1], op[2])
changes += done
report.extend(blocks)
elif kind == "advert":
flood = len(op) > 1 and bool(op[1])
await device.send_advert(flood)
@@ -257,6 +281,7 @@ async def apply_ops(
ctx.advert_store.mark_sent(public_key, flood=flood)
kind_label = "flood" if flood else "zero-hop"
ctx.ui.ack(f"[ok]✓[/ok] {kind_label} advertisement sent")
report.append(_acted("advert", sent=True, flood=flood))
elif kind == "advert_cadence":
from ..core.advert_store import cadence_label
@@ -270,35 +295,52 @@ async def apply_ops(
f"[brand]{cadence_label(hours)}[/brand]"
)
changes += 1
report.append(_acted("cadence", changes=1, flood=flood, hours=hours))
else: # pragma: no cover - SELF_INFO always carries the key on real firmware
raise DeviceCommandError(
"the device reported no public key — the advert cadence was not saved"
)
elif kind == "share":
_share_contact(ctx, snapshot)
report.append(_share_contact(ctx, snapshot))
elif kind == "sync_clock":
import time as _time
from datetime import datetime
epoch = int(_time.time())
drift = await _try_clock(device)
await device.set_time(epoch)
stamp = datetime.fromtimestamp(epoch).astimezone().strftime("%Y-%m-%d %H:%M:%S")
ctx.ui.ack(f"[ok]✓[/ok] device clock set to [brand]{stamp}[/brand]")
changes += 1
report.append(
_acted(
"clock",
changes=1,
set_at=datetime.fromtimestamp(epoch).astimezone(),
# What the clock was off by *before* the set: the fact worth logging,
# and the one this command destroys by succeeding.
drift_s=None if drift is None else drift - epoch,
)
)
elif kind == "reboot":
await device.reboot()
ctx.ui.ack("[warn]device rebooting[/warn]")
report.append(_acted("reboot", rebooted=True))
elif kind == "export_key":
await _export_key(ctx, device, op[1] if len(op) > 1 else None, artifacts)
report.append(await _export_key(ctx, device, op[1] if len(op) > 1 else None, artifacts))
elif kind == "import_key":
await device.import_private_key(op[1])
ctx.ui.ack("[ok]✓[/ok] private key imported")
changes += 1
report.append(_acted("imported", changes=1, imported=True))
elif kind == "factory_reset":
await device.factory_reset()
ctx.ui.ack("[err]device factory-reset[/err]")
report.append(_acted("factory_reset", factory_reset=True))
else: # pragma: no cover - guarded by the call sites that build ops
raise DeviceCommandError(f"unknown config operation: {kind}")
if applied:
report.append(_applied(applied))
# Drop any session-cached facts these ops may have changed, so the next screen re-reads
# the truth rather than a stale copy (see meshterm.services.device_state.DeviceState). A
# reboot/reconnect resets the whole cache on its own; these cover the in-place edits.
@@ -311,22 +353,25 @@ async def apply_ops(
ctx.devstate.invalidate_config() # self-info fields + path-hash mode
if "set_channel" in kinds:
ctx.devstate.invalidate_channels()
return changes, artifacts
return changes, artifacts, tuple(report)
async def _apply_setting(
ctx: AppContext, device: Device, key: str, raw: Any, snapshot: dict
) -> int:
) -> dict[str, Any]:
"""Parse, apply and record one setting; keep ``snapshot`` consistent.
The local snapshot is updated with the new value so a later coupled change in the
same batch (e.g. another radio field) is rebuilt from current values.
Returns:
``1`` (a change was applied).
The change, as ``{"key", "previous", "value"}``. The *previous* value is read off
the snapshot before it is overwritten: a caller managing a machine's configuration
wants to know what moved, and it is gone the moment this returns.
"""
spec = get_spec(key)
value = parse_value(spec, raw, snapshot)
previous = spec.getter(snapshot)
await spec.apply(device, value, snapshot)
snapshot[key] = value
# Remember what we set, keyed by the device's own public key, so a forgetful device (a
@@ -335,10 +380,10 @@ async def _apply_setting(
if ctx.settings_store is not None:
ctx.settings_store.remember(str(snapshot.get("public_key") or ""), key, value)
ctx.ui.ack(f"[ok]✓[/ok] [brand]{key}[/brand] = {format_value(spec, value)}")
return 1
return {"key": key, "previous": previous, "value": value}
async def _show(ctx: AppContext, device: Device, snapshot: dict) -> None:
async def _show(ctx: AppContext, device: Device, snapshot: dict) -> Listing | None:
"""Print every current setting, plus any custom variables.
The menu shows the annotated table; a scripted run gets one ``key value`` line per
@@ -351,25 +396,55 @@ async def _show(ctx: AppContext, device: Device, snapshot: dict) -> None:
PIN is the one value on it that lets someone else's phone onto the radio.
``config get device_pin`` names it deliberately.
"""
from ..ui import script
from ..ui import fields, script
from ..ui.config_editor import PIN_KEY, conceal, config_table
from ..ui.report import Listing
from ..ui.surface import TuiUi
custom = await device.get_custom_vars()
if isinstance(ctx.ui, TuiUi):
ctx.ui.show(config_table(snapshot, custom))
return
return None
rows: list[tuple[str, str]] = []
rows: list[dict[str, Any]] = []
for _category, specs in settings_by_category():
for spec in specs:
value = _script_value(spec, spec.getter(snapshot))
masked = conceal(value, absent=script.NONE) if spec.key == PIN_KEY else value
rows.append((spec.key, masked))
value = spec.getter(snapshot)
printed = _script_value(spec, value)
redacted = spec.key == PIN_KEY and printed != script.NONE
masked = conceal(printed, absent=script.NONE) if spec.key == PIN_KEY else printed
rows.append(
{
"key": spec.key,
# A masked PIN is *withheld*, which is not a value: the document says
# so with `redacted` and writes `null`, rather than shipping six
# bullets a consumer would have to recognise as a mask.
"value": fields.Rendered(None if redacted else value, masked),
"type": spec.value_type,
"label": (spec.choices or {}).get(value) if spec.value_type == "enum" else None,
"redacted": redacted,
}
)
# Custom variables are experimental firmware fields with no spec, so they are namespaced
# rather than mixed in — a reader can tell which lines `config set` will take.
rows.extend((f"custom.{key}", value) for key, value in sorted(custom.items()))
ctx.ui.show(script.pairs(rows))
rows.extend(
{
"key": f"custom.{key}",
"value": fields.Rendered(value, value),
"type": "str",
"label": None,
"redacted": False,
}
for key, value in sorted(custom.items())
)
return Listing(
key="settings",
columns=_setting_columns(),
rows=rows,
# An array to a parser and a `sysctl -a` block to a reader: a header line over two
# columns whose keys *are* the answer would be furniture.
headed=False,
)
def _script_value(spec: SettingSpec, value: Any) -> str:
@@ -402,95 +477,170 @@ def _script_value(spec: SettingSpec, value: Any) -> str:
return str(value)
def _share_contact(ctx: AppContext, snapshot: dict) -> None:
"""Render this node's contact card — the QR code in the menu, the URI alone on the CLI.
def _share_contact(ctx: AppContext, snapshot: dict) -> Facts:
"""State this node's contact card — with the QR code drawn in the menu.
The QR is for a phone pointed at the screen. Redirected into a file it is a block of
block characters around the one thing that is the answer, so a scripted run prints the
link by itself.
block characters around the one thing that is the answer, so a scripted run states the
link by itself. There is no machine face for the code either, and never will be: it is
a second rendering of ``url``.
"""
from rich.console import Group
from rich.text import Text
from ..ui import fields
from ..ui.config_editor import contact_share_url
from ..ui.qr import qr_text
from ..ui.report import BARE, Facts
from ..ui.surface import TuiUi
public_key = str(snapshot.get("public_key") or "")
if not public_key:
raise DeviceCommandError("the device did not report a public key — nothing to share")
name = str(snapshot.get("name") or "this node")
url = contact_share_url(name, public_key, int(snapshot.get("adv_type") or 1))
adv_type = int(snapshot.get("adv_type") or 1)
url = contact_share_url(name, public_key, adv_type)
if isinstance(ctx.ui, TuiUi):
ctx.ui.show(Group(qr_text(url), Text(""), Text(url, style="accent")))
else:
ctx.ui.note(url)
return Facts(
key="share",
fields=(
fields.word("url", "url"),
fields.word("name", "name"),
fields.hexid("public_key", "public_key"),
fields.integer("type", "type"),
),
values={"url": url, "name": name, "public_key": public_key.lower(), "type": adv_type},
shape=BARE,
bare="url",
)
async def _backup(device: Device, snapshot: dict, path: Path) -> Path:
"""Write a TOML backup of the current configuration."""
async def _backup(device: Device, snapshot: dict, path: Path) -> tuple[Path, dict[str, int]]:
"""Write a TOML backup of the current configuration.
Returns:
The path written and how much went into it, counted the way the file counts —
a setting the firmware never reported is not in the backup and is not in the tally.
"""
from ..core.device_config import DEVICE_SETTINGS
custom = await device.get_custom_vars()
channels = await _read_channels(device)
return backup_config(Path(path), snapshot, custom, channels)
written = backup_config(Path(path), snapshot, custom, channels)
counts = {
"settings": sum(1 for spec in DEVICE_SETTINGS if spec.getter(snapshot) is not None),
"channels": len(channels),
"custom": len(custom),
}
return written, counts
async def _restore(
ctx: AppContext, device: Device, snapshot: dict, path: Path, dry_run: bool
) -> int:
) -> tuple[int, list[Block]]:
"""Apply (or preview) a backup file against the current configuration.
The plan is the answer on a dry run and a record on a real one, so the same listing
serves both — drawn for a reader only when nothing was changed, since a run that *did*
change things already said so, setting by setting, on stderr.
Returns:
The number of changes applied (always ``0`` for a dry run).
The number of changes applied (always ``0`` for a dry run), and the blocks stating
what was planned or done.
"""
from ..ui import fields
from ..ui.report import SILENT, Facts, Listing
backup = read_backup(Path(path))
custom = await device.get_custom_vars()
ops = plan_restore(backup, snapshot, custom)
def plan(applied: list[tuple], *, drawn: bool) -> list[Block]:
listing = Listing(
key="operations",
columns=(
fields.word("operation", "OPERATION"),
fields.word("target", "TARGET"),
fields.word("value", "VALUE"),
),
rows=[
{
"operation": op[0],
"target": str(op[1]),
"value": str(op[2]) if len(op) > 2 else None,
}
for op in applied
],
plain_only=not drawn,
)
facts = Facts(
key="restore",
fields=(fields.flag("dry_run", "dry_run"), fields.integer("changes", "changes")),
values={"dry_run": dry_run, "changes": 0 if dry_run else len(applied)},
shape=SILENT,
)
return [facts, listing]
if not ops:
ctx.ui.ack("[muted]restore: device already matches the backup.[/muted]")
return 0
return 0, plan([], drawn=False)
if dry_run:
from ..ui import script
table = script.columns("OPERATION", "TARGET", "VALUE")
for op in ops:
table.add_row(op[0], str(op[1]), str(op[2]) if len(op) > 2 else script.NONE)
ctx.ui.show(table)
ctx.ui.ack("[muted]dry run — nothing was changed.[/muted]")
return 0
return 0, plan(ops, drawn=True)
changes = 0
for op in ops:
if op[0] == "set":
changes += await _apply_setting(ctx, device, op[1], op[2], snapshot)
await _apply_setting(ctx, device, op[1], op[2], snapshot)
changes += 1
elif op[0] == "set_custom":
await device.set_custom_var(op[1], op[2])
changes += 1
elif op[0] == "set_channel":
await device.set_channel(op[1], op[2], op[3])
changes += 1
return changes
return changes, plan(ops, drawn=False)
async def _export_key(
ctx: AppContext, device: Device, out: Path | None, artifacts: list[str]
) -> None:
"""Export the private key, to a file if ``out`` is given, else to the console."""
) -> Facts:
"""Export the private key, to a file if ``out`` is given, else as the answer itself.
Either way the plain face prints **one bare line** — the key, or the path it was
written to — because ``config export-key > key.hex`` is expected to hold the key and
nothing else, and a caller that asked for a file wants the name of it. The warning
that comes with either belongs beside it on screen, not in the file, so it goes to
stderr with every other acknowledgement.
This document carries a private key when one was asked for. That is what the command
is, exactly as on the plain face, and no extra gate is introduced here.
"""
from ..ui import fields
from ..ui.report import BARE, Facts
key_hex = await device.export_private_key()
written: Path | None = None
if out is not None:
path = Path(out)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(key_hex, encoding="utf-8")
artifacts.append(str(path))
written = Path(out)
written.parent.mkdir(parents=True, exist_ok=True)
written.write_text(key_hex, encoding="utf-8")
artifacts.append(str(written))
ctx.ui.ack("[warn]private key written — keep this file secret.[/warn]")
else:
# The key itself is the answer, so it is output rather than an acknowledgement:
# `meshterm config export-key > key.hex` should hold the key and nothing else. The
# warning that comes with it belongs beside it on screen, not in the file.
ctx.ui.ack("[warn]private key (keep secret):[/warn]")
# Bare, on both faces: this line *is* the export, and `config export-key` with no
# `--out` is expected to be redirected into a file.
ctx.ui.note(key_hex)
return Facts(
key="key",
fields=(fields.hexid("private_key", "private_key"), fields.word("path", "path")),
values={
"private_key": None if written else key_hex,
"path": str(written) if written else None,
},
shape=BARE,
bare="path" if written else "private_key",
)
async def _read_channels(device: Device) -> list[dict]:
@@ -524,3 +674,138 @@ def _require_yes(yes: bool, what: str) -> None:
"""
if not yes:
raise typer.BadParameter(f"{what}. Re-run with --yes to confirm.")
def _setting_columns() -> tuple:
"""One setting row's columns, shared by ``config show`` and ``config get``."""
from ..ui import fields
return (
fields.word("key", "key"),
fields.rendered("value", "value"),
fields.hidden("type"),
fields.hidden("label"),
fields.hidden("redacted"),
)
def _one_setting(spec: SettingSpec, value: Any) -> Facts:
"""One setting: the bare value plain, the whole object in the document.
``config get device_pin`` is the one place the PIN is named deliberately, so it is
**not** redacted here — the caller asked for that key by name, which is a different
act from dumping every setting into a file.
"""
from ..ui import fields
from ..ui.report import BARE, Facts
return Facts(
key="setting",
fields=_setting_columns(),
values={
"key": spec.key,
"value": fields.Rendered(value, _script_value(spec, value)),
"type": spec.value_type,
"label": (spec.choices or {}).get(value) if spec.value_type == "enum" else None,
"redacted": False,
},
shape=BARE,
bare="value",
)
def _applied(applied: list[dict[str, Any]]) -> Facts:
"""What a ``set`` changed — nothing plain, a record for whoever is logging it.
The exit status is the plain answer and the acknowledgement on stderr is the reassurance;
neither is something a script can read back later to find out *what moved*.
"""
from ..ui import fields
from ..ui.report import SILENT, Column, Facts
return Facts(
key="applied",
fields=(fields.integer("changes", "changes"), Column(key="applied")),
values={"changes": len(applied), "applied": applied},
shape=SILENT,
)
def _channel_written(idx: int, name: str, secret: bytes | None) -> Facts:
"""What ``config channel`` wrote into a slot."""
from ..core.channels import channel_hash, derive_secret, is_public_channel
from ..ui import fields
from ..ui.fields import ChannelRef
from ..ui.report import SILENT, Facts
key = secret or derive_secret(name)
return Facts(
key="channel",
fields=(fields.integer("changes", "changes"), fields.channel("channel", lanes=())),
values={
"changes": 1,
"channel": ChannelRef(
slot=idx,
name=name,
public=is_public_channel(name, key),
hash=channel_hash(key),
),
},
shape=SILENT,
)
def _backed_up(written: Path, counts: dict[str, int]) -> Facts:
"""What ``config backup`` wrote, and how much of it.
The plain face prints the path bare — it is what the caller keeps, and what the thing
that ran the command reads next.
"""
from ..ui import fields
from ..ui.report import BARE, Facts
return Facts(
key="backup",
fields=(
fields.word("path", "path"),
fields.integer("settings", "settings"),
fields.integer("channels", "channels"),
fields.integer("custom", "custom"),
),
values={"path": str(written), **counts},
shape=BARE,
bare="path",
)
def _acted(key: str, **facts: Any) -> Facts:
"""An action, stated as data: nothing plain, a document for whoever wanted a record.
Every one of these prints nothing on the plain face today and should keep doing so —
``config advert`` says all it has to say in its exit status. But "prints nothing" is an
answer a person can act on and a program cannot, so the machine face gets the fact.
"""
from ..ui import fields
from ..ui.report import SILENT, Column, Facts
def column(name: str, value: Any) -> Column:
if isinstance(value, bool):
return fields.flag(name, name)
if isinstance(value, int) or value is None:
return fields.integer(name, name)
return Column(key=name)
return Facts(
key=key,
fields=tuple(column(name, value) for name, value in facts.items()),
values=dict(facts),
shape=SILENT,
)
async def _try_clock(device: Device) -> int | None:
"""The device's current clock, or ``None`` where the firmware will not answer."""
try:
return await device.get_time()
except Exception: # noqa: BLE001 - optional read; the drift is a nicety
return None
+49 -21
View File
@@ -17,9 +17,8 @@ from ..core.models import NODE_TYPE_LABELS
from .base import Tool, ToolResult, register
if TYPE_CHECKING: # pragma: no cover - typing only
from rich.table import Table
from ..core.models import Contact
from ..ui.report import Listing
@register
@@ -85,10 +84,10 @@ class ContactsTool(Tool):
return ToolResult(summary={"contacts": len(contacts)})
sort = ContactsSort.from_name(sort_name)
if contacts:
ctx.ui.show(_listing(ordered_contacts(contacts, counts, sort), counts))
listing = _listing(ordered_contacts(contacts, counts, sort), counts, prefix_bytes)
return ToolResult(
summary={"contacts": len(contacts)},
report=(listing,),
exit_code=exitcodes.OK if contacts else exitcodes.NO_RESULT,
)
@@ -117,32 +116,61 @@ class ContactsTool(Tool):
run_tool_command(self, {"sort": sort})
def _listing(contacts: list[Contact], counts: dict[str, int]) -> Table:
"""The scripted contact list: one record per contact, one line each.
def _listing(contacts: list[Contact], counts: dict[str, int], prefix_bytes: int) -> Listing:
"""The contact list, stated once for both faces.
``NAME`` is quoted so a name holding a space or a comma stays one field; ``TYPE`` is
the node's advertised role in words, where the menu draws a coloured glyph; ``HEARD``
is an absolute local timestamp, where the menu draws a relative age in recency heat;
and ``KEY`` is the full public key, never elided — a truncated key is not something a
caller can pass back to ``--path`` or ``--to``.
``TYPE`` is the node's advertised role in words, where the menu draws a coloured glyph:
a monochrome ``▲`` would need a legend and the CLI has no legends. ``HEARD`` is a
relative age, because "recently?" is the question this listing is opened to answer.
``HASH`` is the token ``--path`` and ``--to`` take — it used to have to be sliced out
of ``KEY`` by hand — and ``LOCATION`` is a fact the model has always carried and no
column had room for. ``KEY`` stays full and last, so it runs off the right harmlessly
and is never elided: a truncated key is not something a caller can hand back.
Args:
contacts: The contacts, already in the requested order.
counts: Overheard-packet tallies keyed by node id (from passive monitoring).
prefix_bytes: The device's path-hash width, which is what makes ``HASH`` the
token a forced path addresses this node by rather than an arbitrary slice.
Returns:
The scripted table (see :func:`meshterm.ui.script.columns`).
The listing.
"""
from ..ui import script
from ..ui import fields
from ..ui.report import Listing
from ..ui.widgets import _contact_pkts
table = script.columns("NAME", "TYPE", "HEARD", "PKTS", "KEY", right=("PKTS",))
rows = []
for contact in contacts:
table.add_row(
script.name(contact.name),
NODE_TYPE_LABELS.get(contact.node_type, "unknown"),
script.stamp(contact.last_seen),
script.number(_contact_pkts(contact, counts)),
(contact.public_key or contact.key_prefix or "").lower() or script.NONE,
key = (contact.public_key or "").lower()
rows.append(
{
"node": fields.NodeRef(
name=contact.name,
key=key or None,
hash=(key or contact.key_prefix.lower())[: prefix_bytes * 2] or None,
type=NODE_TYPE_LABELS.get(contact.node_type),
),
"heard_at": contact.last_seen,
"packets": _contact_pkts(contact, counts),
"position": (
fields.Position(contact.lat, contact.lon) if contact.has_location else None
),
}
)
return table
return Listing(
key="contacts",
columns=(
fields.node(
"node",
lanes=(("name", "NAME"), ("type", "TYPE"), ("hash", "HASH"), ("key", "KEY")),
),
fields.when("heard_at", "HEARD", absent="never"),
fields.integer("packets", "PKTS"),
fields.position(),
),
rows=rows,
# The node's four lanes are not contiguous: what a reader scans for — who, what,
# how recently, how much — comes first, and the two long hex fields go right.
order=("NAME", "TYPE", "HEARD", "PKTS", "HASH", "LOCATION", "KEY"),
)
+153 -39
View File
@@ -17,16 +17,19 @@ of the outbox screen's own Send-now, Cancel, and Clear-finished actions.
from __future__ import annotations
from typing import Any
from typing import TYPE_CHECKING, Any
import typer
from ..context import AppContext
from ..core import exitcodes
from ..core.connection import DeviceCommandError
from ..ui import script
from .base import Tool, ToolResult, register
if TYPE_CHECKING: # pragma: no cover - typing only
from ..core.courier_store import QueuedMessage
from ..ui.report import Facts, Listing
@register
class CourierTool(Tool):
@@ -124,48 +127,23 @@ class CourierTool(Tool):
f"[ok]queued[/ok] #{message.ident} for [brand]{contact.name}[/brand] — "
f"delivers {when} (an interactive session's courier does the sending)"
)
# The entry's id is the one fact a caller has to keep: it is what `courier send`
# and `courier cancel` take.
ctx.ui.show(script.pairs([("id", str(message.ident))]))
return ToolResult(
summary={"queued": message.ident, "contact": contact.name, "at": params.get("at")}
summary={"queued": message.ident, "contact": contact.name, "at": params.get("at")},
report=(_queued(message, contact.name, not_before),),
)
async def _cli_list(self, ctx: AppContext) -> ToolResult:
"""Print the outbox — waiting entries then finished ones — as a table.
"""State the outbox — waiting entries then finished ones.
A read-only view over the stored outbox: it needs no device, so it works the same
whether or not a radio is attached (the scriptable face of the outbox screen).
"""
from ..core.courier_store import DELIVERED, QUEUED
entries = ctx.courier_store.entries()
if not entries:
return ToolResult(summary={"entries": 0}, exit_code=exitcodes.NO_RESULT)
table = script.columns(
"ID", "STATE", "NODE", "SCHEDULED", "FINISHED", "TEXT", right=("ID",)
return ToolResult(
summary={"entries": len(entries)},
report=(_outbox(entries),),
exit_code=exitcodes.OK if entries else exitcodes.NO_RESULT,
)
for m in entries:
if m.status == QUEUED:
state = "waiting"
elif m.status == DELIVERED:
state = "delivered"
else:
state = "gave up"
table.add_row(
str(m.ident),
state,
script.name(m.node_name),
# An entry with no scheduled time goes as soon as the contact is heard,
# which is not a time and so is not one here either.
script.stamp(m.not_before),
script.stamp(m.finished),
# Not shortened the way the outbox screen shortens it: nothing wraps here.
m.text,
)
ctx.ui.show(table)
return ToolResult(summary={"entries": len(entries)})
async def _cli_send(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
"""Force one delivery attempt for a waiting entry, right now.
@@ -188,18 +166,26 @@ class CourierTool(Tool):
"gone": "[muted]that entry is no longer waiting[/muted]",
}
ctx.ui.ack(notes.get(outcome, outcome))
# The outcome word is the answer, and it is one of a closed set the help lists.
ctx.ui.show(script.pairs([("outcome", outcome)]))
return ToolResult(summary={"id": ident, "outcome": outcome})
return ToolResult(
summary={"id": ident, "outcome": outcome},
report=(_attempted(ident, outcome),),
)
async def _cli_cancel(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
"""Remove a waiting entry from the outbox (finished ones use ``clear``)."""
ident = int(params["id"])
if ctx.courier_store.cancel(ident):
ctx.ui.ack(f"[warn]cancelled outbox entry #{ident}[/warn]")
return ToolResult(summary={"id": ident, "cancelled": True})
return ToolResult(
summary={"id": ident, "cancelled": True},
report=(_did(("id", ident), ("cancelled", True)),),
)
ctx.ui.ack(f"[muted]no waiting entry #{ident} to cancel[/muted]")
return ToolResult(summary={"id": ident, "cancelled": False}, exit_code=exitcodes.NO_RESULT)
return ToolResult(
summary={"id": ident, "cancelled": False},
report=(_did(("id", ident), ("cancelled", False)),),
exit_code=exitcodes.NO_RESULT,
)
async def _cli_clear(self, ctx: AppContext) -> ToolResult:
"""Drop every finished (delivered / given-up) entry, leaving the queue untouched."""
@@ -212,6 +198,7 @@ class CourierTool(Tool):
ctx.ui.ack("[muted]no finished entries to clear[/muted]")
return ToolResult(
summary={"cleared": cleared},
report=(_did(("cleared", cleared)),),
exit_code=exitcodes.OK if cleared else exitcodes.NO_RESULT,
)
@@ -265,3 +252,130 @@ class CourierTool(Tool):
run_tool_command(self, {"cli_action": "clear"})
app.add_typer(courier_app, name=self.name)
def _recipient(name: str | None, node_key: str | None) -> object:
"""The outbox's addressee as the shared node shape.
The outbox stores a name and the 12-hex id it addresses, so three of the five fields
are honestly ``null`` — and the id goes in ``hash`` rather than ``key``, because that
is what it is. A key is the full value and is never truncated by this app's choosing;
calling a six-byte prefix one would hand a caller something ``--to`` cannot take.
"""
from ..ui.fields import NodeRef
return NodeRef(name=name, hash=(node_key or "").lower() or None)
def _queued(message: QueuedMessage, name: str, not_before: object) -> Facts:
"""What ``courier queue`` put in the outbox.
The plain face prints the **id and nothing else**, because the id is the one fact a
caller has to keep — it is what ``send`` and ``cancel`` take — and everything else on
the line is what they just typed. The document carries the rest, because a caller
logging the queue has no other record of it.
"""
from ..ui import fields
from ..ui.report import Facts
return Facts(
key="queued",
fields=(
fields.integer("id", "id"),
fields.node("node", lanes=()),
fields.hidden("scheduled_at"),
fields.hidden("text"),
),
values={
"id": message.ident,
"node": _recipient(name, message.node_key),
"scheduled_at": not_before,
"text": message.text,
},
)
def _outbox(entries: list) -> Listing:
"""The outbox, waiting entries then finished ones.
The deliberate asymmetry between the two time columns: ``SCHEDULED`` is an
*appointment* the caller set with ``--at``, so it stays a wall-clock instant, and
``FINISHED`` is an age, because "how long ago did it go" is what anyone reading a
delivered row wants. The mixed widths are the price of both columns being right, and
``--absolute`` makes them uniform for anyone who dislikes it.
"""
from ..core.courier_store import DELIVERED, QUEUED
from ..ui import fields
from ..ui.report import Listing
rows = []
for m in entries:
if m.status == QUEUED:
state = "waiting"
elif m.status == DELIVERED:
state = "delivered"
else:
state = "gave up"
rows.append(
{
"id": m.ident,
"state": state,
"node": _recipient(m.node_name, m.node_key),
# An entry with no scheduled time goes as soon as the contact is heard,
# which is not a time and so is not one here either.
"scheduled_at": m.not_before,
"finished_at": m.finished,
# Not shortened the way the outbox screen shortens it: nothing wraps here.
"text": m.text,
}
)
return Listing(
key="outbox",
columns=(
fields.integer("id", "ID"),
fields.word("state", "STATE"),
fields.node("node", lanes=(("name", "NODE"),)),
fields.instant("scheduled_at", "SCHEDULED"),
fields.when("finished_at", "FINISHED"),
fields.free("text", "TEXT"),
),
rows=rows,
)
def _attempted(ident: int, outcome: str) -> Facts:
"""What one forced delivery attempt came to.
``outcome`` is one of the closed set the help lists, kept verbatim — spaces and all —
so both faces say the same word and a script matching ``no ack`` matches what a person
reads.
"""
from ..ui import fields
from ..ui.report import Facts
return Facts(
key="attempt",
fields=(fields.hidden("id"), fields.word("outcome", "outcome")),
values={"id": ident, "outcome": outcome},
)
def _did(*facts: tuple[str, object]) -> Facts:
"""An acknowledgement as data: nothing plain, a document for whoever is logging.
``cancel`` and ``clear`` say everything they have to say in their exit status, which
is the right plain answer — but a caller that wants to record *what* was cancelled has
nowhere to read it, and that is what the document is for.
"""
from ..ui import fields
from ..ui.report import SILENT, Facts
def column(key: str, value: object): # noqa: ANN202 - one of two shapes
return fields.flag(key, key) if isinstance(value, bool) else fields.integer(key, key)
return Facts(
key="outcome",
fields=tuple(column(key, value) for key, value in facts),
values=dict(facts),
shape=SILENT,
)
+59 -51
View File
@@ -11,16 +11,18 @@ the prompt shown when the menu launches (which smoke-tests the choice before con
from __future__ import annotations
from typing import Any
from typing import TYPE_CHECKING, Any
import typer
from rich.table import Table
from ..context import AppContext
from ..core import exitcodes
from ..core.discovery import discover_all, discover_devices
from .base import Tool, ToolResult, register
if TYPE_CHECKING: # pragma: no cover - typing only
from ..ui.report import Listing
#: The ``MESHCORE`` verdict per discovery confidence tier, for devices we have *not* yet
#: confirmed. The USB vendor ID is only a hint — a native-USB board or a bare bridge chip
#: is a "maybe", never a "yes" — so nothing is billed as MeshCore until a connection proves
@@ -40,7 +42,7 @@ class DevicesTool(Tool):
menu_visible = False # CLI-only: a startup diagnostic with no place in a connected session
async def run(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
"""Enumerate serial devices and render them as a table.
"""Enumerate attached and advertising devices, and state them as a listing.
Args:
ctx: Shared application context.
@@ -56,44 +58,25 @@ class DevicesTool(Tool):
active = ctx.selected_device
active_target = ctx.ble_override or ctx.port_override or (active.target if active else None)
if ctx.json_output:
import json
payload = [
{
"transport": d.transport,
"port": d.port,
"address": d.address,
"target": d.target,
"label": d.label,
"vendor": d.vendor_label,
"model": (known[d.stable_id].hardware_model if d.stable_id in known else ""),
"likely_lora": d.is_likely_lora,
"confidence": d.confidence,
"confirmed": d.stable_id in known,
"serial_number": d.serial_number,
"stable_id": d.stable_id,
"remembered": remembered is not None and remembered.matches(d),
"active": d.target == active_target,
}
for d in devices
]
ctx.console.print_json(json.dumps(payload))
return ToolResult(summary={"count": len(devices)})
if not devices:
# Not an error: the scan ran and found nothing. Said on stderr so a caller
# redirecting stdout still hears it, and reported as NO_RESULT so a script can
# branch on it without matching prose.
# branch on it without matching prose. The empty listing still reaches the
# machine face as `[]`, which is a document a consumer can read — where an
# empty stdout is a parse error it would have to tell apart from a real one.
from ..ui import script
script.stderr_console().print(
"meshterm: no companion devices detected", style="warn", highlight=False
)
return ToolResult(summary={"count": 0}, exit_code=exitcodes.NO_RESULT)
ctx.ui.show(_listing(devices, known, active_target))
return ToolResult(summary={"count": len(devices)})
return ToolResult(
summary={"count": len(devices)},
report=(_listing(devices, known, remembered, active_target),),
# Changed from the 0 this used to return under --json: a rendering flag has no
# business changing the report, and the plain path has always said 5 here.
exit_code=exitcodes.OK if devices else exitcodes.NO_RESULT,
)
def register_cli(self, app: typer.Typer) -> None:
"""Register the ``devices`` subcommand (inventory only; no connection).
@@ -119,8 +102,8 @@ class DevicesTool(Tool):
run_tool_command(self, {"ble": ble})
def _listing(devices: list, known: dict, active_target: str | None) -> Table:
"""The scripted device inventory: one line per attached or advertising device.
def _listing(devices: list, known: dict, remembered: object, active_target: str | None) -> Listing:
"""The device inventory, stated once for both faces.
``TARGET`` leads because it is the field a caller acts on — it is what ``--port`` and
``--ble`` take, verbatim. The menu's two markers become columns of their own
@@ -130,32 +113,57 @@ def _listing(devices: list, known: dict, active_target: str | None) -> Table:
``MESHCORE`` is three-valued and stays that way: ``yes`` only once a connection has
proved the device speaks the protocol, ``maybe`` for a USB vendor ID that suggests a
LoRa board or a bridge chip, ``no`` for anything else. A vendor ID is a hint, and the
column would be lying if it rounded one up.
column would be lying if it rounded one up — which is why the machine face carries the
same three words rather than the boolean it used to.
Args:
devices: The discovered devices.
known: Remembered device records, keyed by stable id.
remembered: The remembered default device, if there is one.
active_target: The target this invocation is (or would be) using.
Returns:
The scripted table (see :func:`meshterm.ui.script.columns`).
The listing.
"""
from ..ui import script
from ..ui import fields
from ..ui.report import Listing
table = script.columns(
"TARGET", "TRANSPORT", "NAME", "HARDWARE", "MESHCORE", "SERIAL", "ACTIVE"
)
rows = []
for device in devices:
confirmed = known.get(device.stable_id)
name = (confirmed.node_name if confirmed else "") or device.label
hardware = (confirmed.hardware_model if confirmed else "") or device.vendor_label
table.add_row(
device.target,
device.transport,
script.name(name),
script.quote(hardware) if hardware else script.NONE,
"yes" if confirmed else _MAYBE.get(device.confidence, "no"),
device.serial_number or script.NONE,
"yes" if device.target == active_target else "no",
rows.append(
{
"target": device.target,
"transport": device.transport,
"port": device.port,
"address": device.address,
"label": (confirmed.node_name if confirmed else "") or device.label,
"hardware": (confirmed.hardware_model if confirmed else "") or device.vendor_label,
"meshcore": "yes" if confirmed else _MAYBE.get(device.confidence, "no"),
"confidence": device.confidence,
"serial_number": device.serial_number,
"stable_id": device.stable_id,
"confirmed": confirmed is not None,
"remembered": remembered is not None and remembered.matches(device),
"active": device.target == active_target,
}
)
return table
return Listing(
key="devices",
columns=(
fields.word("target", "TARGET"),
fields.word("transport", "TRANSPORT"),
fields.name("label", "NAME"),
fields.name("hardware", "HARDWARE"),
fields.word("meshcore", "MESHCORE"),
# What the "maybe" was derived from. A person reading the column has the
# vendor label beside it and can see for themselves; a program cannot.
fields.hidden("confidence"),
fields.word("serial_number", "SERIAL"),
fields.hidden("stable_id"),
fields.hidden("confirmed"),
fields.hidden("remembered"),
fields.flag("active", "ACTIVE"),
),
rows=rows,
)
+112 -50
View File
@@ -3,8 +3,9 @@
from __future__ import annotations
import time
from collections.abc import Callable
from datetime import datetime, timezone
from typing import Any
from typing import TYPE_CHECKING, Any
import typer
from rich.panel import Panel
@@ -14,6 +15,9 @@ from ..context import AppContext
from ..core.connection import Device
from .base import Tool, ToolResult, register
if TYPE_CHECKING: # pragma: no cover - typing only
from ..ui.report import Column, Facts
#: MeshCore advert-type byte -> human-readable node role.
_ROLES = {1: "companion", 2: "repeater", 3: "room server", 4: "sensor"}
@@ -55,10 +59,10 @@ class InfoTool(Tool):
# well would be the same table twice under two different spellings.
session = getattr(ctx.ui, "session", None)
if session is None:
from ..ui import script
ctx.ui.show(script.pairs(await status_pairs(device, snapshot)))
return ToolResult(summary={"name": snapshot.get("name")})
return ToolResult(
summary={"name": snapshot.get("name")},
report=(await status_facts(device, snapshot),),
)
from ..ui.device_info_screen import DeviceInfoScreen
@@ -99,7 +103,7 @@ async def _read_status(device: Device) -> dict[str, Any]:
The one place the device is asked. Both faces build from what it returns: the menu's
panel composes prose rows out of these values, the CLI prints them one fact per line
(see :func:`_status_panel` and :func:`status_pairs`). Every read is optional —
(see :func:`_status_panel` and :func:`status_facts`). Every read is optional —
firmware predating a query simply contributes nothing — so this works on any device.
Args:
@@ -192,65 +196,123 @@ def _firmware(info: dict) -> str:
return " ".join(str(info[k]) for k in ("ver", "fw_build") if info.get(k))
async def status_pairs(device: Device, snapshot: dict) -> list[tuple[str, str]]:
"""The device's live status as scripted key/value facts.
async def status_facts(device: Device, snapshot: dict) -> Facts:
"""The device's live status as one set of facts.
The CLI's face of :func:`_read_status`. Where the panel writes a phrase per row, this
writes one value per key and puts the unit in the key — ``battery_v``, ``uptime_s``,
``noise_floor_dbm`` — so nothing has to be pulled back out of prose. A reading the
firmware did not answer for is absent rather than :data:`~meshterm.ui.script.NONE`:
these are optional queries, and a missing key says "this device does not report it"
where a dash would claim it reported nothing.
writes one value per key and puts the unit *in the key* — ``battery_v``, ``uptime_s``,
``noise_floor_dbm`` — so nothing has to be pulled back out of prose. A number never
carries a unit in its value, on either face.
The three second-valued readings and the clock drift carry a **gloss** beside the
figure: ``uptime_s 93784 (1d 2h)``. This is the one key/value block that may, and
the reason is that it is the one with no ``get`` and no ``set`` behind it — there is no
round trip to protect, the key still says what the number counts, and ``93784`` is a
reading nobody can hold in their head. The document carries the number alone, having
nowhere useful to put a phrase.
A reading the firmware did not answer for is **left out** of the plain block rather
than dashed: these are optional queries, and a missing key says "this device does not
report it" where a dash would claim it reported nothing. The document keeps the key and
writes ``null`` — see :attr:`~meshterm.ui.report.Facts.omit_absent` for why the two
faces disagree here on purpose.
Args:
device: The connected device to query.
snapshot: The already-built settings snapshot (for role and identity).
Returns:
``(key, value)`` pairs in reading order.
The facts block.
"""
from ..ui import fields
from ..ui.report import Facts
readings = await _read_status(device)
info, battery, stats = readings["info"], readings["battery"], readings["stats"]
pairs: list[tuple[str, str]] = []
def add(key: str, value: Any) -> None:
"""Record one fact, skipping a reading the firmware did not answer for."""
if value is not None and value != "":
pairs.append((key, str(value)))
# Identity first: which radio this is. Everything else it *can be set to* is
# `config show`'s answer, not this one. The name is not quoted here the way it is in
# a listing: a key/value line has exactly two fields, so the value is the rest of the
# line whatever it contains, and quotes would only be something to strip back off.
add("name", snapshot.get("name"))
add("public_key", str(snapshot.get("public_key") or "").lower() or None)
adv_type = snapshot.get("adv_type")
add("role", _ROLES.get(int(adv_type), f"type {adv_type}") if adv_type is not None else None)
add("model", info.get("model"))
add("firmware", _firmware(info) or None)
level = battery.get("level")
add("battery_v", f"{int(level) / 1000:.2f}" if level else None)
if battery.get("total_kb"):
add("storage_used_kb", battery.get("used_kb", 0))
add("storage_total_kb", battery["total_kb"])
adv_type = snapshot.get("adv_type")
clock = readings["clock"]
if clock:
add("clock", datetime.fromtimestamp(clock, tz=timezone.utc).astimezone().isoformat())
add("clock_drift_s", clock - int(time.time()))
values: dict[str, Any] = {
# Identity first: which radio this is. Everything it *can be set to* is
# `config show`'s answer, not this one.
"name": snapshot.get("name"),
"public_key": str(snapshot.get("public_key") or "").lower() or None,
"role": _ROLES.get(int(adv_type), f"type {adv_type}") if adv_type is not None else None,
"model": info.get("model"),
"firmware": _firmware(info) or None,
"battery_v": round(int(level) / 1000, 2) if level else None,
"storage_used_kb": battery.get("used_kb", 0) if battery.get("total_kb") else None,
"storage_total_kb": battery.get("total_kb"),
"clock_at": (
datetime.fromtimestamp(clock, tz=timezone.utc).astimezone() if clock else None
),
"clock_drift_s": (clock - int(time.time())) if clock else None,
"uptime_s": stats.get("uptime_secs"),
"noise_floor_dbm": stats.get("noise_floor"),
"last_rssi_dbm": stats.get("last_rssi"),
"last_snr_db": stats.get("last_snr"),
"tx_air_s": stats.get("tx_air_secs"),
"rx_air_s": stats.get("rx_air_secs"),
"packets_sent": stats.get("sent"),
"packets_received": stats.get("recv"),
"receive_errors": stats.get("recv_errors"),
}
return Facts(
key="info",
fields=(
fields.word("name", "name"),
fields.hexid("public_key", "public_key"),
fields.word("role", "role"),
fields.word("model", "model"),
fields.word("firmware", "firmware"),
fields.decimal("battery_v", "battery_v", ".2f"),
fields.integer("storage_used_kb", "storage_used_kb"),
fields.integer("storage_total_kb", "storage_total_kb"),
fields.instant("clock_at", "clock_at"),
_glossed("clock_drift_s", _drift_gloss),
_glossed("uptime_s", _span_gloss),
fields.integer("noise_floor_dbm", "noise_floor_dbm"),
fields.integer("last_rssi_dbm", "last_rssi_dbm"),
fields.snr("last_snr_db", "last_snr_db"),
_glossed("tx_air_s", _span_gloss),
_glossed("rx_air_s", _span_gloss),
fields.integer("packets_sent", "packets_sent"),
fields.integer("packets_received", "packets_received"),
fields.integer("receive_errors", "receive_errors"),
),
values=values,
omit_absent=True,
)
add("uptime_s", stats.get("uptime_secs"))
add("noise_floor_dbm", stats.get("noise_floor"))
add("last_rssi_dbm", stats.get("last_rssi"))
add("last_snr_db", f"{stats['last_snr']:+.1f}" if stats.get("last_snr") is not None else None)
add("tx_air_s", stats.get("tx_air_secs"))
add("rx_air_s", stats.get("rx_air_secs"))
add("packets_sent", stats.get("sent"))
add("packets_received", stats.get("recv"))
add("receive_errors", stats.get("recv_errors"))
return pairs
def _glossed(key: str, gloss: Callable[[int], str]) -> Column:
"""A seconds reading printed as its own number with a readable span beside it."""
from ..ui import script
from ..ui.report import Column, Lane
def cell(value: Any) -> str:
if value is None:
return script.NONE
return f"{value} ({gloss(int(value))})"
return Column(key=key, lanes=(Lane(header=key, render=cell),))
def _span_gloss(seconds: int) -> str:
"""``93784`` → ``1d 2h``: the figure a person can hold in their head."""
from ..ui import script
return script.duration(seconds)
def _drift_gloss(seconds: int) -> str:
"""The device clock's error, in the direction a reader would say it."""
from ..ui import script
if abs(seconds) < 2:
return "in sync"
return f"{script.duration(seconds)} {'fast' if seconds > 0 else 'slow'}"
async def _try(read) -> Any | None:
+151 -56
View File
@@ -18,17 +18,21 @@ summarizes the window when it ends.
from __future__ import annotations
import asyncio
from typing import Any
from typing import TYPE_CHECKING, Any
import typer
from rich.table import Table
from ..context import AppContext
from ..core import exitcodes
from ..core.events import EventKind, MeshEvent
from ..core.models import HeardNode, Observation
from ..core.models import NODE_TYPE_LABELS, HeardNode, Observation
from ..ui import renderers
from ..ui.fields import NodeRef, Position
from .base import Tool, ToolResult, register
if TYPE_CHECKING: # pragma: no cover - typing only
from ..ui.report import Listing
@register
class MonitorTool(Tool):
@@ -88,50 +92,59 @@ class MonitorTool(Tool):
highlight=False,
)
seen: list[Observation] = []
# One header for the live stream, then a record per packet as it arrives. The
# stream cannot be column-aligned — the widths are not known until it ends — so
# the fields are separated by the gutter and the name carries its quotes, which is
# what keeps the line splittable.
ctx.console.print("TIME NODE NAME SNR_DB RSSI_DBM LAT LON", highlight=False)
def on_observation(event: MeshEvent) -> None:
obs = event.observation
if obs is None:
return
seen.append(obs)
fields = [
script.stamp(obs.observed_at),
obs.node or script.NONE,
script.name(obs.name),
script.number(obs.snr, "+.1f"),
script.number(obs.rssi, ".0f"),
script.number(obs.lat, ".5f"),
script.number(obs.lon, ".5f"),
]
ctx.console.print((" " * script.GUTTER).join(fields), highlight=False)
emit(
{
"observed_at": obs.observed_at,
"node": NodeRef(
name=obs.name,
key=(obs.public_key or "").lower() or None,
hash=obs.node,
type=NODE_TYPE_LABELS.get(obs.node_type),
),
"kind": obs.kind,
"snr_db": obs.snr,
"rssi_dbm": obs.rssi,
"position": _position(obs.lat, obs.lon),
"path": _relays(obs.path),
}
)
unsubscribe = ctx.events.subscribe(on_observation, EventKind.OBSERVATION)
try:
if seconds:
await asyncio.sleep(seconds)
else:
await asyncio.Event().wait() # until Ctrl-C / cancellation
except (KeyboardInterrupt, asyncio.CancelledError): # pragma: no cover - interactive
pass
finally:
unsubscribe()
await ctx.monitor.stop() # close the run row so the capture is a full record
with renderers.stream(ctx, _live_packets()) as emit:
unsubscribe = ctx.events.subscribe(on_observation, EventKind.OBSERVATION)
try:
if seconds:
await asyncio.sleep(seconds)
else:
await asyncio.Event().wait() # until Ctrl-C / cancellation
except (KeyboardInterrupt, asyncio.CancelledError): # pragma: no cover - interactive
pass
finally:
unsubscribe()
await ctx.monitor.stop() # close the run row so the capture is a full record
by_node: dict[str, list[Observation]] = {}
for obs in seen:
by_node.setdefault(obs.node, []).append(obs)
nodes = [HeardNode.from_observations(node, group) for node, group in by_node.items()]
if nodes:
nodes.sort(key=lambda n: n.last_seen, reverse=True)
ctx.ui.show(script.blank())
ctx.ui.show(_heard_table(nodes))
nodes.sort(key=lambda n: n.last_seen, reverse=True)
# The window is over, so its facts are worth saying — on stderr, where everything
# about a run goes, keeping the packets a caller redirected the run for on their own.
script.stderr_console().print(
f"captured {len(seen)} packet{'' if len(seen) == 1 else 's'} "
f"from {len(nodes)} node{'' if len(nodes) == 1 else 's'}"
+ (f" in {seconds}s" if seconds else ""),
style="muted",
highlight=False,
)
return ToolResult(
summary={"seconds": seconds, "packets": len(seen), "nodes": len(nodes)},
report=(_heard(nodes),),
exit_code=exitcodes.OK if seen else exitcodes.NO_RESULT,
)
@@ -152,38 +165,120 @@ class MonitorTool(Tool):
run_tool_command(self, {"seconds": seconds})
def _heard_table(heard: list[HeardNode]) -> Table:
def _position(lat: float | None, lon: float | None) -> Position | None:
"""A reception's shared position, or ``None`` where the packet carried none."""
return Position(lat, lon) if lat is not None and lon is not None else None
def _relays(path: str | None) -> list[str] | None:
"""A packet's relay chain as hashes, in propagation order.
Hashes rather than the shared node shape, and that is the honest answer: nothing
resolves a relay to a node at reception time, so a node object per hop would be four
``null``s wrapped round a hash. ``[]`` is a direct reception (zero hops); ``None``
means the packet class carries no path at all, which is a different fact.
"""
if path is None:
return None
return [hop for hop in path.split(",") if hop]
def _live_packets() -> Listing:
"""The shape the live capture streams: one record per packet, as it lands.
``TIME`` is absolute where a listing's times are ages, and for the reason the whole
rule turns on: every row of a live tail would read ``now``. The **name goes last**, so
the one field with no width cannot push a lane — which is what finally let the two
streams have columns at all, and with them the quoting could go.
"""
from dataclasses import replace
from ..ui import fields
from ..ui.report import Listing
def pin(column, width: int): # noqa: ANN001, ANN202 - one column in, one column out
return replace(column, lanes=tuple(replace(lane, width=width) for lane in column.lanes))
node = fields.node("node", lanes=(("hash", "NODE"), ("name", "NAME")))
node = replace(
node,
lanes=(replace(node.lanes[0], width=8), node.lanes[1]),
)
return Listing(
key="observations",
columns=(
pin(fields.instant("observed_at", "TIME"), 25),
node,
# The packet class, which the plain stream has never had room for. New
# surface: the simulator does not exercise every class, so a consumer should
# treat an unfamiliar word as a word rather than an error.
fields.hidden("kind"),
pin(fields.snr("snr_db", "SNR_DB"), 6),
pin(fields.decimal("rssi_dbm", "RSSI_DBM", ".0f"), 8),
pin(fields.position(), 19),
fields.hidden("path"),
),
order=("TIME", "NODE", "SNR_DB", "RSSI_DBM", "LOCATION", "NAME"),
)
def _heard(heard: list[HeardNode]) -> Listing:
"""The capture window's per-node summary: what each node did over the whole window.
The aggregate the live stream cannot give — a count, a median, a best — one record per
node heard, most recently heard first.
Drawn **for a person only**. Every figure in it is computable from the records that
scrolled past above it, and emitting it as a second *shape* of line in the middle of an
otherwise homogeneous NDJSON stream would cost every consumer a discriminator it would
never otherwise need.
Args:
heard: Aggregated per-node statistics for the window.
Returns:
The scripted table (see :func:`meshterm.ui.script.columns`).
The listing.
"""
from ..ui import script
from ..ui import fields
from ..ui.report import Listing
table = script.columns(
"NODE",
"NAME",
"PKTS",
"MEDIAN_SNR_DB",
"BEST_SNR_DB",
"RSSI_DBM",
"LAST_HEARD",
right=("PKTS", "MEDIAN_SNR_DB", "BEST_SNR_DB", "RSSI_DBM"),
return Listing(
key="heard",
columns=(
fields.node("node", lanes=(("hash", "NODE"), ("name", "NAME"))),
fields.integer("packets", "PKTS"),
fields.snr("median_snr_db", "MEDIAN_SNR_DB"),
fields.snr("best_snr_db", "BEST_SNR_DB"),
fields.decimal("last_rssi_dbm", "RSSI_DBM", ".0f"),
fields.when("last_heard_at", "HEARD"),
fields.position(),
),
rows=[
{
"node": NodeRef(
name=node.name,
key=(node.public_key or "").lower() or None,
hash=node.node,
type=NODE_TYPE_LABELS.get(node.node_type),
),
"packets": node.count,
"median_snr_db": node.median_snr,
"best_snr_db": node.best_snr,
"last_rssi_dbm": node.last_rssi,
"last_heard_at": node.last_seen,
"position": _position(node.lat, node.lon),
}
for node in heard
],
order=(
"NODE",
"NAME",
"PKTS",
"MEDIAN_SNR_DB",
"BEST_SNR_DB",
"RSSI_DBM",
"HEARD",
"LOCATION",
),
plain_only=True,
)
for node in heard:
table.add_row(
node.node or script.NONE,
script.name(node.name),
str(node.count),
script.number(node.median_snr, "+.1f"),
script.number(node.best_snr, "+.1f"),
script.number(node.last_rssi, ".0f"),
script.stamp(node.last_seen),
)
return table
+115 -22
View File
@@ -33,7 +33,7 @@ from ..core.preferences import (
from .base import Tool, ToolResult, register
if TYPE_CHECKING: # pragma: no cover - typing only
from rich.table import Table
from ..ui.report import Block, Facts, Listing
@register
@@ -88,19 +88,23 @@ class PreferencesTool(Tool):
prefs = ctx.preferences
scripted = not isinstance(ctx.ui, TuiUi)
ops: list[tuple] = list(params.get("ops") or [])
blocks: list[Block] = []
applied: list[dict[str, Any]] = []
changes = 0
for op in ops:
action = op[0]
if action == "show":
ctx.ui.show(
_listing(prefs) if scripted else preferences_table(prefs, ctx.console.width)
)
if scripted:
blocks.append(_listing(prefs))
else:
ctx.ui.show(preferences_table(prefs, ctx.console.width))
elif action == "get":
# The bare value and nothing else — the caller named the key, and
# whether it is overridden is what `show`'s DEFAULT lane answers.
# whether it is overridden is what `show`'s DEFAULT lane answers. The
# document keeps both, since a parser has no lane to read them off.
spec = get_spec(op[1])
if scripted:
ctx.ui.note(_script_value(spec, prefs.get(spec.key)))
blocks.append(_one(prefs, spec))
else:
where = "changed" if prefs.is_overridden(spec.key) else "default"
shown = format_value(spec, prefs.get(spec.key))
@@ -110,11 +114,19 @@ class PreferencesTool(Tool):
after = prefs.set(op[1], op[2])
if after != before:
changes += 1
applied.append({"key": op[1], "previous": before, "value": after})
elif action == "reset":
was = {s.key: prefs.get(s.key) for s in PREFERENCES if prefs.is_overridden(s.key)}
changes += prefs.reset()
applied.extend(
{"key": key, "previous": before, "value": prefs.get(key)}
for key, before in was.items()
)
if changes:
prefs.save()
if scripted and applied is not None and any(op[0] in ("set", "reset") for op in ops):
blocks.append(_written(applied, prefs))
plural = "" if changes == 1 else "s"
message = (
@@ -123,7 +135,9 @@ class PreferencesTool(Tool):
if changes
else None
)
return ToolResult(summary={"changes": changes}, message=message)
return ToolResult(
summary={"changes": changes}, message=message, report=tuple(blocks) or None
)
# -- CLI --------------------------------------------------------------------
@@ -197,26 +211,105 @@ def _script_value(spec: PrefSpec, value: Any) -> str:
return str(value)
def _listing(prefs: Preferences) -> Table:
"""Every preference, its current value, and the built-in it would fall back to.
def _row(prefs: Preferences, spec: PrefSpec) -> dict[str, Any]:
"""One preference as a report row, in the shared setting shape.
Three facts per line and no fourth: the descriptions the page carries explain a
preference to someone deciding whether to change it, and they are what forces the
table to wrap. A scripted reader has decided.
``overridden`` is the fact the plain face makes a reader derive by comparing two
columns; a document that carried the same two numbers and left the comparison to the
consumer would be handing over homework it has the answer to.
"""
from ..ui.fields import Rendered
value = prefs.get(spec.key)
return {
"key": spec.key,
"value": Rendered(value, _script_value(spec, value)),
"type": spec.value_type,
# The reader's word for an enum's own key — never what `set` takes, which is the
# key itself and rides in `value`.
"label": (spec.choices or {}).get(value) if spec.value_type == "enum" else None,
"redacted": False,
"default": Rendered(spec.default, _script_value(spec, spec.default)),
"overridden": prefs.is_overridden(spec.key),
"help": spec.description,
}
def _columns() -> tuple:
"""The preference row's columns, shared by ``show`` and ``get``."""
from ..ui import fields
return (
fields.word("key", "PREFERENCE"),
fields.rendered("value", "VALUE"),
fields.hidden("type"),
fields.hidden("label"),
fields.hidden("redacted"),
fields.rendered("default", "DEFAULT"),
fields.hidden("overridden"),
fields.note("help", "DESCRIPTION"),
)
def _listing(prefs: Preferences) -> Listing:
"""Every preference, its value, its built-in default, and what it is for.
``DESCRIPTION`` comes straight from :attr:`PrefSpec.help` — the text is already
written, one short line each, and as the *last* column it can never disturb an
alignment or wrap. It used to be cut for parser hygiene, which is a bargain the plain
face no longer has to make: a listing is read by a person now, and a person choosing a
preference is exactly who that sentence was written for.
Args:
prefs: The preference set to report.
Returns:
The scripted table (see :func:`meshterm.ui.script.columns`).
The listing.
"""
from ..ui import script
from ..ui.report import Listing
table = script.columns("PREFERENCE", "VALUE", "DEFAULT")
for spec in PREFERENCES:
table.add_row(
spec.key,
_script_value(spec, prefs.get(spec.key)),
_script_value(spec, spec.default),
)
return table
return Listing(
key="preferences",
columns=_columns(),
rows=[_row(prefs, spec) for spec in PREFERENCES],
order=("PREFERENCE", "VALUE", "DEFAULT", "DESCRIPTION"),
)
def _one(prefs: Preferences, spec: PrefSpec) -> Facts:
"""One preference: the bare value plain, the whole setting object in the document."""
from ..ui.report import BARE, Facts
return Facts(
key="preference",
fields=_columns(),
values=_row(prefs, spec),
shape=BARE,
bare="value",
)
def _written(applied: list[dict[str, Any]], prefs: Preferences) -> Facts:
"""What ``set`` or ``reset`` changed — nothing plain, a record for whoever wants one.
The plain face says it in the exit status and in the acknowledgement on stderr, which
is the whole answer for a person. A caller managing a machine's configuration wants to
know *what moved*, and had no way to ask.
"""
from ..ui import fields
from ..ui.report import SILENT, Column, Facts
return Facts(
key="applied",
fields=(
fields.integer("changes", "changes"),
Column(key="applied"),
fields.word("path", "path"),
),
values={
"changes": len(applied),
"applied": applied,
"path": str(prefs.path) if prefs.path else None,
},
shape=SILENT,
)
+106 -63
View File
@@ -14,18 +14,21 @@ The CLI face prints the stored boards: ``meshterm records`` reads the tables so
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any
import typer
from ..context import AppContext
from ..core import exitcodes
from ..services.records import CATEGORIES, CATEGORY_BY_ID
from ..services.records import CATEGORIES, CATEGORY_BY_ID, Category
from ..ui.script import NONE
from .base import Tool, ToolResult, register
if TYPE_CHECKING: # pragma: no cover - typing only
from ..persistence.repository import DiscoveredPath
from ..services.records import Category
from ..ui.fields import NodeRef
from ..ui.report import Column
@register
@@ -76,7 +79,7 @@ class TrophyCaseTool(Tool):
def _show_records(
self, ctx: AppContext, *, category: str | None, width: int | None
) -> ToolResult:
"""Print the stored record boards (no device, no transmissions).
"""State the stored record boards (no device, no transmissions).
Args:
ctx: Shared application context.
@@ -87,44 +90,61 @@ class TrophyCaseTool(Tool):
A :class:`ToolResult` with the record count.
"""
from ..services import trace_runner
from ..ui import script
from ..ui import fields
from ..ui.report import Listing
wanted = [CATEGORY_BY_ID[category]] if category in CATEGORY_BY_ID else CATEGORIES
# Names come from stored history alone — no contacts, because this command reads
# the database and must work with no radio attached at all.
resolve = trace_runner.make_node_resolver(None, ctx.repo.node_names())
table = script.columns(
"CATEGORY",
"WIDTH",
"SCORE",
"UNIT",
"RECORDED",
"VERSION",
"ROUTE",
right=("WIDTH", "SCORE"),
)
total = 0
rows: list[dict] = []
for cat in wanted:
rows = ctx.repo.discoveries(cat.id, width_bytes=width)
rows.sort(key=lambda r: (r.width_bytes, r.score if cat.ascending else -r.score))
for record in rows:
table.add_row(
cat.id,
str(record.width_bytes),
_score(cat, record),
cat.unit,
script.stamp(record.discovered_at),
record.app_version,
_route(record, resolve),
found = ctx.repo.discoveries(cat.id, width_bytes=width)
found.sort(key=lambda r: (r.width_bytes, r.score if cat.ascending else -r.score))
for record in found:
floor = cat.id == "long_haul" and not record.stats.get("km_complete", True)
rows.append(
{
"category": cat.id,
"hash_bytes": record.width_bytes,
"score": _Score(record.score, cat, floor),
"unit": cat.unit,
"lower_bound": floor,
"recorded_at": record.discovered_at,
"version": record.app_version,
"path": record.spec or None,
"route": _route(record, resolve),
}
)
total += 1
if total:
ctx.ui.show(table)
listing = Listing(
key="records",
columns=(
fields.word("category", "CATEGORY"),
# The contract's word for the concept. The plain heading stays WIDTH,
# which is what a column of 1/2/4 reads as.
fields.integer("hash_bytes", "WIDTH"),
_score_column(),
fields.word("unit", "UNIT"),
# `>=273.0` is not a number, and a consumer comparing scores must be able
# to see the qualifier without string-matching a prefix off its own data.
fields.hidden("lower_bound"),
fields.when("recorded_at", "RECORDED"),
fields.word("version", "VERSION"),
# The transmitted spec, hop-aligned with the route beside it. It has no
# column — a route line already runs off the right — but it is what a
# caller re-walks with, so the document carries it.
fields.hidden("path"),
fields.route(),
),
rows=rows,
order=("CATEGORY", "WIDTH", "SCORE", "UNIT", "RECORDED", "VERSION", "ROUTE"),
)
return ToolResult(
summary={"records": total},
message=f"{total} record{'s' if total != 1 else ''} stored",
exit_code=exitcodes.OK if total else exitcodes.NO_RESULT,
summary={"records": len(rows)},
message=f"{len(rows)} record{'s' if len(rows) != 1 else ''} stored",
report=(listing,),
exit_code=exitcodes.OK if rows else exitcodes.NO_RESULT,
)
def register_cli(self, app: typer.Typer) -> None:
@@ -167,54 +187,77 @@ class TrophyCaseTool(Tool):
run_tool_command(self, tool_params)
def _score(category: Category, record: DiscoveredPath) -> str:
"""A record's score as a bare number, in the category's unit (which is its own column).
@dataclass(frozen=True, slots=True)
class _Score:
"""A record's score, and the one thing about it a bare number cannot carry.
``format_score`` writes ``273.0 km`` for a reader; the unit belongs in the ``UNIT``
column here so the ``SCORE`` column is one comparable number per row.
``format_score`` writes ``273.0 km`` for a reader; the unit is its own column here so
the ``SCORE`` column is one comparable number per row. What has to travel *with* the
number is the ``>=`` on a Longest-distance walk whose kilometre total is incomplete —
some hop on it has no known position, so the distance is a floor rather than a
measurement, and dropping the qualifier would turn a lower bound into a claim.
The one qualifier that survives is the ``>=`` on a Longest-distance walk whose
kilometre total is incomplete — some hop on it has no known position, so the distance
is a floor rather than a measurement. Dropping it would turn a lower bound into a
claim.
Args:
category: The discipline the record was set in.
record: The record.
Returns:
The score, prefixed ``>=`` where it is a lower bound.
The plain face wears the prefix; the document keeps the number a number and says
``lower_bound`` beside it, because a consumer comparing scores cannot be asked to
string-match a prefix off the front of its own data.
"""
if category.unit == "nodes":
value = str(int(record.score))
elif category.unit == "dB":
value = f"{record.score:+.1f}"
else:
value = f"{record.score:.1f}"
incomplete = category.id == "long_haul" and not record.stats.get("km_complete", True)
return f">={value}" if incomplete else value
value: float
category: Category
lower_bound: bool
@property
def text(self) -> str:
"""The score as the SCORE column prints it, qualifier and all."""
if self.category.unit == "nodes":
drawn = str(int(self.value))
elif self.category.unit == "dB":
drawn = f"{self.value:+.1f}"
else:
drawn = f"{self.value:.1f}"
return f">={drawn}" if self.lower_bound else drawn
def _route(record: DiscoveredPath, resolve: Callable[[str], str | None]) -> str:
"""A record's walk as the CLI's path line.
def _score_column() -> Column:
"""The SCORE column: the qualified figure plain, the bare number in the document."""
from ..ui.report import Column, Lane
return Column(
key="score",
lanes=(
Lane(
header="SCORE",
render=lambda s: s.text if s is not None else NONE,
align="right",
),
),
json=lambda s: None if s is None else s.value,
)
def _route(record: DiscoveredPath, resolve: Callable[[str], str | None]) -> list[NodeRef]:
"""A record's walk as the shared route shape.
Each hop is named where history knows the node and carries the hash it was actually
transmitted at — the spec's own hop, at this record's width — so the line can be read
against the ``spec`` a re-walk would take. Where the two lists disagree in length the
against the ``path`` a re-walk would take. Where the two lists disagree in length the
node id stands in, which is the only identity that hop has.
Our own node never appears in a record's walk: a discovery is scored on what it
reached, and the two ends of it are somebody else's.
Args:
record: The record whose walk to render.
resolve: Maps a node id to a friendly name, or ``None`` when unknown.
Returns:
The path line (see :func:`meshterm.ui.script.path`).
The hops in propagation order.
"""
from ..ui import script
from ..ui.fields import NodeRef
spec = record.spec.split(",") if record.spec else []
hops: list[tuple[str, str | None]] = []
hops: list[NodeRef] = []
for index, node in enumerate(record.route):
addressed = spec[index] if index < len(spec) else node
hops.append((resolve(node) or node, addressed))
return script.path(hops)
hops.append(NodeRef(name=resolve(node), hash=addressed))
return hops
+44 -8
View File
@@ -14,17 +14,19 @@ per invocation, the trace tool's rule.
from __future__ import annotations
from typing import Any
from typing import TYPE_CHECKING, Any
import typer
from rich.text import Text
from ..context import AppContext
from ..core import exitcodes
from ..core.connection import DeviceCommandError
from ..core.models import LoginResult
from ..core.models import NODE_TYPE_LABELS, Contact, LoginResult
from .base import Tool, ToolResult, register
if TYPE_CHECKING: # pragma: no cover - typing only
from ..ui.report import Facts
@register
class RepeaterAdminTool(Tool):
@@ -93,12 +95,8 @@ class RepeaterAdminTool(Tool):
command = str(params["command"])
ctx.remote_store.append_history(node, command)
reply = await device.send_remote_command(node, command, timeout=10.0)
if reply is not None:
# A remote node's reply is its own text, and it is the whole answer: printed as
# a Text, not as markup, so a reply containing a square bracket is a reply
# containing a square bracket.
ctx.ui.show(Text(reply))
return ToolResult(
report=(_answered(node, command, reply),),
summary={"node": name, "command": command, "replied": reply is not None},
# Silence is not failure — the command may well have landed — but there is
# nothing to report, and a script waiting on output should know which it got.
@@ -125,3 +123,41 @@ class RepeaterAdminTool(Tool):
if password is not None:
tool_params["password"] = password
run_tool_command(self, tool_params)
def _answered(node: Contact, command: str, reply: str | None) -> Facts:
"""What a remote node said back.
``reply`` is the node's own text, whole and verbatim — the plain face prints it bare
(as text and never as markup, so a reply holding a square bracket is a reply holding a
square bracket), and the document carries it raw, multi-line where the node sent
several lines.
It is deliberately **not parsed**. MeshTerm does not know the remote node's CLI
grammar, and a document that pretended to would be inventing structure a consumer
would then depend on.
"""
from ..ui import fields
from ..ui.fields import NodeRef
from ..ui.report import BARE, Facts
return Facts(
key="remote",
fields=(
fields.node("node", lanes=()),
fields.hidden("command"),
fields.word("reply", "reply"),
),
values={
"node": NodeRef(
name=node.name,
key=(node.public_key or "").lower() or None,
hash=(node.key_prefix or "").lower() or None,
type=NODE_TYPE_LABELS.get(node.node_type),
),
"command": command,
"reply": reply,
},
shape=BARE,
bare="reply",
)
+79 -57
View File
@@ -448,8 +448,9 @@ async def _trace_once_cli(
device_hash=device_hash,
)
)
else:
_print_trace(ctx, result, target, path, device_label, resolve, device_hash)
report = None
if not isinstance(ctx.ui, TuiUi):
report = _trace_report(result, target, path, device_label, resolve, device_hash)
summary: dict[str, Any] = {
"target": target,
@@ -477,38 +478,47 @@ async def _trace_once_cli(
return ToolResult(
summary=summary,
message=message,
report=report,
exit_code=exitcodes.OK if result.success else exitcodes.NO_RESULT,
)
def _print_trace(
ctx: AppContext,
def _trace_report(
result: TraceResult,
target: str,
path: str | None,
device_label: str,
resolve: Any,
device_hash: str | None,
) -> None:
"""Print one trace as scripted facts: the walk's outcome, then its per-hop readings.
) -> tuple:
"""State one trace: the walk's outcome, then its per-hop readings.
Two blocks. The first is what the walk *did* — one fact per line, ``route`` among them
as the CLI's path line, every node named and carrying the hash it was addressed by.
The second is a record per hop, and it names its two ends by that same hash rather
than repeating the names: the route line above is where the names are, and the hash is
what joins the two blocks (it is what carries a node's identity here, the way its
colour does on a screen).
as a drawn line, every node named and carrying the hash it was addressed by. The
second is a record per hop, and it names its two ends by that same **hash** rather than
repeating the names: the route line above is where the names are, and the hash is what
joins the two blocks (it is what carries a node's identity here, the way its colour
does on a screen). The document embeds the whole node at both ends instead, because a
structural join needs no key and each edge should be readable on its own.
Our own node is a hop like any other, named and hashed. The menu draws it as ``★``
because a reader never has to be told which node is theirs; this line is as often read
out of a file by somebody who was not at the prompt when it ran.
Args:
ctx: Shared application context.
result: The trace that ran.
target: The label it was addressed to.
path: The forced route as hex, or ``None`` when the device routed it.
device_label: Our own node's name, at both ends of the walk.
resolve: Maps a hop hash to a friendly name when known.
device_hash: Our own public key, so our ends carry a hash like every other hop.
Returns:
The report's blocks.
"""
from ..ui import script
from ..ui import fields
from ..ui.fields import NodeRef
from ..ui.report import Facts, Listing
hash_bytes = result.path_hash_bytes
@@ -519,53 +529,65 @@ def _print_trace(
raw = value.lower().removeprefix("0x")
return raw[: hash_bytes * 2] if hash_bytes else raw
facts: list[tuple[str, str]] = [
("target", target if target != PATH_TRACE_TARGET else script.NONE),
# "auto" is not a path spec, so it cannot be confused for one: every real value
# here is comma-separated hex.
("path", path or "auto"),
("success", "yes" if result.success else "no"),
("hops", script.number(result.hop_count if result.success else None)),
("min_snr_db", script.number(result.min_snr, "+.1f")),
("rtt_ms", script.number(result.round_trip_ms, ".0f")),
]
def node(label: str) -> NodeRef:
"""One end of a hop as the shared node shape."""
if not label or label == device_label:
return NodeRef(name=device_label, hash=short(device_hash), is_self=True)
return NodeRef(name=resolve(label) or None, hash=short(label))
edges = result.edges(device_label)
route = None
if edges:
nodes = [edges[0].origin] + [edge.destination for edge in edges]
facts.append(
(
"route",
script.path(
[
(device_label, short(device_hash))
if (not node or node == device_label)
else (resolve(node) or node, short(node))
for node in nodes
]
),
)
)
ctx.ui.show(script.pairs(facts))
route = [node(edges[0].origin)] + [node(edge.destination) for edge in edges]
if not edges:
return
ctx.ui.show(script.blank())
table = script.columns("HOP", "FROM", "TO", "SNR_DB", right=("HOP", "SNR_DB"))
def end(node: str) -> str:
"""A hop end as its addressed hash — our own device by its key, like any other."""
if not node or node == device_label:
return short(device_hash) or script.NONE
return short(node) or script.NONE
for edge in edges:
table.add_row(
str(edge.index),
end(edge.origin),
end(edge.destination),
script.number(edge.snr, "+.1f"),
)
ctx.ui.show(table)
facts = Facts(
key="trace",
fields=(
fields.word("target", "target"),
fields.spec(),
fields.flag("success", "success"),
fields.integer("hops", "hops"),
fields.snr("min_snr_db", "min_snr_db"),
fields.decimal("rtt_ms", "rtt_ms", ".0f"),
# New surface on the machine face: the model has always carried the TX power
# the walk ran at, and the plain facts block has no room for it.
fields.hidden("tx_dbm"),
# What `hash` means on this walk's nodes, so a consumer can join two traces
# that addressed the same node at different widths.
fields.hidden("hash_bytes"),
fields.route("route", "route"),
),
values={
"target": target if target != PATH_TRACE_TARGET else None,
"path": path,
"success": result.success,
"hops": result.hop_count if result.success else None,
"min_snr_db": result.min_snr,
"rtt_ms": result.round_trip_ms,
"tx_dbm": result.tx_power,
"hash_bytes": hash_bytes,
"route": route,
},
)
hops = Listing(
key="edges",
columns=(
fields.integer("index", "HOP"),
fields.node("from", lanes=(("hash", "FROM"),)),
fields.node("to", lanes=(("hash", "TO"),)),
fields.snr("snr_db", "SNR_DB"),
),
rows=[
{
"index": edge.index,
"from": node(edge.origin),
"to": node(edge.destination),
"snr_db": edge.snr,
}
for edge in edges
],
)
return (facts, hops)
def _last_traced_by_name(
+77 -34
View File
@@ -24,7 +24,7 @@ import typer
from ..context import AppContext
from ..core import exitcodes
from ..core.connection import DeviceCommandError
from ..core.models import Contact, LoginResult, TxOptResult
from ..core.models import NODE_TYPE_LABELS, Contact, LoginResult, TxOptResult
from ..services import trace_runner, tx_optimizer
from ..ui.widgets import tx_opt_summary, tx_opt_table
from .base import Tool, ToolResult, register
@@ -319,11 +319,12 @@ class TxOptimizeTool(Tool):
if result.applied:
ctx.log.info("set TX power %s on %s", result.best_tx, admin_node.name)
report = None
if isinstance(ctx.ui, TuiUi):
ctx.ui.show(tx_opt_table(result))
ctx.ui.show(tx_opt_summary(result))
else:
_print_sweep(ctx, result, path)
report = _sweep_report(result, path, admin_node, contacts)
if no_result:
restored = (
@@ -341,6 +342,7 @@ class TxOptimizeTool(Tool):
)
return ToolResult(
report=report,
# No level got a trace through, so nothing was measured and nothing was tuned:
# the sweep ran and has nothing to report.
exit_code=exitcodes.NO_RESULT if no_result else exitcodes.OK,
@@ -535,46 +537,87 @@ def _fmt_snr(snr: float | None) -> str:
return f"{snr:+.1f}" if snr is not None else " n/a"
def _print_sweep(ctx: AppContext, result: TxOptResult, path: str) -> None:
"""Print a TX sweep as scripted facts: the outcome, then every level measured.
def _sweep_report(
result: TxOptResult, path: str, admin_node: Contact, contacts: list[Contact]
) -> tuple:
"""State a TX sweep: the outcome, then every level measured.
The winner comes first, because it is what the command was asked for and what a caller
acts on; the per-level records follow, so the choice can be checked against the
measurements it was made from. The menu's ``★`` on the winning row has no column here
— ``optimal_tx`` above the table already names it, and a mark is something to look at
rather than something to test.
— ``optimal_tx_dbm`` above the table already names it, and a mark is something to look
at rather than something to test.
The two nodes become the shared node shape, and here that matters more than anywhere
else: on the plain line they are told apart by nothing but their names, and a caller
correlating a sweep with a trace has no key to join on.
Args:
ctx: Shared application context.
result: The completed sweep.
path: The forced route the traces walked, as hex hops.
"""
from ..ui import script
admin_node: The node whose TX power was tuned.
contacts: The contact list, for placing the target by name.
ctx.ui.show(
script.pairs(
[
("tuning_node", result.admin_node),
("target", result.target),
("path", path),
("optimal_tx_dbm", script.number(result.best_tx)),
("target_snr_db", script.number(result.best_snr, "+.1f")),
("reliability", f"{result.best_success_rate:.2f}"),
("previous_tx_dbm", script.number(result.original_tx)),
("applied", "yes" if result.applied else "no"),
]
Returns:
The report's blocks.
"""
from ..ui import fields
from ..ui.fields import NodeRef
from ..ui.report import Facts, Listing
def node(name: str) -> NodeRef | None:
"""One end of the tuned link as the shared node shape."""
contact = next((c for c in contacts if c.name == name), None)
if contact is None:
return NodeRef(name=name) if name else None
return NodeRef(
name=contact.name,
key=(contact.public_key or "").lower() or None,
hash=(contact.key_prefix or "").lower() or None,
type=NODE_TYPE_LABELS.get(contact.node_type),
)
facts = Facts(
key="tx_optimize",
fields=(
fields.node("tuning_node", lanes=(("name", "tuning_node"),)),
fields.node("target", lanes=(("name", "target"),)),
fields.spec(),
fields.integer("optimal_tx_dbm", "optimal_tx_dbm"),
fields.snr("target_snr_db", "target_snr_db"),
fields.decimal("reliability", "reliability", ".2f"),
fields.integer("previous_tx_dbm", "previous_tx_dbm"),
fields.flag("applied", "applied"),
),
values={
"tuning_node": node(result.admin_node) or NodeRef(name=admin_node.name),
"target": node(result.target),
"path": path,
"optimal_tx_dbm": result.best_tx,
"target_snr_db": result.best_snr,
# A fraction in [0, 1], not a formatted "1.00": the plain column rounds for
# the eye and the document keeps the number a number.
"reliability": result.best_success_rate,
"previous_tx_dbm": result.original_tx,
"applied": result.applied,
},
)
if not result.levels:
return
ctx.ui.show(script.blank())
lanes = ("TX_DBM", "TARGET_SNR_DB", "SUCCESSES", "SAMPLES")
table = script.columns(*lanes, right=lanes)
for level in result.sorted_by_tx():
table.add_row(
str(level.tx_power),
script.number(level.target_snr, "+.1f"),
str(level.successes),
str(level.samples),
)
ctx.ui.show(table)
levels = Listing(
key="levels",
columns=(
fields.integer("tx_dbm", "TX_DBM"),
fields.snr("target_snr_db", "TARGET_SNR_DB"),
fields.integer("successes", "SUCCESSES"),
fields.integer("samples", "SAMPLES"),
),
rows=[
{
"tx_dbm": level.tx_power,
"target_snr_db": level.target_snr,
"successes": level.successes,
"samples": level.samples,
}
for level in result.sorted_by_tx()
],
)
return (facts, levels)
+1 -1
View File
@@ -1003,7 +1003,7 @@ async def _run_now(
"""
from ..tools.config import apply_ops
changes, artifacts = await apply_ops(ctx, device, snapshot, ops)
changes, artifacts, _report = await apply_ops(ctx, device, snapshot, ops)
for artifact in artifacts:
ctx.ui.note(f"[ok]●[/ok] wrote [accent]{artifact}[/accent]")
await ctx.ui.present(title=title)
+58
View File
@@ -106,6 +106,30 @@ class ChannelRef:
hash: str | None = None
@dataclass(frozen=True, slots=True)
class Rendered:
"""A typed value beside the one plain rendering only its own registry can produce.
Almost every cell is derivable from its value alone, which is what lets a
:class:`~meshterm.ui.report.Lane` be a function of one argument. A *setting* is the
exception: what ``config set`` takes back for a given key is a question only that key's
spec can answer — an enum wants its number, an empty string wants ``""``, a preference
drops the unit its page shows — and the spec is per row, not per column.
So the row carries both halves, computed once where the spec is in hand. The plain
face prints ``text`` and the document emits ``value``, and neither has to parse the
other's output to get at what it needs, which is the same rule the rest of this module
keeps by other means.
Attributes:
value: The typed value, or ``None`` where the source never reported one.
text: What the plain face prints, in the form the matching ``set`` accepts back.
"""
value: Any
text: str
def node_json(ref: NodeRef | None) -> dict[str, Any] | None:
"""One node as its machine object, or ``null`` where there is no node at all."""
if ref is None:
@@ -352,6 +376,21 @@ def spec(key: str = "path", header: str = "path") -> Column:
)
def rendered(key: str, header: str) -> Column:
"""A setting's value: what ``set`` takes back plain, and the typed value in the document.
Takes a :class:`Rendered` (see there for why this one concept needs both halves in the
row). The round-trip claim is the plain face's whole design for these commands —
``config show > f`` and feeding ``f`` back in must work — and the document's typed
value is the machine face's largest single gain over it: ``9`` and not ``"9"``.
"""
return Column(
key=key,
lanes=(Lane(header=header, render=_rendered_cell),),
json=lambda r: None if r is None else r.value,
)
def hexid(key: str, header: str) -> Column:
"""A hash or a key on its own, lowercase, never truncated and never elided."""
return Column(
@@ -372,6 +411,20 @@ def note(key: str, header: str) -> Column:
return Column(key=key, lanes=(Lane(header=header, render=script.text),), json=None)
def plain_only(column: Column) -> Column:
"""The same concept, drawn for a reader and left out of the document.
For a lane the machine face says *better* somewhere else rather than not at all. A
transcript's ``PEER`` is the case: one column holding whichever of a channel or a
contact the conversation had, because a reader has the conversation in front of them —
where the document keeps ``channel`` and ``node`` apart, since a parser has to know
which of the two it is holding and cannot tell from the label.
"""
from dataclasses import replace
return replace(column, json=None)
def hidden(key: str) -> Column:
"""A fact the document carries and the plain face has no room for.
@@ -382,6 +435,11 @@ def hidden(key: str) -> Column:
return Column(key=key)
def _rendered_cell(value: Rendered | None) -> str:
"""One setting's cell: the text its own spec produced, or the absent token."""
return script.NONE if value is None else value.text
def _position_cell(where: Position | None) -> str:
"""One position's cell: the pair, or the absent token."""
return script.NONE if where is None else script.location(where.lat, where.lon)
+51 -143
View File
@@ -1,41 +1,50 @@
"""The scripted CLI's output language — plain text, aligned, unadorned.
"""The plain CLI's output language — text for a person at a prompt, aligned and unadorned.
MeshTerm has two front ends and they are read by different things. The menu is read by a
person sitting in front of it, and everything else in ``meshterm/ui/`` is built for that
reader: colour that carries meaning, glyphs that stand for concepts, frames that group.
The CLI is read by ``awk``. So it gets its own vocabulary, and this module is all of it.
MeshTerm has two front ends and a third face. The menu is read by a person sitting in
front of it, and everything else in ``meshterm/ui/`` is built for that reader: colour that
carries meaning, glyphs that stand for concepts, frames that group. ``--json`` is read by
a program, often somewhere else and often later. This module is the third: **someone at a
prompt who typed a command to find something out, and wants the answer legible in one
glance.**
The rules, and why each one:
That is a change of premise. The plain face used to be a serialisation format a person
could squint at, and several of its rules existed only to make *splitting* safe — quoted
names, absolute timestamps, a comma between path hops. Splitting is the machine face's job
now (see :mod:`meshterm.ui.report` and :mod:`meshterm.ui.renderers`), so those rules were
paying rent they could not afford. What survives is everything that makes the output a
Unix utility's output, and one new rule in place of the quoting.
* **Alignment is the delimiter.** A column that lines up is already one field to the eye,
so a name is bare (:func:`name`). The *escaping* under the quoting is not gone and never
can be (:func:`_escaped`): a name is broadcast by its own node and a message body is
filled in by a stranger, and neither may end the record it sits in.
* **No colour.** ``stdout`` is a pipe more often than a terminal, and a colour that
survives into a file is noise in it. The console this module builds has
``color_system=None``, so Rich emits no escape sequence at all — not a dim, not a bold.
That is stricter than ``no_color``, which keeps the attributes and drops only the hues.
A *highlight* colour is reserved for the one thing that would be unreadable without it
(a route graph's emphasised path); nothing in the CLI draws one, and nothing claims the
colour speculatively.
* **No wrapping.** A record is a line. Wrapping turns one record into two and puts the
second one's fields under the wrong headings, so the console is made :data:`WIDTH`
cells wide — far past any real line — and every column is ``no_wrap``. Long output runs
off the right, and the terminal (or ``less -S``, or the reader's pipe) decides what to
do about it.
second one's fields under the wrong headings, so the console is made :data:`WIDTH` cells
wide — far past any real line — and every column is ``no_wrap``. The one deliberate
exception is a *written page* (``about``, ``support``), which is drawn on a console
:data:`PAGE_WIDTH` cells wide instead: a paragraph is not a record, it has no headings
to land under, and unwrapped it is a single 600-cell line no terminal can read.
* **No frames.** No panel borders, no table boxes, no rules, no titles. The command the
reader typed is the title; a box around the answer is a second one.
* **Aligned columns, one header line.** ``ps`` and ``df``'s shape: an uppercase header
row, then records, fields padded apart by :data:`GUTTER` spaces. See :func:`columns`.
* **Names are quoted where they share a line.** A node name can hold a space, a comma,
even a quote, so in a listing a bare name is not a field — it is however many fields the
name happens to split into. Every name in a :func:`columns` row or a :func:`path` line
goes through :func:`quote`. A :func:`pairs` line is the exception and needs no quoting:
it has exactly two fields, so the value is the rest of the line whatever it holds, and
quotes there would only be something to strip back off.
* **Times are absolute.** ``3h`` is for a person watching a screen; a script wants
something it can sort and subtract, so every timestamp is local ISO-8601 to the second
(:func:`stamp`). The relative ages and their recency heat stay in the menu, where the
colour they are half made of still exists.
* **One token for "nothing".** :data:`NONE` — a lone ``-`` — in every column, so an
absent value never has to be told apart from an empty one, an ``—``, an ``n/a`` or a
blank.
* **A time is an age.** ``5m``, ``3h``, ``never`` — the same ladder the menu's columns use
(:func:`age`), because "how long ago" is the question a person is actually asking. An
absolute instant survives where the instant *is* the fact: the device clock, a scheduled
appointment, a live capture's own clock (:func:`stamp`). ``--absolute`` swaps the whole
language back for anyone who wants it, and the machine face is absolute UTC regardless.
* **A route is drawn, a path is typed.** The lexicon already splits these: a *path* is an
ordered hop spec you compose or force, a *route* is the concrete node sequence a walk
took. So a path keeps its commas (:func:`spec`) because it round-trips into ``--path``,
and a route takes arrows (:func:`route`) because it is a picture and visibly is not a
spec. Rendering both with commas promised a round trip only one of them has.
* **One token for "nothing".** :data:`NONE` — a lone ``-`` — in every column, so an absent
value never has to be told apart from an empty one, an ``—``, an ``n/a`` or a blank.
``never`` is a *value*, not an absence: a node that has never been heard is a fact.
* **No trailing whitespace.** Padding the last column to its width would put invisible
spaces at the end of every line; the console trims them on the way out
(:class:`_Trimmed`).
@@ -105,6 +114,10 @@ def _escaped(text: str) -> str:
true of a value that arrived over the air. Named escapes where one reads (``\n``),
numeric escapes everywhere else, and the backslash doubled so the whole thing reads
back unambiguously.
This is what survived the retirement of the quoting. The quotes made a name *one
field* for a splitter, and alignment does that now; nothing else can stop a name
holding a newline from ending its own record early.
"""
out: list[str] = []
for ch in text:
@@ -117,60 +130,27 @@ def _escaped(text: str) -> str:
return "".join(out)
def quote(value: str | None) -> str:
r"""Wrap a name in doublequotes, escaping the ones inside it.
A node name is user-supplied text that can hold a space, a comma, or a quote, so it
is never a field on its own — quoting is what makes it one. The escaping is the
ordinary backslash convention and covers the backslash and the line breaks too, so the
value reads back unambiguously and cannot end the record it sits in.
Args:
value: The name to quote. ``None`` and the empty string both give ``""``.
Returns:
The quoted name, e.g. ``"Yagi-Repeater"`` or ``"He said \"hi\""``.
"""
return '"' + _escaped(value or "").replace('"', '\\"') + '"'
def name(value: str | None) -> str:
r"""A node's name as a field: quoted when there is one, :data:`NONE` when there isn't.
r"""A node's name as a field: bare and escaped, or :data:`NONE` when it has none.
:func:`quote` renders a missing name as ``""``, which is a different claim — *this node
is called nothing* rather than *this node never told us what it is called*. The CLI has
one token for the second, and seven of the ten places that printed a name were making
the first claim by accident, so the same unnamed node read as ``""`` in one listing and
``-`` in the next.
The quotes are gone. They existed so a name holding a space or a comma stayed one
field for whatever was splitting the line; the splitter reads JSON now, and in plain
text the column alignment already says where a field ends. What survives is the
escaping — a node broadcasts its own name, so every name here is remote data, and a
newline in one must still not end the record it sits in.
It is also the distinction the JSON rendering needs: an absent name is ``null`` and an
empty one is ``""``, and it cannot be recovered later from a field that flattened both.
An empty name is still :data:`NONE` rather than an empty cell: ``""`` says *called
nothing* where the token says *never said*, and those are different facts. It is also
the distinction the machine face needs — an absent name is ``null``, an empty one is
``""`` — and it cannot be recovered from a field that flattened both.
Args:
value: The name, or ``None``/empty where the node never supplied one.
Returns:
The quoted name, or :data:`NONE`.
The escaped name, or :data:`NONE`.
"""
return quote(value) if value else NONE
def oneline(value: str | None) -> str:
r"""``value`` as a field that can end a record without breaking it.
For the one field a listing leaves unquoted because it *is* the rest of the line — a
message body. "The rest of the line" and "anything at all" are not compatible: a body
holding a newline ended its record early and left the remainder indented under the
other columns, reading as a second record with an empty ``TIME``. A body is also the
field a stranger fills in, so that is not a hypothetical shape of message.
Args:
value: The free text. ``None`` gives :data:`NONE`.
Returns:
The text with its line breaks and tabs escaped, and no quotes around it.
"""
return NONE if value is None else _escaped(value)
return _escaped(value) if value else NONE
def text(value: str | None) -> str:
@@ -359,34 +339,6 @@ def spec(hops: Iterable[str]) -> str:
return ",".join(parts) if parts else NONE
def path(hops: Iterable[tuple[str, str | None]]) -> str:
"""Render a hop sequence as the CLI's path line.
``"Origin" (3d),"Relay" (f2),"Us" (a1)`` — each node's name quoted (:func:`quote`),
its hash in parentheses *outside* the quotes, hops separated by a bare comma.
The hash is there because the CLI has no colour. On a screen a route's hops are told
apart by their key-derived hues, and matched to a route graph's one-byte labels the
same way; with the colour gone, the hash is what carries that identity.
Our own node is a hop like any other — named and hashed, never the menu's ``★``. The
star says "you already know who this is", which is true of a reader and false of a
parser.
The comma is the separator ``--path`` already takes, and the quoting is what keeps the
line splittable when a name contains one.
Args:
hops: ``(label, hash)`` pairs in propagation order; a ``None`` hash prints the
name alone, for a node whose identity is genuinely unknown.
Returns:
The joined path line, or :data:`NONE` when there are no hops.
"""
parts = [f"{quote(label)} ({value})" if value else quote(label) for label, value in hops]
return ",".join(parts) if parts else NONE
def columns(*headers: str, right: Sequence[str] = ()) -> Table:
"""Build the scripted table: an uppercase header line, then padded records.
@@ -508,50 +460,6 @@ def stream(*widths: tuple[str, int], right: Sequence[str] = ()) -> Lanes:
return Lanes(widths=widths, right=frozenset(right))
def wrap(body: str, width: int = PAGE_WIDTH) -> str:
"""Fold a written page's paragraphs to ``width``, leaving its structure alone.
The deliberate exception to "no wrapping", and the reason the rule exists says why:
wrapping splits a *record* and lands half its fields under the wrong headings. A
written page has no records and no headings — and left unwrapped, ``meshterm about``
prints paragraphs six hundred cells long that no terminal can read.
A line's own indent is kept and carried onto its continuations, so a bullet stays a
bullet and an indented block stays indented; a blank line stays blank; a line that is
already short is untouched.
Args:
body: The page, as its plain renderer laid it out.
width: The cell width to fold to.
Returns:
The page with every over-long line folded under its own indent.
"""
import textwrap
out: list[str] = []
for line in body.splitlines():
stripped = line.lstrip()
if not stripped:
out.append("")
continue
indent = line[: len(line) - len(stripped)]
# A bullet's continuations hang under its text, not under its marker.
hanging = indent + (" " if stripped[:1] in "-*•" else "")
out.extend(
textwrap.wrap(
stripped,
width=max(width - len(indent), 20),
initial_indent=indent,
subsequent_indent=hanging,
break_long_words=False,
break_on_hyphens=False,
)
or [indent + stripped]
)
return "\n".join(out)
# -- the console ---------------------------------------------------------------------
+1 -1
View File
@@ -194,7 +194,7 @@ async def test_advert_cadence_op_writes_the_store(ctx: AppContext) -> None:
device = await ctx.device()
ctx._ui = _NoteUi() # type: ignore[assignment]
snapshot = dict(await device.get_self_info())
changes, _ = await apply_ops(
changes, _artifacts, _report = await apply_ops(
ctx, device, snapshot, [("advert_cadence", True, 168), ("advert_cadence", False, OFF)]
)
assert changes == 2
+122 -23
View File
@@ -1,15 +1,18 @@
"""End-to-end tests for what the CLI actually prints and returns.
:mod:`tests.test_script_output` covers the vocabulary; this covers the commands built out
of it — the real Typer app, driven against the simulator, asserting the four things a
script depends on: no escape sequences, no wrapped records, a header its records line up
under, and a documented exit status.
:mod:`tests.test_script_output` covers the vocabulary and :mod:`tests.test_report` the
seam; this covers the commands built out of both — the real Typer app, driven against the
simulator, asserting the things a caller depends on: no escape sequences, no wrapped
records, a header its records line up under, a documented exit status — and, under
``--json``, a document that parses and a status that did not change because of a rendering
flag.
Every command here is read-only or simulator-backed, so nothing transmits.
"""
from __future__ import annotations
import json
import re
from pathlib import Path
@@ -26,21 +29,26 @@ _ANSI = re.compile(r"\x1b\[")
#: MeshTerm writes itself is no longer consumed either, and lands in the caller's pipe.
_MARKUP = re.compile(r"\[/?(?:ok|err|warn|muted|accent|brand|bold|dim|reverse)\b[^\]]*\]")
#: The shape of a scripted path line: one or more quoted node names, each optionally
#: followed by its hash in parentheses *outside* the quotes, joined by a bare comma. This
#: is the whole grammar — a line matching it splits, whatever the names hold.
_HOP = r'"(?:[^"\\]|\\.)*"(?: \([^)]*\))?'
_PATH_LINE = re.compile(f"{_HOP}(?:,{_HOP})*")
#: The shape of a drawn route: one or more hops, each a bare name with its hash in
#: parentheses (or one of the two halves alone, where history knows only one), joined by
#: the spaced arrow. This is the whole grammar, and the arrow is what makes it one — a
#: name holding a comma no longer needs quoting to stay one hop.
_HOP = r"[^\n→]+?"
_ROUTE_LINE = re.compile(f"{_HOP}(?: → {_HOP})*")
#: The shape of a path *spec*: comma-joined lowercase hex, exactly what ``--path`` takes
#: back. The one line on either face that round-trips, so nothing may creep into it.
_PATH_SPEC = re.compile(r"[0-9a-f]+(?:,[0-9a-f]+)*")
#: The commands whose output is a listing or a key/value block, and the fields a caller
#: should find in each. Every one runs against the mock companion or the database alone.
_LISTINGS: list[tuple[str, list[str], list[str]]] = [
# (command, arguments, headings or keys the output must carry)
("contacts", [], ["NAME", "TYPE", "HEARD", "PKTS", "KEY"]),
("contacts", [], ["NAME", "TYPE", "HEARD", "PKTS", "HASH", "LOCATION", "KEY"]),
("info", [], ["name", "public_key", "role"]),
("config", ["show"], ["name", "radio_freq", "tx_power"]),
("preferences", ["show"], ["PREFERENCE", "VALUE", "DEFAULT"]),
("chat", ["list"], ["CONVERSATION", "KIND", "UNREAD", "LAST_TIME", "LAST_TEXT"]),
("preferences", ["show"], ["PREFERENCE", "VALUE", "DEFAULT", "DESCRIPTION"]),
("chat", ["list"], ["CONVERSATION", "KIND", "UNREAD", "LAST", "LAST_TEXT"]),
("platform", [], ["platform", "icons"]),
]
@@ -97,6 +105,10 @@ def run(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): # noqa: ANN201 - a cl
return invoke
#: Commands that refuse a machine-readable face outright, and say so as a usage error.
_REFUSES_JSON = {("specimen",)}
# -- the four rules -------------------------------------------------------------------
@@ -162,6 +174,70 @@ def test_every_registered_command_obeys_the_rules(run, leaf) -> None: # noqa: A
assert line == line.rstrip(), f"{' '.join(leaf)} padded a line: {line!r}"
@pytest.mark.parametrize(
"leaf",
[c for c in _leaf_commands() if c not in _NOT_ONE_SHOT | _KEEPS_ITS_COLOUR],
ids=lambda leaf: "-".join(leaf),
)
def test_every_registered_command_answers_json_that_parses(run, leaf) -> None: # noqa: ANN001
"""``--json`` is a rendering, so it belongs to the CLI rather than to a list of tools.
It reached two commands out of twenty and stopped, because each one that wanted it had
to restate its whole answer above the rendering. The same sweep that keeps the plain
rules honest now runs every command a second time and reads what comes back: **every
line on stdout is one complete document**, on success and on empty alike, and a
command that fails writes nothing there at all.
"""
result = run("--json", *leaf)
for line in result.stdout.splitlines():
json.loads(line) # raises, with the offending line, if anything else got out
if result.exit_code in (exitcodes.OK, exitcodes.NO_RESULT):
assert result.stdout.strip(), f"{' '.join(leaf)} exited {result.exit_code} silently"
@pytest.mark.parametrize(
"leaf",
[c for c in _leaf_commands() if c not in _NOT_ONE_SHOT | _KEEPS_ITS_COLOUR],
ids=lambda leaf: "-".join(leaf),
)
def test_the_two_faces_never_disagree_about_the_exit_status(run, leaf) -> None: # noqa: ANN001
"""``--json`` changes the rendering, never the report.
The status is the report — it is what a caller branches on — and a flag about *how the
answer looks* has no business moving it. This is the cross-check that stops the two
faces drifting the way they did before: ``devices --json`` used to exit ``0`` where the
plain path said ``5``, so a script could not tell an empty scan from a full one by
asking the same question twice.
"""
assert run("--json", *leaf).exit_code == run(*leaf).exit_code, " ".join(leaf)
@pytest.mark.parametrize("leaf", sorted(_REFUSES_JSON), ids=lambda leaf: "-".join(leaf))
def test_a_command_whose_output_is_the_colour_refuses_json_loudly(run, leaf) -> None: # noqa: ANN001
"""``specimen``'s output *is* the colour; a document of it would be an empty gesture.
A loud refusal is the honest answer, and it is the same one this project already gives
for the map, the dashboard and the live feed — which simply have no subcommand to
refuse from. It is a usage error, so stdout stays empty and the sentence goes where
every other failure's does.
"""
result = run("--json", *leaf)
assert result.exit_code == exitcodes.USAGE
assert result.stdout == ""
assert "machine-readable" in result.stderr
def test_the_menu_has_no_document_and_says_so(run) -> None: # noqa: ANN001
"""``meshterm --json`` with no subcommand would launch a full-screen session.
There is no document an interactive session can emit, and starting one would hang
whatever was waiting to parse it.
"""
result = run("--json")
assert result.exit_code == exitcodes.USAGE
assert result.stdout == ""
def test_the_sweep_actually_covers_the_whole_command_surface() -> None:
"""A derived list is only a guarantee while it is still finding the commands.
@@ -177,14 +253,25 @@ def test_the_sweep_actually_covers_the_whole_command_surface() -> None:
# -- listings -------------------------------------------------------------------------
def test_contacts_quotes_every_name_and_stamps_every_time(run) -> None: # noqa: ANN001
"""A record is quoted where a name could hold a space, and timed absolutely."""
def test_contacts_names_every_node_bare_and_ages_every_time(run) -> None: # noqa: ANN001
"""The two rules that reversed, checked on the listing they were reversed for.
A name is bare — the alignment is what makes it one field now — and ``HEARD`` is an
age, because "heard recently?" is the question this listing is opened to ask and an ISO
instant makes the reader do arithmetic to answer it. ``--absolute`` puts the instants
back, for anyone who wants them.
"""
header, *records = result_lines(run("contacts"))
assert header.split() == ["NAME", "TYPE", "HEARD", "PKTS", "KEY"]
assert header.split() == ["NAME", "TYPE", "HEARD", "PKTS", "HASH", "LOCATION", "KEY"]
assert records, "the simulator always has contacts"
for line in records:
assert line.startswith('"')
stamp = line.split()[-3]
assert not line.startswith('"')
age = line.split()[2]
assert age in ("now", "never") or age[-1] in "mhdw"
absolute = result_lines(run("--absolute", "contacts"))[1:]
for line in absolute:
stamp = line.split()[2]
assert stamp == "-" or stamp[:4].isdigit()
@@ -280,20 +367,32 @@ def test_preferences_get_prints_a_value_its_own_set_would_take(run) -> None: #
# -- path lines -----------------------------------------------------------------------
def test_a_trace_prints_its_route_as_the_scripted_path_line(run) -> None: # noqa: ANN001
"""Quoted names, hashes outside the quotes, comma-separated, no star for our own node."""
result = run("trace", "--target", "Alice")
def test_a_trace_draws_its_route_with_arrows_and_keeps_its_path_in_commas(run) -> None: # noqa: ANN001
"""The sharpest of the reversals, and the reason is the project's own lexicon.
A *path* is a spec you compose and can paste back into ``--path``, so it stays
comma-separated and round-trippable. A *route* is what a walk actually did; it is not a
spec, and rendering it with commas promised a round trip it does not have. Our own node
is still named like any other hop — the menu's ``★`` says "you already know who this
is", which is true of the reader and false of whoever opens the file later.
"""
result = run("trace-path", "--path", "a1,d4,a1")
assert result.exit_code == exitcodes.OK, result.output
facts = dict(line.split(None, 1) for line in result_lines(result) if line and " " in line)
assert facts["success"] == "yes"
route = facts["route"]
assert "★" not in route and "→" not in route
assert _PATH_LINE.fullmatch(route), route
assert "★" not in route
assert "→" in route
assert _ROUTE_LINE.fullmatch(route), route
assert _PATH_SPEC.fullmatch(facts["path"]), facts["path"]
assert "→" not in facts["path"]
def test_a_trace_reports_its_per_hop_readings_under_a_header(run) -> None: # noqa: ANN001
"""The second block is a record per hop, joined to the route line by hash."""
lines = result_lines(run("trace", "--target", "Alice"))
lines = result_lines(run("trace-path", "--path", "a1,d4,a1"))
header = next(line for line in lines if line.startswith("HOP"))
assert header.split() == ["HOP", "FROM", "TO", "SNR_DB"]
+10 -2
View File
@@ -13,6 +13,7 @@ from types import SimpleNamespace
import pytest
from meshterm.core import exitcodes
from meshterm.core.courier_store import DONE_CAP, QUEUED, CourierStore
from meshterm.core.models import ChatMessage, Contact, utcnow
from meshterm.core.watch_store import WatchStore
@@ -273,7 +274,11 @@ async def test_cli_list_renders_waiting_and_finished(tmp_path: Path) -> None:
result = await CourierTool().run(ctx, {"cli_action": "list"})
assert result.summary == {"entries": 2}
assert ctx.ui.shown # a table was rendered
# The outbox is *stated*, not printed: the tool hands back the rows and the CLI
# boundary picks a renderer for them (see meshterm.ui.report).
listing = result.report[0]
assert sorted(row["state"] for row in listing.rows) == ["delivered", "waiting"]
assert sorted(row["text"] for row in listing.rows) == ["landed", "still waiting"]
async def test_cli_list_empty_notes_and_counts_zero(tmp_path: Path) -> None:
@@ -283,7 +288,10 @@ async def test_cli_list_empty_notes_and_counts_zero(tmp_path: Path) -> None:
ctx = _ToolCtx(tmp_path)
result = await CourierTool().run(ctx, {"cli_action": "list"})
assert result.summary == {"entries": 0}
assert ctx.ui.shown == []
assert result.exit_code == exitcodes.NO_RESULT
# An empty listing, not an absent one: the plain face draws nothing from it and the
# machine face gets `[]`, which is a document a consumer can read.
assert result.report[0].rows == []
async def test_cli_cancel_removes_a_waiting_entry(tmp_path: Path) -> None:
+211 -71
View File
@@ -1,9 +1,11 @@
"""Tests for the scripted CLI's output language (:mod:`meshterm.ui.script`).
"""Tests for the plain CLI's output language (:mod:`meshterm.ui.script`).
The CLI is read by ``awk``, and every rule that makes that possible is testable: no
colour, no wrapping, no framing, quoted names, absolute times, one token for absent, one
documented exit status per outcome. These are the enforcement point for
:mod:`meshterm.ui.script` the way ``test_gallery`` is for the screens.
The plain face is read by a person at a prompt — ``--json`` carries the machine now — and
every rule that makes it legible is testable: no colour, no wrapping, no framing, bare
names that still cannot break a record, relative ages, arrows for a route and commas for a
path, one token for absent, one documented exit status per outcome. These are the
enforcement point for :mod:`meshterm.ui.script` the way ``test_gallery`` is for the
screens, and :mod:`tests.test_report` is for the seam above them.
"""
from __future__ import annotations
@@ -35,35 +37,42 @@ def _rendered(*renderables: object) -> str:
return buffer.getvalue()
# -- quoting -------------------------------------------------------------------------
# -- names ---------------------------------------------------------------------------
@pytest.mark.parametrize(
"value,expected",
[
("Yagi-Repeater", '"Yagi-Repeater"'),
("YUL Cartierville", '"YUL Cartierville"'),
("Node, Inc", '"Node, Inc"'),
('He said "hi"', '"He said \\"hi\\""'),
("back\\slash", '"back\\\\slash"'),
("", '""'),
(None, '""'),
("Yagi-Repeater", "Yagi-Repeater"),
("YUL Cartierville", "YUL Cartierville"),
("Node, Inc", "Node, Inc"),
('He said "hi"', 'He said "hi"'),
("back\\slash", "back\\\\slash"),
],
)
def test_quote_makes_any_name_exactly_one_field(value: str | None, expected: str) -> None:
"""A name is quoted and its quotes and backslashes escaped, whatever it holds.
def test_a_name_is_bare_because_alignment_is_the_delimiter(value: str, expected: str) -> None:
"""The quotes are retired; a column that lines up is already one field to the eye.
This is what lets a listing be split: an unquoted ``YUL Cartierville`` is two fields,
and an unquoted ``Node, Inc`` is two hops of a path line.
They existed so a name holding a space or a comma stayed one field for whatever was
splitting the line. The splitter reads JSON now, and in plain text the padding says
where a field ends — so a reader gets back the six cells the quotes were spending on
every row, and a name reads as the name the node broadcast.
"""
assert script.quote(value) == expected
assert script.name(value) == expected
def test_a_quoted_name_survives_a_round_trip_through_the_escaping() -> None:
"""The escaping is the ordinary backslash convention, so the value reads back."""
for name in ('a "quoted" name', "back\\slash", 'both\\"kinds'):
body = script.quote(name)[1:-1]
assert body.replace('\\"', '"').replace("\\\\", "\\") == name
def test_a_name_still_cannot_end_the_record_it_sits_in() -> None:
"""The escaping under the quoting stays, and it is the half that was load-bearing.
A node broadcasts its own name, so every name here is remote data. Nothing about
alignment stops a newline, a tab or an ESC in one from ending the record early — and
an ESC inside a name is a live colour run written into the caller's file, which is the
one thing "not a single escape sequence" exists to prevent.
"""
assert script.name("two\nlines") == "two\\nlines"
assert script.name("tabbed\there") == "tabbed\\there"
assert script.name("bell\a") == "bell\\x07"
assert "\x1b" not in script.name("esc\x1b[31m")
@pytest.mark.parametrize(
@@ -85,13 +94,8 @@ def test_a_message_body_can_never_end_its_own_record(body: str, expected: str) -
also the one field a stranger fills in, so this is not a hypothetical message. Quotes
inside it are left alone: nothing wraps it, so nothing needs escaping from.
"""
assert script.oneline(body) == expected
assert "\n" not in script.oneline(body)
def test_a_name_holding_a_newline_cannot_end_its_record_either() -> None:
"""The same hazard one field over — and a name is broadcast by its own node."""
assert script.quote("two\nlines") == '"two\\nlines"'
assert script.text(body) == expected
assert "\n" not in script.text(body)
@pytest.mark.parametrize("absent", [None, ""])
@@ -100,12 +104,12 @@ def test_a_node_that_never_gave_a_name_reads_as_absent_not_as_empty(absent: str
Seven of the ten places that printed a name made the first claim by accident, so the
same unnamed node read as ``""`` in one listing and ``-`` in the next. One helper now
answers it, and the distinction is the one the JSON rendering needs too — an absent
name is ``null``, an empty one is ``""``, and a field that flattened both cannot say
which it meant.
answers it, and the distinction is the one the machine face needs too — an absent name
is ``null``, an empty one is ``""``, and a field that flattened both cannot say which
it meant.
"""
assert script.name(absent) == script.NONE
assert script.name("Alice") == '"Alice"'
assert script.name("Alice") == "Alice"
# -- timestamps ----------------------------------------------------------------------
@@ -128,19 +132,97 @@ def test_stamp_reads_an_unknown_offset_as_absent() -> None:
assert script.stamp(datetime(2026, 9, 7, 22, 22, 41)) == script.NONE
def test_no_relative_age_ever_reaches_the_scripted_output() -> None:
"""``3h`` is for a screen; the CLI's stamp is absolute even for a fresh time."""
assert script.stamp(datetime.now(timezone.utc) - timedelta(seconds=5)) != "now"
def test_a_time_is_an_age_unless_the_instant_is_itself_the_fact() -> None:
"""The rule that reversed, and the reason it reversed.
A person at a prompt reading ``2026-09-07T19:58:53-04:00`` is doing arithmetic to
answer "recently?", which is the question they typed the command to ask. So an age is
the default now. :func:`~meshterm.ui.script.stamp` survives for the three places the
instant *is* the answer — the device clock, an appointment set with ``--at``, and a
live capture's own ``TIME`` column, every row of which would otherwise read ``now`` —
and for every column under ``--absolute``.
"""
fresh = datetime.now(timezone.utc) - timedelta(seconds=5)
assert script.age(fresh) == "now"
assert script.stamp(fresh)[:4].isdigit()
@pytest.mark.parametrize(
"seconds,expected",
[
(0, "now"),
(59, "now"),
(60, "1m"),
(3599, "59m"),
(3600, "1h"),
(86400, "1d"),
(604800, "1w"),
],
)
def test_the_age_ladder_steps_where_a_person_would_step(seconds: int, expected: str) -> None:
"""``now``, ``5m``, ``3h``, ``2d``, ``4w`` — the menu's own column ladder.
Delegated to :func:`~meshterm.ui.widgets._format_age` rather than re-derived, so the
two faces of the same age cannot drift apart.
"""
assert script.age(datetime.now(timezone.utc) - timedelta(seconds=seconds)) == expected
def test_an_age_reads_an_unknown_offset_as_absent_exactly_as_a_stamp_does() -> None:
"""A naive datetime is a time whose offset we don't know, which is not a fact."""
assert script.age(None) == script.NONE
assert script.age(datetime(2026, 9, 7, 22, 22, 41)) == script.NONE
def test_never_is_a_value_and_the_absent_token_is_not() -> None:
"""A node that has never been heard is a *fact*; a row with no such time is not.
Flattening both onto ``-`` made "we have not heard from it" and "this kind of row has
no heard time" read the same, which is the distinction the column exists to draw.
"""
assert script.age(None, absent="never") == "never"
assert script.age(None) == script.NONE
@pytest.mark.parametrize(
"seconds,expected",
[(42, "42s"), (360, "6m"), (125, "2m 5s"), (93784, "1d 2h"), (3600, "1h"), (-125, "2m 5s")],
)
def test_a_duration_reads_in_two_units_at_most(seconds: int, expected: str) -> None:
"""``93784`` is a number nobody can hold in their head; ``1d 2h`` is the same fact.
Two adjacent units, largest first, so the magnitude arrives at a glance and the
precision never outruns what anyone would act on. Only ever a gloss beside the raw
figure — the key says what the number counts, and a caller reading the key must still
find a number under it.
"""
assert script.duration(seconds) == expected
def test_a_location_is_one_field_because_it_is_one_fact() -> None:
"""A coordinate pair is what goes into a map's search box, verbatim.
Two columns would also mean two ``-`` for every node that has never shared a position,
which is worse than one.
"""
assert script.location(45.50190, -73.56740) == "45.50190,-73.56740"
assert script.location(45.5, None) == script.NONE
assert script.location(None, None) == script.NONE
# -- the absent token -----------------------------------------------------------------
def test_one_token_stands_for_every_kind_of_absence() -> None:
"""A reader (and a parser) learns ``-`` once, and never meets an em dash or an n/a."""
"""A reader learns ``-`` once, and never meets an em dash, an ``n/a`` or a blank.
``never`` is the one word that is *not* an absence: it is a value, and the age lane is
where it is spelled (see the ladder tests above).
"""
assert script.NONE == "-"
assert script.number(None) == script.NONE
assert script.path([]) == script.NONE
assert script.route([]) == script.NONE
assert script.spec([]) == script.NONE
def test_number_formats_with_its_spec_when_there_is_one() -> None:
@@ -150,33 +232,91 @@ def test_number_formats_with_its_spec_when_there_is_one() -> None:
assert script.number(7) == "7"
# -- path lines -----------------------------------------------------------------------
# -- routes and paths ------------------------------------------------------------------
def test_a_path_line_quotes_each_node_and_hangs_its_hash_outside_the_quotes() -> None:
"""The CLI's path shape: ``"Name" (hash)``, hops joined by a bare comma."""
line = script.path([("Alice", "3d"), ("Yagi-Repeater", "f2")])
assert line == '"Alice" (3d),"Yagi-Repeater" (f2)'
def test_a_route_is_drawn_with_arrows_and_a_path_keeps_its_commas() -> None:
"""The lexicon's own split, finally drawn: a *route* is walked, a *path* is typed.
Both used to be comma-separated, which made a route look like something you could
paste back into ``--path``. You cannot — the names are in it — so that comma was
promising a round trip only the spec has. The arrow says "this is a picture", and
carries the direction the packet travelled while it is there.
"""
assert (
script.route([("Alice", "3d"), ("Yagi-Repeater", "f2")])
== "Alice (3d) → Yagi-Repeater (f2)"
)
assert script.spec(["a1", "d4", "a1"]) == "a1,d4,a1"
def test_a_path_line_names_our_own_node_like_any_other() -> None:
"""No star: the glyph says "you already know this one", which is false of a parser."""
line = script.path([("MockCompanion", "a1"), ("Alice", "3d"), ("MockCompanion", "a1")])
def test_a_route_names_our_own_node_like_any_other_hop() -> None:
"""No star, and the reason is who reads the line.
The menu draws our own node as ``★`` because a reader never has to be told which node
is theirs. A route line is as often read out of a file, by somebody who was not at the
prompt when it ran — and by then the star names nothing.
"""
line = script.route([("MockCompanion", "00"), ("Alice", "3d"), ("MockCompanion", "00")])
assert "★" not in line
assert line.startswith('"MockCompanion" (a1)')
assert line.endswith('"MockCompanion" (a1)')
assert line.startswith("MockCompanion (00)")
assert line.endswith("MockCompanion (00)")
def test_a_path_line_prints_a_name_alone_when_no_hash_places_it() -> None:
"""An unidentified node has a name and nothing else; it does not get an empty ``()``."""
assert script.path([("Alice", None)]) == '"Alice"'
def test_a_route_hop_never_shows_an_empty_pair_of_parentheses() -> None:
"""Half an identity prints as the half there is — a name alone, or a hash alone.
Which is how the menu's own :class:`~meshterm.ui.pathline.PathLine` labels an
unresolved hop, and the reason both halves are optional: history often knows one.
"""
assert script.route([("Alice", None)]) == "Alice"
assert script.route([(None, "3d")]) == "3d"
assert script.route([(None, None)]) == script.NONE
def test_a_path_line_splits_even_when_a_name_holds_a_comma() -> None:
"""The quoting is what makes the bare comma separator safe."""
line = script.path([("Node, Inc", "3d"), ("Bob", "f2")])
assert line.count(",") == 2 # the one inside the name, and the one separating hops
assert line == '"Node, Inc" (3d),"Bob" (f2)'
def test_the_arrow_is_what_makes_the_quoting_unnecessary() -> None:
"""The claim moved: the quotes made a comma safe, and now nothing has to.
A comma inside a node's name can no longer be read as a hop boundary, because the hop
boundary is not a comma any more.
"""
line = script.route([("Node, Inc", "3d"), ("Bob", "f2")])
assert line == "Node, Inc (3d) → Bob (f2)"
assert line.count("→") == 1
def test_a_spec_is_the_one_line_on_either_face_that_round_trips() -> None:
"""It is what ``--path`` takes back, so it grows no name, no arrow and no space."""
assert script.spec(["a1b2", "d4e5", "a1b2"]) == "a1b2,d4e5,a1b2"
assert " " not in script.spec(["a1", "d4"])
# -- live streams ----------------------------------------------------------------------
def test_a_stream_pins_its_lanes_because_it_cannot_measure_them() -> None:
"""A live capture cannot size its columns, because it has not seen the records yet.
:func:`~meshterm.ui.script.columns` sizes each lane to its widest value once every
record is in. ``monitor`` prints a row the instant a packet lands, and the row after it
may be twice as wide — so those two streams were gutter-joined with no alignment at
all, which is the last thing the quoting was holding up.
"""
lanes = script.stream(("TIME", 19), ("NODE", 8), ("SNR_DB", 6), right=("SNR_DB",))
assert lanes.header == "TIME NODE SNR_DB"
assert lanes.record("2026-09-08T02:25:45", "a1b2c3d4", "+7.0") == (
"2026-09-08T02:25:45 a1b2c3d4 +7.0"
)
def test_a_streamed_value_wider_than_its_lane_overruns_rather_than_lying() -> None:
"""One row out of many is simply wider, and nothing is elided.
Which is why the one unbounded field — a node's name, a message body — goes last: it
can then push nothing at all.
"""
lanes = script.stream(("NODE", 8), ("NAME", 6))
assert lanes.record("a1", "A Very Long Node Name") == "a1 A Very Long Node Name"
# -- the console ----------------------------------------------------------------------
@@ -204,9 +344,9 @@ def test_a_node_name_is_never_read_as_rich_markup(name: str) -> None:
parser over. Neither is hypothetical: a name is the one field a stranger controls.
"""
table = script.columns("NAME", "TYPE")
table.add_row(script.quote(name), "node")
table.add_row(script.name(name), "node")
record = _rendered(table).splitlines()[1]
assert record.startswith(script.quote(name)), record
assert record.startswith(script.name(name)), record
def test_the_scripted_console_never_wraps_a_record_onto_a_second_line() -> None:
@@ -221,8 +361,8 @@ def test_the_scripted_console_never_wraps_a_record_onto_a_second_line() -> None:
def test_the_scripted_console_leaves_no_trailing_whitespace() -> None:
"""Padding the last column to its width would put invisible spaces on every line."""
table = script.columns("NAME", "TYPE")
table.add_row(script.quote("a-long-name"), "node")
table.add_row(script.quote("b"), "repeater")
table.add_row(script.name("a-long-name"), "node")
table.add_row(script.name("b"), "repeater")
for line in _rendered(table).splitlines():
assert line == line.rstrip(), repr(line)
@@ -235,21 +375,21 @@ def test_the_gutter_still_separates_columns_after_the_trimming() -> None:
columns (see :class:`meshterm.ui.script._Trimmed`).
"""
table = script.columns("NAME", "TYPE", "PKTS", right=("PKTS",))
table.add_row(script.quote("Alice"), "node", "12")
table.add_row(script.name("Alice"), "node", "12")
header, record = _rendered(table).splitlines()
assert header.split() == ["NAME", "TYPE", "PKTS"]
assert record.split() == ['"Alice"', "node", "12"]
assert record.split() == ["Alice", "node", "12"]
# Spelled out, not written as `" " * script.GUTTER`: an assertion that reads the
# constant it is checking narrows to `" " in record` when the gutter narrows, which
# is exactly the state the gutter exists to prevent.
assert '"Alice" node' in record
assert "Alice node" in record
def test_columns_align_so_every_record_splits_the_same_way() -> None:
"""Every field lands in its heading's column, whatever the widths."""
table = script.columns("NAME", "TYPE")
table.add_row(script.quote("Yagi-Repeater"), "repeater")
table.add_row(script.quote("Al"), "node")
table.add_row(script.name("Yagi-Repeater"), "repeater")
table.add_row(script.name("Al"), "node")
header, *records = _rendered(table).splitlines()
starts = {line.index("re") if "repeater" in line else line.index("node") for line in records}
assert starts == {header.index("TYPE")} # every TYPE value opens under its heading
@@ -265,8 +405,8 @@ def test_a_numeric_column_really_right_aligns() -> None:
still aligns. Checking where a field *starts* cannot see this; only its end can.
"""
table = script.columns("NAME", "PKTS", right=("PKTS",))
table.add_row(script.quote("Alice"), "1174")
table.add_row(script.quote("Bob"), "3")
table.add_row(script.name("Alice"), "1174")
table.add_row(script.name("Bob"), "3")
header, *records = _rendered(table).splitlines()
assert len({len(line) for line in records}) == 1, records
assert {len(line) for line in records} == {len(header)}
@@ -296,10 +436,10 @@ def test_the_trimmer_keeps_the_gutter_when_the_stream_flushes_between_columns()
"""
buffer = io.StringIO()
stream = script._Trimmed(buffer)
for segment in ('"Alice"', " ", "node", " ", "12", "\n"):
for segment in ("Alice", " ", "node", " ", "12", "\n"):
stream.write(segment)
stream.flush()
assert buffer.getvalue() == '"Alice" node 12\n'
assert buffer.getvalue() == "Alice node 12\n"
def test_the_trimmer_still_drops_the_run_that_ends_a_line() -> None:
@@ -359,7 +499,7 @@ def test_a_group_is_unframed_member_by_member() -> None:
def test_no_rendered_line_is_a_box_drawing_rule() -> None:
"""No borders and no header rules — a listing is its header line and its records."""
table = script.columns("NAME", "TYPE")
table.add_row(script.quote("Alice"), "node")
table.add_row(script.name("Alice"), "node")
assert "─" not in _rendered(table)
+1 -1
View File
@@ -224,7 +224,7 @@ async def test_apply_setting_remembers_through_the_config_executor(ctx) -> None:
pubkey = (await device.get_self_info())["public_key"]
snapshot = await build_snapshot(device)
changes, _ = await apply_ops(
changes, _artifacts, _report = await apply_ops(
ctx, device, snapshot, [("set", "name", "Ops-Node"), ("set", "tx_power", "18")]
)
assert changes == 2