Trophy case + route graph: node-type marks, SNR colour, reliability

Route graph (both the record card and the Message paths dialog):
- Relays now wear their own map marker — ▲ repeater, ■ room, ◉ sensor —
  instead of a generic ●, via a node-type resolver threaded into
  route_graph_style (built from contacts; an untyped relay keeps the dot).
- A shared node-type key (★ you ▲ repeater ● node ■ room ◉ sensor) sits
  under every fan, minted once as widgets.node_type_legend.
- The drawn path is coloured by SNR (the SNR palette): the record's walk
  by its weakest link, the Message paths selection by its reception SNR;
  unselected paths stay gray. Per-path, not per-hop — overheard frames
  only carry the reception reading.

Record card:
- New 'reliability' stat: the success rate of traces aimed at the walk's
  far node (the one delivery figure history can answer — a failed trace
  records its target, not its route), with the sample count beside it.
- Area drawing reworked per feedback: our star is drawn last so it is
  never hidden, hops are pinned in their node-name hue (star stays
  yellow), the fill is one flat tone (even-odd shades only the enclosed
  side; no dimmer interior wash), and the box is larger.

New shared helpers: theme.snr_rgb, widgets.name_rgb,
trace_runner.make_node_type_resolver, repository.target_trace_counts.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Jean-Pierre MartineauandClaude Opus 4.8 committed 2026-07-15 20:26:47 -04:00
1 parent 8a1c069a96
commit ed391ab16e
11 files changed
+336 -72

No files matched your search

+21
View File
@@ -462,6 +462,27 @@ class Repository:
).fetchall()
return [row["target"] for row in rows]
def target_trace_counts(self) -> dict[str, tuple[int, int]]:
"""Per trace target, its ``(successes, total)`` across every stored trace.
The substrate for a route's observed reliability: unlike a walk's route, a target
is on every trace row — timed-out attempts included — so a target's success rate is
the one delivery figure the history can honestly answer (a failed trace records no
path, only the target it was aimed at). Keyed by the raw target string the trace
was filed under (a contact name or a hex hash); the caller matches its node against
those keys. The hand-composed path walks filed under
:data:`~meshterm.core.models.PATH_TRACE_TARGET` are excluded — they name no target.
Returns:
``target → (successes, total)`` for every distinct non-path-walk target.
"""
rows = self._conn.execute(
"SELECT target, SUM(success) AS ok, COUNT(*) AS n FROM traces "
"WHERE target != ? GROUP BY target",
(PATH_TRACE_TARGET,),
).fetchall()
return {row["target"]: (int(row["ok"] or 0), int(row["n"])) for row in rows}
def trace_paths(self, *, limit: int = 2000) -> list[TracedPath]:
"""Return the walked paths of recent successful traces, newest first.
+35
View File
@@ -77,6 +77,41 @@ def make_node_resolver(
return resolve
def make_node_type_resolver(
contacts: Optional[list[Contact]],
) -> Callable[[Optional[str]], Optional[int]]:
"""Build a resolver that names a hop's *node type* from its key-prefix hash.
The type counterpart to :func:`make_node_resolver`: it maps a trace/relay hash to the
contact's advertised node type (a ``NODE_TYPE_*`` constant), so a route graph can mark
a repeater with its own glyph rather than a generic dot. Matches the hash against each
contact's key the same prefix-either-way way the name resolver does; a hash we can't
place, or a contact with no type, resolves to ``None``.
Args:
contacts: Known contacts to resolve against.
Returns:
A callable taking a hop hash and returning its node type, or ``None`` when unknown.
"""
entries: list[tuple[str, int]] = []
for c in contacts or []:
ident = (c.public_key or c.key_prefix or "").lower().removeprefix("0x")
if ident and c.node_type is not None:
entries.append((ident, c.node_type))
def type_of(label: Optional[str]) -> Optional[int]:
if not label:
return None
needle = label.lower().removeprefix("0x")
for ident, node_type in entries:
if ident.startswith(needle) or needle.startswith(ident):
return node_type
return None
return type_of
def parse_trace_path(spec: str, contacts: Optional[list[Contact]] = None) -> str:
"""Parse a user path spec into the hex path string ``send_trace`` expects.
+4 -4
View File
@@ -896,9 +896,9 @@ async def _make_paths_presenter(
from .timemachine_screen import _routing_prefix_bytes
session = ctx.ui.session
resolve = trace_runner.make_node_resolver(
await ctx.devstate.contacts(), ctx.repo.node_names()
)
contacts = await ctx.devstate.contacts()
resolve = trace_runner.make_node_resolver(contacts, ctx.repo.node_names())
type_of = trace_runner.make_node_type_resolver(contacts)
prefix_bytes = await _routing_prefix_bytes(ctx)
self_name: Optional[str] = None
try:
@@ -954,7 +954,7 @@ async def _make_paths_presenter(
MessagePathsScreen(
message, arrivals, matched=matched, resolve=resolve,
prefix_bytes=prefix_bytes, self_name=self_name, summary=summary,
source=source or None,
source=source or None, type_of=type_of,
)
)
+26 -14
View File
@@ -7,10 +7,11 @@ detail read together:
* a **graph** up top draws every distinct path the message took through the shared
route-graph widget (:mod:`~meshterm.ui.pathgraph`): the currently selected path
white, the unused paths gray beneath it. The origin sits at the left, we sit at
the right, and every relay in between gets its marker plus the first byte of its
hash — set straight above or below the marker, in the mesh name's own colour, so
the byte reads as that node and never crowds the line running through it. The row
coloured by its reception SNR (the SNR palette), the unused paths gray beneath it.
The origin sits at the left, we sit at the right, and every relay in between gets its
own node-type marker (``▲`` repeater, …) plus the first byte of its hash — set straight
above or below the marker, in the mesh name's own colour, so the byte reads as that node
and never crowds the line running through it. A node-type key sits under the fan. The row
list carries the full names, so the graph's labels stay two cells wide and a
many-path graph stays readable.
* the **arrival list** beneath is one row per logged copy — time, reception SNR, and
@@ -33,15 +34,16 @@ from rich.text import Text
from ..core.models import ChatMessage
from ..services.message_paths import Arrival
from .pathgraph import PathLayer, render_path_graph
from .theme import snr_style
from .theme import snr_rgb, snr_style
from .tui.render import crop_cells, render_to_ansi
from .tui.screen import Screen
from .widgets import NodeResolver, path_text, route_graph_style
from .widgets import NodeResolver, TypeOf, node_type_legend, path_text, route_graph_style
#: Cells one ←/→ press shifts the selected row by.
_HSTEP = 4
#: Edge colours: the selected path draws white over the unused paths' gray.
#: Edge colours: the selected path draws in its arrival's SNR hue (white when it carried no
#: reading) over the unused paths' gray.
_EDGE_SELECTED = (255, 255, 255)
_EDGE_UNUSED = (110, 110, 110)
@@ -60,6 +62,7 @@ class MessagePathsScreen(Screen):
self_name: Optional[str],
summary: str,
source: Optional[str] = None,
type_of: Optional[TypeOf] = None,
) -> None:
"""Build the dialog over one message's matched arrivals.
@@ -76,6 +79,8 @@ class MessagePathsScreen(Screen):
source: Display name of the message's origin — the sender parsed from a
channel message, us for an outbound one, the peer for a direct chat
(``None`` reads as an unknown ``?`` origin).
type_of: Maps a relay hash to its node type, so the graph marks a repeater
``▲`` (etc.) instead of a generic dot; ``None`` keeps the plain dots.
"""
super().__init__()
self.title = "Message paths"
@@ -87,6 +92,7 @@ class MessagePathsScreen(Screen):
self._self_name = self_name
self._summary = summary
self._source = source
self._type_of = type_of
self._index = 0
#: Cells the selected row is shifted left by (reset whenever ↑↓ move), and
#: how far it *can* shift, measured against the width of the last render.
@@ -166,9 +172,10 @@ class MessagePathsScreen(Screen):
lines.append("")
lines.extend(self._graph_lines(width))
caption = Text("origin → you · white = selected path · labels = hash byte",
caption = Text("origin → you · colour = SNR of selected · labels = hash byte",
style="faint")
lines.append(render_to_ansi(caption, width, no_wrap=True))
lines.append(render_to_ansi(node_type_legend(), width, no_wrap=True))
lines.append("")
for i, arrival in enumerate(self._arrivals):
if i == self._index:
@@ -246,26 +253,31 @@ class MessagePathsScreen(Screen):
return seen
def _graph_lines(self, width: int) -> list[str]:
"""Draw every distinct path origin → us, the selected one white over gray.
"""Draw every distinct path origin → us, the selected one in its SNR hue over gray.
The shared route-graph widget does the layout (lanes fanned from the centre
in first-heard order, shared relays averaged together); this just maps each
distinct path to a :class:`~meshterm.ui.pathgraph.PathLayer` — the selected
one white on top, the rest gray beneath — and hands it the shared route-graph
callbacks (:func:`~meshterm.ui.widgets.route_graph_style`): endpoints named,
relays marked and labelled by their first hash byte in the node's own hue.
one on top coloured by the selected arrival's reception SNR (the SNR palette,
white when it carried no reading), the rest gray beneath — and hands it the shared
route-graph callbacks (:func:`~meshterm.ui.widgets.route_graph_style`): endpoints
named, relays their own map marker where the type is known, each labelled by its
first hash byte in the node's own hue.
"""
selected = self._arrivals[self._index].hops
arrival = self._arrivals[self._index]
selected = arrival.hops
selected_color = snr_rgb(arrival.snr) if arrival.snr is not None else _EDGE_SELECTED
layers = [
PathLayer(
hops=path,
color=_EDGE_SELECTED if path == selected else _EDGE_UNUSED,
color=selected_color if path == selected else _EDGE_UNUSED,
priority=3 if path == selected else 2,
)
for path in self._paths()
]
glyph_of, label_of, label_rgb_of = route_graph_style(
resolve=self._resolve, self_name=self._self_name, source=self._source,
type_of=self._type_of,
)
return render_path_graph(
layers, width,
+119 -42
View File
@@ -39,12 +39,15 @@ from ..services.records import CATEGORIES, CATEGORY_BY_ID, Category, _local_xy
from .mapcanvas import RGB, MapCanvas
from .menus import back_rows, section_heading
from .pathgraph import PathLayer, render_path_graph
from .theme import name_style, snr_style
from .theme import name_style, snr_rgb, snr_style
from .tui.render import render_hanging, render_to_ansi
from .tui.screen import Screen
from .widgets import (
NodeResolver,
TypeOf,
name_rgb,
node_marker,
node_type_legend,
path_text,
route_graph_style,
self_marker,
@@ -61,16 +64,19 @@ _DESC_WRAP = 64
#: on the canvas, so it needs no colour to set it apart — white just reads as "the walk".
_WALK_EDGE = (255, 255, 255)
#: The area drawing's two body colours: a dim slate wash for the enclosed interior, and a
#: near-white outline for the walked loop itself (the coloured node markers ride on top).
_AREA_FILL: RGB = (71, 85, 105)
_AREA_EDGE: RGB = (226, 232, 240)
#: The area drawing's one body tone — a plain muted slate the enclosed region and its loop
#: both draw in. Deliberately a single shade (no dimmer interior wash under a brighter rim):
#: an even-odd fill shades only the enclosed side, and the coloured node pins carry the
#: colour, so the shape reads as one flat silhouette rather than two brightnesses.
_AREA_TONE: RGB = (148, 163, 184)
#: The area drawing's cell box, to the right of the stats: a width that scales with the
#: dialog but stays inside these bounds, and a floor of columns below which it is dropped
#: (the stats reclaim the full width) rather than squeezed into illegibility.
_POLY_MIN_W = 14
_POLY_MAX_W = 22
#: dialog but stays inside these bounds, extra rows spent below the stats so the shape has
#: room, and a floor of columns below which it is dropped (the stats reclaim the full
#: width) rather than squeezed into illegibility.
_POLY_MIN_W = 16
_POLY_MAX_W = 28
_POLY_EXTRA_ROWS = 2
_SIDE_BY_SIDE_MIN = 44
@@ -100,11 +106,12 @@ class WalkVertex:
class RecordDialog(Screen):
"""One record's full story, floating over the screen beneath.
Every stat the walk was measured by — the far point named with the node it reached —
with the walk's enclosed area drawn beside them on a braille mini-map (us at the
origin, every positioned hop pinned in its map colour, no labels) when the terminal has
the room; below, the walk on THE route graph (us at both ends, the same shape the
Message paths dialog draws), the route in full (wrapped, never truncated), and the
Every stat the walk was measured by — a reliability read from the far node's trace
history, the far point named with the node it reached — with the walk's enclosed area
drawn beside them on a braille mini-map (us at the origin, every positioned hop pinned
in its own name hue, no labels) when the terminal has the room; below, the walk on THE
route graph (us at both ends, coloured by its weakest link, relays wearing their map
marker over a node-type key), the route in full (wrapped, never truncated), and the
record's provenance — when it was set and by which app version. The card scrolls
(PgUp/PgDn/Home/End) when it outgrows the frame, while the arrows drive the actions.
Two actions besides Back: *Trace this path* reopens Trace path with the record's route
@@ -123,6 +130,8 @@ class RecordDialog(Screen):
device_hash: Optional[str],
far_label: Optional[str] = None,
shape: Optional[Sequence[WalkVertex]] = None,
reliability: Optional[tuple[float, int, int]] = None,
type_of: Optional[TypeOf] = None,
) -> None:
"""Build the dialog for one stored record.
@@ -137,6 +146,10 @@ class RecordDialog(Screen):
no positioned hop was named.
shape: The walk's positioned circuit, projected for the area drawing; ``None``
below the three points a polygon needs.
reliability: ``(rate, successes, total)`` of traces to the walk's far node, or
``None`` when that node was never a trace target.
type_of: Maps a route hash to its node type, so relays draw their map marker
(``▲`` repeater, …) on the graph; ``None`` falls back to generic dots.
"""
super().__init__()
self.title = f"Record — {category.title} #{rank}"
@@ -148,6 +161,8 @@ class RecordDialog(Screen):
self._device_hash = device_hash
self._far_label = far_label
self._shape = list(shape) if shape else None
self._reliability = reliability
self._type_of = type_of
self._actions = ("trace", "delete", "back")
self._index = 0
self._cursor: Optional[int] = None
@@ -192,7 +207,7 @@ class RecordDialog(Screen):
def _lane(self, label: str, value: Text) -> Text:
"""One label/value stat lane (label lane fixed so values align)."""
row = Text(f"{label:<11}", style="muted")
row = Text(f"{label:<12}", style="muted")
row.append_text(value)
return row
@@ -200,12 +215,14 @@ class RecordDialog(Screen):
"""Draw the walked route as one path on THE route graph — us at both ends.
A scored walk leaves home and comes back, so it draws us (left) → its relays →
us (right) as a single white path through the shared fan-lane widget
us (right) as a single path through the shared fan-lane widget
(:func:`~meshterm.ui.pathgraph.render_path_graph`), the same shape the Message
paths dialog draws a delivery over. Nodes the walk passed through more than once
— a boomerang's mirrored return leg — collapse to their first appearance: the
depth-ordered graph can only seat a node once, and the route line below still
carries every hop, revisits and all.
paths dialog draws a delivery over. The path is coloured by the walk's weakest link
(the SNR palette), so the loop reads red when a hop barely carried; relays wear
their own map marker (``▲`` repeater, …) where the type is known. Nodes the walk
passed through more than once — a boomerang's mirrored return leg — collapse to
their first appearance: the depth-ordered graph can only seat a node once, and the
route line below still carries every hop, revisits and all.
"""
seen: set[str] = set()
hops: list[str] = []
@@ -217,9 +234,12 @@ class RecordDialog(Screen):
resolve=self._resolve,
self_name=self._device_label,
source=self._device_label,
type_of=self._type_of,
)
min_snr = self._record.stats.get("min_snr")
color = snr_rgb(min_snr) if min_snr is not None else _WALK_EDGE
return render_path_graph(
[PathLayer(hops=tuple(hops), color=_WALK_EDGE, priority=3)],
[PathLayer(hops=tuple(hops), color=color, priority=3)],
width,
glyph_of=glyph_of, label_of=label_of, label_rgb_of=label_rgb_of,
)
@@ -240,6 +260,12 @@ class RecordDialog(Screen):
score = "≥ " + score
lanes.append(self._lane("score", Text(score, style="accent bold")))
if self._reliability is not None:
rate, ok, total = self._reliability
value = Text(f"{rate:.0%}", style=snr_style(20 * rate - 10))
value.append(f" · {ok}/{total} traces to far node", style="muted")
lanes.append(self._lane("reliability", value))
hops = stats.get("hop_count", len(record.route))
distinct = stats.get("distinct_nodes", len(set(record.route)))
walk = Text(f"{hops} hop{'s' if hops != 1 else ''}")
@@ -284,15 +310,16 @@ class RecordDialog(Screen):
return [render_to_ansi(lane, width, no_wrap=True) for lane in lanes]
poly_w = min(_POLY_MAX_W, max(_POLY_MIN_W, width // 3))
left_w = width - poly_w - 2
poly = self._shape_lines(poly_w, len(lanes))
# Spend a couple of rows beyond the stat lanes so the shape has room; the extra
# rows hang just below the last stat, in the air above the route graph.
poly = self._shape_lines(poly_w, len(lanes) + _POLY_EXTRA_ROWS)
out: list[str] = []
for i, lane in enumerate(lanes):
row = lane.copy()
for i in range(len(poly)):
row = lanes[i].copy() if i < len(lanes) else Text()
row.no_wrap = True
row.truncate(left_w, overflow="ellipsis", pad=True)
row.append(" ")
if i < len(poly):
row.append_text(Text.from_ansi(poly[i]))
row.append_text(Text.from_ansi(poly[i]))
out.append(render_to_ansi(row, width, no_wrap=True))
return out
@@ -324,10 +351,19 @@ class RecordDialog(Screen):
for v in verts
]
closed = ring + ring[:1]
canvas.fill_polygon([closed], _AREA_FILL, priority=0)
canvas.draw_line(closed, _AREA_EDGE, priority=1)
# One flat tone for the enclosed side and its loop: even-odd shades only the odd
# (enclosed) region — a self-crossing walk's crossing stays unshaded — and nothing
# is drawn darker than anything else, so there is no dimmer sub-region.
canvas.fill_polygon([closed], _AREA_TONE, priority=0)
canvas.draw_line(closed, _AREA_TONE, priority=0)
# Hops first, then our own node last, so the star always wins its cell — a hop that
# projects onto the same cell can never hide us.
for v, (dx, dy) in zip(verts, ring):
canvas.marker(int(round(dx)), int(round(dy)), v.glyph, v.color)
if not v.is_self:
canvas.marker(int(round(dx)), int(round(dy)), v.glyph, v.color)
for v, (dx, dy) in zip(verts, ring):
if v.is_self:
canvas.marker(int(round(dx)), int(round(dy)), v.glyph, v.color)
return canvas.to_ansi_lines()
def render_body(self, width: int) -> list[str]:
@@ -341,6 +377,7 @@ class RecordDialog(Screen):
lines.extend(self._graph_lines(width))
caption = Text("you → … → you · labels = hash byte", style="faint")
lines.append(render_to_ansi(caption, width, no_wrap=True))
lines.append(render_to_ansi(node_type_legend(), width, no_wrap=True))
lines.append("")
route = path_text(
@@ -351,12 +388,12 @@ class RecordDialog(Screen):
hash_bytes=record.width_bytes,
device_hash=self._device_hash,
)
lines.extend(render_hanging(Text("route ", style="muted"), route, width,
indent=11))
lines.extend(render_hanging(Text("route ", style="muted"), route, width,
indent=12))
spec = Text(record.spec, style="brand")
spec.append(f" ({record.width_bytes}-byte hops)", style="muted")
lines.extend(render_hanging(Text("spec ", style="muted"), spec, width,
indent=11))
lines.extend(render_hanging(Text("spec ", style="muted"), spec, width,
indent=12))
stamp = record.discovered_at.astimezone().strftime("%b %d %Y %H:%M")
when = Text(stamp)
@@ -428,6 +465,10 @@ async def open_records(ctx: "AppContext") -> dict:
lat, lon = _as_float(self_info.get("adv_lat")), _as_float(self_info.get("adv_lon"))
self_pos = (lat, lon) if lat is not None and lon is not None and (lat or lon) else None
# Per-target trace outcomes, for a record's reliability (the success rate of traces to
# its far node). Read once here; the far node is matched against these keys per record.
target_counts = ctx.repo.target_trace_counts()
# Node positions and types for a record's area drawing, gathered live the same way the
# trace tools gather them to score a walk: adverts we've heard, contacts over them. A
# record stores canonical ids, so match those the resolver's way (a prefix either side).
@@ -464,33 +505,67 @@ async def open_records(ctx: "AppContext") -> dict:
def walk_drawing(
record: DiscoveredPath,
) -> tuple[Optional[str], Optional[list[WalkVertex]]]:
"""The farthest node's name and the walk's projected polygon (or ``None`` each).
) -> tuple[Optional[str], Optional[str], Optional[list[WalkVertex]]]:
"""The farthest node (name and id) and the walk's projected polygon.
Mirrors the scoring geometry (see :mod:`~meshterm.services.records`): our node at
the plane's origin, every positioned hop projected onto the same local km plane in
walk order, the farthest hop named. A polygon needs three points — us plus two
positioned hops — so a sparser walk yields no drawing but still names its far point.
walk order, the farthest hop named. That farthest hop is the walk's natural target,
so its id is handed back for the reliability lookup. A polygon needs three points —
us plus two positioned hops — so a sparser walk yields no drawing but still names
its far point.
"""
if self_pos is None:
return None, None
return None, None, None
glyph, color = self_marker()
verts = [WalkVertex(0.0, 0.0, glyph, color, True)]
far_label: Optional[str] = None
far_id: Optional[str] = None
far_dist = -1.0
for node_id in record.route:
pos, ntype = node_geo(node_id)
if pos is None:
continue
east, north = _local_xy(self_pos, pos)
hop_glyph, hop_color = node_marker(ntype)
# The node-type glyph in the node's *name* hue — the pin reads as that mesh
# name, the way the route line and graph labels colour it; an unnamed hop keeps
# the neutral shape tone so it never masquerades as a coloured name.
name = resolve(node_id)
hop_glyph = node_marker(ntype)[0]
hop_color = name_rgb(name) if name and name != node_id else _AREA_TONE
verts.append(WalkVertex(east, north, hop_glyph, hop_color, False))
dist = haversine_km(self_pos[0], self_pos[1], pos[0], pos[1])
if dist > far_dist:
far_dist = dist
name = resolve(node_id)
far_id = node_id
far_label = name if name and name != node_id else None
return far_label, (verts if len(verts) >= 3 else None)
return far_label, far_id, (verts if len(verts) >= 3 else None)
def walk_reliability(
far_id: Optional[str], far_label: Optional[str]
) -> Optional[tuple[float, int, int]]:
"""A record's observed reliability: the success rate of traces to its far node.
A walk's route can't be counted from history — a timed-out trace records no path —
but its farthest node *is* the target a boomerang was aimed at, and every trace
carries its target, failures included. So the far node's success rate (matched
against the filed targets by name or hex hash) is an honest delivery figure. Thin
by nature (the count is shown alongside); ``None`` when the node was never a target.
"""
if not far_id:
return None
needle = far_id.lower().removeprefix("0x")
wanted = far_label.lower() if far_label else None
ok = total = 0
for target, (t_ok, t_n) in target_counts.items():
key = target.lower().removeprefix("0x")
is_hex = bool(key) and all(c in "0123456789abcdef" for c in key)
if (wanted is not None and key == wanted) or (
is_hex and (key.startswith(needle) or needle.startswith(key))
):
ok += t_ok
total += t_n
return (ok / total, ok, total) if total else None
def ranked(category: Category) -> list[DiscoveredPath]:
"""One discipline's records, ranked best-first across every hash width."""
@@ -621,11 +696,13 @@ async def open_records(ctx: "AppContext") -> dict:
ctx.repo.delete_discoveries()
continue
_verb, category, rank, record = picked
far_label, shape = walk_drawing(record)
far_label, far_id, shape = walk_drawing(record)
action = await session.run_screen(RecordDialog(
record, category, rank,
resolve=resolve, device_label=device_label, device_hash=device_hash,
far_label=far_label, shape=shape,
reliability=walk_reliability(far_id, far_label),
type_of=lambda node_id: node_geo(node_id)[1],
))
if action == "trace":
await open_trace_path(ctx, spec=record.spec)
+15
View File
@@ -176,3 +176,18 @@ def snr_style(snr: float | None) -> str:
if snr >= -5:
return "snr.ok"
return "snr.bad"
#: The truecolour behind each :func:`snr_style` band, matching its ``snr.*`` theme hue, for
#: rasters that plot SNR directly (the route graph colouring a path by its link quality).
_SNR_RGB: dict[str, tuple[int, int, int]] = {
"snr.good": (74, 222, 128), # #4ade80
"snr.ok": (251, 191, 36), # #fbbf24
"snr.bad": (248, 113, 113), # #f87171
"muted": (148, 163, 184), # #94a3b8
}
def snr_rgb(snr: float | None) -> tuple[int, int, int]:
"""The ``(r, g, b)`` of an SNR reading's quality band (see :func:`snr_style`)."""
return _SNR_RGB[snr_style(snr)]
+42 -4
View File
@@ -402,11 +402,28 @@ def _name_rgb(name: str) -> RGB:
return parse_hex(name_style(name).split()[-1])
def name_rgb(name: str) -> RGB:
"""The truecolour of a node name's stable palette hue, for a braille canvas.
The public face of :func:`_name_rgb` — the same hue :func:`~meshterm.ui.theme.name_style`
paints a name in, as an ``(r, g, b)`` tuple a raster can plot. A caller colouring node
markers by their mesh name (the trophy case's area drawing) mints them through here.
"""
return _name_rgb(name)
#: Maps a relay hash to its node type (see the ``NODE_TYPE_*`` constants), or ``None`` when
#: the type is unknown — the hook that lets the route graph mark a repeater ``▲`` and not
#: just a generic ``●``. Endpoint sentinels are never passed to it.
TypeOf = Callable[[str], Optional[int]]
def route_graph_style(
*,
resolve: NodeResolver,
self_name: Optional[str],
source: Optional[str],
type_of: Optional[TypeOf] = None,
) -> tuple[GlyphOf, LabelOf, LabelRgbOf]:
"""Build the per-node callbacks that draw a route on THE route graph (``pathgraph``).
@@ -424,6 +441,10 @@ def route_graph_style(
source: The left endpoint's display name (a message's origin, or us for a walk
that starts at home — pass ``self_name`` to draw both ends as us). ``None``
reads as an unknown ``?`` origin.
type_of: Maps a relay's hash to its node type, so a relay draws its own map marker
(``▲`` repeater, ``■`` room, ``◉`` sensor) in the shared palette instead of a
generic dot. ``None``, or a hash whose type it can't resolve, keeps the old
named-dot / unknown-ring fallback.
Returns:
The ``(glyph_of, label_of, label_rgb_of)`` triple to hand to
@@ -432,11 +453,15 @@ def route_graph_style(
src_is_self = bool(source) and source == self_name
def glyph_of(node: str) -> tuple[str, str]:
"""Us a star, a named node a dot, an unidentified node a ring."""
"""Us a star, a typed relay its map marker, a named node a dot, else a ring."""
if node == DST_NODE:
return _SELF
if node == SRC_NODE:
return _SELF if src_is_self else (_NODE if source else _UNKNOWN)
if type_of is not None:
node_type = type_of(node)
if node_type is not None:
return _NODE_GLYPHS.get(node_type, _DEFAULT_GLYPH)
named = resolve(node)
return _NODE if named and named != node else _UNKNOWN
@@ -780,9 +805,17 @@ def _sort_header(label: str, column: str, sort: NodesSort) -> str:
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
def node_type_legend(indent: str = "") -> Text:
"""The one-line key to the node-type marks: ``★ you ▲ repeater ● node …``.
Every glyph in its shared map colour (see :data:`_NODE_GLYPHS`), each named muted after
it. THE legend for any surface that draws typed node markers — the nodes list under its
table, the route graph under its fan — so one glyph means one thing app-wide.
Args:
indent: Leading spaces to sit the legend under a table or graph body.
"""
legend = Text(indent)
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):
@@ -793,6 +826,11 @@ def _nodes_legend() -> Text:
return legend
def _nodes_legend() -> Text:
"""The node-type legend, indented to sit under the nodes table body."""
return node_type_legend(indent=" ")
def nodes_table(
self_name: str,
self_key: str,
+16 -5
View File
@@ -49,7 +49,8 @@ def test_paths_screen_renders_graph_rows_and_cursor() -> None:
assert "Alice" in body and "Homestead" in body # origin and us, on the graph
assert body.count("via") == 2 # one row per arrival
assert "❯" in body
assert "white = selected path" in body
assert "colour = SNR of selected" in body # the graph caption
assert "★ you" in body and "▲ repeater" in body # the node-type legend
assert screen.cursor_line() is not None
@@ -66,12 +67,22 @@ def test_paths_screen_labels_every_relay_with_its_hash_byte() -> None:
assert "Waymarker (a1)" in body
def test_paths_screen_draws_selected_path_white_over_gray() -> None:
"""The selected path's edges render white; the unused path's edges gray."""
screen = _screen(_arrivals())
def test_paths_screen_colours_selected_path_by_snr_over_gray() -> None:
"""The selected path's edges take its arrival's SNR hue; the unused path stays gray."""
screen = _screen(_arrivals()) # selected arrival's SNR is +4.0 dB → the "ok" band
raw = "\n".join(screen.render_body(76)).split("origin →")[0]
assert "38;2;255;255;255" in raw # the selected path
assert "38;2;251;191;36" in raw # snr.ok amber — the selected path, by SNR
assert "38;2;110;110;110" in raw # the other path, gray beneath it
screen.handle("down") # select the weaker arrival (−2.0 dB, still "ok")
raw2 = "\n".join(screen.render_body(76)).split("origin →")[0]
assert "38;2;251;191;36" in raw2 # the newly selected path recolours
def test_paths_screen_marks_a_repeater_relay_with_its_triangle() -> None:
"""A relay whose type resolves to a repeater draws ▲, not the generic dot."""
screen = _screen(_arrivals(), type_of=lambda h: 2 if h == "3d63" else None)
graph = _plain(screen.render_body(76)).split("origin →")[0]
assert "▲" in graph # the repeater relay wears its map glyph
screen.handle("down")
raw = "\n".join(screen.render_body(76)).split("origin →")[0]
assert "38;2;255;255;255" in raw and "38;2;110;110;110" in raw
+25 -3
View File
@@ -75,11 +75,25 @@ def test_record_dialog_shows_stats_route_and_the_trace_action() -> None:
def test_record_dialog_draws_the_walk_as_a_route_graph() -> None:
"""The walk is drawn on THE route graph — us starred at both ends, over a caption."""
body = _plain(_dialog(_record()).render_body(60))
assert body.count("★") == 2 # our node marks both endpoints of the round trip
graph = body[: body.index("labels = hash byte")]
assert graph.count("★") == 2 # our node marks both endpoints of the round trip
assert "labels = hash byte" in body # the graph's caption
assert any("⠀" <= ch <= "⣿" for ch in body) # braille edges are drawn
def test_record_dialog_shows_the_node_type_legend() -> None:
"""The standard node-type key sits under the graph, so its markers read."""
body = _plain(_dialog(_record()).render_body(60))
assert "★ you" in body and "▲ repeater" in body and "◉ sensor" in body
def test_record_graph_marks_a_repeater_relay_with_its_triangle() -> None:
"""A relay whose type is a repeater draws ▲, not the generic ● dot."""
dialog = _dialog(_record(), type_of=lambda h: 2 if h == HUB_ID else None)
graph = _plain(dialog._graph_lines(60))
assert "▲" in graph # the repeater relay, in its map glyph
def test_record_graph_collapses_a_revisited_route_to_distinct_nodes() -> None:
"""A boomerang that doubles back seats each node once — the depth graph can't stack a
node on itself — so the revisiting route draws the same graph as its distinct one."""
@@ -126,8 +140,16 @@ def test_record_dialog_down_moves_to_delete() -> None:
def test_record_dialog_names_the_far_point() -> None:
"""The reached node's name rides beside the far-point distance."""
body = _plain(_dialog(_record(), far_label="Far").render_body(60))
assert "far point 1.5 km" in body
assert "far point 1.5 km Far" in body
assert re.search(r"far point\s+1\.5 km\s+Far", body)
def test_record_dialog_shows_reliability_when_the_far_node_has_history() -> None:
"""The reliability lane reads the % and its sample count; absent without history."""
with_rate = _plain(_dialog(_record(), reliability=(0.875, 7, 8)).render_body(60))
assert re.search(r"reliability\s+88%", with_rate)
assert "7/8" in with_rate
without = _plain(_dialog(_record()).render_body(60))
assert "reliability" not in without
def test_record_dialog_draws_the_enclosed_area_beside_the_stats() -> None:
+15
View File
@@ -179,6 +179,21 @@ def test_traced_targets_excludes_path_walks(repo: Repository) -> None:
assert latest is not None and latest.target == PATH_TRACE_TARGET
def test_target_trace_counts_tallies_successes_and_totals(repo: Repository) -> None:
"""Per target, (successes, total) over every trace — failures included, path walks not."""
run = repo.start_run("trace", {})
repo.record_trace(run, _make_trace("Alice", ("r", 1.0), success=True))
repo.record_trace(run, _make_trace("Alice", success=False))
repo.record_trace(run, _make_trace("Alice", ("r", 2.0), success=True))
repo.record_trace(run, _make_trace("Bob", ("r", 1.0), success=True))
repo.record_trace(run, _make_trace(PATH_TRACE_TARGET, ("r", 3.0), success=True))
counts = repo.target_trace_counts()
assert counts["Alice"] == (2, 3) # two of three came home
assert counts["Bob"] == (1, 1)
assert PATH_TRACE_TARGET not in counts # path walks name no target
# -- latest_trace edge cases -----------------------------------------------
+18
View File
@@ -79,6 +79,24 @@ def test_recent_targets_folds_hex_to_names_and_dedupes() -> None:
assert _recent_targets(stored, contacts) == ["Alice", "cafe", "beefbeef"]
def test_node_type_resolver_matches_hop_hash_to_contact_type() -> None:
"""A relay's hash resolves to its contact's node type (prefix-matched either way)."""
from meshterm.core.models import NODE_TYPE_REPEATER
type_of = trace_runner.make_node_type_resolver(
[
Contact(name="Repeater", public_key="3d63c6" + "00" * 26,
key_prefix="3d63c6429436", node_type=NODE_TYPE_REPEATER),
Contact(name="Typeless", public_key="a1b2c3" + "00" * 26,
key_prefix="a1b2c3d4"),
]
)
assert type_of("3d63") == NODE_TYPE_REPEATER # a short hash prefixes the full key
assert type_of("a1b2") is None # known contact, but no advertised type
assert type_of("ffff") is None # unknown node
assert type_of(None) is None # our own device, passed through
def test_path_hash_flags_power_of_two_only() -> None:
"""Trace flags encode the hash width as 1 << s, so only 1/2/4/8 bytes map."""
assert trace_runner.path_hash_flags(1) == 0