diff --git a/meshtools/persistence/repository.py b/meshtools/persistence/repository.py index 937c1c0..0b55f52 100644 --- a/meshtools/persistence/repository.py +++ b/meshtools/persistence/repository.py @@ -230,27 +230,46 @@ class Repository: "SELECT * FROM traces WHERE target = ? ORDER BY id DESC LIMIT ?", (target, limit), ).fetchall() - traces: list[TraceResult] = [] - for row in rows: - hop_rows = self._conn.execute( - "SELECT hop_index, node, snr FROM trace_hops WHERE trace_id = ? " - "ORDER BY hop_index", - (row["id"],), - ).fetchall() - hops = [ - Hop(index=h["hop_index"], node=h["node"], snr=h["snr"]) for h in hop_rows - ] - traces.append( - TraceResult( - target=row["target"], - success=bool(row["success"]), - hops=hops, - round_trip_ms=row["round_trip_ms"], - tx_power=row["tx_power"], - timestamp=datetime.fromisoformat(row["created_at"]), - ) - ) - return traces + return [self._hydrate_trace(row) for row in rows] + + def all_traces(self, *, limit: int = 500) -> list[TraceResult]: + """Return recent traces across every target, newest first, rehydrated with hops. + + Used to build a whole-mesh route graph rather than a single target's history. + + Args: + limit: Maximum number of traces to return. + + Returns: + The matching :class:`TraceResult` objects, newest first. + """ + rows = self._conn.execute( + "SELECT * FROM traces ORDER BY id DESC LIMIT ?", (limit,) + ).fetchall() + return [self._hydrate_trace(row) for row in rows] + + def _hydrate_trace(self, row: sqlite3.Row) -> TraceResult: + """Rebuild a :class:`TraceResult` (with hops) from a ``traces`` row. + + Args: + row: A ``traces`` table row. + + Returns: + The rehydrated trace. + """ + hop_rows = self._conn.execute( + "SELECT hop_index, node, snr FROM trace_hops WHERE trace_id = ? ORDER BY hop_index", + (row["id"],), + ).fetchall() + hops = [Hop(index=h["hop_index"], node=h["node"], snr=h["snr"]) for h in hop_rows] + return TraceResult( + target=row["target"], + success=bool(row["success"]), + hops=hops, + round_trip_ms=row["round_trip_ms"], + tx_power=row["tx_power"], + timestamp=datetime.fromisoformat(row["created_at"]), + ) def traced_targets(self) -> list[str]: """Return the distinct trace destinations recorded, most-traced first. diff --git a/meshtools/services/route_stability.py b/meshtools/services/route_stability.py new file mode 100644 index 0000000..53d1c2e --- /dev/null +++ b/meshtools/services/route_stability.py @@ -0,0 +1,190 @@ +"""Route-stability aggregation: turn trace history into a mesh graph with churn metrics. + +The MeshCore app traces a route once. Because MeshTools persists every trace, it can +instead show how *stable* routing is over time — which hops are always used and which the +mesh flaps between. This service aggregates a set of traces into: + +* a **graph** of nodes and directed links, each link weighted by how often it was + traversed and its median SNR; and +* per-target **churn** metrics: how many distinct routes were seen and what share the + single most-common route held (the higher that share, the steadier the routing). + +It is pure logic over :class:`~meshtools.core.models.TraceResult` lists; the pyvis +renderer in :mod:`meshtools.viz.route_graph` draws the result. +""" + +from __future__ import annotations + +import math +import statistics +from dataclasses import dataclass, field +from typing import Callable, Optional + +from ..core.models import LOCAL_DEVICE_LABEL, TraceResult + +NodeResolver = Callable[[Optional[str]], Optional[str]] + + +@dataclass(slots=True) +class RouteLink: + """A directed link between two nodes, aggregated across traces. + + Attributes: + origin: Transmitting node label. + destination: Receiving node label. + count: Number of trace hops that traversed this link. + median_snr: Median SNR (dB) measured on it. + """ + + origin: str + destination: str + count: int + median_snr: float + + +@dataclass(slots=True) +class RouteNode: + """A node in the route graph. + + Attributes: + label: Display label for the node. + is_self: Whether this is our own companion device. + degree: Number of distinct links (in or out) touching the node. + """ + + label: str + is_self: bool + degree: int + + +@dataclass(slots=True) +class RouteGraph: + """An aggregated route graph plus churn metrics. + + Attributes: + target: The single target analyzed, or ``None`` for a whole-mesh graph. + nodes: The graph's nodes. + links: The graph's directed links. + total_routes: Number of successful traces aggregated. + distinct_routes: Number of unique full-path signatures seen. + dominant_share: Fraction of traces that took the single most-common route. + churn_entropy: Shannon entropy (bits) of the route-choice distribution — ``0`` + when one route is always chosen, growing as traffic spreads across more + equally-likely routes. + """ + + target: Optional[str] + nodes: list[RouteNode] = field(default_factory=list) + links: list[RouteLink] = field(default_factory=list) + total_routes: int = 0 + distinct_routes: int = 0 + dominant_share: float = 0.0 + churn_entropy: float = 0.0 + + @property + def stability(self) -> float: + """A 0-1 stability score: ``1`` when one route always wins, lower as it flaps. + + Defined as the dominant route's share, so a single stable path scores ``1.0`` and + an even split across many routes trends toward ``0``. + """ + return self.dominant_share + + +def build_route_graph( + traces: list[TraceResult], + *, + target: Optional[str] = None, + resolve: Optional[NodeResolver] = None, + device_label: str = LOCAL_DEVICE_LABEL, +) -> RouteGraph: + """Aggregate traces into a route graph with churn metrics. + + Args: + traces: The traces to aggregate (only successful ones contribute edges/routes). + target: The target these traces share, or ``None`` for a multi-target graph. + resolve: Optional resolver mapping a hop hash to a friendly node name. + device_label: Label for our own device at the path endpoints. + + Returns: + A populated :class:`RouteGraph`. With no successful traces the graph is empty and + its metrics are zero. + """ + successes = [t for t in traces if t.success and t.hops] + link_snrs: dict[tuple[str, str], list[float]] = {} + route_counts: dict[tuple[str, ...], int] = {} + node_links: dict[str, set[tuple[str, str]]] = {} + + for trace in successes: + edges = trace.edges(device_label) + signature: list[str] = [] + for edge in edges: + origin = _name(edge.origin, resolve) + destination = _name(edge.destination, resolve) + key = (origin, destination) + link_snrs.setdefault(key, []).append(edge.snr) + node_links.setdefault(origin, set()).add(key) + node_links.setdefault(destination, set()).add(key) + if not signature: + signature.append(origin) + signature.append(destination) + route_counts[tuple(signature)] = route_counts.get(tuple(signature), 0) + 1 + + links = [ + RouteLink( + origin=origin, + destination=destination, + count=len(snrs), + median_snr=statistics.median(snrs), + ) + for (origin, destination), snrs in link_snrs.items() + ] + links.sort(key=lambda link_item: link_item.count, reverse=True) + + self_label = _name(None, resolve) or device_label + nodes = [ + RouteNode(label=label, is_self=(label == self_label), degree=len(keys)) + for label, keys in sorted(node_links.items(), key=lambda kv: len(kv[1]), reverse=True) + ] + + total = len(successes) + dominant_share = (max(route_counts.values()) / total) if total else 0.0 + return RouteGraph( + target=target, + nodes=nodes, + links=links, + total_routes=total, + distinct_routes=len(route_counts), + dominant_share=dominant_share, + churn_entropy=_route_entropy(route_counts, total), + ) + + +def _route_entropy(route_counts: dict[tuple[str, ...], int], total: int) -> float: + """Return the Shannon entropy (bits) of a route-choice distribution. + + Args: + route_counts: Map of route signature to how often it occurred. + total: Total number of routes counted. + + Returns: + The entropy in bits (``0`` for a single route). + """ + if total <= 0: + return 0.0 + entropy = 0.0 + for count in route_counts.values(): + p = count / total + if p > 0: + entropy -= p * math.log2(p) + return entropy + + +def _name(label: Optional[str], resolve: Optional[NodeResolver]) -> str: + """Resolve a hop label to a friendly name, leaving our own device's label intact.""" + if label is None: + resolved = resolve(None) if resolve else None + return resolved or LOCAL_DEVICE_LABEL + if resolve is None: + return label + return resolve(label) or label diff --git a/meshtools/tools/route_map.py b/meshtools/tools/route_map.py new file mode 100644 index 0000000..f89f6f6 --- /dev/null +++ b/meshtools/tools/route_map.py @@ -0,0 +1,228 @@ +"""The ``route-map`` tool: visualize how stable mesh routing is over time. + +Where the app's traceroute is a single snapshot, this aggregates a target's (or the whole +mesh's) trace history into an interactive graph and churn metrics: which links are always +used, which the mesh flaps between, and how often the single most-common route actually +wins. It reads stored traces and can optionally take a few fresh ones first. +""" + +from __future__ import annotations + +from typing import Any, Optional + +import questionary +import typer +from rich.panel import Panel +from rich.table import Table +from rich.text import Text + +from ..context import AppContext +from ..services import route_stability, trace_runner +from ..services.route_stability import RouteGraph +from ..ui.theme import snr_style +from ..ui.widgets import make_progress +from ..viz.route_graph import render_route_graph +from .base import Tool, ToolResult, register + +_WHOLE_MESH = "(whole mesh)" + + +@register +class RouteMapTool(Tool): + """Aggregate trace history into a route-stability graph and churn metrics.""" + + name = "route-map" + help = "Map how stable mesh routing is over time and render an interactive graph." + category = "Optimization" + order = 30 + + async def prompt_params(self, ctx: AppContext) -> Optional[dict[str, Any]]: + """Choose a target (or the whole mesh) and an optional fresh-sample count. + + Args: + ctx: Shared application context. + + Returns: + A parameter dict, or ``None`` if the user cancelled. + """ + targets = ctx.repo.traced_targets() + choice = await questionary.select( + "Map routing for:", + choices=[_WHOLE_MESH, *targets] if targets else [_WHOLE_MESH], + ).ask_async() + if choice is None: + return None + samples = await questionary.text( + "Take how many fresh traces first? (0 = use stored history only)", + default="0", + validate=_is_nonneg_int, + ).ask_async() + if samples is None: + return None + params: dict[str, Any] = {"samples": int(samples)} + if choice != _WHOLE_MESH: + params["target"] = choice + return params + + async def run(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult: + """Optionally trace, load history, build the graph, render it, and summarize. + + Args: + ctx: Shared application context. + params: optional ``target``/``samples``/``path``/``viz`` and ``_run_id``. + + Returns: + A :class:`ToolResult` with the churn metrics and the graph path. + """ + target = params.get("target") + run_id = params["_run_id"] + samples = int(params.get("samples", 0)) + resolve = None + + if samples > 0 and target: + device = await ctx.device() + contacts = await device.get_contacts() + resolve = trace_runner.make_node_resolver(contacts) + path = ( + trace_runner.parse_trace_path(params["path"], contacts) + if params.get("path") + else None + ) + with make_progress(ctx.console) as progress: + task = progress.add_task(f"tracing {target}", total=samples) + await trace_runner.run_traces( + device, target, samples=samples, path=path, + cooldown_s=ctx.settings.trace_cooldown_s, + on_result=lambda *_: progress.advance(task), + persist=lambda t: ctx.repo.record_trace(run_id, t), + ) + + traces = ctx.repo.recent_traces(target) if target else ctx.repo.all_traces() + if resolve is None: + try: + contacts = await (await ctx.device()).get_contacts() + resolve = trace_runner.make_node_resolver(contacts) + except Exception: # noqa: BLE001 - the graph reads fine with raw hashes + resolve = None + + graph = route_stability.build_route_graph(traces, target=target, resolve=resolve) + + ctx.console.print(_graph_panel(graph)) + + artifacts: list[str] = [] + if graph.total_routes and params.get("viz", True): + assert ctx.settings.output_dir is not None + out = render_route_graph(graph, ctx.settings.output_dir) + artifacts.append(str(out)) + + scope = target or "the mesh" + if not graph.total_routes: + message = ( + f"[muted]no successful traces stored for[/muted] [brand]{scope}[/brand] — " + "run some traces first." + ) + else: + message = ( + f"[ok]✓[/ok] [brand]{scope}[/brand]: {graph.stability:.0%} of " + f"{graph.total_routes} traces took the top route " + f"({graph.distinct_routes} distinct)" + ) + + return ToolResult( + summary={ + "target": target, + "total_routes": graph.total_routes, + "distinct_routes": graph.distinct_routes, + "stability": round(graph.stability, 3), + "churn_entropy_bits": round(graph.churn_entropy, 3), + "nodes": len(graph.nodes), + "links": len(graph.links), + }, + message=message, + artifacts=artifacts, + ) + + def register_cli(self, app: typer.Typer) -> None: + """Register the ``route-map`` subcommand. + + Args: + app: The Typer application. + """ + from ..cli import run_tool_command + + @app.command(name=self.name, help=self.help) + def _route_map( + target: Optional[str] = typer.Option( + None, "--target", "-t", help="Target to map (omit for the whole mesh)." + ), + samples: int = typer.Option( + 0, "--samples", "-n", help="Fresh traces to take first (requires --target)." + ), + path: Optional[str] = typer.Option( + None, "--path", "-p", help="Force a path for the fresh traces." + ), + viz: bool = typer.Option(True, "--viz/--no-viz", help="Render the HTML graph."), + ) -> None: + tool_params: dict[str, Any] = {"samples": samples, "viz": viz} + if target: + tool_params["target"] = target + if path: + tool_params["path"] = path + run_tool_command(self, tool_params) + + +def _graph_panel(graph: RouteGraph) -> Panel: + """Render churn metrics and the busiest links as a panel. + + Args: + graph: The aggregated route graph. + + Returns: + A Rich :class:`Panel` summarizing stability and top links. + """ + scope = graph.target or "whole mesh" + stab_style = "ok" if graph.stability >= 0.8 else ("warn" if graph.stability >= 0.5 else "err") + header = Text.assemble( + ("scope ", "muted"), (f"{scope}\n", "brand"), + ("traces ", "muted"), (f"{graph.total_routes}\n", ""), + ("distinct routes ", "muted"), (f"{graph.distinct_routes}\n", ""), + ("stability ", "muted"), (f"{graph.stability:.0%}", stab_style), ("\n", ""), + ("route entropy ", "muted"), (f"{graph.churn_entropy:.2f} bits", ""), + ) + if not graph.links: + return Panel( + header, title="[accent]route stability[/accent]", border_style="muted", expand=False + ) + + table = Table(box=None, padding=(0, 2, 0, 0), expand=False) + table.add_column("link") + table.add_column("uses", justify="right") + table.add_column("median SNR", justify="right") + for link in graph.links[:10]: + table.add_row( + f"{link.origin} → {link.destination}", + str(link.count), + Text(f"{link.median_snr:+.1f} dB", style=snr_style(link.median_snr)), + ) + + from rich.console import Group + + body = Group(header, Text("\nbusiest links", style="muted"), table) + return Panel( + body, title="[accent]route stability[/accent]", border_style="accent", expand=False + ) + + +def _is_nonneg_int(value: str) -> bool | str: + """Validate a non-negative integer for a questionary text answer. + + Args: + value: Raw input. + + Returns: + ``True`` if valid, otherwise an error message string. + """ + try: + return True if int(value) >= 0 else "Enter zero or a positive number." + except ValueError: + return "Enter a whole number." diff --git a/meshtools/viz/route_graph.py b/meshtools/viz/route_graph.py new file mode 100644 index 0000000..a31c9cf --- /dev/null +++ b/meshtools/viz/route_graph.py @@ -0,0 +1,130 @@ +"""pyvis renderer for the route-stability graph. + +Draws the aggregated mesh routing as a self-contained, interactive HTML graph: nodes are +mesh devices (our companion highlighted), directed edges are links whose width grows with +how often the link was traversed and whose colour tracks median SNR. The title carries the +churn metrics so the picture and the numbers travel together. +""" + +from __future__ import annotations + +from datetime import datetime +from pathlib import Path + +from ..services.route_stability import RouteGraph + +_SELF_COLOR = "#4ade80" +_NODE_COLOR = "#5eead4" +_BG = "#0f172a" +_FONT = "#e2e8f0" + + +def render_route_graph(graph: RouteGraph, output_dir: Path) -> Path: + """Render a route graph to an interactive, self-contained HTML file. + + Args: + graph: The aggregated route graph to draw. + output_dir: Directory the HTML file is written to (created if needed). + + Returns: + Path to the written HTML file. + """ + from pyvis.network import Network + + # ``cdn_resources='in_line'`` inlines vis.js so the file works offline, matching the + # self-contained Plotly charts elsewhere. + net = Network( + height="720px", + width="100%", + directed=True, + bgcolor=_BG, + font_color=_FONT, + cdn_resources="in_line", + ) + + for node in graph.nodes: + color = _SELF_COLOR if node.is_self else _NODE_COLOR + net.add_node( + node.label, + label=node.label, + color=color, + shape="star" if node.is_self else "dot", + size=14 + 3 * node.degree, + title=f"{node.label} · {node.degree} link(s)", + ) + + max_count = max((link.count for link in graph.links), default=1) + for link in graph.links: + width = 1.0 + 6.0 * (link.count / max_count) + net.add_edge( + link.origin, + link.destination, + value=link.count, + width=width, + color=_snr_color(link.median_snr), + title=f"{link.origin} → {link.destination}: " + f"{link.count}× · median {link.median_snr:+.1f} dB", + ) + + net.set_options(_OPTIONS) + + scope = graph.target or "mesh" + heading = ( + f"Route stability — {scope} · {graph.total_routes} traces, " + f"{graph.distinct_routes} route(s), {graph.stability:.0%} on the top route" + ) + output_dir.mkdir(parents=True, exist_ok=True) + stamp = datetime.now().strftime("%Y%m%d-%H%M%S") + path = output_dir / f"route-map_{_slug(scope)}_{stamp}.html" + html = net.generate_html(notebook=False) + banner = f"

{heading}

" + html = html.replace("", f"{banner}") + path.write_text(html, encoding="utf-8") + return path + + +def _snr_color(snr: float) -> str: + """Map a median SNR (dB) to an edge colour from red (weak) to green (strong). + + Args: + snr: Median SNR in dB. + + Returns: + A hex colour string. + """ + if snr >= 5: + return "#4ade80" # strong + if snr >= 0: + return "#facc15" # marginal + if snr >= -8: + return "#fb923c" # weak + return "#f87171" # very weak + + +def _slug(value: str) -> str: + """Make a filesystem-safe slug from a label. + + Args: + value: Arbitrary node name or scope label. + + Returns: + A lowercase slug containing only alphanumerics, dashes, and underscores. + """ + return "".join(c if c.isalnum() or c in "-_" else "-" for c in value).strip("-").lower() + + +#: vis.js physics/layout tuning for a readable, settled mesh graph. +_OPTIONS = """ +{ + "physics": { + "solver": "forceAtlas2Based", + "forceAtlas2Based": {"gravitationalConstant": -60, "springLength": 120}, + "stabilization": {"iterations": 200} + }, + "edges": { + "arrows": {"to": {"enabled": true, "scaleFactor": 0.6}}, + "smooth": {"type": "curvedCW", "roundness": 0.15} + }, + "nodes": {"font": {"color": "#e2e8f0", "size": 16}} +} +""" diff --git a/tests/test_route_stability.py b/tests/test_route_stability.py new file mode 100644 index 0000000..c45d4fa --- /dev/null +++ b/tests/test_route_stability.py @@ -0,0 +1,83 @@ +"""Tests for route-stability aggregation and the pyvis renderer.""" + +from __future__ import annotations + +from pathlib import Path + +from meshtools.core.models import Hop, TraceResult +from meshtools.services import route_stability + + +def _trace(*nodes: str) -> TraceResult: + """Build a successful trace whose forward path visits ``nodes`` then returns to us. + + Each hop SNR is fixed; the final hash-less hop models the reply returning to us. + """ + hops = [Hop(i, node, 5.0) for i, node in enumerate(nodes)] + hops.append(Hop(len(nodes), None, 5.0)) + return TraceResult(target=nodes[-1] if nodes else "x", success=True, hops=hops) + + +def test_stable_route_scores_one() -> None: + """A single route taken every time scores full stability and zero entropy.""" + traces = [_trace("3d", "f2") for _ in range(6)] + graph = route_stability.build_route_graph(traces, target="f2") + assert graph.total_routes == 6 + assert graph.distinct_routes == 1 + assert graph.stability == 1.0 + assert graph.churn_entropy == 0.0 + + +def test_flapping_routes_lower_stability_and_raise_entropy() -> None: + """An even split between two routes halves stability and gives 1 bit of entropy.""" + traces = [_trace("3d", "f2") for _ in range(3)] + [_trace("aa", "f2") for _ in range(3)] + graph = route_stability.build_route_graph(traces, target="f2") + assert graph.distinct_routes == 2 + assert graph.stability == 0.5 + assert graph.churn_entropy == 1.0 # two equally-likely routes + + +def test_links_aggregate_counts_and_self_node() -> None: + """Links carry traversal counts and our own device is marked is_self.""" + traces = [_trace("3d", "f2") for _ in range(4)] + graph = route_stability.build_route_graph(traces, target="f2", device_label="us") + by_pair = {(link.origin, link.destination): link for link in graph.links} + assert by_pair[("us", "3d")].count == 4 # every trace starts us → 3d + assert by_pair[("f2", "us")].count == 4 # and returns f2 → us + self_nodes = [n for n in graph.nodes if n.is_self] + assert len(self_nodes) == 1 and self_nodes[0].label == "us" + + +def test_resolver_relabels_nodes() -> None: + """A resolver renames hop hashes throughout the graph.""" + traces = [_trace("3d", "f2") for _ in range(3)] + graph = route_stability.build_route_graph( + traces, target="f2", resolve=lambda h: {"3d": "Repeater", "f2": "Alice"}.get(h, h) + ) + labels = {n.label for n in graph.nodes} + assert "Repeater" in labels + assert "Alice" in labels + + +def test_empty_history_is_safe() -> None: + """Failed/empty traces yield an empty graph with zeroed metrics, no errors.""" + graph = route_stability.build_route_graph( + [TraceResult(target="x", success=False)], target="x" + ) + assert graph.total_routes == 0 + assert graph.stability == 0.0 + assert graph.links == [] + + +def test_render_route_graph_writes_html(tmp_path: Path) -> None: + """The pyvis renderer produces a self-contained HTML file.""" + from meshtools.viz.route_graph import render_route_graph + + traces = [_trace("3d", "f2") for _ in range(3)] + [_trace("aa", "f2")] + graph = route_stability.build_route_graph(traces, target="f2") + out = render_route_graph(graph, tmp_path) + assert out.exists() + assert out.suffix == ".html" + text = out.read_text(encoding="utf-8") + assert "Route stability" in text # heading injected + assert out.stat().st_size > 1000 # vis.js inlined