From eba264385b4f00040bae3f692921e7fdacb475b7 Mon Sep 17 00:00:00 2001 From: Jean-Pierre Martineau Date: Sun, 21 Jun 2026 15:48:17 -0400 Subject: [PATCH] Add passive mesh monitor (longitudinal advert/telemetry logger) Adds a Device.listen capability (real + mock), an Observation/HeardNode domain model, an observations table with Repository.record_observation / heard_nodes aggregation, and a `monitor` tool that listens for a window and logs every overheard advert/telemetry (SNR, RSSI, shared location) to the database. This is the longitudinal substrate later features read back. The real-hardware event-payload mapping is best-effort and, like the trace mapping, flagged for validation against the firmware's event schema; the mock device synthesizes plausible adverts so the tool and tests run without a radio. Co-Authored-By: Claude Opus 4.8 --- README.md | 1 + meshtools/core/connection.py | 171 ++++++++++++++++++++++- meshtools/core/models.py | 96 +++++++++++++ meshtools/persistence/db.py | 18 ++- meshtools/persistence/repository.py | 72 +++++++++- meshtools/tools/monitor.py | 202 ++++++++++++++++++++++++++++ tests/test_monitor.py | 115 ++++++++++++++++ 7 files changed, 671 insertions(+), 4 deletions(-) create mode 100644 meshtools/tools/monitor.py create mode 100644 tests/test_monitor.py diff --git a/README.md b/README.md index f9719a2..992d197 100644 --- a/README.md +++ b/README.md @@ -18,6 +18,7 @@ turned into interactive HTML visualizations. | **History / DB** | ✅ skeleton | Every tool execution and measurement is persisted and queryable. | | **TX-power optimization** | ✅ working | Sweep transmit power (coarse + refine), converge on the best signal, optionally apply, render an interactive chart. | | **Link budget** | ✅ working | Predict time-on-air, receiver sensitivity, range, and duty-cycle/dwell headroom for a radio config — offline, no radio needed. | +| **Passive monitor** | ✅ working | Listen on the air and log every overheard advert/telemetry (SNR, RSSI, location) to a longitudinal history. | | **Path optimization** | 🚧 planned | Discover and score mesh paths from neighbors outward. | | **Visualizations** | 🟡 partial | Plotly TX-sweep chart (HTML) done; pyvis mesh graph + time-series planned. | diff --git a/meshtools/core/connection.py b/meshtools/core/connection.py index a49b6a5..698de5c 100644 --- a/meshtools/core/connection.py +++ b/meshtools/core/connection.py @@ -16,9 +16,12 @@ from __future__ import annotations import asyncio import random from abc import ABC, abstractmethod -from typing import Optional +from typing import Callable, Optional -from .models import Contact, Hop, TraceResult +from .models import Contact, Hop, Observation, TraceResult + +#: Callback invoked with each :class:`Observation` as passive monitoring captures it. +ObservationCallback = Callable[[Observation], None] TX_POWER_MIN = 1 TX_POWER_MAX = 22 @@ -142,6 +145,31 @@ class Device(ABC): A :class:`TraceResult`; ``success`` is ``False`` on timeout. """ + # -- passive monitoring ------------------------------------------------------ + + @abstractmethod + async def listen( + self, + duration_s: float, + *, + on_observation: Optional["ObservationCallback"] = None, + ) -> list[Observation]: + """Passively capture packets the companion overhears for ``duration_s`` seconds. + + Adverts (and, where the firmware surfaces them, telemetry frames) heard on the + air are returned as :class:`~meshtools.core.models.Observation` records with + whatever SNR/RSSI/location the radio reported. The radio is not asked to transmit; + it simply listens. + + Args: + duration_s: How long to listen, in seconds. + on_observation: Optional callback invoked with each observation as it arrives + (e.g. to advance a live progress display). + + Returns: + Every observation captured during the window, in arrival order. + """ + # -- configuration: extra reads --------------------------------------------- @abstractmethod @@ -587,6 +615,60 @@ class MeshCoreDevice(Device): return b"".join(hops), path_hash_flags(trace_size) or 0 return None + async def listen( # noqa: D102 - inherited docstring + self, + duration_s: float, + *, + on_observation: Optional[ObservationCallback] = None, + ) -> list[Observation]: + from meshcore import EventType + + mc = self._require() + subscribe = getattr(mc, "subscribe", None) + if subscribe is None: # pragma: no cover - depends on installed meshcore build + raise DeviceCommandError( + "this meshcore build doesn't expose event subscription, so passive " + "monitoring isn't available. Upgrade the 'meshcore' library." + ) + + observations: list[Observation] = [] + + def make_handler(kind: str): # type: ignore[no-untyped-def] + def handler(event) -> None: # noqa: ANN001 + obs = observation_from_event(event, kind) + if obs is None: + return + observations.append(obs) + if on_observation is not None: + on_observation(obs) + + return handler + + # Subscribe to whichever advert/telemetry event types this firmware/library build + # exposes. The event payload field names below are best-effort and, like the trace + # mapping, should be validated against your firmware's event schema. + subs = [] + for attr, kind in ( + ("ADVERTISEMENT", "advert"), + ("ADVERT", "advert"), + ("NEW_CONTACT", "advert"), + ("TELEMETRY_RESPONSE", "telemetry"), + ): + etype = getattr(EventType, attr, None) + if etype is not None: + subs.append(subscribe(etype, make_handler(kind))) + try: + await asyncio.sleep(duration_s) + finally: + for sub in subs: + unsub = getattr(sub, "unsubscribe", None) + if unsub is not None: + try: + unsub() + except Exception: # noqa: BLE001 - best-effort cleanup + pass + return observations + @staticmethod def _ok(event): # type: ignore[no-untyped-def] """Return ``event`` if it succeeded, else raise its error payload. @@ -905,6 +987,42 @@ class MockDevice(Device): self._custom_vars.clear() self._channels.clear() + async def listen( # noqa: D102 - inherited docstring + self, + duration_s: float, + *, + on_observation: Optional[ObservationCallback] = None, + ) -> list[Observation]: + # Synthesize plausible adverts from the known contacts, cycling through them. + # Two of them ("…Repeater") advertise a fixed location so location-aware features + # (e.g. the coverage map) have data. The wall-clock duration is honored only + # loosely — the simulator emits a count proportional to it and paces minimally so + # tests stay fast. + repeats = max(1, round(duration_s / 5.0)) + locations = { + "Yagi-Repeater": (45.5019, -73.5674), + "Local-Repeater": (45.4768, -73.5990), + } + observations: list[Observation] = [] + for r in range(repeats): + for c in self._contacts: + kind = "telemetry" if (r + len(observations)) % 4 == 3 else "advert" + lat_lon = locations.get(c.name) + obs = Observation( + node=c.key_prefix or c.public_key[:12], + name=c.name, + kind=kind, + snr=round(self._rng.gauss(6.0, 3.0), 1), + rssi=round(self._rng.gauss(-95.0, 8.0), 1), + lat=lat_lon[0] if lat_lon else None, + lon=lat_lon[1] if lat_lon else None, + ) + observations.append(obs) + if on_observation is not None: + on_observation(obs) + await asyncio.sleep(0) + return observations + def _expected_snr(self, hop_index: int) -> float: """Model SNR for a hop as an inverted-U in TX power plus distance falloff. @@ -1010,6 +1128,55 @@ def parse_trace_hops(payload: dict) -> list[Hop]: return hops +def observation_from_event(event, kind: str) -> Optional[Observation]: # noqa: ANN001 + """Map a meshcore advert/telemetry event into an :class:`Observation`. + + The companion reports a node identifier, optionally a name and shared location, and + the SNR/RSSI of the reception. Field names vary across firmware and library versions, + so several common spellings are tried for each value. Like the trace mapping this is + best-effort and should be validated against your firmware's event schema. + + Args: + event: A meshcore event (anything exposing a ``payload`` mapping). + kind: The observation class to tag the record with (e.g. ``advert``). + + Returns: + The parsed :class:`Observation`, or ``None`` if the payload carried no node id. + """ + payload = dict(getattr(event, "payload", {}) or {}) + node = ( + payload.get("public_key") + or payload.get("pubkey") + or payload.get("hash") + or payload.get("key_prefix") + ) + if not node: + return None + node = str(node).lower().removeprefix("0x")[:12] + lat = payload.get("adv_lat", payload.get("lat")) + lon = payload.get("adv_lon", payload.get("lon")) + return Observation( + node=node, + name=payload.get("adv_name") or payload.get("name"), + kind=kind, + snr=_as_float(payload.get("snr")), + rssi=_as_float(payload.get("rssi")), + lat=_as_float(lat) if lat else None, + lon=_as_float(lon) if lon else None, + raw=payload, + ) + + +def _as_float(value: object) -> Optional[float]: + """Best-effort float conversion, returning ``None`` on missing/garbage values.""" + if value is None: + return None + try: + return float(value) # type: ignore[arg-type] + except (TypeError, ValueError): + return None + + def _mock_pub(prefix: str) -> str: """Build a 32-byte mock public key from a short hex prefix (simulator only). diff --git a/meshtools/core/models.py b/meshtools/core/models.py index a7d7391..4987547 100644 --- a/meshtools/core/models.py +++ b/meshtools/core/models.py @@ -42,6 +42,102 @@ class Contact: last_seen: Optional[datetime] = None +@dataclass(slots=True) +class Observation: + """A single over-the-air reception heard while passively monitoring the mesh. + + Adverts, telemetry frames, and other packets the companion overhears are captured as + observations and logged longitudinally, so the mesh's behavior can be reviewed over + time rather than only at the instant a command runs. + + Attributes: + node: Key prefix / hash identifying the transmitting node, if known. + name: Friendly name advertised by the node, if carried. + kind: Packet class, e.g. ``advert`` or ``telemetry``. + snr: Signal-to-noise ratio (dB) the companion measured, if reported. + rssi: Received signal strength (dBm), if reported. + lat: Advertised latitude (decimal degrees), when the node shares location. + lon: Advertised longitude (decimal degrees), when the node shares location. + observed_at: When the packet was heard. + raw: Optional raw event payload for debugging/replay. + """ + + node: Optional[str] + name: Optional[str] = None + kind: str = "advert" + snr: Optional[float] = None + rssi: Optional[float] = None + lat: Optional[float] = None + lon: Optional[float] = None + observed_at: datetime = field(default_factory=utcnow) + raw: Optional[dict] = None + + +@dataclass(slots=True) +class HeardNode: + """Aggregated reception statistics for one node across many observations. + + Attributes: + node: Key prefix / hash of the node (``None`` only if never identified). + name: Most recent friendly name seen for the node, if any. + count: Number of observations aggregated. + median_snr: Median SNR (dB) across observations that reported one. + best_snr: Strongest SNR (dB) seen, if any. + last_rssi: Most recent RSSI (dBm), if any. + last_seen: Timestamp of the most recent observation. + lat: Most recent advertised latitude, if the node shared one. + lon: Most recent advertised longitude, if the node shared one. + """ + + node: Optional[str] + name: Optional[str] + count: int + median_snr: Optional[float] + best_snr: Optional[float] + last_rssi: Optional[float] + last_seen: datetime + lat: Optional[float] = None + lon: Optional[float] = None + + @property + def has_location(self) -> bool: + """Whether this node reported a usable latitude/longitude.""" + return self.lat is not None and self.lon is not None + + @classmethod + def from_observations( + cls, node: Optional[str], observations: list["Observation"] + ) -> "HeardNode": + """Aggregate one node's observations into reception statistics. + + Args: + node: The node identifier these observations belong to. + observations: The observations for ``node`` (must be non-empty). + + Returns: + A :class:`HeardNode` summarizing them. The most recent observation supplies + the name, RSSI, and location; SNR is summarized robustly (median + best). + """ + ordered = sorted(observations, key=lambda o: o.observed_at) + latest = ordered[-1] + snrs = [o.snr for o in ordered if o.snr is not None] + located = next( + (o for o in reversed(ordered) if o.lat is not None and o.lon is not None), None + ) + name = next((o.name for o in reversed(ordered) if o.name), None) + return cls( + node=node, + name=name, + count=len(ordered), + median_snr=statistics.median(snrs) if snrs else None, + best_snr=max(snrs) if snrs else None, + last_rssi=latest.rssi, + last_seen=latest.observed_at, + lat=located.lat if located else None, + lon=located.lon if located else None, + ) + + @dataclass(slots=True) class Hop: """A single hop in a path trace. diff --git a/meshtools/persistence/db.py b/meshtools/persistence/db.py index 15e1da3..d8d0774 100644 --- a/meshtools/persistence/db.py +++ b/meshtools/persistence/db.py @@ -10,7 +10,7 @@ from __future__ import annotations import sqlite3 from pathlib import Path -SCHEMA_VERSION = 1 +SCHEMA_VERSION = 2 _SCHEMA = """ CREATE TABLE IF NOT EXISTS schema_meta ( @@ -76,9 +76,25 @@ CREATE TABLE IF NOT EXISTS path_candidates ( created_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, + run_id INTEGER NOT NULL REFERENCES runs(id) ON DELETE CASCADE, + node TEXT, + name TEXT, + kind TEXT NOT NULL DEFAULT 'advert', + snr REAL, + rssi REAL, + lat REAL, + lon REAL, + observed_at TEXT NOT NULL +); + 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_observations_run ON observations(run_id); +CREATE INDEX IF NOT EXISTS idx_observations_node ON observations(node); """ diff --git a/meshtools/persistence/repository.py b/meshtools/persistence/repository.py index 147f387..b56155b 100644 --- a/meshtools/persistence/repository.py +++ b/meshtools/persistence/repository.py @@ -14,7 +14,14 @@ from typing import Any, Optional from datetime import datetime -from ..core.models import Hop, TraceResult, TxLevelResult, utcnow +from ..core.models import ( + HeardNode, + Hop, + Observation, + TraceResult, + TxLevelResult, + utcnow, +) from . import db @@ -232,3 +239,66 @@ class Repository: ), ) self._conn.commit() + + # -- observations (passive monitoring) -------------------------------------- + + def record_observation(self, run_id: int, obs: Observation) -> None: + """Persist one overheard packet from a monitoring run. + + Args: + run_id: The owning run. + obs: The observation to store. + """ + self._conn.execute( + "INSERT INTO observations " + "(run_id, node, name, kind, snr, rssi, lat, lon, observed_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + ( + run_id, + obs.node, + obs.name, + obs.kind, + obs.snr, + obs.rssi, + obs.lat, + obs.lon, + obs.observed_at.isoformat(), + ), + ) + self._conn.commit() + + def heard_nodes(self, *, since: Optional[datetime] = None) -> list[HeardNode]: + """Aggregate stored observations into per-node reception statistics. + + Spans every monitoring run (optionally limited to recent history), so the result + is a longitudinal view of which nodes have been heard, how strongly, and where — + the substrate for the monitor summary and the coverage map. + + Args: + since: Only include observations at or after this time, if given. + + Returns: + One :class:`HeardNode` per distinct node, ordered by most-recently heard. + """ + sql = "SELECT node, name, snr, rssi, lat, lon, observed_at FROM observations" + params: list[Any] = [] + if since is not None: + sql += " WHERE observed_at >= ?" + params.append(since.isoformat()) + rows = self._conn.execute(sql, params).fetchall() + + grouped: dict[Optional[str], list[Observation]] = {} + for row in rows: + obs = Observation( + node=row["node"], + name=row["name"], + snr=row["snr"], + rssi=row["rssi"], + lat=row["lat"], + lon=row["lon"], + observed_at=datetime.fromisoformat(row["observed_at"]), + ) + grouped.setdefault(row["node"], []).append(obs) + + nodes = [HeardNode.from_observations(node, obs) for node, obs in grouped.items()] + return sorted(nodes, key=lambda n: n.last_seen, reverse=True) diff --git a/meshtools/tools/monitor.py b/meshtools/tools/monitor.py new file mode 100644 index 0000000..4f54833 --- /dev/null +++ b/meshtools/tools/monitor.py @@ -0,0 +1,202 @@ +"""The ``monitor`` tool: passively log adverts/telemetry the mesh emits over time. + +Standard MeshCore tooling is point-in-time. This tool sits on the companion radio and +records every advert and telemetry frame it overhears — with SNR, RSSI, and any shared +location — to the database, building the longitudinal history that the coverage map and +link-quality alerting read back. It transmits nothing; it only listens. +""" + +from __future__ import annotations + +from typing import Any, Optional + +import questionary +import typer +from rich.table import Table +from rich.text import Text + +from ..context import AppContext +from ..core.models import HeardNode, Observation +from ..services import trace_runner +from ..ui.widgets import make_progress +from .base import Tool, ToolResult, register + +#: Cap a single listening window so an interactive run can't block indefinitely; longer +#: passive logging is still possible by re-running or scripting the subcommand. +MAX_DURATION_S = 3600 + + +@register +class MonitorTool(Tool): + """Listen for a window and log every overheard advert/telemetry to the database.""" + + name = "monitor" + help = "Passively log adverts/telemetry heard on the mesh (longitudinal history)." + category = "Diagnostics" + order = 20 + + async def prompt_params(self, ctx: AppContext) -> Optional[dict[str, Any]]: + """Ask how long to listen. + + Args: + ctx: Shared application context. + + Returns: + A parameter dict, or ``None`` if the user cancelled. + """ + duration = await questionary.text( + f"Listen for how many seconds? (1-{MAX_DURATION_S})", + default="60", + validate=_is_valid_duration, + ).ask_async() + if duration is None: + return None + return {"duration": int(duration)} + + async def run(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult: + """Capture observations for the window, persist them, and summarize heard nodes. + + Args: + ctx: Shared application context. + params: ``duration`` (seconds) and the injected ``_run_id``. + + Returns: + A :class:`ToolResult` noting how many nodes/packets were heard. + """ + duration = max(1, min(MAX_DURATION_S, int(params.get("duration", 60)))) + run_id = params["_run_id"] + device = await ctx.device() + + # Resolve node hashes to friendly names where we know the contact. + contacts = await device.get_contacts() + resolve = trace_runner.make_node_resolver(contacts) + + observations: list[Observation] = [] + with make_progress(ctx.console) as progress: + task = progress.add_task(f"listening {duration}s", total=duration) + + def on_observation(obs: Observation) -> None: + observations.append(obs) + ctx.repo.record_observation(run_id, obs) + progress.update( + task, + description=f"heard {len(observations)} pkts · " + f"{_distinct(observations)} nodes", + ) + + await device.listen(duration, on_observation=on_observation) + progress.update(task, completed=duration) + + heard = _aggregate(observations) + if heard: + ctx.console.print(_heard_table(heard, resolve)) + else: + ctx.console.print("[muted]no packets heard during the window[/muted]") + + located = sum(1 for n in heard if n.has_location) + return ToolResult( + summary={ + "duration_s": duration, + "observations": len(observations), + "nodes_heard": len(heard), + "nodes_with_location": located, + }, + message=( + f"[ok]✓[/ok] heard [brand]{len(observations)}[/brand] packets from " + f"[brand]{len(heard)}[/brand] nodes in {duration}s" + ), + ) + + def register_cli(self, app: typer.Typer) -> None: + """Register the ``monitor`` subcommand. + + Args: + app: The Typer application. + """ + from ..cli import run_tool_command + + @app.command(name=self.name, help=self.help) + def _monitor( + duration: int = typer.Option( + 60, "--duration", "-d", help=f"Seconds to listen (max {MAX_DURATION_S})." + ), + ) -> None: + run_tool_command(self, {"duration": duration}) + + +def _distinct(observations: list[Observation]) -> int: + """Return the number of distinct nodes among observations.""" + return len({o.node for o in observations}) + + +def _aggregate(observations: list[Observation]) -> list[HeardNode]: + """Group observations by node into :class:`HeardNode` records, freshest first. + + Args: + observations: The observations captured this run. + + Returns: + One aggregate per node, ordered by most-recently heard. + """ + grouped: dict[Optional[str], list[Observation]] = {} + for obs in observations: + grouped.setdefault(obs.node, []).append(obs) + nodes = [HeardNode.from_observations(node, obs) for node, obs in grouped.items()] + return sorted(nodes, key=lambda n: n.last_seen, reverse=True) + + +def _heard_table(heard: list[HeardNode], resolve) -> Table: # noqa: ANN001 + """Render the heard-node summary table. + + Args: + heard: Aggregated per-node statistics. + resolve: Maps a raw node hash to a friendly contact name when known. + + Returns: + A Rich :class:`Table` of node, packet count, SNR, RSSI, and location. + """ + from ..ui.theme import snr_style + + table = Table(title=f"Heard nodes ({len(heard)})", border_style="muted", expand=False) + table.add_column("Node") + table.add_column("Pkts", justify="right") + table.add_column("Median SNR", justify="right") + table.add_column("Best SNR", justify="right") + table.add_column("RSSI", justify="right") + table.add_column("Loc", justify="center") + for node in heard: + label = node.name or resolve(node.node) or node.node or "?" + median = node.median_snr + best = node.best_snr + median_cell = ( + Text(f"{median:+.1f}", style=snr_style(median)) if median is not None else Text("—") + ) + best_cell = ( + Text(f"{best:+.1f}", style=snr_style(best)) if best is not None else Text("—") + ) + rssi_cell = f"{node.last_rssi:.0f}" if node.last_rssi is not None else "—" + loc_cell = Text("●", style="ok") if node.has_location else Text("·", style="muted") + table.add_row( + str(label), str(node.count), median_cell, best_cell, rssi_cell, loc_cell + ) + return table + + +def _is_valid_duration(value: str) -> bool | str: + """Validate a listening duration in ``1..MAX_DURATION_S`` for a text answer. + + Args: + value: Raw input. + + Returns: + ``True`` if valid, otherwise an error message string. + """ + try: + seconds = int(value) + except ValueError: + return "Enter a whole number of seconds." + if seconds < 1: + return "Enter a number greater than zero." + if seconds > MAX_DURATION_S: + return f"Maximum {MAX_DURATION_S} seconds." + return True diff --git a/tests/test_monitor.py b/tests/test_monitor.py new file mode 100644 index 0000000..9b7f0e2 --- /dev/null +++ b/tests/test_monitor.py @@ -0,0 +1,115 @@ +"""Tests for the passive mesh monitor: listening, aggregation, and persistence. + +All run against the :class:`MockDevice` simulator and an on-disk SQLite database, no +hardware required. +""" + +from __future__ import annotations + +from datetime import timedelta +from pathlib import Path + +from meshtools.core.connection import MockDevice, observation_from_event +from meshtools.core.models import HeardNode, Observation, utcnow +from meshtools.persistence.repository import Repository + + +async def test_mock_listen_emits_observations_and_callbacks() -> None: + """Listening yields observations and fires the per-observation callback.""" + device = MockDevice() + await device.connect() + + seen: list[Observation] = [] + observations = await device.listen(0.2, on_observation=seen.append) + + assert observations # the simulator advertises its known contacts + assert seen == observations # every observation was reported via the callback + assert all(o.node for o in observations) + # At least one node advertises a location (for the coverage map). + assert any(o.lat is not None and o.lon is not None for o in observations) + + +def test_heard_node_aggregation() -> None: + """from_observations summarizes count, robust SNR, latest RSSI, and location.""" + now = utcnow() + obs = [ + Observation(node="a1", name="Yagi", snr=4.0, rssi=-100.0, observed_at=now), + Observation( + node="a1", name="Yagi", snr=8.0, rssi=-90.0, lat=45.5, lon=-73.5, + observed_at=now + timedelta(seconds=5), + ), + Observation( + node="a1", name="Yagi", snr=6.0, rssi=-95.0, observed_at=now + timedelta(seconds=2) + ), + ] + node = HeardNode.from_observations("a1", obs) + assert node.count == 3 + assert node.median_snr == 6.0 # median of [4, 8, 6] + assert node.best_snr == 8.0 + assert node.last_seen == now + timedelta(seconds=5) # freshest observation + assert node.last_rssi == -90.0 # RSSI from that freshest observation + assert node.has_location # location carried by one observation is retained + + +def test_heard_node_latest_wins_for_rssi_and_location() -> None: + """The freshest observation supplies RSSI; the freshest *located* one supplies coords.""" + now = utcnow() + obs = [ + Observation(node="b2", snr=1.0, rssi=-80.0, lat=10.0, lon=20.0, observed_at=now), + Observation(node="b2", snr=2.0, rssi=-70.0, observed_at=now + timedelta(seconds=10)), + ] + node = HeardNode.from_observations("b2", obs) + assert node.last_rssi == -70.0 # latest reading + assert (node.lat, node.lon) == (10.0, 20.0) # latest one that actually had a fix + + +def test_observation_from_event_parses_advert() -> None: + """The event parser pulls node id, name, SNR/RSSI, and location from a payload.""" + + class _Event: + payload = { + "public_key": "AABBCC0011", + "adv_name": "Repeater", + "snr": "7.5", + "rssi": -92, + "adv_lat": 45.5, + "adv_lon": -73.6, + } + + obs = observation_from_event(_Event(), "advert") + assert obs is not None + assert obs.node == "aabbcc0011" + assert obs.name == "Repeater" + assert obs.snr == 7.5 + assert obs.rssi == -92.0 + assert (obs.lat, obs.lon) == (45.5, -73.6) + + +def test_observation_from_event_without_node_is_skipped() -> None: + """A payload carrying no node identifier yields no observation.""" + + class _Event: + payload = {"snr": 3.0} + + assert observation_from_event(_Event(), "advert") is None + + +async def test_observations_round_trip_and_aggregate(tmp_path: Path) -> None: + """Observations persist and heard_nodes reaggregates them across runs.""" + repo = Repository(tmp_path / "obs.db") + run_id = repo.start_run("monitor", {"duration": 1}) + + device = MockDevice() + await device.connect() + observations = await device.listen( + 0.2, on_observation=lambda o: repo.record_observation(run_id, o) + ) + assert observations + + nodes = repo.heard_nodes() + assert nodes + # Every node aggregated should have been heard at least once. + assert sum(n.count for n in nodes) == len(observations) + # Nodes are ordered most-recently-heard first. + assert nodes == sorted(nodes, key=lambda n: n.last_seen, reverse=True) + repo.close()