diff --git a/README.md b/README.md index dc3a874..ef2e8b5 100644 --- a/README.md +++ b/README.md @@ -18,7 +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. | +| **Passive monitor** | ✅ working | Toggle a non-blocking background logger that records every overheard advert/telemetry (SNR, RSSI, location) to a longitudinal history while you keep using the app; on/off is remembered between sessions and live packet counts show in the menu. | | **Link health** | ✅ working | Compare a target's recent traces to its rolling baseline and flag SNR/reliability regressions, end-to-end and per hop. | | **Route map** | ✅ working | Aggregate trace history into an interactive route-stability graph — which links are stable, which the mesh flaps between, with churn metrics. | | **Path optimization** | 🚧 planned | Discover and score mesh paths from neighbors outward. | diff --git a/meshtools/context.py b/meshtools/context.py index 849b505..166c5da 100644 --- a/meshtools/context.py +++ b/meshtools/context.py @@ -8,7 +8,7 @@ never open a serial port. from __future__ import annotations from dataclasses import dataclass, field -from typing import Optional +from typing import TYPE_CHECKING, Optional from rich.console import Console @@ -21,6 +21,9 @@ from .core.selection import resolve_device from .persistence.logging import get_logger from .persistence.repository import Repository +if TYPE_CHECKING: + from .services.monitor_service import MonitorService + @dataclass(slots=True) class AppContext: @@ -55,12 +58,30 @@ class AppContext: selected_device: Optional[DiscoveredDevice] = None explicit_selection: bool = False _device: Optional[Device] = field(default=None, init=False, repr=False) + _monitor: "Optional[MonitorService]" = field(default=None, init=False, repr=False) @property def profile_name(self) -> Optional[str]: """Name of the active profile, if any.""" return self.profile.name if self.profile else None + @property + def monitor(self) -> "MonitorService": + """Return the session's passive-monitor service, creating it on first use. + + The service is built lazily so the (cheap) database read for the "total" + observation count and the on/off preference load happen only once, when + monitoring is first referenced. + """ + if self._monitor is None: + from .core.monitor_store import MonitorStore + from .services.monitor_service import MonitorService + + self._monitor = MonitorService( + self, MonitorStore(self.settings.config_dir / "monitor.json") + ) + return self._monitor + @property def log(self): # type: ignore[no-untyped-def] """The application logger.""" @@ -105,7 +126,9 @@ class AppContext: return self._device async def aclose(self) -> None: - """Disconnect the device (if connected) and close the repository.""" + """Stop monitoring, disconnect the device (if connected), and close the repo.""" + if self._monitor is not None: + await self._monitor.aclose() if self._device is not None: await self._device.disconnect() self._device = None diff --git a/meshtools/core/connection.py b/meshtools/core/connection.py index 698de5c..91a1f0f 100644 --- a/meshtools/core/connection.py +++ b/meshtools/core/connection.py @@ -23,6 +23,14 @@ from .models import Contact, Hop, Observation, TraceResult #: Callback invoked with each :class:`Observation` as passive monitoring captures it. ObservationCallback = Callable[[Observation], None] +#: Zero-argument callable returned by :meth:`Device.subscribe_observations` that stops the +#: subscription and releases its resources when invoked. +Unsubscribe = Callable[[], None] + +#: How often the :class:`MockDevice` simulator emits a fresh burst of synthetic packets +#: while a passive-monitor subscription is open (seconds). +_MOCK_MONITOR_INTERVAL_S = 0.05 + TX_POWER_MIN = 1 TX_POWER_MAX = 22 @@ -148,18 +156,38 @@ class Device(ABC): # -- passive monitoring ------------------------------------------------------ @abstractmethod + async def subscribe_observations( + self, on_observation: "ObservationCallback" + ) -> "Unsubscribe": + """Begin passively capturing overheard packets, without blocking the caller. + + Subscribes to the adverts (and, where the firmware surfaces them, telemetry + frames) the companion overhears and invokes ``on_observation`` with each + :class:`~meshtools.core.models.Observation` as it arrives. Capture continues in + the background until the returned callable is invoked to stop it; the radio is + never asked to transmit, it only listens. This is the primitive behind both the + always-on background monitor and the bounded :meth:`listen` helper. + + Args: + on_observation: Callback invoked with each observation as it is heard. + + Returns: + A zero-argument callable that stops capture and releases the subscription. + """ + async def listen( self, duration_s: float, *, on_observation: Optional["ObservationCallback"] = None, ) -> list[Observation]: - """Passively capture packets the companion overhears for ``duration_s`` seconds. + """Capture observations for a bounded window, then stop and return them. - 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. + A thin, foreground convenience wrapper over :meth:`subscribe_observations` for + one-shot captures (e.g. the scripted ``monitor --seconds`` command): it + subscribes, waits ``duration_s`` seconds, unsubscribes, and returns everything + heard. For continuous, non-blocking capture, use :meth:`subscribe_observations` + directly. Args: duration_s: How long to listen, in seconds. @@ -169,6 +197,19 @@ class Device(ABC): Returns: Every observation captured during the window, in arrival order. """ + captured: list[Observation] = [] + + def collect(obs: Observation) -> None: + captured.append(obs) + if on_observation is not None: + on_observation(obs) + + unsubscribe = await self.subscribe_observations(collect) + try: + await asyncio.sleep(duration_s) + finally: + unsubscribe() + return captured # -- configuration: extra reads --------------------------------------------- @@ -615,12 +656,9 @@ 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]: + async def subscribe_observations( # noqa: D102 - inherited docstring + self, on_observation: ObservationCallback + ) -> Unsubscribe: from meshcore import EventType mc = self._require() @@ -631,15 +669,10 @@ class MeshCoreDevice(Device): "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: + if obs is not None: on_observation(obs) return handler @@ -657,9 +690,8 @@ class MeshCoreDevice(Device): etype = getattr(EventType, attr, None) if etype is not None: subs.append(subscribe(etype, make_handler(kind))) - try: - await asyncio.sleep(duration_s) - finally: + + def unsubscribe() -> None: for sub in subs: unsub = getattr(sub, "unsubscribe", None) if unsub is not None: @@ -667,7 +699,8 @@ class MeshCoreDevice(Device): unsub() except Exception: # noqa: BLE001 - best-effort cleanup pass - return observations + + return unsubscribe @staticmethod def _ok(event): # type: ignore[no-untyped-def] @@ -848,6 +881,9 @@ class MockDevice(Device): self._channels: dict[int, dict] = {} self._device_pin = 0 self._private_key = "11" * 32 + # Background emitter tasks spawned by ``subscribe_observations``; tracked so they + # can be cancelled on disconnect and are never garbage-collected while pending. + self._bg_tasks: set[asyncio.Task] = set() async def connect(self) -> None: # noqa: D102 - inherited docstring await asyncio.sleep(0) @@ -855,6 +891,9 @@ class MockDevice(Device): async def disconnect(self) -> None: # noqa: D102 - inherited docstring self._connected = False + for task in list(self._bg_tasks): + task.cancel() + self._bg_tasks.clear() async def get_self_info(self) -> dict: # noqa: D102 - inherited docstring return {**self._info, "tx_power": self._tx_power} @@ -987,41 +1026,71 @@ 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 + async def subscribe_observations( # noqa: D102 - inherited docstring + self, on_observation: ObservationCallback + ) -> Unsubscribe: + # Simulate a live packet stream. Emit a burst of synthetic adverts/telemetry from + # the known contacts *synchronously* here, then keep emitting at a steady cadence + # from a background task until unsubscribed. The immediate first burst means even + # a zero-length capture window always sees every contact (two of which carry a + # location), keeping bounded ``listen`` runs and their tests deterministic. + stop = asyncio.Event() + seq = 0 + + def emit_burst() -> None: + nonlocal seq + for contact in self._contacts: + on_observation(self._synth_observation(contact, seq)) + seq += 1 + + emit_burst() + + async def emit_loop() -> None: + while not stop.is_set(): + try: + await asyncio.wait_for(stop.wait(), _MOCK_MONITOR_INTERVAL_S) + except asyncio.TimeoutError: + pass # cadence tick elapsed; emit the next burst + if not stop.is_set(): + emit_burst() + + task = asyncio.create_task(emit_loop()) + self._bg_tasks.add(task) + task.add_done_callback(self._bg_tasks.discard) + + def unsubscribe() -> None: + stop.set() # wakes the loop's wait immediately; it exits on the next check + + return unsubscribe + + #: Fixed locations advertised by the two simulated repeaters, so location-aware + #: features (e.g. the coverage map) always have coordinates to work with. + _MOCK_LOCATIONS = { + "Yagi-Repeater": (45.5019, -73.5674), + "Local-Repeater": (45.4768, -73.5990), + } + + def _synth_observation(self, contact: Contact, seq: int) -> Observation: + """Build one plausible synthetic observation for ``contact`` (simulator only). + + Args: + contact: The contact to synthesize a reception from. + seq: Monotonic emission counter, used to vary the packet kind. + + Returns: + A noisy :class:`Observation` tagged ``telemetry`` on every fourth packet and + ``advert`` otherwise, carrying a location for the simulated repeaters. + """ + lat_lon = self._MOCK_LOCATIONS.get(contact.name) + return Observation( + node=contact.key_prefix or contact.public_key[:12], + name=contact.name, + kind="telemetry" if seq % 4 == 3 else "advert", + 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, + ) def _expected_snr(self, hop_index: int) -> float: """Model SNR for a hop as an inverted-U in TX power plus distance falloff. diff --git a/meshtools/core/monitor_store.py b/meshtools/core/monitor_store.py new file mode 100644 index 0000000..d071ebc --- /dev/null +++ b/meshtools/core/monitor_store.py @@ -0,0 +1,54 @@ +"""Persistence for the passive-monitor on/off preference. + +Whether background monitoring is enabled is remembered between sessions in a small JSON +file in the config directory (``/monitor.json``), following the same pattern +as :class:`~meshtools.core.device_store.DeviceStore`: it is machine/user state managed by +the app rather than user-authored config, and it must survive a per-invocation ``--db`` +override, so it lives beside the config rather than in the swappable database. + +The preference defaults to *off*: a fresh install never captures until the user opts in. +""" + +from __future__ import annotations + +import json +from pathlib import Path + + +class MonitorStore: + """Reads and writes the remembered passive-monitor on/off preference.""" + + def __init__(self, path: Path) -> None: + """Open the store against a JSON file location. + + Args: + path: Path to the JSON state file (created lazily on first write). + """ + self._path = path + + def load_enabled(self) -> bool: + """Return whether monitoring is enabled, defaulting to ``False``. + + A missing or corrupt file is treated as "disabled" rather than an error, so a + stray edit never blocks startup and never silently starts capturing. + + Returns: + ``True`` if monitoring was left enabled, otherwise ``False``. + """ + try: + data = json.loads(self._path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return False + return bool(data.get("enabled", False)) if isinstance(data, dict) else False + + def save_enabled(self, enabled: bool) -> None: + """Persist the monitoring on/off preference. + + Args: + enabled: Whether background monitoring should be on for future sessions. + """ + self._path.parent.mkdir(parents=True, exist_ok=True) + # Atomic-ish replace so a crash mid-write can't leave a truncated file. + tmp = self._path.with_suffix(self._path.suffix + ".tmp") + tmp.write_text(json.dumps({"enabled": bool(enabled)}, indent=2), encoding="utf-8") + tmp.replace(self._path) diff --git a/meshtools/persistence/repository.py b/meshtools/persistence/repository.py index 0b55f52..6d2a7b0 100644 --- a/meshtools/persistence/repository.py +++ b/meshtools/persistence/repository.py @@ -335,6 +335,18 @@ class Repository: ) self._conn.commit() + def observation_count(self) -> int: + """Return the total number of overheard packets stored across every run. + + Backs the monitor's "total" counter, so it counts all observations ever logged, + not just the current session's. + + Returns: + The row count of the ``observations`` table. + """ + row = self._conn.execute("SELECT COUNT(*) AS n FROM observations").fetchone() + return int(row["n"]) if row else 0 + def heard_nodes(self, *, since: Optional[datetime] = None) -> list[HeardNode]: """Aggregate stored observations into per-node reception statistics. diff --git a/meshtools/services/monitor_service.py b/meshtools/services/monitor_service.py new file mode 100644 index 0000000..e1f659c --- /dev/null +++ b/meshtools/services/monitor_service.py @@ -0,0 +1,189 @@ +"""The passive-monitor background service. + +Passive monitoring records every advert/telemetry frame the companion overhears to the +database, building the longitudinal history the coverage map and link-quality alerting +read back. Unlike a one-shot capture, it runs as a non-blocking background subscription: +once started it logs observations as they arrive while the user keeps using the app, and +keeps going across menu actions until stopped or the session ends. + +The service is session-scoped state on the :class:`~meshtools.context.AppContext` +(``ctx.monitor``). It owns the on/off preference (persisted via +:class:`~meshtools.core.monitor_store.MonitorStore`), the live subscription, and the +counters shown live in the menu. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Optional + +from ..core.connection import Unsubscribe +from ..core.models import Observation + +if TYPE_CHECKING: + from ..context import AppContext + from ..core.monitor_store import MonitorStore + + +class MonitorService: + """Manages the always-on, non-blocking passive-monitor subscription. + + Attributes are private; interact through the properties and the async lifecycle + methods (:meth:`enable`, :meth:`disable`, :meth:`toggle`, :meth:`start`, :meth:`stop`, + :meth:`aclose`). + """ + + def __init__(self, ctx: "AppContext", store: "MonitorStore") -> None: + """Initialize the service and load the persisted on/off preference. + + Args: + ctx: The shared application context (device, repository, logger). + store: Persistence for the on/off preference. + """ + self._ctx = ctx + self._store = store + self._enabled = store.load_enabled() + self._unsubscribe: Optional[Unsubscribe] = None + self._run_id: Optional[int] = None + self._session_count = 0 + self._run_start_count = 0 + # Total observations already in the database when the session began; the live + # "total" is this plus what we capture this session (this process is the only + # writer during an interactive session), avoiding a DB count on every repaint. + self._start_total = ctx.repo.observation_count() + self._error: Optional[str] = None + + @property + def enabled(self) -> bool: + """Whether monitoring is the desired state (persisted across sessions).""" + return self._enabled + + @property + def active(self) -> bool: + """Whether a live subscription is currently capturing.""" + return self._unsubscribe is not None + + @property + def session_count(self) -> int: + """Observations captured since this process started.""" + return self._session_count + + @property + def last_error(self) -> Optional[str]: + """The reason the last :meth:`start` attempt failed, if any.""" + return self._error + + def total_count(self) -> int: + """Return the total observations logged, all time (including this session).""" + return self._start_total + self._session_count + + def status_text(self) -> str: + """Return a compact one-line status for the live menu header. + + Returns: + A glyph-prefixed summary: capture counts when active, a waiting note when + enabled but not yet capturing, or an off marker. + """ + if self.active: + return ( + f"● monitor ON · {self._session_count} this session " + f"· {self.total_count()} total" + ) + if self._enabled: + return "● monitor ON (waiting for a device)" + return "○ monitor OFF" + + async def start(self) -> None: + """Open a background subscription and begin logging observations. + + Idempotent: a no-op if already active. Opens the device connection (which may + raise if no device can be selected) and records a ``monitor`` run that the + captured observations are linked to. + + Raises: + Exception: Propagates any device/subscription error after recording it; the + on/off preference is left unchanged so the caller can surface the problem. + """ + if self.active: + return + try: + device = await self._ctx.device() + except Exception as exc: # noqa: BLE001 - remember why, then re-raise + self._error = str(exc) + raise + run_id = self._ctx.repo.start_run( + "monitor", {"mode": "background"}, self._ctx.profile_name + ) + self._run_start_count = self._session_count + + def on_observation(obs: Observation) -> None: + # Runs on the event loop as packets arrive; keep it cheap and defensive so a + # single bad write can never take down the subscription. + self._session_count += 1 + try: + self._ctx.repo.record_observation(run_id, obs) + except Exception as exc: # noqa: BLE001 - never let logging break capture + self._ctx.log.debug("monitor: failed to record observation: %s", exc) + + try: + self._unsubscribe = await device.subscribe_observations(on_observation) + except Exception as exc: # noqa: BLE001 - record, remember, and re-raise + self._ctx.repo.finish_run(run_id, "error", {"error": str(exc)}) + self._error = str(exc) + raise + self._run_id = run_id + self._error = None + self._ctx.log.info("passive monitor started (run %s)", run_id) + + async def stop(self) -> None: + """Stop the background subscription and close its run record. + + Idempotent: a no-op if not active. + """ + if not self.active: + return + try: + assert self._unsubscribe is not None + self._unsubscribe() + finally: + self._unsubscribe = None + if self._run_id is not None: + captured = self._session_count - self._run_start_count + self._ctx.repo.finish_run(self._run_id, "ok", {"observations": captured}) + self._ctx.log.info("passive monitor stopped (run %s, %s pkts)", self._run_id, captured) + self._run_id = None + + async def enable(self) -> None: + """Turn monitoring on, persist the preference, and start capturing. + + Raises: + Exception: If capture could not start (e.g. no device); the preference is + still persisted as *on* so it resumes once a device is available. + """ + self._enabled = True + self._store.save_enabled(True) + await self.start() + + async def disable(self) -> None: + """Turn monitoring off, persist the preference, and stop capturing.""" + self._enabled = False + self._store.save_enabled(False) + await self.stop() + + async def toggle(self) -> bool: + """Flip the on/off state, starting or stopping capture accordingly. + + Returns: + The new enabled state (``True`` if now on). + + Raises: + Exception: Propagates a failure to start capture when turning on. + """ + if self._enabled: + await self.disable() + else: + await self.enable() + return self._enabled + + async def aclose(self) -> None: + """Stop capture at session end without changing the persisted preference.""" + await self.stop() diff --git a/meshtools/tools/monitor.py b/meshtools/tools/monitor.py index 4f54833..58d9968 100644 --- a/meshtools/tools/monitor.py +++ b/meshtools/tools/monitor.py @@ -1,9 +1,19 @@ -"""The ``monitor`` tool: passively log adverts/telemetry the mesh emits over time. +"""The ``monitor`` tool: control the passive background logger and review its history. -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. +Passive monitoring records every advert and telemetry frame the companion 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. + +Capture itself runs as a non-blocking background subscription owned by +:class:`~meshtools.services.monitor_service.MonitorService` (``ctx.monitor``), so it never +blocks the menu. This tool is the control panel for it: toggling capture on/off (a +preference remembered between sessions) and reviewing the heard-node history. The live +packet counters are shown in the main-menu header, not here. + +The CLI subcommand additionally offers a one-shot foreground capture (``--seconds``) for +scripting, since a short-lived CLI process has no long-running menu to host a background +subscription. """ from __future__ import annotations @@ -21,57 +31,215 @@ 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. +#: Cap a single foreground capture window so a scripted run can't block indefinitely; +#: longer passive logging is done by enabling the background monitor instead. MAX_DURATION_S = 3600 @register class MonitorTool(Tool): - """Listen for a window and log every overheard advert/telemetry to the database.""" + """Toggle the passive background monitor and review the nodes it has heard.""" name = "monitor" - help = "Passively log adverts/telemetry heard on the mesh (longitudinal history)." + help = "Toggle passive background monitoring on/off and review heard nodes." category = "Diagnostics" order = 20 async def prompt_params(self, ctx: AppContext) -> Optional[dict[str, Any]]: - """Ask how long to listen. + """Show the monitor status and offer to toggle it or review heard nodes. Args: ctx: Shared application context. Returns: - A parameter dict, or ``None`` if the user cancelled. + A parameter dict naming the chosen ``action``, or ``None`` if cancelled. """ - duration = await questionary.text( - f"Listen for how many seconds? (1-{MAX_DURATION_S})", - default="60", - validate=_is_valid_duration, + monitor = ctx.monitor + toggle_label = "Turn monitoring OFF" if monitor.enabled else "Turn monitoring ON" + choice = await questionary.select( + f"Passive monitor — {monitor.status_text()}", + choices=[ + questionary.Choice(toggle_label, value="toggle"), + questionary.Choice("View heard nodes (all time)", value="view"), + questionary.Choice("Back", value="__back__"), + ], ).ask_async() - if duration is None: + if choice in (None, "__back__"): return None - return {"duration": int(duration)} + return {"action": choice} + + async def execute(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult: + """Run the control-panel action directly, without a logged ``runs`` row. + + The monitor entry is a control panel, not a measurement, so it is deliberately + not wrapped in run-logging (unlike the base :meth:`Tool.execute`). The background + capture session records its own ``monitor`` run instead. + + Args: + ctx: Shared application context. + params: The chosen ``action``. + + Returns: + The :class:`ToolResult` from :meth:`run`. + """ + return await self.run(ctx, params) async def run(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult: - """Capture observations for the window, persist them, and summarize heard nodes. + """Dispatch the selected control-panel action. Args: ctx: Shared application context. - params: ``duration`` (seconds) and the injected ``_run_id``. + params: ``action`` — ``"toggle"`` or ``"view"``. Returns: - A :class:`ToolResult` noting how many nodes/packets were heard. + A :class:`ToolResult` describing the outcome. """ - duration = max(1, min(MAX_DURATION_S, int(params.get("duration", 60)))) - run_id = params["_run_id"] + if params.get("action") == "view": + return self._view(ctx) + return await self._toggle(ctx) + + @staticmethod + async def _toggle(ctx: AppContext) -> ToolResult: + """Flip background monitoring on or off, surfacing any start failure. + + Args: + ctx: Shared application context. + + Returns: + A :class:`ToolResult` reporting the new state. + """ + try: + now_on = await ctx.monitor.toggle() + except Exception as exc: # noqa: BLE001 - report cleanly, keep the menu alive + return ToolResult( + summary={"enabled": ctx.monitor.enabled, "active": ctx.monitor.active, + "error": str(exc)}, + message=f"[warn]monitoring enabled but capture couldn't start:[/warn] {exc}", + ) + state = "[ok]● ON[/ok]" if now_on else "[muted]○ OFF[/muted]" + return ToolResult( + summary={"enabled": now_on, "active": ctx.monitor.active}, + message=f"passive monitor is now {state}", + ) + + @staticmethod + def _view(ctx: AppContext) -> ToolResult: + """Render the all-time heard-node summary from stored observations. + + Reads only the database, so it needs no device connection and works whether or + not monitoring is currently active. + + Args: + ctx: Shared application context. + + Returns: + A :class:`ToolResult` summarizing the heard nodes. + """ + heard = ctx.repo.heard_nodes() + if heard: + # Names come from the stored observations, so no device lookup is needed. + ctx.console.print(_heard_table(heard, lambda _node: None)) + else: + ctx.console.print("[muted]no packets heard yet — turn monitoring on[/muted]") + + packets = sum(n.count for n in heard) + located = sum(1 for n in heard if n.has_location) + return ToolResult( + summary={ + "nodes_heard": len(heard), + "observations": packets, + "nodes_with_location": located, + }, + message=( + f"[ok]✓[/ok] heard [brand]{packets}[/brand] packets from " + f"[brand]{len(heard)}[/brand] nodes (all time)" + ), + ) + + def register_cli(self, app: typer.Typer) -> None: + """Register the ``monitor`` subcommand. + + Args: + app: The Typer application. + """ + from ..core.monitor_store import MonitorStore + + @app.command(name=self.name, help=self.help) + def _monitor( + on: bool = typer.Option( + False, "--on", help="Enable background monitoring for future sessions." + ), + off: bool = typer.Option( + False, "--off", help="Disable background monitoring." + ), + seconds: Optional[int] = typer.Option( + None, + "--seconds", + "-d", + help=f"Capture in the foreground for N seconds now, then exit " + f"(1-{MAX_DURATION_S}).", + ), + ) -> None: + from ..cli import _state + + if on and off: + raise typer.BadParameter("Pass only one of --on / --off.") + assert _state is not None # set by the callback before any subcommand runs + ctx = _state + store = MonitorStore(ctx.settings.config_dir / "monitor.json") + + if on or off: + store.save_enabled(on) + state = "[ok]on[/ok]" if on else "[muted]off[/muted]" + ctx.console.print( + f"[ok]✓[/ok] background monitoring set {state} for future " + "interactive sessions" + ) + return + if seconds is not None: + _run_foreground_capture(ctx, seconds) + return + # No flags: report the current preference and history size. + enabled = "[ok]on[/ok]" if store.load_enabled() else "[muted]off[/muted]" + ctx.console.print( + f"background monitoring is {enabled} · " + f"[brand]{ctx.repo.observation_count()}[/brand] observations logged all-time" + ) + + +def _run_foreground_capture(ctx: AppContext, seconds: int) -> None: + """Drive a one-shot foreground capture on its own event loop. + + Args: + ctx: Shared application context. + seconds: Requested capture duration (clamped to ``1..MAX_DURATION_S``). + """ + import asyncio + + from ..cli import _drive + + asyncio.run(_drive(_foreground_capture(ctx, seconds), ctx)) + + +async def _foreground_capture(ctx: AppContext, seconds: int) -> None: + """Capture overheard packets for a bounded window and print a heard-node table. + + This is the scripted, blocking counterpart to the background monitor: it records to + its own ``monitor`` run and returns when the window elapses. Used by ``monitor + --seconds``. + + Args: + ctx: Shared application context. + seconds: Requested capture duration (clamped to ``1..MAX_DURATION_S``). + """ + duration = max(1, min(MAX_DURATION_S, int(seconds))) + run_id = ctx.repo.start_run( + "monitor", {"mode": "foreground", "duration": duration}, ctx.profile_name + ) + observations: list[Observation] = [] + try: 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] = [] + resolve = trace_runner.make_node_resolver(await device.get_contacts()) with make_progress(ctx.console) as progress: task = progress.add_task(f"listening {duration}s", total=duration) @@ -86,42 +254,31 @@ class MonitorTool(Tool): await device.listen(duration, on_observation=on_observation) progress.update(task, completed=duration) + except Exception as exc: # noqa: BLE001 - record then re-raise for the CLI handler + ctx.repo.finish_run(run_id, "error", {"error": str(exc)}) + raise - heard = _aggregate(observations) - if heard: - ctx.console.print(_heard_table(heard, resolve)) - else: - ctx.console.print("[muted]no packets heard during the window[/muted]") + 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}) + located = sum(1 for n in heard if n.has_location) + ctx.repo.finish_run( + run_id, + "ok", + { + "duration_s": duration, + "observations": len(observations), + "nodes_heard": len(heard), + "nodes_with_location": located, + }, + ) + ctx.console.print( + f"[ok]✓[/ok] heard [brand]{len(observations)}[/brand] packets from " + f"[brand]{len(heard)}[/brand] nodes in {duration}s" + ) def _distinct(observations: list[Observation]) -> int: @@ -180,23 +337,3 @@ def _heard_table(heard: list[HeardNode], resolve) -> Table: # noqa: ANN001 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/meshtools/ui/menu.py b/meshtools/ui/menu.py index 2dbf29e..39b7b55 100644 --- a/meshtools/ui/menu.py +++ b/meshtools/ui/menu.py @@ -24,13 +24,39 @@ _MENU_STYLE = questionary.Style( ] ) +#: How often (seconds) prompt_toolkit repaints the menu, so the live monitor counters in +#: the header tick up while the user sits at the prompt without pressing a key. +_MENU_REFRESH_S = 1.0 + + +class _MenuPrompt: + """The menu's question line, re-rendered with live monitor counts on every repaint. + + questionary formats the ``message`` into the prompt on each render, so an object + whose ``__str__`` reads the current counters gives a genuinely live header (paired + with the select's ``refresh_interval``) at no cost when nothing is happening. + """ + + def __init__(self, ctx: AppContext) -> None: + """Bind the prompt to the application context. + + Args: + ctx: The shared application context, read for live monitor state. + """ + self._ctx = ctx + + def __str__(self) -> str: + """Return the prompt text plus the current passive-monitor status line.""" + return f"What would you like to do? [{self._ctx.monitor.status_text()}]" + async def run_menu(ctx: AppContext) -> None: """Run the interactive menu loop until the user quits. Before showing the menu, prompts for a companion device (unless the simulator is in use or a port/profile was already chosen explicitly), so the banner reflects the - target and tools connect to the right radio. + target and tools connect to the right radio. If passive monitoring was left enabled + in a previous session, it resumes in the background here. Args: ctx: The shared application context. @@ -38,33 +64,58 @@ async def run_menu(ctx: AppContext) -> None: console = ctx.console await _select_device_at_startup(ctx) console.print(banner(ctx.profile_name, ctx.mock, ctx.selected_device)) + await _resume_monitor(ctx) - while True: - choices: list[questionary.Choice | questionary.Separator] = [] - current_category: str | None = None - for tool in all_tools(): - if tool.category != current_category: - current_category = tool.category - choices.append(questionary.Separator(f"── {current_category} ──")) - choices.append( - questionary.Choice(title=f"{tool.name} — {tool.help}", value=tool.name) - ) - choices.append(questionary.Separator(" ")) - choices.append(questionary.Choice(title="quit", value="__quit__")) + prompt = _MenuPrompt(ctx) + try: + while True: + choices: list[questionary.Choice | questionary.Separator] = [] + current_category: str | None = None + for tool in all_tools(): + if tool.category != current_category: + current_category = tool.category + choices.append(questionary.Separator(f"── {current_category} ──")) + choices.append( + questionary.Choice(title=f"{tool.name} — {tool.help}", value=tool.name) + ) + choices.append(questionary.Separator(" ")) + choices.append(questionary.Choice(title="quit", value="__quit__")) - selection = await questionary.select( - "What would you like to do?", - choices=choices, - style=_MENU_STYLE, - qmark="◆", - ).ask_async() + selection = await questionary.select( + prompt, + choices=choices, + style=_MENU_STYLE, + qmark="◆", + refresh_interval=_MENU_REFRESH_S, + ).ask_async() - if selection in (None, "__quit__"): - console.print("[muted]bye 73![/muted]") - return + if selection in (None, "__quit__"): + console.print("[muted]bye 73![/muted]") + return - await _run_selection(ctx, selection) - console.print() + await _run_selection(ctx, selection) + console.print() + finally: + # Stop background capture and close its run record even on an unexpected exit. + await ctx.monitor.aclose() + + +async def _resume_monitor(ctx: AppContext) -> None: + """Resume passive monitoring if it was left enabled, warning if it can't start. + + A failure to start (typically no companion device selected) is non-fatal: the + preference stays on, so monitoring resumes automatically once a device is available. + + Args: + ctx: The shared application context. + """ + if not ctx.monitor.enabled: + return + try: + await ctx.monitor.start() + ctx.console.print("[muted]passive monitor resumed in the background[/muted]") + except Exception as exc: # noqa: BLE001 - surface, don't crash the menu + ctx.console.print(f"[warn]passive monitor could not start:[/warn] {exc}") async def _select_device_at_startup(ctx: AppContext) -> None: diff --git a/tests/test_monitor.py b/tests/test_monitor.py index 9b7f0e2..721ca42 100644 --- a/tests/test_monitor.py +++ b/tests/test_monitor.py @@ -6,12 +6,38 @@ hardware required. from __future__ import annotations +import asyncio +import logging from datetime import timedelta from pathlib import Path -from meshtools.core.connection import MockDevice, observation_from_event +import pytest + +from meshtools.core.connection import ( + _MOCK_MONITOR_INTERVAL_S as _MOCK_INTERVAL, +) +from meshtools.core.connection import ( + MockDevice, + observation_from_event, +) from meshtools.core.models import HeardNode, Observation, utcnow +from meshtools.core.monitor_store import MonitorStore from meshtools.persistence.repository import Repository +from meshtools.services.monitor_service import MonitorService + + +class _StubContext: + """Minimal stand-in for :class:`~meshtools.context.AppContext` for service tests.""" + + def __init__(self, repo: Repository, device: MockDevice) -> None: + self.repo = repo + self._device = device + self.profile_name = None + self.log = logging.getLogger("test.monitor") + + async def device(self) -> MockDevice: + await self._device.connect() + return self._device async def test_mock_listen_emits_observations_and_callbacks() -> None: @@ -29,6 +55,40 @@ async def test_mock_listen_emits_observations_and_callbacks() -> None: assert any(o.lat is not None and o.lon is not None for o in observations) +async def test_subscribe_observations_streams_then_stops() -> None: + """The subscription emits an immediate burst, keeps streaming, and halts on unsub.""" + device = MockDevice() + await device.connect() + + seen: list[Observation] = [] + unsubscribe = await device.subscribe_observations(seen.append) + try: + # The first burst is emitted synchronously, before any await. + first_burst = len(seen) + assert first_burst >= 1 + assert any(o.lat is not None for o in seen) # a located repeater is in the burst + + await asyncio.sleep(_MOCK_INTERVAL * 3) + assert len(seen) > first_burst # more packets arrived while we did nothing + finally: + unsubscribe() + + frozen = len(seen) + await asyncio.sleep(_MOCK_INTERVAL * 3) + assert len(seen) == frozen # nothing more after unsubscribing + + +async def test_disconnect_stops_background_emitter() -> None: + """Disconnecting cancels any live subscription task rather than leaking it.""" + device = MockDevice() + await device.connect() + await device.subscribe_observations(lambda _obs: None) + assert device._bg_tasks # a background emitter is running + await device.disconnect() + await asyncio.sleep(0) # let the cancellation settle + assert not device._bg_tasks + + def test_heard_node_aggregation() -> None: """from_observations summarizes count, robust SNR, latest RSSI, and location.""" now = utcnow() @@ -113,3 +173,70 @@ async def test_observations_round_trip_and_aggregate(tmp_path: Path) -> None: # Nodes are ordered most-recently-heard first. assert nodes == sorted(nodes, key=lambda n: n.last_seen, reverse=True) repo.close() + + +def test_observation_count_totals_every_run(tmp_path: Path) -> None: + """observation_count sums observations across all runs, empty database included.""" + repo = Repository(tmp_path / "count.db") + assert repo.observation_count() == 0 + run_id = repo.start_run("monitor", {}) + repo.record_observation(run_id, Observation(node="a1", snr=1.0)) + repo.record_observation(run_id, Observation(node="b2", snr=2.0)) + assert repo.observation_count() == 2 + repo.close() + + +def test_monitor_store_defaults_off_and_persists(tmp_path: Path) -> None: + """The preference defaults to off and round-trips through a fresh store instance.""" + path = tmp_path / "monitor.json" + assert MonitorStore(path).load_enabled() is False # nothing saved yet → off + MonitorStore(path).save_enabled(True) + assert MonitorStore(path).load_enabled() is True # remembered across instances + MonitorStore(path).save_enabled(False) + assert MonitorStore(path).load_enabled() is False + + +async def test_monitor_service_captures_in_background(tmp_path: Path) -> None: + """Enabling starts a live subscription that logs observations without blocking.""" + repo = Repository(tmp_path / "svc.db") + ctx = _StubContext(repo, MockDevice()) + store = MonitorStore(tmp_path / "monitor.json") + service = MonitorService(ctx, store) + + assert not service.enabled and not service.active + await service.enable() + assert service.enabled and service.active + assert store.load_enabled() is True # preference persisted for next session + + await asyncio.sleep(_MOCK_INTERVAL * 3) # let packets stream in while we "do other work" + assert service.session_count > 0 + # Started from an empty database, so total equals what this session captured. + assert service.total_count() == service.session_count + + await service.disable() + assert not service.enabled and not service.active + assert store.load_enabled() is False + assert repo.observation_count() == service.session_count # everything was logged + + frozen = service.session_count + await asyncio.sleep(_MOCK_INTERVAL * 3) + assert service.session_count == frozen # capture really stopped + repo.close() + + +async def test_monitor_service_enable_without_device_keeps_preference(tmp_path: Path) -> None: + """If capture can't start, the on preference is still remembered so it can resume.""" + + class _NoDevice(_StubContext): + async def device(self) -> MockDevice: + raise ValueError("no companion device selected") + + repo = Repository(tmp_path / "nodev.db") + service = MonitorService(_NoDevice(repo, MockDevice()), MonitorStore(tmp_path / "m.json")) + + with pytest.raises(ValueError): + await service.enable() + assert service.enabled is True # preference stays on to resume later + assert service.active is False + assert service.last_error == "no companion device selected" + repo.close()