From 10f7db0a2c560fc88c82cebda43031c454412ba4 Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Sun, 4 Oct 2026 01:20:11 -0400 Subject: [PATCH 1/2] Local dashboard: plain-language status checklist and C2-free listening - /api/health: ordered checklist (password, C2 approval, MQTT, SDR, source, OP25, decoding, audio, Icecast) with what's wrong and what to do; pure build_checks() so every state (incl. Toughny's unapproved node) is tested - /api/listen: pick P25 control channels / one analog freq / a cached C2 system, validated in MHz, stored as a local override; OP25 restarted with the start failure reason surfaced - /api/config/system no longer writes a raw blob OP25 can't read - OP25 resumes after reboot on a locally chosen system, not only C2-assigned - op25-container: analog (NBFM) config generation fixed (AttributeError + wrong field names), /op25/log tail, start refuses without a config, status detects a dead multi_rx ("crashed") -- node-26#12 - index.html rebuilt on UI_REDESIGN tokens; scanner drops DMR, shows System key Co-Authored-By: Claude Opus 5.5 --- drb-edge-node/app/internal/health.py | 299 +++++ .../app/internal/metadata_watcher.py | 33 + drb-edge-node/app/internal/op25_client.py | 41 +- drb-edge-node/app/main.py | 7 +- drb-edge-node/app/routers/api.py | 173 ++- drb-edge-node/app/templates/index.html | 1135 ++++++++--------- drb-edge-node/app/templates/scanner.html | 9 +- drb-edge-node/tests/test_local_dashboard.py | 247 ++++ op25-container/app/models.py | 300 ++--- op25-container/app/routers/op25_controller.py | 449 ++++--- op25-container/run_multi-rx_service.sh | 11 +- 11 files changed, 1721 insertions(+), 983 deletions(-) create mode 100644 drb-edge-node/app/internal/health.py create mode 100644 drb-edge-node/tests/test_local_dashboard.py diff --git a/drb-edge-node/app/internal/health.py b/drb-edge-node/app/internal/health.py new file mode 100644 index 0000000..be95a1e --- /dev/null +++ b/drb-edge-node/app/internal/health.py @@ -0,0 +1,299 @@ +""" +Plain-language health checklist for the local dashboard (/api/health). + +A first-time owner opens http:/// and needs three answers: is this node +working, if not what is wrong, and what do I do about it. Each check below +answers one of those for one link in the chain + + password -> C2 enrollment -> MQTT -> SDR -> what to listen to -> OP25 + -> decoding -> audio (PulseAudio) -> local stream (Icecast) + +`gather_facts()` does the I/O; `build_checks()` is a pure function of those +facts so every state (unapproved, no SDR, OP25 crashed, …) is unit-testable. + +Read-only: nothing here changes node state, and it never polls the OP25 +terminal itself (that drains OP25's call_log, which metadata_watcher owns) — +it reads metadata_watcher.console_state instead. +""" +import time +from typing import Any, Dict, List, Optional + +import httpx + +from app.config import settings +from app.internal.logger import logger + +OK, WARN, ERROR, OFF = "ok", "warn", "error", "off" + +# OP25's console answers within a second or two of a healthy multi_rx; past +# this the console is considered silent. +CONSOLE_STALE_SECONDS = 5.0 +# P25 control channels send several messages a second; this long without one +# means OP25 is not decoding. +TSBK_STALE_SECONDS = 15.0 + + +def _check(cid: str, label: str, state: str, summary: str, action: str = "") -> Dict[str, str]: + return {"id": cid, "label": label, "state": state, "summary": summary, "action": action} + + +def _mhz(freq: Any) -> Optional[str]: + try: + f = float(freq) + except (TypeError, ValueError): + return None + if f <= 0: + return None + return f"{f / 1e6:.4f} MHz" if f > 1e5 else f"{f:.4f} MHz" + + +def _ago(seconds: float) -> str: + seconds = max(0, int(seconds)) + if seconds < 60: + return f"{seconds}s ago" + if seconds < 3600: + return f"{seconds // 60} min ago" + return f"{seconds // 3600} h ago" + + +def active_config(node_cfg) -> Any: + if node_cfg.override_system_id and node_cfg.override_config: + return node_cfg.override_config + return node_cfg.system_config + + +def build_checks(f: Dict[str, Any]) -> List[Dict[str, str]]: + """Facts -> ordered checklist. See gather_facts() for the keys.""" + now = f.get("now", time.time()) + checks: List[Dict[str, str]] = [] + + # 1. Dashboard password + if f["default_password"]: + checks.append(_check( + "password", "Dashboard password", WARN, + "Still the factory default — anyone on this network can log in.", + "Set DASHBOARD_PASSWORD in the node's .env file, then restart the edge-node container.", + )) + else: + checks.append(_check("password", "Dashboard password", OK, "Changed from the default.")) + + # 2. C2 enrollment / approval + c2_url, has_key = f["c2_url"], f["has_api_key"] + if not c2_url: + checks.append(_check( + "c2", "Server (C2) link", OFF, + "Standalone — no C2 server is set. Local listening below still works.", + "To join a DRB server, re-run install.sh with an enrollment token.", + )) + elif not has_key: + checks.append(_check( + "c2", "Server (C2) link", WARN, + f"Waiting for approval — this node has no API key from {c2_url} yet, so nothing is uploaded.", + f"Ask a DRB admin to approve node \"{f['node_id']}\" on the Nodes page, then re-run " + "install.sh (or restart the edge-node container). Local listening below works meanwhile.", + )) + else: + checks.append(_check("c2", "Server (C2) link", OK, f"Approved — API key on file for {c2_url}.")) + + # 3. MQTT + broker = f"{f['mqtt_broker']}:{f['mqtt_port']}" + if f["mqtt_connected"]: + checks.append(_check("mqtt", "Live link (MQTT)", OK, f"Connected to {broker}.")) + elif not c2_url and not has_key: + checks.append(_check( + "mqtt", "Live link (MQTT)", OFF, f"Not connected to {broker} (expected for a standalone node).", + )) + elif not has_key: + checks.append(_check( + "mqtt", "Live link (MQTT)", WARN, + f"Not connected to {broker} — expected until the node is approved (the broker only admits approved nodes).", + "Fixes itself once the node is approved and has its API key (see above).", + )) + else: + checks.append(_check( + "mqtt", "Live link (MQTT)", ERROR, + f"Not connected to {broker} even though an API key is on file.", + "If the key was just installed or the node was re-approved, restart the edge-node container " + "so it loads the key. If it still fails, the key may be stale (every re-approval issues a new " + "one) — re-run install.sh, and check the node can reach the broker.", + )) + + # 4. SDR + sdr_count = f.get("sdr_count") + if sdr_count is None: + checks.append(_check( + "sdr", "SDR receiver", WARN, "Couldn't check — the SDR service isn't answering.", + "Check the secondary-sdr container is running.", + )) + elif sdr_count == 0: + checks.append(_check( + "sdr", "SDR receiver", ERROR, "No RTL-SDR dongle detected.", + "Plug the SDR into a USB port (avoid unpowered hubs), then restart the stack.", + )) + else: + checks.append(_check( + "sdr", "SDR receiver", OK, f"{sdr_count} SDR{'s' if sdr_count != 1 else ''} detected.", + )) + + # 5. What to listen to + src = f.get("source") + if not src: + checks.append(_check( + "source", "What to listen to", WARN, "Nothing chosen yet.", + "Pick a system or enter a frequency under \"Listen locally\" below.", + )) + else: + origin = "chosen on this dashboard" if src["local"] else "assigned by the server" + checks.append(_check("source", "What to listen to", OK, f"{src['name']} ({src['type']}) — {origin}.")) + + # 6. OP25 + op25 = f.get("op25_status") + running = op25 == "running" + if op25 is None: + checks.append(_check( + "op25", "Radio decoder (OP25)", ERROR, "The op25 container isn't answering.", + "Check the op25 container is running; its log usually says why it stopped.", + )) + elif running: + checks.append(_check("op25", "Radio decoder (OP25)", OK, "Running.")) + elif op25 == "crashed": + checks.append(_check( + "op25", "Radio decoder (OP25)", ERROR, "Stopped unexpectedly.", + "Open \"OP25 log\" below for the reason — usually the SDR is unplugged or in use. Then press Start.", + )) + elif not src: + checks.append(_check("op25", "Radio decoder (OP25)", OFF, "Stopped — choose what to listen to first.")) + else: + checks.append(_check("op25", "Radio decoder (OP25)", WARN, "Stopped.", "Press Start.")) + + # 7. Decoding + console = f.get("console") or {} + last_poll = console.get("last_poll_epoch") + is_analog = bool(src) and str(src.get("type", "")).upper() in ("NBFM", "ANALOG") + if not running: + checks.append(_check("decode", "Decoding", OFF, "Starts when OP25 runs.")) + elif last_poll is None or now - last_poll > CONSOLE_STALE_SECONDS: + checks.append(_check( + "decode", "Decoding", WARN, "OP25 is running but hasn't reported in yet.", + "Give it 20 seconds after a start. If this stays, open the OP25 log below.", + )) + elif is_analog: + chans = console.get("channels") or [] + freq = _mhz(chans[0].get("freq")) if chans else None + checks.append(_check( + "decode", "Decoding", OK, f"Tuned to {freq or 'the analog frequency'} (analog FM — no decoding needed).", + )) + else: + trunk = console.get("trunk") or [] + tsbk = [float(t.get("last_tsbk") or 0) for t in trunk] + if not trunk: + checks.append(_check("decode", "Decoding", WARN, "Waiting for OP25 to report control-channel status…")) + elif max(tsbk) > 0 and now - max(tsbk) < TSBK_STALE_SECONDS: + sys_ = trunk[tsbk.index(max(tsbk))] + cc = _mhz(sys_.get("rxchan")) + checks.append(_check( + "decode", "Decoding", OK, f"Locked on the control channel{' ' + cc if cc else ''}.", + )) + else: + checks.append(_check( + "decode", "Decoding", ERROR, "Not decoding the control channel.", + "Usually a wrong control-channel frequency, a weak signal/antenna, or SDR frequency error (PPM). " + "Try another control channel for the same system.", + )) + + # 8. Audio (PulseAudio — recording + Discord path) + if not f["audio_capturing"]: + checks.append(_check( + "audio", "Audio", ERROR if running else OFF, + "No audio reaching the recorder." if running else "Nothing to hear until OP25 runs.", + "The op25 container owns the audio path (PulseAudio); restart it if this persists." if running else "", + )) + else: + last_voice = f.get("last_voice_epoch") + if last_voice is None: + checks.append(_check("audio", "Audio", OK, "Audio path up — no voice heard yet.")) + elif now - last_voice < 3: + checks.append(_check("audio", "Audio", OK, "Voice right now.")) + else: + checks.append(_check("audio", "Audio", OK, f"Audio path up — last voice {_ago(now - last_voice)}.")) + + # 9. Local stream (Icecast) + ice = f.get("icecast_live") + if ice is None: + checks.append(_check( + "stream", "Local stream", ERROR, f"Icecast isn't answering on port {f['icecast_port']}.", + "Check the icecast container is running.", + )) + elif ice: + checks.append(_check("stream", "Local stream", OK, "Live — press play in the player below.")) + elif running: + checks.append(_check( + "stream", "Local stream", WARN, "OP25 is running but nothing is feeding the stream yet.", + "It normally appears within ~20 seconds of a start; if not, check the OP25 log below.", + )) + else: + checks.append(_check("stream", "Local stream", OFF, "Goes live when OP25 runs.")) + + return checks + + +async def icecast_live() -> Optional[bool]: + """True if the configured mount has a source, False if not, None if Icecast is down.""" + url = f"http://{settings.icecast_host}:{settings.icecast_port}/status-json.xsl" + try: + async with httpx.AsyncClient(timeout=3) as client: + r = await client.get(url) + r.raise_for_status() + sources = (r.json().get("icestats") or {}).get("source") or [] + except Exception as e: + logger.debug(f"Icecast status check failed: {e}") + return None + if isinstance(sources, dict): + sources = [sources] + return any(str(s.get("listenurl", "")).endswith(settings.icecast_mount) for s in sources) + + +async def gather_facts() -> Dict[str, Any]: + from app.internal import auth, credentials + from app.internal.call_recorder import call_recorder + from app.internal.config_manager import load_node_config + from app.internal.metadata_watcher import metadata_watcher + from app.internal.mqtt_manager import mqtt_manager + from app.internal.op25_client import op25_client + from app.internal.secondary_sdr_client import secondary_sdr_client + + node_cfg = load_node_config() + cfg = active_config(node_cfg) + op25 = await op25_client.status() + + devs = await secondary_sdr_client.devices() + sdr_count: Optional[int] = len(devs) if devs is not None else None + if not sdr_count: + # op25's own lsusb count — its :stable image may lack lsusb and say 0, + # so only trust a positive answer from it. + op25_devs = await op25_client.devices() + if op25_devs and op25_devs.get("count"): + sdr_count = op25_devs["count"] + + activity = call_recorder.audio_activity() + return { + "now": time.time(), + "node_id": settings.node_id, + "default_password": auth.is_using_default_password(), + "c2_url": settings.c2_url, + "has_api_key": bool(credentials.get_api_key()), + "mqtt_connected": mqtt_manager.is_connected, + "mqtt_broker": settings.mqtt_broker, + "mqtt_port": settings.mqtt_port, + "sdr_count": sdr_count, + "source": None if cfg is None else { + "name": cfg.name, "type": cfg.type, "system_id": cfg.system_id, + "local": bool(node_cfg.override_system_id and node_cfg.override_config), + }, + "op25_status": (op25 or {}).get("status") if op25 is not None else None, + "console": metadata_watcher.console_state, + "audio_capturing": call_recorder.is_capturing, + "last_voice_epoch": activity.last_voice_epoch, + "icecast_live": await icecast_live(), + "icecast_port": settings.icecast_port, + } diff --git a/drb-edge-node/app/internal/metadata_watcher.py b/drb-edge-node/app/internal/metadata_watcher.py index b1d88cb..2718bdf 100644 --- a/drb-edge-node/app/internal/metadata_watcher.py +++ b/drb-edge-node/app/internal/metadata_watcher.py @@ -235,6 +235,13 @@ class MetadataWatcher: self._last_tx_end: Optional[float] = None # epoch of the srcaddr 1→0 edge self._last_ok_poll: float = 0.0 + # Dashboard-only view of what the OP25 console last said. Read by + # /api/health; never consulted by segmentation or attribution. + self._terminal_epoch: Optional[float] = None + self._terminal_channels: List[Dict[str, Any]] = [] + self._terminal_trunk: List[Dict[str, Any]] = [] + self._last_heard: Optional[ConsoleEvent] = None + # Rolling console history for close-time attribution. self._console: deque[ConsoleEvent] = deque(maxlen=CONSOLE_HISTORY_MAX) @@ -294,6 +301,7 @@ class MetadataWatcher: self._last_ok_poll = now self._record_console(update, now) + self._note_terminal(update, now) activity = self._snapshot() if activity is None or not activity.capturing: @@ -350,6 +358,18 @@ class MetadataWatcher: while self._console and self._console[0].epoch < cutoff: self._console.popleft() + def _note_terminal(self, update: Any, now: float) -> None: + """Keep the latest console snapshot for the dashboard. No side effects + on segment state — trunk_update/channel_update are not sent on every + poll, so the last non-empty one is kept.""" + self._terminal_epoch = now + if update.channels: + self._terminal_channels = update.channels + if getattr(update, "trunk", None): + self._terminal_trunk = update.trunk + if self._console: + self._last_heard = self._console[-1] + def _is_duplicate_grant(self, event: ConsoleEvent) -> bool: for index in range(len(self._console) - 1, -1, -1): if len(self._console) - index > _GRANT_DEDUPE_DEPTH: @@ -898,6 +918,19 @@ class MetadataWatcher: def is_active(self) -> bool: return self._active_call_id is not None + @property + def console_state(self) -> Dict[str, Any]: + """Last thing the OP25 console reported, for /api/health only.""" + heard = self._last_heard + return { + "last_poll_epoch": self._terminal_epoch, + "channels": self._terminal_channels, + "trunk": self._terminal_trunk, + "last_heard": None if heard is None else { + "epoch": heard.epoch, "tgid": heard.tgid, "name": heard.name, "freq": heard.freq, + }, + } + @property def unattributed_segments(self) -> int: """Orphan-audio segments discarded since start. Surfaced on /api/status.""" diff --git a/drb-edge-node/app/internal/op25_client.py b/drb-edge-node/app/internal/op25_client.py index c7956d7..7a6f0f1 100644 --- a/drb-edge-node/app/internal/op25_client.py +++ b/drb-edge-node/app/internal/op25_client.py @@ -11,6 +11,11 @@ from app.internal.logger import logger # each id is also a top-level key holding that channel's dict # (freq/tgid/tag/srcaddr/svcopts/hold_tgid/…). # +# trunk_update — per-system trunking state keyed "0", "1", …; each carries +# `last_tsbk` (OP25's time.time() of the last control-channel +# message decoded) — the only direct proof OP25 is DECODING, +# not merely running. Surfaced on the dashboard only. +# # call_log — an EVENT QUEUE, not a snapshot. `log` holds entries appended # by tk_p25.log_call() at channel-grant time, each stamped with # OP25's own time.time(). get_call_log() DRAINS the deque, so @@ -18,7 +23,7 @@ from app.internal.logger import logger # it forever. The deque is capped at CALL_LOG_MAX_LEN = 10, so # the consumer must keep up. # -# Everything else (trunk_update, rx_update, terminal_config, …) is ignored. +# Everything else (rx_update, terminal_config, …) is ignored. TERMINAL_UPDATE_COMMAND = [{"command": "update", "arg1": 0, "arg2": 0}] @@ -28,6 +33,7 @@ class TerminalUpdate: channels: List[Dict[str, Any]] = field(default_factory=list) call_log: List[Dict[str, Any]] = field(default_factory=list) + trunk: List[Dict[str, Any]] = field(default_factory=list) class OP25Client: @@ -45,6 +51,23 @@ class OP25Client: logger.error(f"OP25 start failed: {e}") return False + async def start_with_reason(self) -> Optional[str]: + """Start OP25; None on success, else a human-readable reason.""" + try: + async with httpx.AsyncClient(timeout=10) as client: + r = await client.post(f"{self.api_url}/op25/start") + except Exception as e: + logger.error(f"OP25 start failed: {e}") + return "The op25 container isn't answering." + if r.is_success: + return None + try: + detail = r.json().get("detail") + except Exception: + detail = None + logger.error(f"OP25 start failed: HTTP {r.status_code} {detail or r.text[:200]}") + return detail or f"op25 container returned HTTP {r.status_code}." + async def stop(self) -> bool: try: async with httpx.AsyncClient(timeout=10) as client: @@ -65,6 +88,17 @@ class OP25Client: logger.error(f"OP25 status failed: {e}") return None + async def log(self, lines: int = 80) -> Optional[List[str]]: + """Tail of multi_rx/liquidsoap output, kept by the op25 container.""" + try: + async with httpx.AsyncClient(timeout=5) as client: + r = await client.get(f"{self.api_url}/op25/log", params={"lines": lines}) + r.raise_for_status() + return r.json().get("lines", []) + except Exception as e: + logger.error(f"OP25 log fetch failed: {e}") + return None + async def devices(self) -> Optional[Dict[str, Any]]: try: async with httpx.AsyncClient(timeout=5) as client: @@ -144,6 +178,11 @@ def parse_terminal_messages(messages: Any) -> TerminalUpdate: if isinstance(channel, dict): update.channels.append(channel) + elif json_type == "trunk_update": + for key, system in msg.items(): + if key != "json_type" and isinstance(system, dict): + update.trunk.append(system) + elif json_type == "call_log": for entry in msg.get("log") or []: if isinstance(entry, dict): diff --git a/drb-edge-node/app/main.py b/drb-edge-node/app/main.py index eadb50e..7b00c73 100644 --- a/drb-edge-node/app/main.py +++ b/drb-edge-node/app/main.py @@ -327,9 +327,12 @@ async def lifespan(app: FastAPI): if (node_cfg.override_system_id and node_cfg.override_config) else node_cfg.system_config ) - if node_cfg.configured and active_config: + # Not gated on `configured` (set only by a C2 assignment): a standalone or + # not-yet-approved node listening to a system chosen on the local + # dashboard (an override) must come back on it after a reboot too. + if active_config: from app.internal.op25_client import op25_client - logger.info("Node is configured — waiting for OP25 API then generating config.") + logger.info("Node has a system to listen to — waiting for OP25 API then generating config.") for attempt in range(10): if await _generate_op25_config(active_config): await op25_client.start() diff --git a/drb-edge-node/app/routers/api.py b/drb-edge-node/app/routers/api.py index 7cbf696..bba4175 100644 --- a/drb-edge-node/app/routers/api.py +++ b/drb-edge-node/app/routers/api.py @@ -1,12 +1,12 @@ from fastapi import APIRouter, Depends, HTTPException, Body from pydantic import BaseModel -from typing import Dict, List, Optional +from typing import Dict, List, Literal, Optional import asyncio import httpx from app.config import settings from app.models import SystemConfig, SECONDARY_SDR_MODES from app.internal.op25_client import op25_client -from app.internal.config_manager import load_node_config, save_node_config, apply_system_config +from app.internal.config_manager import load_node_config, save_node_config from app.internal.call_recorder import call_recorder from app.internal.discord_radio import radio_bot from app.internal.metadata_watcher import metadata_watcher @@ -36,8 +36,10 @@ async def get_status(): if (node_cfg.override_system_id and node_cfg.override_config) else node_cfg.system_config ) + system_type = None if active_config: system_name = active_config.name + system_type = active_config.type if active_tgid: # Reverse engineer the TGID to talkgroup name based on active system object tgs = active_config.config.get("talkgroups", []) @@ -58,6 +60,11 @@ async def get_status(): "configured": node_cfg.configured, "assigned_system_id": node_cfg.assigned_system_id, "system_name": system_name, + "system_type": system_type, + # True when what OP25 is tuned to was chosen on this dashboard rather + # than assigned by C2 (a local override, revertible to C2's system). + "local_choice": bool(node_cfg.override_system_id and node_cfg.override_config), + "has_c2_system": node_cfg.system_config is not None, "is_recording": call_recorder.is_recording, # Health of the PulseAudio capture that feeds every recording — the single # most useful signal when recordings come back empty. Segment boundaries @@ -77,15 +84,27 @@ async def get_status(): "icecast_url": ( f"http://{settings.icecast_host}:{settings.icecast_port}{settings.icecast_mount}" ), + # The browser builds its own URL from these + window.location.hostname; + # icecast_host is "localhost" from the node's point of view. + "icecast_port": settings.icecast_port, + "icecast_mount": settings.icecast_mount, "op25": op25_status, } +@router.get("/health") +async def get_health(): + """Plain-language setup/status checklist for the dashboard (see internal/health.py).""" + from app.internal import health + facts = await health.gather_facts() + return {"checks": health.build_checks(facts), "last_heard": facts["console"].get("last_heard")} + + @router.post("/op25/start") async def start_op25(): - ok = await op25_client.start() - if not ok: - raise HTTPException(500, "Failed to start OP25") + err = await op25_client.start_with_reason() + if err: + raise HTTPException(502, err) return {"ok": True} @@ -93,30 +112,135 @@ async def start_op25(): async def stop_op25(): ok = await op25_client.stop() if not ok: - raise HTTPException(500, "Failed to stop OP25") + raise HTTPException(502, "The op25 container did not answer the stop request.") return {"ok": True} +@router.get("/op25/log") +async def op25_log(lines: int = 80): + out = await op25_client.log(lines) + if out is None: + raise HTTPException(502, "The op25 container isn't answering.") + return {"lines": out} + + @router.get("/config") async def get_config(): return load_node_config() +async def _restart_op25_with(config: SystemConfig) -> None: + """Generate OP25's config for `config` and restart OP25 on it.""" + from app.main import _generate_op25_config + if not await _generate_op25_config(config): + raise HTTPException( + 502, f"The op25 container did not accept the config for {config.name} — is it running?", + ) + await op25_client.stop() + await asyncio.sleep(2) + err = await op25_client.start_with_reason() + if err: + raise HTTPException(502, f"Config saved, but OP25 did not start: {err}") + + +async def _apply_override(config: SystemConfig) -> None: + """Make `config` what this node listens to, chosen locally. + + Stored as an override so a C2-assigned system (if any) survives and can + be reverted to, and so C2 sees the node is overridden. A later C2 config + push replaces it. + """ + node_cfg = load_node_config() + node_cfg.override_system_id = config.system_id + node_cfg.override_config = config + save_node_config(node_cfg) + await _restart_op25_with(config) + if mqtt_manager.is_connected: + await mqtt_manager.publish_checkin() + + @router.post("/config/system") async def set_system_config(config: SystemConfig): """ - Apply a system config locally — called by the web UI or pushed by C2. - Writes the OP25 config and persists the node config. + Assign a system as if C2 had (persists as the node's own system, not an + override) and restart OP25 on it. + + Used to write the raw `config` blob straight to active.cfg.json, which is + not an OP25 config at all — OP25 could not start from it. """ + _validate_type(config.type) node_cfg = load_node_config() node_cfg.assigned_system_id = config.system_id node_cfg.system_config = config node_cfg.configured = True save_node_config(node_cfg) - apply_system_config(config) + await _restart_op25_with(config) return {"ok": True} +def _validate_type(system_type: str) -> None: + if str(system_type).upper() == "DMR": + raise HTTPException(400, "DMR isn't supported by this node's decoder yet — use P25 or analog (NBFM).") + if str(system_type).upper() not in ("P25", "NBFM"): + raise HTTPException(400, f"Unknown system type {system_type!r} — use P25 or NBFM.") + + +class Talkgroup(BaseModel): + id: int + name: str = "" + + +class ListenBody(BaseModel): + """What to listen to, chosen on the local dashboard.""" + mode: Literal["system", "p25", "analog"] + system_id: Optional[str] = None # mode=system: a cached C2 system + name: Optional[str] = None + frequencies: List[float] = [] # MHz; p25 = control channels, analog = exactly one + talkgroups: List[Talkgroup] = [] # p25 only; empty = every talkgroup + + +# RTL-SDR (R820T/R828D) tuning range. +_MIN_MHZ, _MAX_MHZ = 24.0, 1766.0 + + +@router.post("/listen") +async def listen(body: ListenBody): + """Point OP25 at a system or frequency and start it — no C2 needed.""" + if body.mode == "system": + from app.internal.system_cacher import get_cached_system + cached = get_cached_system(body.system_id or "") + if not cached: + raise HTTPException(404, "That system isn't in this node's saved list.") + config = SystemConfig(**cached) + _validate_type(config.type) + else: + freqs = body.frequencies + if not freqs: + raise HTTPException(400, "Enter at least one frequency in MHz.") + bad = [f for f in freqs if not (_MIN_MHZ <= f <= _MAX_MHZ)] + if bad: + raise HTTPException( + 400, f"{bad[0]:g} isn't a frequency an RTL-SDR can tune — enter MHz between " + f"{_MIN_MHZ:g} and {_MAX_MHZ:g} (e.g. 851.0125).", + ) + if body.mode == "analog" and len(freqs) != 1: + raise HTTPException(400, "Analog listens to exactly one frequency.") + is_p25 = body.mode == "p25" + first = f"{freqs[0]:.5f}".rstrip("0").rstrip(".") + label = (body.name or "").strip() or (f"P25 {first} MHz" if is_p25 else f"Analog {first} MHz") + config = SystemConfig( + system_id=f"local-{body.mode}", + name=label, + type="P25" if is_p25 else "NBFM", + config={ + "control_channels": freqs, + "talkgroups": [tg.model_dump() for tg in body.talkgroups] if is_p25 else [], + }, + ) + await _apply_override(config) + return {"ok": True, "name": config.name} + + @router.get("/systems") async def get_systems(): from app.internal.system_cacher import load_cached_systems, fetch_and_cache_systems @@ -129,9 +253,6 @@ async def set_override( system_id: Optional[str] = Body(None), system_config: Optional[dict] = Body(None) ): - node_cfg = load_node_config() - config = None - if system_id: from app.internal.system_cacher import get_cached_system cached = get_cached_system(system_id) @@ -145,19 +266,8 @@ async def set_override( else: raise HTTPException(400, "Must specify system_id or system_config.") - node_cfg.override_system_id = config.system_id - node_cfg.override_config = config - save_node_config(node_cfg) - - from app.main import _generate_op25_config - if not await _generate_op25_config(config): - raise HTTPException(500, f"Failed to generate OP25 config for override: {config.name}") - - await op25_client.stop() - await asyncio.sleep(2) - await op25_client.start() - - await mqtt_manager._publish_checkin() + _validate_type(config.type) + await _apply_override(config) return {"ok": True} @@ -172,15 +282,14 @@ async def revert_config(): save_node_config(node_cfg) if node_cfg.system_config: - from app.main import _generate_op25_config - if not await _generate_op25_config(node_cfg.system_config): - raise HTTPException(500, "Failed to regenerate original OP25 config.") - + await _restart_op25_with(node_cfg.system_config) + else: + # Nothing from C2 to go back to: stop rather than keep decoding a + # system the dashboard now says is cleared. await op25_client.stop() - await asyncio.sleep(2) - await op25_client.start() - await mqtt_manager._publish_checkin() + if mqtt_manager.is_connected: + await mqtt_manager.publish_checkin() return {"ok": True} diff --git a/drb-edge-node/app/templates/index.html b/drb-edge-node/app/templates/index.html index cfe08d5..4121d20 100644 --- a/drb-edge-node/app/templates/index.html +++ b/drb-edge-node/app/templates/index.html @@ -3,681 +3,568 @@ - DRB Edge Node Dashboard - + DRB Edge Node + -
-
-
-

DRB Edge Node

-

ID: Loading…

-
-
- - - Scanner Mode - - - - Logout - -
-
+
+
+
+

DRB Edge Node

+
Node ID …
+
+
+ Scanner view + Log out +
+
-