diff --git a/meshterm/core/connection.py b/meshterm/core/connection.py index ce425fa..7757850 100644 --- a/meshterm/core/connection.py +++ b/meshterm/core/connection.py @@ -18,7 +18,7 @@ import logging import random import sys from abc import ABC, abstractmethod -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from typing import TYPE_CHECKING, Callable, Optional from .channels import CHANNEL_SLOT_PROBE_CAP @@ -30,8 +30,10 @@ from .models import ( Contact, Hop, Message, + NeighbourInfo, Observation, TraceResult, + utcnow, ) if TYPE_CHECKING: @@ -347,6 +349,29 @@ class Device(ABC): DeviceCommandError: If the node rejected the command. """ + @abstractmethod + async def fetch_neighbours(self, node: Contact) -> list[NeighbourInfo]: + """Ask a remote node for its neighbour table: who it hears directly, and how well. + + A second vantage point for the topology graph: every entry is a link + ``node ↔ neighbour`` with SNR measured *at the remote node*, including nodes we + have never received anything from ourselves. Requires an authenticated session + (:meth:`admin_login` first) — firmware silently ignores the request from guests, + which surfaces here as a timeout. + + Args: + node: The contact to query (must already be logged in). + + Returns: + The reported neighbour entries; empty when the node's table is empty (a + normal answer — repeaters forget neighbours across reboots and only relearn + them as adverts arrive). + + Raises: + DeviceCommandError: If the node never answered (not logged in, out of + reach, or firmware without neighbour tables). + """ + @abstractmethod async def run_trace( self, @@ -1263,6 +1288,39 @@ class MeshCoreDevice(Device): # since some firmware answers tersely or drops the ack under duty-cycle limits. await self._send_admin_cmd(node, f"set tx {value}") + async def fetch_neighbours(self, node: Contact) -> list[NeighbourInfo]: # noqa: D102 + mc = self._require() + pub = self._node_pubkey(node) + # The library pages through the table (one binary request per ~25 entries) and + # concatenates; ``min_timeout`` keeps slow multi-hop replies from being cut off + # at the companion's optimistic suggested timeout. + result = await mc.commands.fetch_all_neighbours(pub, min_timeout=20) + if result is None: + raise DeviceCommandError( + f"{node.name!r} did not answer the neighbour request. Firmware ignores " + "it without an admin login (log in first), and firmware older than " + "~v1.15 has no neighbour table at all." + ) + now = utcnow() + neighbours: list[NeighbourInfo] = [] + for entry in result.get("neighbours") or []: + pubkey = str(entry.get("pubkey") or "").lower() + if not pubkey: + continue + snr = entry.get("snr") + secs_ago = entry.get("secs_ago") + heard_at = None + if isinstance(secs_ago, (int, float)) and secs_ago >= 0: + heard_at = now - timedelta(seconds=float(secs_ago)) + neighbours.append( + NeighbourInfo( + node=pubkey, + snr=float(snr) if snr is not None else None, + heard_at=heard_at, + ) + ) + return neighbours + async def run_trace( # noqa: D102 - inherited docstring self, target: str, @@ -1789,6 +1847,21 @@ class MockDevice(Device): self._admin_sessions: set[str] = set() self._remote_tx: dict[str, int] = {} self._default_remote_tx = 20 + # Simulated neighbour tables, keyed by the repeater's key prefix: what each + # repeater "hears directly" as ``(neighbour_prefix, snr_db, secs_ago)``. The + # ``e5f6a7b8`` entry is deliberately absent from the contact list, so the + # fetched-evidence flow exercises discovering a node we never received from. + self._neighbour_tables: dict[str, list[tuple[str, float, int]]] = { + "a1b2c3d4": [ + ("d4e5f6a7", 6.5, 300), + ("b2c3d4e5", -3.25, 1200), + ("e5f6a7b8", 2.0, 3600), + ], + "b2c3d4e5": [ + ("c3d4e5f6", 8.0, 240), + ("a1b2c3d4", -3.25, 900), + ], + } # Mutable configuration state, keyed exactly like the real SELF_INFO payload so the # settings registry behaves identically on the simulator and on hardware. self._info: dict = { @@ -1889,6 +1962,28 @@ class MockDevice(Device): ) self._remote_tx[key] = value + async def fetch_neighbours(self, node: Contact) -> list[NeighbourInfo]: # noqa: D102 + await asyncio.sleep(0) + key = self._mock_key(node) + # Mirrors real firmware (verified on v1.15): without a login the request is + # silently dropped, which the caller experiences as a timeout. + if key not in self._admin_sessions: + raise DeviceCommandError( + f"{node.name!r} did not answer the neighbour request. Firmware ignores " + "it without an admin login (log in first)." + ) + table = self._neighbour_tables.get(key[:8]) + if table is None: + raise DeviceCommandError( + f"{node.name!r} did not answer the neighbour request (no neighbour " + "table on this node type)." + ) + now = utcnow() + return [ + NeighbourInfo(node=prefix, snr=snr, heard_at=now - timedelta(seconds=ago)) + for prefix, snr, ago in table + ] + @staticmethod def _mock_key(node: Contact) -> str: """Return the lookup key for a remote node (its public key, else key prefix).""" diff --git a/meshterm/core/models.py b/meshterm/core/models.py index 6846275..f379a51 100644 --- a/meshterm/core/models.py +++ b/meshterm/core/models.py @@ -80,6 +80,28 @@ class Contact: return self.node_type == NODE_TYPE_REPEATER +@dataclass(slots=True) +class NeighbourInfo: + """One entry of a remote repeater's neighbour table, fetched over the mesh. + + Repeater firmware keeps a table of the nodes it has *heard directly* (zero-hop), + each with the SNR measured at the repeater — a vantage point ours can't reproduce. + :meth:`~meshterm.core.connection.Device.fetch_neighbours` returns these; the values + describe the link ``repeater ↔ node`` as observed *at the repeater*. + + Attributes: + node: The neighbour's hex public-key prefix, lowercased (the width the firmware + replied at — typically 8 hex). + snr: SNR (dB) the repeater measured receiving this neighbour, if reported. + heard_at: When the repeater last heard the neighbour (derived from the reply's + seconds-ago field), or ``None`` when it didn't say. + """ + + node: str + snr: Optional[float] = None + heard_at: Optional[datetime] = None + + @dataclass(slots=True) class Observation: """A single over-the-air reception heard while passively monitoring the mesh. diff --git a/meshterm/persistence/db.py b/meshterm/persistence/db.py index f45ad4c..f2b0a74 100644 --- a/meshterm/persistence/db.py +++ b/meshterm/persistence/db.py @@ -10,7 +10,7 @@ from __future__ import annotations import sqlite3 from pathlib import Path -SCHEMA_VERSION = 7 +SCHEMA_VERSION = 8 _SCHEMA = """ CREATE TABLE IF NOT EXISTS schema_meta ( @@ -83,6 +83,19 @@ CREATE TABLE IF NOT EXISTS path_candidates ( created_at TEXT NOT NULL ); +-- One entry of a remote repeater's neighbour table, fetched over the mesh (v8). Each row +-- is a link the repeater reported hearing directly, with SNR measured at the repeater; +-- refetching appends a new snapshot and readers take the latest row per pair. +CREATE TABLE IF NOT EXISTS neighbour_reports ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + run_id INTEGER NOT NULL REFERENCES runs(id) ON DELETE CASCADE, + repeater TEXT NOT NULL, -- canonical id of the repeater that was asked + neighbour TEXT NOT NULL, -- reported neighbour's key-prefix hex, as replied + snr REAL, -- dB, measured at the repeater + heard_at TEXT, -- when the repeater last heard the neighbour + fetched_at TEXT NOT NULL +); + -- One packet overheard while passively monitoring the mesh (advert/telemetry/...). CREATE TABLE IF NOT EXISTS observations ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -120,6 +133,7 @@ CREATE TABLE IF NOT EXISTS messages ( CREATE INDEX IF NOT EXISTS idx_traces_run ON traces(run_id); CREATE INDEX IF NOT EXISTS idx_trace_hops_trace ON trace_hops(trace_id); CREATE INDEX IF NOT EXISTS idx_tx_samples_run ON tx_samples(run_id); +CREATE INDEX IF NOT EXISTS idx_neighbour_reports_pair ON neighbour_reports(repeater, neighbour); CREATE INDEX IF NOT EXISTS idx_observations_run ON observations(run_id); CREATE INDEX IF NOT EXISTS idx_observations_node ON observations(node); CREATE INDEX IF NOT EXISTS idx_messages_peer ON messages(peer); diff --git a/meshterm/persistence/repository.py b/meshterm/persistence/repository.py index 8ea824f..a4f93b6 100644 --- a/meshterm/persistence/repository.py +++ b/meshterm/persistence/repository.py @@ -17,6 +17,7 @@ from ..core.models import ( ChatMessage, HeardNode, Hop, + NeighbourInfo, Observation, TraceResult, TxLevelResult, @@ -93,6 +94,29 @@ class PacketPath: hops: list[str] +@dataclass(slots=True) +class NeighbourLink: + """One repeater-reported direct link, as evidence for the topology graph. + + The fetched counterpart of :class:`TracedPath`/:class:`PacketPath`: instead of being + inferred from what *we* received, this link was asserted by a remote repeater about + its own reception (see :meth:`Repository.neighbour_links`). + + Attributes: + when: The link's recency — when the repeater last heard the neighbour, falling + back to when we fetched the table. + repeater: Canonical id of the repeater that reported the link. + neighbour: The neighbour's hex hash as the repeater replied it (any width; the + topology layer canonicalizes). + snr: SNR (dB) measured at the repeater, if reported. + """ + + when: datetime + repeater: str + neighbour: str + snr: Optional[float] + + @dataclass(slots=True) class RunRecord: """A summary row from the ``runs`` table. @@ -403,6 +427,83 @@ class Repository: ) return paths + def record_neighbours( + self, run_id: int, repeater: str, neighbours: list[NeighbourInfo] + ) -> None: + """Persist one fetched neighbour-table snapshot from a remote repeater. + + Every entry becomes a row; refetching the same repeater later appends a fresh + snapshot rather than overwriting, and :meth:`neighbour_links` reads back only + the latest row per ``(repeater, neighbour)`` pair — a neighbour table is the + repeater's *current* state, so a new snapshot supersedes the old one instead + of stacking as extra evidence. + + Args: + run_id: The owning run. + repeater: Canonical id of the repeater the table came from. + neighbours: The fetched entries (may be empty — recorded as no rows). + """ + fetched_at = utcnow().isoformat() + self._conn.executemany( + "INSERT INTO neighbour_reports " + "(run_id, repeater, neighbour, snr, heard_at, fetched_at) " + "VALUES (?, ?, ?, ?, ?, ?)", + [ + ( + run_id, + repeater, + n.node, + n.snr, + n.heard_at.isoformat() if n.heard_at is not None else None, + fetched_at, + ) + for n in neighbours + ], + ) + self._conn.commit() + + def neighbour_links(self, *, limit: int = 2000) -> list[NeighbourLink]: + """Return the current repeater-reported links, newest report first. + + The fetched side of the topology evidence. Only the most recent row per + ``(repeater, neighbour)`` pair is returned (see :meth:`record_neighbours`), so + repeatedly refreshing a table never inflates a link's sample count. + + Args: + limit: Maximum number of links to load. + + Returns: + One :class:`NeighbourLink` per currently-reported link. + """ + rows = self._conn.execute( + "SELECT r.repeater, r.neighbour, r.snr, r.heard_at, r.fetched_at " + "FROM neighbour_reports r " + "JOIN (SELECT MAX(id) AS id FROM neighbour_reports " + " GROUP BY repeater, neighbour) latest ON latest.id = r.id " + "ORDER BY r.id DESC LIMIT ?", + (limit,), + ).fetchall() + links: list[NeighbourLink] = [] + for row in rows: + when: Optional[datetime] = None + for stamp in (row["heard_at"], row["fetched_at"]): + try: + when = datetime.fromisoformat(stamp) + break + except (TypeError, ValueError): + continue + if when is None: + continue # a malformed stray contributes no evidence + links.append( + NeighbourLink( + when=when, + repeater=row["repeater"], + neighbour=row["neighbour"], + snr=row["snr"], + ) + ) + return links + def record_tx_sample(self, run_id: int, level: TxLevelResult) -> None: """Persist one robust TX-power level from an optimization sweep. diff --git a/meshterm/services/topology.py b/meshterm/services/topology.py index 96b7e62..272df8f 100644 --- a/meshterm/services/topology.py +++ b/meshterm/services/topology.py @@ -4,9 +4,12 @@ MeshCore never tells us the network's shape outright, but almost everything we * carries a fragment of it: a successful trace demonstrably crossed every link in its path (out and back, with an SNR reading at each hop), the firmware's per-contact ``out_path`` is a route distilled from received floods, and — when the companion's packet logging is -on — every overheard frame reports the relay chain it rode in on. This module folds those -fragments into one undirected evidence graph and answers the questions the trace path -composer asks of it: +on — every overheard frame reports the relay chain it rode in on. A fourth fragment can +be *asked for*: a repeater we hold admin rights on reports its own neighbour table (who +it hears directly, at what SNR) when queried over the mesh — a second vantage point that +reveals links, and whole nodes, our radio has never received anything from. This module +folds those fragments into one undirected evidence graph and answers the questions the +trace path composer asks of it: * *who neighbours whom* — :meth:`MeshTopology.next_hops`, the suggestion list while composing a path hop by hop, sorted by the strongest observed link first; @@ -34,7 +37,7 @@ from datetime import datetime from typing import Optional from ..core.models import Contact, utcnow -from ..persistence.repository import PacketPath, TracedPath +from ..persistence.repository import NeighbourLink, PacketPath, TracedPath #: Canonical id for our own device in the graph (its 12-hex key prefix when known). #: Kept distinct from hop hashes by construction: it is derived from the full public key. @@ -96,7 +99,8 @@ class Link: samples: How many independent readings crossed this link, any source. snrs: The SNR readings (dB) attributed to the link, either direction. last_seen: When the link was most recently observed. - sources: Which evidence classes saw it (``trace`` / ``route`` / ``packet``). + sources: Which evidence classes saw it (``trace`` / ``route`` / ``packet`` / + ``neighbour``). """ a: str @@ -293,7 +297,8 @@ class MeshTopology: snrs: Optional per-link SNR readings aligned with the links (``snrs[i]`` belongs to the ``nodes[i] → nodes[i+1]`` link, measured at arrival). when: When the walk was observed (stamps every touched link's recency). - source: Evidence class tag (``trace`` / ``route`` / ``packet``). + source: Evidence class tag (``trace`` / ``route`` / ``packet`` / + ``neighbour``). """ for i in range(len(nodes) - 1): a, b = nodes[i], nodes[i + 1] @@ -485,10 +490,11 @@ def build_topology( contacts: list[Contact], trace_paths: list[TracedPath], packet_paths: list[PacketPath], + neighbour_links: tuple[NeighbourLink, ...] | list[NeighbourLink] = (), ) -> MeshTopology: - """Assemble the evidence graph from everything we have received. + """Assemble the evidence graph from everything we have received — or asked for. - Three sources fold in, each one walked through :meth:`MeshTopology.add_walk`: + Four sources fold in, each one walked through :meth:`MeshTopology.add_walk`: * **traces** — each successful trace's hop sequence, bracketed by us at both ends (the boomerang leaves us and returns to us; the final hash-less hop *is* us). @@ -500,12 +506,18 @@ def build_topology( * **packets** — RX-logged frames: the relay chain, prefixed by the originator when known and always ending at us (we heard the last relay). The measured SNR belongs to that final link only. + * **neighbour reports** — links a remote repeater asserted about itself when we + fetched its neighbour table: one two-node walk per entry, SNR as measured at the + repeater. The only evidence here not derived from our own reception — it can + introduce nodes no other source has ever seen. Args: self_id: Our device's key/hash (any width; canonicalized to 12 hex). contacts: The device's known contacts (canonicalization + route evidence). trace_paths: Stored successful trace walks (see ``Repository.trace_paths``). packet_paths: Stored RX-logged packet paths (see ``Repository.packet_paths``). + neighbour_links: Current repeater-reported links (see + ``Repository.neighbour_links``); empty when none have been fetched. Returns: The populated :class:`MeshTopology`. @@ -543,4 +555,12 @@ def build_topology( snrs = [None] * (len(nodes) - 2) + [packet.snr] topo.add_walk(nodes, snrs=snrs, when=packet.when, source="packet") + for reported in neighbour_links: + topo.add_walk( + [topo.canonical(reported.repeater), topo.canonical(reported.neighbour)], + snrs=[reported.snr], + when=reported.when, + source="neighbour", + ) + return topo diff --git a/meshterm/tools/trace.py b/meshterm/tools/trace.py index 382eda6..7d0cf12 100644 --- a/meshterm/tools/trace.py +++ b/meshterm/tools/trace.py @@ -3,10 +3,12 @@ In the interactive menu this opens the live trace screen (:mod:`meshterm.ui.trace_screen`): pick a target from a recency-ordered list and the screen opens armed but idle — Enter runs a burst under an abortable tracing dialog, ``p`` -composes a forced path hop by hop from the topology observed in received traffic, and -``x`` explores ranked path scenarios (with an optional probe that measures each candidate -and offers the winner). On the CLI it stays a scriptable one-shot: run N traces, print -the per-trace table and the aggregate summary, exit. +composes a forced path hop by hop from the topology observed in received traffic (and, +standing on a repeater with stored admin credentials, can fetch that repeater's live +neighbour table for suggestions our own radio never heard), and ``x`` explores ranked +path scenarios (with an optional probe that measures each candidate and offers the +winner). On the CLI it stays a scriptable one-shot: run N traces, print the per-trace +table and the aggregate summary, exit. Both front ends persist identically: one ``runs`` row per burst with every trace recorded under it, so stored history reads the same no matter where it came from. diff --git a/meshterm/ui/path_composer.py b/meshterm/ui/path_composer.py index a208f9b..5e6fe92 100644 --- a/meshterm/ui/path_composer.py +++ b/meshterm/ui/path_composer.py @@ -15,6 +15,11 @@ Interaction, following the reorder screen's cursor-over-rows-and-actions pattern even-length hex, an *add custom hop* row appears, so a node we have never observed (or a bare hash from another tool) can be forced into the route. * Backspace erases the filter first; with the filter empty it removes the last hop. +* When the path's tail is a repeater we hold admin credentials for, a *fetch + neighbours* row asks that repeater over the mesh for its own neighbour table — + fresh second-vantage evidence exactly where composing ran out of it. The dialog + resolves :class:`FetchNeighbours` and the owning flow fetches, refreshes the + topology, and reopens the composer mid-thought (hops preserved). * Enter on **Use this path** commits the composed spec; **Auto** hands routing back to the device; Esc cancels with no change. @@ -24,6 +29,7 @@ automatically, so the dialog shows the outbound route only and says so. from __future__ import annotations +from dataclasses import dataclass from typing import Optional from rich.cells import cell_len @@ -49,8 +55,24 @@ _AUTO = "auto" _CANCEL = "cancel" #: One-letter tags for the evidence classes backing a link, shown beside each -#: suggestion: T(race), R(oute — the firmware's learned out_path), P(acket log). -_SOURCE_TAGS = {"trace": "T", "route": "R", "packet": "P"} +#: suggestion: T(race), R(oute — the firmware's learned out_path), P(acket log), +#: N(eighbour table fetched from a repeater). +_SOURCE_TAGS = {"trace": "T", "route": "R", "packet": "P", "neighbour": "N"} + + +@dataclass(frozen=True, slots=True) +class FetchNeighbours: + """The composer's resolution when the user asks a repeater for its neighbours. + + The dialog itself never touches the radio; it resolves this marker and the owning + flow performs the login + fetch, refreshes the topology, and reopens the composer + with :attr:`PathComposerScreen.hops` re-seeded. + + Attributes: + node: Canonical id of the repeater to query (the path's tail when committed). + """ + + node: str class PathComposerScreen(Screen): @@ -74,6 +96,7 @@ class PathComposerScreen(Screen): topology: MeshTopology, width_bytes: int, hops: Optional[list[str]] = None, + fetch_nodes: frozenset[str] = frozenset(), ) -> None: """Build the composer. @@ -88,6 +111,9 @@ class PathComposerScreen(Screen): width_bytes: Preferred per-hop path-hash width (bytes) for the emitted spec. hops: Canonical ids of already-composed intermediate hops (reopening the dialog resumes where the user left off). + fetch_nodes: Canonical ids whose live neighbour table can be fetched + (repeater contacts with a public key); when the path's tail is one of + them, the *fetch neighbours* row appears. """ super().__init__() self.title = f"compose path · {target_label}" @@ -98,9 +124,15 @@ class PathComposerScreen(Screen): self._topology = topology self._width_bytes = width_bytes self._hops: list[str] = list(hops or []) + self._fetch_nodes = fetch_nodes self._entry = "" self._index = 0 + @property + def hops(self) -> list[str]: + """The composed intermediate hops so far (for re-seeding after a fetch).""" + return list(self._hops) + # --- state ----------------------------------------------------------------- def _tail(self) -> str: @@ -134,12 +166,17 @@ class PathComposerScreen(Screen): return needle if _is_hex(needle) else None def _rows(self) -> list[tuple[str, object]]: - """The cursor-addressable rows: custom hop, suggestions, then the actions.""" + """The cursor-addressable rows: custom hop, suggestions, fetch, then actions.""" rows: list[tuple[str, object]] = [] custom = self._custom_hex() if custom: rows.append(("custom", custom)) rows.extend(("hop", s) for s in self._suggestions()) + # Standing on a repeater we hold credentials for, its live neighbour table is + # one keypress away — placed with the suggestions, because that is what it + # extends: "don't see the node you need? ask the repeater what it hears." + if self._tail() in self._fetch_nodes: + rows.append(("fetch", self._tail())) rows.append(("action", _USE)) rows.append(("action", _AUTO)) rows.append(("action", _CANCEL)) @@ -202,6 +239,12 @@ class PathComposerScreen(Screen): return text if kind == "hop": return self._suggestion_text(payload) + if kind == "fetch": + name = self._topology.display_name(str(payload)) or str(payload)[:12] + text = Text("⇣ Fetch neighbours from ", style="") + text.append(name, style="brand") + text.append(" (asks the repeater over the mesh)", style="muted") + return text if payload == _USE: label = Text.assemble(("✓ ", "ok"), "Use this path") label.append(f" ({self._spec()})", style="muted") @@ -255,7 +298,10 @@ class PathComposerScreen(Screen): cursor_at = len(lines) lines.append(render_to_ansi(text, width)) if not any(kind == "hop" for kind, _ in rows): - note = "(no observed links from here — type a hex hash to force a hop)" + if any(kind == "fetch" for kind, _ in rows): + note = "(no observed links from here — fetch the repeater's neighbours, or type a hex hash)" + else: + note = "(no observed links from here — type a hex hash to force a hop)" lines.append(render_to_ansi(Text(note, style="muted"), width)) self._cursor = cursor_at @@ -282,6 +328,8 @@ class PathComposerScreen(Screen): self._hops.append(payload.node) # type: ignore[union-attr] self._entry = "" self._index = 0 + elif kind == "fetch": + self.resolve(FetchNeighbours(node=str(payload))) elif payload == _USE: self.resolve(self._spec()) elif payload == _AUTO: diff --git a/meshterm/ui/trace_screen.py b/meshterm/ui/trace_screen.py index 11631ca..4978556 100644 --- a/meshterm/ui/trace_screen.py +++ b/meshterm/ui/trace_screen.py @@ -11,7 +11,10 @@ whatever you make it, and nothing transmits until you say so. Three verbs drive * **p** opens the path composer (:mod:`~meshterm.ui.path_composer`): build the outbound route hop by hop, each step suggested from the links observed in *received* traffic — traces, firmware-learned contact routes, and RX-logged packet paths — strongest first, - with raw hex entry for nodes the data has never seen. Only the outbound leg is + with raw hex entry for nodes the data has never seen. Standing on a repeater you hold + admin credentials for, the composer can also *fetch that repeater's neighbour table* + over the mesh (login required; firmware ignores guests): second-vantage evidence, + persisted and folded straight back into the suggestions. Only the outbound leg is composed: the trace protocol replies back along the reversed path automatically. * **x** explores scenarios: ranked candidate routes to the target straight from the topology evidence (the device's own learned route, the direct shot, and the strongest @@ -128,6 +131,9 @@ class TracingDialog(Screen): self.status = "transmitting…" #: The most recent trace result, echoed beneath the status line. self.last: Optional[TraceResult] = None + #: Whether to render the last-reply line. Trace/probe owners stream replies + #: through it; request-shaped owners (a neighbour fetch) have none to show. + self.show_last = True @property def dialog_width(self) -> int: @@ -159,7 +165,8 @@ class TracingDialog(Screen): chip = self._spinner.text() chip.append(f" {self.status}", style="") lines = render_lines(chip, width) - lines.extend(render_lines(self._last_line(), width)) + if self.show_last: + lines.extend(render_lines(self._last_line(), width)) lines.append("") lines.append(render_to_ansi(Text("❯ [ Abort ]", style="selected"), width)) return lines @@ -554,10 +561,11 @@ async def open_trace(ctx: "AppContext", target: str) -> int: Raises: RuntimeError: If called outside the interactive menu (no full-screen session). """ - from ..core.models import LOCAL_DEVICE_LABEL + from ..core.connection import DeviceAuthenticationError + from ..core.models import LOCAL_DEVICE_LABEL, Contact, NeighbourInfo from ..services.path_probe import ProbeCandidate, ProbeOutcome, probe_paths from ..services.topology import MeshTopology, PathScenario, _is_hex, build_topology - from .path_composer import PathComposerScreen + from .path_composer import FetchNeighbours, PathComposerScreen from .surface import TuiUi from .tui import CANCEL, Choice, SelectScreen, Separator @@ -609,6 +617,7 @@ async def open_trace(ctx: "AppContext", target: str) -> int: contacts=contacts, trace_paths=ctx.repo.trace_paths(), packet_paths=ctx.repo.packet_paths(), + neighbour_links=ctx.repo.neighbour_links(), ) async def unaddressable() -> None: @@ -621,8 +630,122 @@ async def open_trace(ctx: "AppContext", target: str) -> int: title="no destination hash", ) + async def fetch_neighbours_via(repeater: Contact, repeater_id: str) -> bool: + """Log in to a repeater, fetch its neighbour table, and persist the snapshot. + + The composer's fetch action lands here: password from the admin store (or a + one-time prompt, remembered on success — the tx-optimize convention), then the + login + fetch run under a floating spinner dialog with Abort. Every outcome + closes its own ``runs`` row; a rejected login also forgets the stored password + so the next attempt asks fresh. + + Args: + repeater: The repeater contact to query (carries the public key). + repeater_id: Its canonical id, the key the snapshot is stored under. + + Returns: + ``True`` when new neighbour links were recorded (the caller should rebuild + the topology); ``False`` on cancel, failure, or an empty table. + """ + password = ctx.admin_store.get(repeater) + if password is None: + password = await session.text( + f"Admin password for {repeater.name}", + prompt="The repeater ignores neighbour requests without an admin login.", + password=True, + ) + if not password: + return False + run_id = ctx.repo.start_run( + "trace", + {"mode": "neighbours", "repeater": repeater.name}, + ctx.profile_name, + ) + spinner = Spinner() + dialog = TracingDialog( + f"neighbours · {repeater.name}", spinner=spinner, on_abort=lambda: None + ) + dialog.show_last = False # a fetch has no streaming replies to echo + dialog.status = f"logging in to {repeater.name}…" + + async def work() -> list[NeighbourInfo]: + if not await device.admin_login(repeater, password): + raise DeviceAuthenticationError( + f"{repeater.name!r} rejected the admin login (wrong password?). " + "The saved password was cleared; retry to enter a new one." + ) + ctx.admin_store.remember(repeater, password) + dialog.status = "fetching the neighbour table…" + session.invalidate() + return await device.fetch_neighbours(repeater) + + task = asyncio.ensure_future(work()) + dialog.on_abort = task.cancel + + async def animate() -> None: + while True: + await asyncio.sleep(_SPINNER_INTERVAL) + spinner.tick() + session.invalidate() + + ticker = asyncio.ensure_future(animate()) + session.push(dialog) + error: Optional[BaseException] = None + aborted = False + entries: list[NeighbourInfo] = [] + try: + entries = await task + except asyncio.CancelledError: + aborted = True + except Exception as exc: # noqa: BLE001 - surface in a dialog, keep composing + error = exc + finally: + ticker.cancel() + try: + await ticker + except asyncio.CancelledError: + pass + except Exception: # noqa: BLE001 - a spinner hiccup must never break a fetch + pass + session.pop(dialog) + if aborted: + ctx.repo.finish_run(run_id, "error", {"error": "aborted"}) + return False + if error is not None: + if isinstance(error, DeviceAuthenticationError): + ctx.admin_store.forget(repeater) # bad password: don't keep reusing it + ctx.repo.finish_run( + run_id, "error", {"error": str(error) or type(error).__name__} + ) + await session.message_dialog( + Text(str(error), style="err"), title="fetch neighbours" + ) + return False + ctx.repo.record_neighbours(run_id, repeater_id, entries) + ctx.repo.finish_run( + run_id, "ok", {"repeater": repeater.name, "neighbours": len(entries)} + ) + if not entries: + # Verified on real firmware: an empty table is a normal answer (repeaters + # forget neighbours across reboots and relearn them from adverts). + await session.message_dialog( + Text( + f"{repeater.name} answered, but its neighbour table is empty — " + "it relearns neighbours from received adverts, so ask again later.", + style="muted", + ), + title="fetch neighbours", + ) + return False + return True + async def compose(current: str) -> Optional[str]: - """Open the hop-by-hop composer seeded with the current spec's hops.""" + """Open the hop-by-hop composer seeded with the current spec's hops. + + Runs the composer in a loop: a :class:`FetchNeighbours` resolution performs the + fetch, rebuilds the topology with the new evidence, and reopens the composer + exactly where the user stood (same hops, refreshed suggestions). + """ if target_hash is None: await unaddressable() return None @@ -634,8 +757,17 @@ async def open_trace(ctx: "AppContext", target: str) -> int: for h in [p for p in current.split(",") if p.strip()][:-1] if (cid := topo.canonical(h.strip())) is not None ] - result = await session.run_screen( - PathComposerScreen( + while True: + # Nodes whose neighbour table can be asked for: repeater contacts with a + # public key to log in against (our own node has nothing new to tell us). + fetchable: dict[str, Contact] = {} + for c in contacts: + if not c.is_repeater or not (c.public_key or "").strip(): + continue + cid = topo.canonical(c.public_key) + if cid is not None and cid != topo.self_id: + fetchable[cid] = c + screen = PathComposerScreen( target_id=target_id, target_hash=target_hash, target_label=target_label, @@ -643,9 +775,20 @@ async def open_trace(ctx: "AppContext", target: str) -> int: topology=topo, width_bytes=width_bytes, hops=seed, + fetch_nodes=frozenset(fetchable), ) - ) - return None if result is CANCEL else result + result = await session.run_screen(screen) + if result is CANCEL: + return None + if isinstance(result, FetchNeighbours): + seed = screen.hops # resume mid-thought after the fetch + repeater = fetchable.get(result.node) + if repeater is not None and await fetch_neighbours_via( + repeater, result.node + ): + topo = fresh_topology() # fold the new reports into suggestions + continue + return result def scenario_title(scenario: PathScenario, topo: MeshTopology) -> Text: """One scenario as a select row: source, route, and its observed evidence.""" diff --git a/tests/test_topology.py b/tests/test_topology.py index 58f7d8b..d37b0ee 100644 --- a/tests/test_topology.py +++ b/tests/test_topology.py @@ -10,11 +10,19 @@ from __future__ import annotations from datetime import timedelta -from meshterm.core.models import Contact, Observation, utcnow -from meshterm.persistence.repository import PacketPath, Repository, TracedPath +import pytest + +from meshterm.core.connection import DeviceCommandError, MockDevice +from meshterm.core.models import Contact, NeighbourInfo, Observation, utcnow +from meshterm.persistence.repository import ( + NeighbourLink, + PacketPath, + Repository, + TracedPath, +) from meshterm.services.path_probe import ProbeCandidate, ProbeOutcome, probe_paths from meshterm.services.topology import build_topology, collapse_width -from meshterm.ui.path_composer import AUTO_SPEC, PathComposerScreen +from meshterm.ui.path_composer import AUTO_SPEC, FetchNeighbours, PathComposerScreen from meshterm.ui.tui.screen import CANCEL US = "aaaaaaaaaaaa" @@ -23,12 +31,13 @@ FAR = Contact(name="Far", public_key="f2c24f54551e" + "0" * 52, key_prefix="f2c2 LEAF = Contact(name="Leaf", public_key="27d4396a2967" + "0" * 52, key_prefix="27d4396a2967") -def _topo(trace_paths=(), packet_paths=(), contacts=None): # noqa: ANN001 +def _topo(trace_paths=(), packet_paths=(), neighbour_links=(), contacts=None): # noqa: ANN001 return build_topology( self_id=US + "0" * 52, contacts=list(contacts) if contacts is not None else [REPEATER, FAR, LEAF], trace_paths=list(trace_paths), packet_paths=list(packet_paths), + neighbour_links=list(neighbour_links), ) @@ -82,6 +91,23 @@ def test_contact_routes_and_packet_paths_feed_the_graph() -> None: assert packet_link.snrs == [] +def test_neighbour_reports_feed_the_graph_as_fetched_evidence() -> None: + """A repeater's fetched table adds links at its vantage point, tagged ``neighbour``.""" + reported = [ + NeighbourLink(when=utcnow(), repeater="3d63c6429436", neighbour="f2c2", snr=7.0), + NeighbourLink(when=utcnow(), repeater="3d63c6429436", neighbour="beefbeef", snr=None), + ] + topo = _topo(neighbour_links=reported) + link = topo.link("3d63c6429436", "f2c24f54551e") # 'f2c2' canonicalizes onto Far + assert link is not None and link.sources == {"neighbour"} + assert link.snrs == [7.0] # the SNR measured at the repeater rides the link + # A neighbour no contact matches still counts — discovery of an unseen node. + unknown = topo.link("3d63c6429436", "beefbeef") + assert unknown is not None and unknown.snrs == [] + suggested = [s.node for s in topo.next_hops("3d63c6429436")] + assert "f2c24f54551e" in suggested and "beefbeef" in suggested + + # --- suggestions and scenarios ---------------------------------------------------- @@ -152,6 +178,54 @@ def test_repository_round_trips_packet_paths_and_candidates(tmp_path) -> None: repo.close() +def test_repository_neighbour_snapshots_keep_latest_per_pair(tmp_path) -> None: # noqa: ANN001 + """A refetched table supersedes the old snapshot instead of stacking as evidence.""" + repo = Repository(tmp_path / "n.db") + run = repo.start_run("trace", {"mode": "neighbours"}) + heard = utcnow() - timedelta(hours=2) + repo.record_neighbours( + run, + "3d63c6429436", + [ + NeighbourInfo(node="f2c24f54", snr=-5.0, heard_at=heard), + NeighbourInfo(node="beefbeef", snr=2.0, heard_at=None), + ], + ) + links = {link.neighbour: link for link in repo.neighbour_links()} + assert len(links) == 2 + assert links["f2c24f54"].snr == -5.0 + assert links["f2c24f54"].when == heard # the repeater's own recency survives + assert links["beefbeef"].when is not None # no heard_at → falls back to fetched_at + + repo.record_neighbours(run, "3d63c6429436", [NeighbourInfo(node="f2c24f54", snr=9.0)]) + links = {link.neighbour: link for link in repo.neighbour_links()} + assert len(links) == 2 # still one link per pair + assert links["f2c24f54"].snr == 9.0 # the fresh snapshot won + repo.close() + + +async def test_mock_fetch_neighbours_is_login_gated_like_real_firmware() -> None: + """No login → ignored (error); logged in → the table, including an unseen node.""" + device = MockDevice() + await device.connect() + contacts = await device.get_contacts() + yagi = next(c for c in contacts if c.name == "Yagi-Repeater") + + with pytest.raises(DeviceCommandError): + await device.fetch_neighbours(yagi) # mirrors v1.15 firmware: guests are ignored + + assert await device.admin_login(yagi, "admin") + entries = await device.fetch_neighbours(yagi) + nodes = {e.node for e in entries} + assert "e5f6a7b8" in nodes # a node absent from the contact list: pure discovery + assert all(e.heard_at is not None for e in entries) + + alice = next(c for c in contacts if c.name == "Alice") + assert await device.admin_login(alice, "admin") + with pytest.raises(DeviceCommandError): + await device.fetch_neighbours(alice) # chat nodes keep no neighbour table + + # --- path probe ---------------------------------------------------------------------- @@ -180,7 +254,7 @@ async def test_probe_paths_ranks_reliability_first() -> None: # --- path composer --------------------------------------------------------------------- -def _composer(topo, hops=None): # noqa: ANN001 +def _composer(topo, hops=None, fetch_nodes=frozenset()): # noqa: ANN001 return PathComposerScreen( target_id="f2c24f54551e", target_hash="f2c24f54551e" + "0" * 52, @@ -189,6 +263,7 @@ def _composer(topo, hops=None): # noqa: ANN001 topology=topo, width_bytes=1, hops=list(hops or []), + fetch_nodes=fetch_nodes, ) @@ -219,6 +294,28 @@ def test_composer_typed_hex_adds_a_custom_hop_and_backspace_removes() -> None: assert screen._hops == [] +async def test_composer_fetch_row_resolves_a_fetch_request() -> None: + """Standing on a fetchable repeater, Enter on the fetch row hands off to the owner.""" + import asyncio + + screen = _composer( + _topo(), hops=["3d63c6429436"], fetch_nodes=frozenset({"3d63c6429436"}) + ) + screen.future = asyncio.get_running_loop().create_future() + body = _rows_plain(screen) + assert "Fetch neighbours from" in body and "Hub" in body + screen.handle("enter") # an empty graph offers no suggestions: fetch is the top row + result = screen.future.result() + assert isinstance(result, FetchNeighbours) and result.node == "3d63c6429436" + assert screen.hops == ["3d63c6429436"] # preserved, so the owner can reopen mid-path + + +def test_composer_hides_fetch_row_off_fetchable_tails() -> None: + """The fetch row tracks the path's tail: at our own node there is nothing to ask.""" + screen = _composer(_topo(), fetch_nodes=frozenset({"3d63c6429436"})) + assert "Fetch neighbours" not in _rows_plain(screen) # tail is us, not the repeater + + async def test_composer_commits_spec_auto_and_cancel() -> None: """Use resolves the target-terminated spec; Auto resolves the empty spec; Esc cancels.""" import asyncio