From ed391ab16e8935930ff0b08abbf9575a026bab5a Mon Sep 17 00:00:00 2001 From: Jean-Pierre Martineau Date: Wed, 15 Jul 2026 20:26:47 -0400 Subject: [PATCH] Trophy case + route graph: node-type marks, SNR colour, reliability MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- meshterm/persistence/repository.py | 21 ++++ meshterm/services/trace_runner.py | 35 ++++++ meshterm/ui/chat.py | 8 +- meshterm/ui/message_paths_screen.py | 40 ++++--- meshterm/ui/records_screen.py | 161 ++++++++++++++++++++-------- meshterm/ui/theme.py | 15 +++ meshterm/ui/widgets.py | 46 +++++++- tests/test_message_paths_screen.py | 21 +++- tests/test_records_screen.py | 28 ++++- tests/test_repository.py | 15 +++ tests/test_skeleton.py | 18 ++++ 11 files changed, 336 insertions(+), 72 deletions(-) diff --git a/meshterm/persistence/repository.py b/meshterm/persistence/repository.py index 76d82d6..8efd6f7 100644 --- a/meshterm/persistence/repository.py +++ b/meshterm/persistence/repository.py @@ -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. diff --git a/meshterm/services/trace_runner.py b/meshterm/services/trace_runner.py index 26e062d..681e903 100644 --- a/meshterm/services/trace_runner.py +++ b/meshterm/services/trace_runner.py @@ -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. diff --git a/meshterm/ui/chat.py b/meshterm/ui/chat.py index d9f2e60..dd4d9b7 100644 --- a/meshterm/ui/chat.py +++ b/meshterm/ui/chat.py @@ -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, ) ) diff --git a/meshterm/ui/message_paths_screen.py b/meshterm/ui/message_paths_screen.py index dfb5061..fb9cad9 100644 --- a/meshterm/ui/message_paths_screen.py +++ b/meshterm/ui/message_paths_screen.py @@ -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, diff --git a/meshterm/ui/records_screen.py b/meshterm/ui/records_screen.py index 75420c3..c4cc070 100644 --- a/meshterm/ui/records_screen.py +++ b/meshterm/ui/records_screen.py @@ -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) diff --git a/meshterm/ui/theme.py b/meshterm/ui/theme.py index d9528f6..5204c0a 100644 --- a/meshterm/ui/theme.py +++ b/meshterm/ui/theme.py @@ -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)] diff --git a/meshterm/ui/widgets.py b/meshterm/ui/widgets.py index 4528d30..9ad7a76 100644 --- a/meshterm/ui/widgets.py +++ b/meshterm/ui/widgets.py @@ -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, diff --git a/tests/test_message_paths_screen.py b/tests/test_message_paths_screen.py index cbff815..4ad6dd0 100644 --- a/tests/test_message_paths_screen.py +++ b/tests/test_message_paths_screen.py @@ -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 diff --git a/tests/test_records_screen.py b/tests/test_records_screen.py index 697e61e..935a415 100644 --- a/tests/test_records_screen.py +++ b/tests/test_records_screen.py @@ -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: diff --git a/tests/test_repository.py b/tests/test_repository.py index 87d99bb..dba77a0 100644 --- a/tests/test_repository.py +++ b/tests/test_repository.py @@ -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 ----------------------------------------------- diff --git a/tests/test_skeleton.py b/tests/test_skeleton.py index 6f91036..cc3db94 100644 --- a/tests/test_skeleton.py +++ b/tests/test_skeleton.py @@ -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