Jean-Pierre MartineauandClaude Opus 4.8
dfd7d2332d
Add a one-cell braille battery indicator to the status bar, shown only when the companion reports a pack. The cell fills bottom-to-top in four tiers — green healthy, orange at a quarter, red near empty — with the true percent beside it; it blinks red↔dim when critically low and sweeps a filling animation while charging. A session-long BatteryService polls the pack, maps millivolts to a state-of-charge estimate off a single-cell LiPo curve, and infers charging from a rising terminal voltage (firmware exposes no charging flag). The header pins the gauge flush right and the activity pulse cedes exactly the cells it needs. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
929 lines
38 KiB
Python
929 lines
38 KiB
Python
"""Reusable Rich widgets: banner, status pill, trace tables, and progress bars."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
from dataclasses import dataclass
|
||
from datetime import datetime
|
||
from typing import TYPE_CHECKING, Callable, Optional, Sequence
|
||
|
||
from rich import box
|
||
from rich.console import Console, Group
|
||
from rich.panel import Panel
|
||
from rich.progress import (
|
||
BarColumn,
|
||
MofNCompleteColumn,
|
||
Progress,
|
||
SpinnerColumn,
|
||
TextColumn,
|
||
TimeElapsedColumn,
|
||
)
|
||
from rich.table import Table
|
||
from rich.text import Text
|
||
|
||
from .. import __version__
|
||
from ..core.channels import is_name_derived, is_public_channel, is_public_name
|
||
from ..core.models import (
|
||
LOCAL_DEVICE_LABEL,
|
||
NODE_TYPE_CHAT,
|
||
NODE_TYPE_LABELS,
|
||
NODE_TYPE_REPEATER,
|
||
NODE_TYPE_ROOM,
|
||
NODE_TYPE_SENSOR,
|
||
Contact,
|
||
HopAggregate,
|
||
TraceResult,
|
||
TraceStats,
|
||
TxOptResult,
|
||
utcnow,
|
||
)
|
||
from .map_render import _NODE, _REPEATER, _SELF
|
||
from .theme import name_style, snr_style
|
||
|
||
if TYPE_CHECKING:
|
||
from ..core.discovery import DiscoveredDevice
|
||
|
||
#: Maps a hop's raw key-prefix hash to a display label (a contact name when known).
|
||
NodeResolver = Callable[[Optional[str]], Optional[str]]
|
||
|
||
|
||
def channel_glyph(name: str, secret: Optional[bytes]) -> str:
|
||
"""The one-character openness marker for a channel, shared across channel-facing screens.
|
||
|
||
``＃`` marks a name-derived (``#``-style) channel, ``🌐`` a fixed-key well-known public
|
||
channel (e.g. the firmware default ``Public``), and ``🔒`` a private one. Every glyph is a
|
||
single double-width cell, so callers can prefix rows with ``"{glyph} "`` without disturbing
|
||
column alignment. When the secret is unknown, the name alone is used to guess.
|
||
|
||
Args:
|
||
name: The channel name.
|
||
secret: The channel's 16-byte secret, or ``None`` when only the name is known.
|
||
|
||
Returns:
|
||
A single-character glyph.
|
||
"""
|
||
if secret is None:
|
||
return "＃" if is_public_name(name) else "🔒"
|
||
if is_name_derived(name, secret):
|
||
return "＃"
|
||
if is_public_channel(name, secret):
|
||
return "🌐"
|
||
return "🔒"
|
||
|
||
def _identity(label: Optional[str]) -> Optional[str]:
|
||
"""Default node resolver: leave labels untouched."""
|
||
return label
|
||
|
||
|
||
def banner(
|
||
profile: Optional[str], mock: bool, selected_device: Optional["DiscoveredDevice"] = None
|
||
) -> Panel:
|
||
"""Build the application header panel.
|
||
|
||
Args:
|
||
profile: Active device profile name, if any.
|
||
mock: Whether the simulator is in use.
|
||
selected_device: The discovered device chosen for this session, if any, shown
|
||
when no named profile is in use.
|
||
|
||
Returns:
|
||
A Rich :class:`Panel` showing the app name, version, and connection target.
|
||
"""
|
||
if mock:
|
||
target = "[warn]simulator[/warn]"
|
||
elif profile:
|
||
target = profile
|
||
elif selected_device is not None:
|
||
target = selected_device.label
|
||
else:
|
||
target = "[muted]no device[/muted]"
|
||
body = Text.from_markup(
|
||
f"[brand]MeshTerm[/brand] [muted]v{__version__}[/muted]\n"
|
||
f"[muted]device:[/muted] {target}"
|
||
)
|
||
return Panel(body, border_style="accent", expand=False, title="[accent]Mesh[/accent]")
|
||
|
||
|
||
def make_progress(console: Console) -> Progress:
|
||
"""Create a themed progress bar for trace sweeps.
|
||
|
||
Args:
|
||
console: The console to render into.
|
||
|
||
Returns:
|
||
A configured :class:`rich.progress.Progress` (use as a context manager).
|
||
"""
|
||
return Progress(
|
||
SpinnerColumn(style="accent"),
|
||
TextColumn("[accent]{task.description}"),
|
||
BarColumn(complete_style="brand", finished_style="ok"),
|
||
MofNCompleteColumn(),
|
||
TimeElapsedColumn(),
|
||
console=console,
|
||
transient=False,
|
||
)
|
||
|
||
|
||
def _link_text(
|
||
origin: Optional[str],
|
||
destination: Optional[str],
|
||
device_label: str,
|
||
resolve: NodeResolver = _identity,
|
||
hash_bytes: Optional[int] = None,
|
||
device_hash: Optional[str] = None,
|
||
) -> Text:
|
||
"""Render an ``origin -> destination`` link through THE path widget.
|
||
|
||
A two-node :func:`path_text` in the trace presentation: a known node as
|
||
``name (hash)``, an unknown one as its bare hash, our own device as its label
|
||
(plus ``device_hash`` when supplied). Hashes are truncated to ``hash_bytes`` so
|
||
they match how the trace command addressed each node.
|
||
|
||
Args:
|
||
origin: The transmitting node (``None`` = our device).
|
||
destination: The receiving node (``None`` = our device).
|
||
device_label: The label used for our own device.
|
||
resolve: Maps a raw hop hash to a friendly name when the node is known.
|
||
hash_bytes: Path-hash width (bytes) to truncate shown hashes to.
|
||
device_hash: Our own device's key/hash, shown alongside its label when known.
|
||
|
||
Returns:
|
||
A :class:`Text` like ``us (a1) → Alice (3d)`` with the arrow muted.
|
||
"""
|
||
ends = [
|
||
None if not node or node == device_label else node
|
||
for node in (origin, destination)
|
||
]
|
||
return path_text(
|
||
ends, resolve, prefix_bytes=hash_bytes or 8, self_name=device_label,
|
||
show_hash=True, hash_bytes=hash_bytes, device_hash=device_hash,
|
||
)
|
||
|
||
|
||
def traces_table(
|
||
traces: list[TraceResult],
|
||
device_label: str = LOCAL_DEVICE_LABEL,
|
||
resolve: NodeResolver = _identity,
|
||
device_hash: Optional[str] = None,
|
||
) -> Table:
|
||
"""Render every trace's per-hop SNR side by side, one column per trace.
|
||
|
||
Hops are framed as ``origin -> destination`` links (the first originates at our
|
||
device, the last returns to it) and aligned by position across traces, so each
|
||
column is one trace's SNR readings down the shared path. Each node is annotated with
|
||
its hash at the command's path-hash width. A trailing ``min`` row shows each trace's
|
||
bottleneck — the per-trace values the run's *median min SNR* summarizes. Traces that
|
||
never replied appear as a ``✗`` column.
|
||
|
||
Args:
|
||
traces: The individual traces to display, in run order.
|
||
device_label: Name to show for our own device at the path's endpoints.
|
||
resolve: Maps a raw hop hash to a friendly contact name when known.
|
||
device_hash: Our own device's key/hash, annotated onto the path's endpoints.
|
||
|
||
Returns:
|
||
A Rich :class:`Table` with a column per trace.
|
||
"""
|
||
target = traces[0].target if traces else ""
|
||
# All traces in a run share the command's path-hash width; take it from the first
|
||
# successful one so node hashes render at the width they were addressed.
|
||
hash_bytes = next((t.path_hash_bytes for t in traces if t.success), None)
|
||
table = Table(title=f"Trace → {target}", border_style="muted", expand=False)
|
||
table.add_column("HOP", justify="right", style="muted")
|
||
table.add_column("FROM → TO")
|
||
for i in range(1, len(traces) + 1):
|
||
table.add_column(f"#{i}", justify="right")
|
||
|
||
# Index each trace's edges by hop position so columns line up even when traces
|
||
# take different-length paths (a missing hop shows as a muted dash).
|
||
edges_by_trace = [
|
||
{e.index: e for e in t.edges(device_label)} if t.success else {} for t in traces
|
||
]
|
||
indices = sorted({idx for edges in edges_by_trace for idx in edges})
|
||
for idx in indices:
|
||
link = next(
|
||
(
|
||
_link_text(
|
||
edges[idx].origin, edges[idx].destination, device_label, resolve,
|
||
hash_bytes, device_hash,
|
||
)
|
||
for edges in edges_by_trace
|
||
if idx in edges
|
||
),
|
||
Text("—", style="muted"),
|
||
)
|
||
row: list[Text] = [Text(str(idx)), link]
|
||
for edges in edges_by_trace:
|
||
edge = edges.get(idx)
|
||
if edge is None:
|
||
row.append(Text("—", style="muted"))
|
||
else:
|
||
row.append(Text(f"{edge.snr:+.1f}", style=snr_style(edge.snr)))
|
||
table.add_row(*row)
|
||
|
||
# Bottleneck per trace: the weakest link the run's median min SNR is taken over.
|
||
min_row: list[Text] = [Text("min", style="muted"), Text("bottleneck", style="muted")]
|
||
for trace in traces:
|
||
if not trace.success:
|
||
min_row.append(Text("✗", style="err"))
|
||
elif trace.min_snr is not None:
|
||
min_row.append(Text(f"{trace.min_snr:+.1f}", style=snr_style(trace.min_snr)))
|
||
else:
|
||
min_row.append(Text("—", style="muted"))
|
||
table.add_section()
|
||
table.add_row(*min_row)
|
||
return table
|
||
|
||
|
||
def path_text(
|
||
hops: Sequence[Optional[str]],
|
||
resolve: NodeResolver = _identity,
|
||
*,
|
||
prefix_bytes: int = 0,
|
||
self_name: Optional[str] = None,
|
||
empty: str = "direct",
|
||
show_hash: bool = False,
|
||
hash_bytes: Optional[int] = None,
|
||
device_hash: Optional[str] = None,
|
||
dim_from: Optional[int] = None,
|
||
) -> Text:
|
||
"""Render a hop sequence compactly on one line — THE path widget.
|
||
|
||
The one way MeshTerm shows a walked, relayed, or planned hop sequence, wherever one
|
||
appears: a packet's ``via`` row, a message's delivery paths, a feed note, a trace's
|
||
walked route or planned spec. Each hop renders as its resolved name — coloured in
|
||
the app-wide per-name hue, our own node pure white — or, unnamed, as its hash
|
||
through :func:`highlighted_hash` (the addressed prefix lit, the rest muted; colour
|
||
stays the "this is a name" signal). Hops are joined by muted ``→`` arrows and, by
|
||
default, no hash is repeated after a name, so the compact form survives a 72-column
|
||
row. The trace-flavoured options: ``show_hash`` annotates each named hop with the
|
||
hash it is addressed by (``Alice (3d63)``), a ``None`` hop is our own device at a
|
||
route's endpoints, and ``dim_from`` fades the tail a caller wants read as automatic
|
||
(a boomerang's mirrored return leg). An empty path reads as ``empty``.
|
||
|
||
Args:
|
||
hops: The hops in propagation order — hex hashes, with ``None`` marking our
|
||
own device (empty strings are skipped; ``dim_from`` counts rendered hops).
|
||
resolve: Maps a hop hash to a friendly name when known.
|
||
prefix_bytes: Path-hash width to light in unnamed hops' hashes (0 = none).
|
||
self_name: Our own node's name — drawn in the white ``you`` style when a
|
||
resolved name matches it, and naming any ``None`` device hop.
|
||
empty: The muted text shown when there are no hops (e.g. ``"direct"``).
|
||
show_hash: Annotate named hops (and, with ``device_hash``, our device) with
|
||
their hash in parentheses — the trace presentation.
|
||
hash_bytes: Truncate shown/annotated hashes to this byte width (the width the
|
||
hops were addressed at); ``None`` shows them whole.
|
||
device_hash: Our own device's key, annotated onto ``None`` hops when
|
||
``show_hash`` is on.
|
||
dim_from: Render hops at/after this index — and the arrows into them — faint
|
||
(a planned route's return leg); ``None`` dims nothing.
|
||
|
||
Returns:
|
||
A one-line :class:`Text`. Space tighter than the path is the caller's call,
|
||
per surface: ellipsize (``no_wrap``), wrap under a hanging indent, or crop
|
||
with horizontal scrolling.
|
||
"""
|
||
shown = [h for h in hops if h is None or h]
|
||
if not shown:
|
||
return Text(empty, style="muted")
|
||
text = Text()
|
||
for i, hop in enumerate(shown):
|
||
dim = dim_from is not None and i >= dim_from
|
||
if i:
|
||
text.append(" → ", style="faint" if dim else "muted")
|
||
text.append_text(
|
||
_path_node(
|
||
hop, resolve, prefix_bytes=prefix_bytes, self_name=self_name,
|
||
show_hash=show_hash, hash_bytes=hash_bytes, device_hash=device_hash,
|
||
dim=dim,
|
||
)
|
||
)
|
||
return text
|
||
|
||
|
||
def _path_node(
|
||
hop: Optional[str],
|
||
resolve: NodeResolver,
|
||
*,
|
||
prefix_bytes: int,
|
||
self_name: Optional[str],
|
||
show_hash: bool,
|
||
hash_bytes: Optional[int],
|
||
device_hash: Optional[str],
|
||
dim: bool,
|
||
) -> Text:
|
||
"""One node of :func:`path_text` (see there for the rendering rules)."""
|
||
note_style = "faint" if dim else "muted"
|
||
if hop is None:
|
||
text = Text(self_name or LOCAL_DEVICE_LABEL, style="faint" if dim else "you")
|
||
annotated = _shorten_hash(device_hash, hash_bytes) if device_hash else ""
|
||
if show_hash and annotated:
|
||
text.append(f" ({annotated})", style=note_style)
|
||
return text
|
||
named = resolve(hop)
|
||
if named and named != hop:
|
||
if dim:
|
||
style = "faint"
|
||
elif self_name and named == self_name:
|
||
style = "you"
|
||
else:
|
||
style = name_style(named)
|
||
text = Text(named, style=style)
|
||
if show_hash:
|
||
text.append(f" ({_shorten_hash(hop, hash_bytes)})", style=note_style)
|
||
return text
|
||
shown = _shorten_hash(hop, hash_bytes)
|
||
if dim:
|
||
return Text(shown, style="faint")
|
||
return highlighted_hash(shown, prefix_bytes)
|
||
|
||
|
||
def highlighted_hash(value: str, prefix_bytes: int, width: Optional[int] = None) -> Text:
|
||
"""Render a hex key with its leading path-hash prefix highlighted — THE hash widget.
|
||
|
||
The one way MeshTerm displays a hash, wherever one appears: the first
|
||
``prefix_bytes`` bytes are the slice other nodes address in a forced trace path
|
||
(the path-hash), shown in the brand colour with the remainder muted, so the
|
||
addressable prefix stands out within the otherwise full key. A ``width`` budget
|
||
shorter than the key ellipsizes it (the ``…`` takes the colour of the digit it
|
||
replaces, so a highlight wider than the budget still reads as one).
|
||
|
||
Args:
|
||
value: The key as hex, optionally ``0x``-prefixed and mixed-case.
|
||
prefix_bytes: Number of leading bytes the current path-hash mode addresses; ``0``
|
||
(or negative) leaves the whole key un-highlighted.
|
||
width: Display budget in cells; a longer key shows ``width - 1`` digits plus an
|
||
ellipsis, a shorter one is right-padded to the budget so lanes stay aligned.
|
||
``None`` shows the key whole, unpadded.
|
||
|
||
Returns:
|
||
A styled :class:`Text` of the key (exactly ``width`` cells when given).
|
||
"""
|
||
raw = value.lower().removeprefix("0x")
|
||
split = max(0, prefix_bytes) * 2
|
||
pad = 0
|
||
ellipsis = False
|
||
if width is not None and len(raw) > width:
|
||
raw, ellipsis = raw[: max(0, width - 1)], True
|
||
elif width is not None:
|
||
pad = width - len(raw)
|
||
text = Text()
|
||
text.append(raw[:split], style="brand")
|
||
text.append(raw[split:], style="muted")
|
||
if ellipsis:
|
||
text.append("…", style="brand" if len(raw) < split else "muted")
|
||
text.append(" " * pad)
|
||
return text
|
||
|
||
|
||
def _shorten_hash(value: str, hash_bytes: Optional[int]) -> str:
|
||
"""Return ``value`` as bare hex truncated to ``hash_bytes`` bytes.
|
||
|
||
Args:
|
||
value: A hex hash, optionally ``0x``-prefixed and mixed-case.
|
||
hash_bytes: Width in bytes to truncate to; the full value when falsy/unknown.
|
||
|
||
Returns:
|
||
Lowercase hex with no ``0x`` prefix, at most ``hash_bytes`` bytes wide.
|
||
"""
|
||
raw = value.lower().removeprefix("0x")
|
||
return raw[: hash_bytes * 2] if hash_bytes else raw
|
||
|
||
|
||
def _route_text(
|
||
result: TraceResult,
|
||
device_label: str = LOCAL_DEVICE_LABEL,
|
||
resolve: NodeResolver = _identity,
|
||
device_hash: Optional[str] = None,
|
||
) -> Text:
|
||
"""Render a trace's walked route through THE path widget.
|
||
|
||
Shows the path the trace actually walked — the forced path, or the route the
|
||
device resolved when auto-routing — as a :func:`path_text` in the trace
|
||
presentation: each node annotated by its hash at the command's path-hash width,
|
||
our own device bracketing both ends, e.g.
|
||
``Me (a1b2) → Alice (3d63) → Bob (f2a1) → Me (a1b2)``.
|
||
|
||
Args:
|
||
result: The trace whose route to display.
|
||
device_label: Name to show for our own device at the path's endpoints.
|
||
resolve: Maps a raw hop hash to a friendly contact name when known.
|
||
device_hash: Our own device's key/hash, annotated onto its endpoints when known.
|
||
|
||
Returns:
|
||
A :class:`Text` with the node sequence, or a muted note when no hops exist.
|
||
"""
|
||
edges = result.edges(device_label)
|
||
if not edges:
|
||
return Text("no hops recorded", style="muted")
|
||
hash_bytes = result.path_hash_bytes
|
||
nodes = [edges[0].origin] + [edge.destination for edge in edges]
|
||
hops = [None if not node or node == device_label else node for node in nodes]
|
||
return path_text(
|
||
hops, resolve, prefix_bytes=hash_bytes or 8, self_name=device_label,
|
||
show_hash=True, hash_bytes=hash_bytes, device_hash=device_hash,
|
||
)
|
||
|
||
|
||
def _hop_medians_table(
|
||
hop_snrs: list[HopAggregate],
|
||
device_label: str,
|
||
resolve: NodeResolver = _identity,
|
||
hash_bytes: Optional[int] = None,
|
||
device_hash: Optional[str] = None,
|
||
) -> Table:
|
||
"""Render the per-hop median SNR aggregated across a run's traces.
|
||
|
||
Args:
|
||
hop_snrs: The per-hop aggregates to display.
|
||
device_label: Name to show for our own device at the path's endpoints.
|
||
resolve: Maps a raw hop hash to a friendly contact name when known.
|
||
hash_bytes: Path-hash width (bytes) to truncate shown node hashes to.
|
||
device_hash: Our own device's key/hash, annotated onto the path's endpoints.
|
||
|
||
Returns:
|
||
A compact Rich :class:`Table` of hop, link, and median SNR.
|
||
"""
|
||
table = Table(box=None, padding=(0, 1, 0, 0), expand=False)
|
||
table.add_column("HOP", justify="right", style="muted")
|
||
table.add_column("FROM → TO")
|
||
table.add_column("MEDIAN SNR", justify="right")
|
||
for agg in hop_snrs:
|
||
table.add_row(
|
||
str(agg.index),
|
||
_link_text(
|
||
agg.origin, agg.destination, device_label, resolve, hash_bytes, device_hash
|
||
),
|
||
Text(f"{agg.median_snr:+.1f} dB", style=snr_style(agg.median_snr)),
|
||
)
|
||
return table
|
||
|
||
|
||
def stats_panel(
|
||
stats: TraceStats,
|
||
device_label: str = LOCAL_DEVICE_LABEL,
|
||
resolve: NodeResolver = _identity,
|
||
route: Optional[TraceResult] = None,
|
||
device_hash: Optional[str] = None,
|
||
) -> Panel:
|
||
"""Summarize aggregated trace statistics in a panel.
|
||
|
||
Includes the median SNR for every hop along the path (not just the bottleneck), so
|
||
a weak link anywhere in the route is visible. When a representative ``route`` trace
|
||
is given, the path it actually walked is shown as a node sequence, e.g.
|
||
``Me → Alice → Bob → Me``.
|
||
|
||
Args:
|
||
stats: The aggregated statistics to display.
|
||
device_label: Name to show for our own device at the path's endpoints.
|
||
resolve: Maps a raw hop hash to a friendly contact name when known.
|
||
route: A representative trace whose walked route to display, if any.
|
||
device_hash: Our own device's key/hash, annotated onto the route endpoints.
|
||
|
||
Returns:
|
||
A Rich :class:`Panel` with the route, success rate, robust SNR/RTT, and
|
||
per-hop medians.
|
||
"""
|
||
snr = stats.median_min_snr
|
||
snr_text = Text(f"{snr:+.1f} dB", style=snr_style(snr)) if snr is not None else Text("n/a")
|
||
rtt = f"{stats.median_rtt_ms:.0f} ms" if stats.median_rtt_ms is not None else "n/a"
|
||
summary = Text.assemble(
|
||
("target ", "muted"), (f"{stats.target}\n", ""),
|
||
("success rate ", "muted"), (f"{stats.success_rate:.0%} "
|
||
f"({stats.successes}/{stats.samples})\n", ""),
|
||
("median min SNR ", "muted"), snr_text, ("\n", ""),
|
||
("median RTT ", "muted"), (rtt, ""),
|
||
)
|
||
sections: list[Text | Table] = []
|
||
if route is not None:
|
||
sections.append(Text("Route", style="accent"))
|
||
sections.append(_route_text(route, device_label, resolve, device_hash))
|
||
sections.append(Text()) # blank line before the stats block
|
||
sections.append(summary)
|
||
if stats.hop_snrs:
|
||
sections.append(Text("\nPer-hop medians", style="accent"))
|
||
hash_bytes = route.path_hash_bytes if route is not None else None
|
||
sections.append(
|
||
_hop_medians_table(stats.hop_snrs, device_label, resolve, hash_bytes, device_hash)
|
||
)
|
||
body: Text | Group = sections[0] if len(sections) == 1 else Group(*sections)
|
||
return Panel(body, title="[accent]Trace summary[/accent]", border_style="accent", expand=False)
|
||
|
||
|
||
# Node-type glyphs and their colours, consistent with the map's marker palette across the
|
||
# whole app (see ui.map_render): our own node is the yellow ``★``, plain nodes the loud pink
|
||
# ``●`` and repeaters the calmer violet ``▲``. The remaining types take map-safe hues that
|
||
# stay distinct from those — a white square for rooms (the house glyph read poorly) and an
|
||
# orange ringed dot for sensors. All are single-width BMP glyphs so columns stay aligned.
|
||
_ROOM_COLOR = "#ffffff"
|
||
_SENSOR_COLOR = "#fb923c"
|
||
_NODE_GLYPHS: dict[int, tuple[str, str]] = {
|
||
NODE_TYPE_REPEATER: (_REPEATER[0], _REPEATER[1]),
|
||
NODE_TYPE_ROOM: ("■", _ROOM_COLOR),
|
||
NODE_TYPE_SENSOR: ("◉", _SENSOR_COLOR),
|
||
NODE_TYPE_CHAT: (_NODE[0], _NODE[1]),
|
||
}
|
||
_DEFAULT_GLYPH: tuple[str, str] = (_NODE[0], _NODE[1])
|
||
|
||
# A heat-map gradient for a contact's name, hottest (most recently heard) to coldest: white
|
||
# → yellow → orange → red → grey. Each stop pairs an age anchor (log10 of seconds since heard)
|
||
# with an RGB colour; :func:`_recency_style` interpolates continuously between them, so the
|
||
# colour glides with recency rather than snapping between a handful of discrete shades.
|
||
_HEAT_STOPS: tuple[tuple[float, tuple[int, int, int]], ...] = (
|
||
(math.log10(300), (255, 255, 255)), # ≤5m — white (fresh)
|
||
(math.log10(3600), (250, 204, 21)), # ~1h — yellow
|
||
(math.log10(21600), (251, 146, 60)), # ~6h — orange
|
||
(math.log10(86400), (248, 113, 113)), # ~1d — red
|
||
(math.log10(604800), (148, 163, 184)), # ~1w — grey
|
||
(math.log10(2592000), (100, 116, 139)), # ~30d+ — cold slate
|
||
)
|
||
_RECENCY_NEVER = "#64748b" # never heard — the coldest slate
|
||
|
||
|
||
def _age_seconds(when: Optional[datetime]) -> Optional[float]:
|
||
"""Seconds since ``when`` (aware UTC), or ``None`` when unknown/naive."""
|
||
if when is None or getattr(when, "tzinfo", None) is None:
|
||
return None
|
||
return max(0.0, (utcnow() - when).total_seconds())
|
||
|
||
|
||
def _format_age(secs: Optional[float]) -> str:
|
||
"""A compact relative age — ``now``, ``5m``, ``3h``, ``2d``, ``4w`` — or ``never``."""
|
||
if secs is None:
|
||
return "never"
|
||
if secs < 60:
|
||
return "now"
|
||
if secs < 3600:
|
||
return f"{int(secs // 60)}m"
|
||
if secs < 86400:
|
||
return f"{int(secs // 3600)}h"
|
||
if secs < 604800:
|
||
return f"{int(secs // 86400)}d"
|
||
return f"{int(secs // 604800)}w"
|
||
|
||
|
||
def format_ago(secs: Optional[float]) -> str:
|
||
"""The relative-age *phrase* — ``now``, ``5m ago``, ``never`` — for running prose.
|
||
|
||
The canonical grammar for every "heard … (…)" and "delivered …" row: a fresh
|
||
sighting reads as bare ``now`` and an unknown one as bare ``never`` (neither takes
|
||
the suffix — "now ago" is nonsense), while any measured age reads ``5m ago``.
|
||
Callers embedding an age in a sentence or parenthetical use this;
|
||
:func:`_format_age` stays the bare column form for aligned age lanes.
|
||
"""
|
||
age = _format_age(secs)
|
||
return age if age in ("now", "never") else f"{age} ago"
|
||
|
||
|
||
def _recency_style(secs: Optional[float]) -> str:
|
||
"""The heat-map name colour for a contact last heard ``secs`` ago (hotter = more recent).
|
||
|
||
Interpolates the RGB channels between the two :data:`_HEAT_STOPS` bracketing ``secs`` (in
|
||
log-age space), clamping to white below the first stop and cold slate above the last.
|
||
"""
|
||
if secs is None:
|
||
return _RECENCY_NEVER
|
||
x = math.log10(max(secs, 0.0) + 1.0)
|
||
if x <= _HEAT_STOPS[0][0]:
|
||
r, g, b = _HEAT_STOPS[0][1]
|
||
elif x >= _HEAT_STOPS[-1][0]:
|
||
r, g, b = _HEAT_STOPS[-1][1]
|
||
else:
|
||
(x0, c0), (x1, c1) = next(
|
||
(lo, hi) for lo, hi in zip(_HEAT_STOPS, _HEAT_STOPS[1:]) if lo[0] <= x <= hi[0]
|
||
)
|
||
f = (x - x0) / (x1 - x0)
|
||
r, g, b = (round(a + (bb - a) * f) for a, bb in zip(c0, c1))
|
||
return f"#{r:02x}{g:02x}{b:02x}"
|
||
|
||
|
||
def _key_id(value: str) -> str:
|
||
"""Normalise a key/prefix to the lowercased 12-hex id used to match heard nodes."""
|
||
return value.lower().removeprefix("0x")[:12]
|
||
|
||
|
||
def _contact_pkts(contact: Contact, counts: dict[str, int]) -> Optional[int]:
|
||
"""The overheard-packet tally for ``contact``, or ``None`` if never overheard."""
|
||
ident = contact.public_key or contact.key_prefix
|
||
return counts.get(_key_id(ident)) if ident else None
|
||
|
||
|
||
# The columns the node list can be sorted by, left-to-right, and the direction each opens on
|
||
# — name A→Z, most-recently-heard first, most packets first — chosen so a fresh sort shows
|
||
# the "interesting" end at the top.
|
||
_SORT_COLUMNS: tuple[str, ...] = ("name", "heard", "packets")
|
||
_SORT_OPENS_ASCENDING: dict[str, bool] = {"name": True, "heard": True, "packets": False}
|
||
|
||
|
||
@dataclass
|
||
class NodesSort:
|
||
"""Which column the node list is sorted by, and in which direction.
|
||
|
||
``column`` is one of :data:`_SORT_COLUMNS`; ``ascending`` sorts the column's underlying
|
||
metric low-to-high — name A→Z, *age* (so ascending = most recently heard first), packet
|
||
count low-to-high. The interactive screen mutates this in place as the user presses the
|
||
arrows.
|
||
"""
|
||
|
||
column: str = "name"
|
||
ascending: bool = True
|
||
|
||
@classmethod
|
||
def from_name(cls, name: str) -> "NodesSort":
|
||
"""Build a sort for ``name``, opening in that column's natural direction."""
|
||
column = name if name in _SORT_COLUMNS else "name"
|
||
return cls(column, _SORT_OPENS_ASCENDING[column])
|
||
|
||
def move(self, delta: int) -> None:
|
||
"""Step the active column ``delta`` places (wrapping), adopting its natural direction."""
|
||
index = (_SORT_COLUMNS.index(self.column) + delta) % len(_SORT_COLUMNS)
|
||
self.column = _SORT_COLUMNS[index]
|
||
self.ascending = _SORT_OPENS_ASCENDING[self.column]
|
||
|
||
|
||
def _ordered_contacts(
|
||
contacts: list[Contact], counts: dict[str, int], sort: NodesSort
|
||
) -> list[Contact]:
|
||
"""Contacts sorted per ``sort`` (our own node is pinned separately, above these).
|
||
|
||
The active column's metric drives the order (reversed for a descending sort); ties always
|
||
break by case-folded name *ascending*, so two nodes sharing a metric (e.g. the same
|
||
``heard`` age) keep a stable A→Z order instead of flipping with the primary direction.
|
||
Never-heard / never-overheard rows carry an extreme metric so they gather at the ascending
|
||
end.
|
||
"""
|
||
if sort.column == "heard":
|
||
# One clock snapshot for the whole sort: per-row utcnow() calls would skew two
|
||
# identical last_seen stamps apart by microseconds and defeat the name tie-break.
|
||
now = utcnow()
|
||
|
||
def metric(c: Contact) -> float:
|
||
if c.last_seen is None or getattr(c.last_seen, "tzinfo", None) is None:
|
||
return float("inf")
|
||
return max(0.0, (now - c.last_seen).total_seconds())
|
||
elif sort.column == "packets":
|
||
def metric(c: Contact) -> float:
|
||
return _contact_pkts(c, counts) or 0
|
||
else:
|
||
def metric(c: Contact) -> object:
|
||
return c.name.casefold()
|
||
|
||
# Name-ascending first, then a stable sort by the primary metric: equal-metric rows retain
|
||
# their A→Z order under both directions (reversing the whole list would flip the tiebreak).
|
||
ordered = sorted(contacts, key=lambda c: c.name.casefold())
|
||
ordered.sort(key=metric, reverse=not sort.ascending)
|
||
return ordered
|
||
|
||
|
||
def _sort_header(label: str, column: str, sort: NodesSort) -> str:
|
||
"""A column header: plain-muted, or cyan with a direction triangle when it's the sort key.
|
||
|
||
The active column's name and its triangle are lit cyan together (so the interactive
|
||
left/right selection is obvious) — ``▲`` for ascending, ``▼`` for descending. The column
|
||
reserves its width (see :func:`nodes_table`) so toggling the sort doesn't shift the row.
|
||
"""
|
||
if column != sort.column:
|
||
return label
|
||
triangle = "▲" if sort.ascending else "▼"
|
||
return f"[bold #22d3ee]{label} {triangle}[/]"
|
||
|
||
|
||
def _nodes_legend() -> Text:
|
||
"""A one-line glyph legend covering every node type (plus us), whatever's listed."""
|
||
legend = Text(" ") # a small indent to sit under the table body
|
||
legend.append(_SELF[0], style=_SELF[1])
|
||
legend.append(" you", style="muted")
|
||
for node_type in (NODE_TYPE_REPEATER, NODE_TYPE_CHAT, NODE_TYPE_ROOM, NODE_TYPE_SENSOR):
|
||
glyph, color = _NODE_GLYPHS[node_type]
|
||
legend.append(" ")
|
||
legend.append(glyph, style=color)
|
||
legend.append(f" {NODE_TYPE_LABELS[node_type]}", style="muted")
|
||
return legend
|
||
|
||
|
||
def nodes_table(
|
||
self_name: str,
|
||
self_key: str,
|
||
contacts: list[Contact],
|
||
prefix_bytes: int,
|
||
counts: dict[str, int],
|
||
sort: Optional[NodesSort] = None,
|
||
) -> Group:
|
||
"""List this node and its known contacts with recency, packets, type, key, and legend.
|
||
|
||
Our own node is the first row (``★``, name in the white ``you`` style); contacts follow
|
||
in ``sort`` order.
|
||
A per-type glyph marks each node in the app's shared colours, the name is coloured by how
|
||
recently it was last heard (brighter = fresher), and the full key is shown with its
|
||
path-hash prefix lit — chopped with an ellipsis only when the terminal is too narrow.
|
||
|
||
Args:
|
||
self_name: This node's advertised name.
|
||
self_key: This node's full public key (hex); blank renders as ``?``.
|
||
contacts: Known contacts, listed after our own node.
|
||
prefix_bytes: Path-hash width in bytes to highlight in every key.
|
||
counts: Overheard-packet counts keyed by lowercased 12-hex node id (from
|
||
monitoring); a contact with no entry shows ``—``.
|
||
sort: The active sort (column + direction); defaults to name-ascending. The sorted
|
||
column's header is lit cyan with an up/down direction triangle.
|
||
|
||
Returns:
|
||
A Rich :class:`Group` of the frameless table and its glyph legend.
|
||
"""
|
||
sort = sort if sort is not None else NodesSort()
|
||
|
||
# expand=True lets the key column (the only flexible one) soak up all spare width and be
|
||
# the sole column Rich squeezes when narrow — the fixed columns keep their natural size.
|
||
table = Table(
|
||
title=f"[accent]Nodes[/accent] [muted]· {len(contacts)} known[/muted]",
|
||
title_justify="left",
|
||
box=box.SIMPLE_HEAD,
|
||
show_edge=False,
|
||
pad_edge=False,
|
||
header_style="muted",
|
||
expand=True,
|
||
padding=(0, 2, 0, 0),
|
||
)
|
||
# Each sortable header reserves two extra columns for its " ▲" direction marker (via
|
||
# min_width = label + 2) so the same width holds whether or not it's the active sort —
|
||
# switching the sort never widens a column and shifts the rest of the row.
|
||
table.add_column("", no_wrap=True) # node-type glyph
|
||
table.add_column(_sort_header("NAME", "name", sort), no_wrap=True, min_width=6)
|
||
table.add_column(
|
||
_sort_header("HEARD", "heard", sort), justify="right", no_wrap=True, min_width=7
|
||
)
|
||
table.add_column(
|
||
_sort_header("PKTS", "packets", sort), justify="right", no_wrap=True, min_width=6
|
||
)
|
||
# The full key, chopped to an ellipsis by Rich only when the row won't otherwise fit.
|
||
table.add_column("KEY", no_wrap=True, overflow="ellipsis", ratio=1, min_width=10)
|
||
|
||
unknown = Text("?", style="muted")
|
||
table.add_row(
|
||
Text(_SELF[0], style=_SELF[1]),
|
||
# Our own name is the app-wide pure-white "you" style, never a palette hue.
|
||
Text.assemble((self_name, "you"), (" (you)", "muted")),
|
||
Text("—", style="faint"),
|
||
Text("—", style="faint"),
|
||
highlighted_hash(self_key, prefix_bytes) if self_key else unknown,
|
||
)
|
||
table.add_section()
|
||
for c in _ordered_contacts(contacts, counts, sort):
|
||
secs = _age_seconds(c.last_seen)
|
||
glyph, glyph_style = _NODE_GLYPHS.get(c.node_type, _DEFAULT_GLYPH)
|
||
pkts = _contact_pkts(c, counts)
|
||
table.add_row(
|
||
Text(glyph, style=glyph_style),
|
||
Text(c.name, style=_recency_style(secs)),
|
||
Text(_format_age(secs), style="muted"),
|
||
Text(str(pkts), style="muted") if pkts else Text("—", style="faint"),
|
||
highlighted_hash(c.public_key, prefix_bytes) if c.public_key else unknown,
|
||
)
|
||
return Group(table, Text(""), _nodes_legend())
|
||
|
||
|
||
def tx_opt_table(result: TxOptResult) -> Table:
|
||
"""Render every measured TX level of an optimization sweep, best row highlighted.
|
||
|
||
Args:
|
||
result: The optimization result to display.
|
||
|
||
Returns:
|
||
A Rich :class:`Table` of TX level, target SNR, success rate, and sample count.
|
||
"""
|
||
table = Table(
|
||
title=f"TX sweep · {result.admin_node} -> {result.target}",
|
||
border_style="muted",
|
||
expand=False,
|
||
)
|
||
table.add_column("TX", justify="right")
|
||
table.add_column("TARGET SNR", justify="right")
|
||
table.add_column("SUCCESS", justify="right")
|
||
table.add_column("TRACES", justify="right")
|
||
for lv in result.sorted_by_tx():
|
||
is_best = lv.tx_power == result.best_tx
|
||
marker = "[ok]★[/ok] " if is_best else " "
|
||
snr = lv.target_snr
|
||
snr_cell = Text(f"{snr:+.1f}", style=snr_style(snr)) if snr is not None else Text("—")
|
||
# Highlight anything short of a perfect success rate — reliability comes first.
|
||
rate = lv.success_rate
|
||
rate_style = "ok" if rate >= 1.0 else ("warn" if rate > 0 else "err")
|
||
rate_cell = Text(f"{rate:.0%}", style=rate_style)
|
||
tx_cell = f"{marker}{lv.tx_power}"
|
||
row_style = "ok" if is_best else None
|
||
table.add_row(
|
||
tx_cell, snr_cell, rate_cell, f"{lv.successes}/{lv.samples}", style=row_style
|
||
)
|
||
return table
|
||
|
||
|
||
def tx_opt_summary(result: TxOptResult) -> Panel:
|
||
"""Summarize a TX optimization outcome.
|
||
|
||
Args:
|
||
result: The optimization result to summarize.
|
||
|
||
Returns:
|
||
A Rich :class:`Panel` stating the chosen optimum and whether it was applied.
|
||
"""
|
||
snr = result.best_snr
|
||
snr_text = Text(f"{snr:+.1f} dB", style=snr_style(snr)) if snr is not None else Text("n/a")
|
||
applied = (
|
||
f"[ok]applied to {result.admin_node}[/ok]"
|
||
if result.applied
|
||
else "[muted]not applied[/muted]"
|
||
)
|
||
body = Text.assemble(
|
||
("tuning node ", "muted"), (f"{result.admin_node}\n", "brand"),
|
||
("target ", "muted"), (f"{result.target}\n", "brand"),
|
||
("optimal TX ", "muted"), (f"{result.best_tx}", "brand"), ("\n", ""),
|
||
("target SNR ", "muted"), snr_text, ("\n", ""),
|
||
("reliability ", "muted"), (f"{result.best_success_rate:.0%}\n", ""),
|
||
("previous TX ", "muted"),
|
||
(f"{result.original_tx if result.original_tx is not None else '?'}\n", ""),
|
||
("status ", "muted"), Text.from_markup(applied),
|
||
)
|
||
return Panel(body, title="[accent]TX optimization[/accent]", border_style="accent", expand=False)
|
||
|
||
|
||
# -- battery gauge --------------------------------------------------------------------------
|
||
|
||
#: Unicode braille pattern base; add an 8-bit dot mask (a 2×4 cell) to get the glyph.
|
||
_BRAILLE_BASE = 0x2800
|
||
|
||
#: Dot bit per (column, row) within a braille cell — the Unicode standard layout, the same
|
||
#: mapping the map canvas rasters with. Kept here so the one-cell battery glyph needn't reach
|
||
#: into the canvas module for two constants.
|
||
_BRAILLE_DOTS = ((0x01, 0x02, 0x04, 0x40), (0x08, 0x10, 0x20, 0x80))
|
||
|
||
#: Battery charge tiers: ``(percent floor, braille rows lit from the bottom, colour)``,
|
||
#: highest first. The glyph is one braille cell (2×4 dots) filled from its bottom row up, so
|
||
#: four coarse levels read as a shrinking fuel gauge: the whole cell above three-quarters,
|
||
#: six dots above half, four above a quarter, two below it — green while healthy, orange at a
|
||
#: quarter, red near empty (the app's ok/warn/err semantics as a gauge).
|
||
_BATTERY_TIERS: tuple[tuple[int, int, str], ...] = (
|
||
(75, 4, "batt.high"),
|
||
(50, 3, "batt.high"),
|
||
(25, 2, "batt.mid"),
|
||
(0, 1, "batt.low"),
|
||
)
|
||
|
||
#: At or below this charge the red glyph blinks with a dim frame — a can't-miss low warning.
|
||
_BATTERY_CRITICAL = 10
|
||
|
||
#: Frames in the charging sweep: empty → four fills → round again (bottom-to-full loop).
|
||
_CHARGE_FRAMES = 5
|
||
|
||
|
||
def _braille_fill(rows: int) -> str:
|
||
"""A single braille cell with its bottom ``rows`` (0–4) dot rows lit."""
|
||
rows = max(0, min(4, rows))
|
||
bits = 0
|
||
for row in range(4 - rows, 4):
|
||
bits |= _BRAILLE_DOTS[0][row] | _BRAILLE_DOTS[1][row]
|
||
return chr(_BRAILLE_BASE + bits)
|
||
|
||
|
||
def _battery_tier(percent: int) -> tuple[int, str]:
|
||
"""The ``(rows lit, colour)`` a charge level draws at (see :data:`_BATTERY_TIERS`)."""
|
||
for floor, rows, color in _BATTERY_TIERS:
|
||
if percent >= floor:
|
||
return rows, color
|
||
return 1, "batt.low"
|
||
|
||
|
||
def battery_cell(percent: int, *, charging: bool = False, frame: int = 0) -> Text:
|
||
"""The status-bar battery gauge: one braille cell coloured by charge, then its ``%``.
|
||
|
||
The braille cell fills from the bottom up in four coarse steps (see
|
||
:data:`_BATTERY_TIERS`), coloured green while healthy, orange at a quarter, red near
|
||
empty. Two live states animate off the caller's ``frame`` counter (advanced one step per
|
||
repaint tick), so the gauge moves without any per-frame plumbing:
|
||
|
||
* **Charging** overrides the fill: the cell sweeps empty-to-full on a loop, the charge
|
||
colour held, so a plugged-in pack visibly climbs — while the number beside it stays the
|
||
*true* charge, never the animation's.
|
||
* **Critically low** (≤ :data:`_BATTERY_CRITICAL`%, not charging) blinks the red cell to a
|
||
dim slate on alternate frames — an unmissable pulse in the corner.
|
||
|
||
Args:
|
||
percent: State of charge, 0–100 (clamped).
|
||
charging: Whether the pack is taking charge (drives the fill sweep).
|
||
frame: A monotonically advancing tick; only its phase is read, so any
|
||
steadily-incrementing integer animates the two live states.
|
||
|
||
Returns:
|
||
A Rich :class:`Text`: the coloured glyph, a space, and ``NN%`` in muted text.
|
||
"""
|
||
pct = max(0, min(100, int(percent)))
|
||
rows, color = _battery_tier(pct)
|
||
if charging:
|
||
# The charge colour is held; only the fill sweeps, so it reads as "filling", not
|
||
# as the charge itself jumping around.
|
||
rows = frame % _CHARGE_FRAMES
|
||
elif pct <= _BATTERY_CRITICAL and frame % 2:
|
||
color = "batt.dim"
|
||
out = Text(_braille_fill(rows), style=color)
|
||
out.append(f" {pct}%", style="muted")
|
||
return out
|