Compare commits

...
Author SHA1 Message Date
Logan CusanoandClaude Opus 5.5 797e14481b Scanner manual entry uses validated /api/listen; drop Hold mock; flag analog as live-only (node-26#20)
CI / lint (push) Successful in 6s
CI / lint (pull_request) Successful in 6s
CI / test (push) Successful in 41s
CI / test (pull_request) Successful in 39s
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-04 01:22:12 -04:00
Logan CusanoandClaude Opus 5.5 10f7db0a2c 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 <noreply@anthropic.com>
2026-10-04 01:20:11 -04:00
Logan CusanoandClaude Opus 5.5 6bf2c9dd93 Merge feat/sdr-pins: pin each SDR service to a dongle by serial; OP25 always by serial (closes #11)
CI / lint (push) Successful in 10s
CI / test (push) Successful in 51s
Build edge-node / build (push) Successful in 58s
Build secondary-sdr / build (push) Successful in 2m39s
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-27 14:41:28 -04:00
Logan CusanoandClaude Opus 5.5 575a99feb5 checkin: report an unreachable secondary-sdr as null, never keep stale 'Running'
QA blocker: when the secondary-sdr container was down, the checkin
omitted secondary_sdr_running / sdr_devices / op25_sdr_serial, and C2
only overwrites keys that are present, so the dashboard kept the last
'Running' forever. Send explicit nulls. The local card also says 'not
reported' rather than 'not plugged in' for a pin it can't check.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-27 14:41:27 -04:00
12 changed files with 1739 additions and 1002 deletions
+300
View File
@@ -0,0 +1,300 @@
"""
Plain-language health checklist for the local dashboard (/api/health).
A first-time owner opens http://<node>/ 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).",
"Live listening only: analog calls are not recorded or uploaded yet (node-26#20).",
))
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,
}
@@ -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."""
@@ -236,6 +236,12 @@ class MQTTManager:
}
payload["sdr_pins"] = config.sdr_pins
secondary = await secondary_sdr_client.status()
# Unreachable secondary-sdr: say so with explicit nulls. C2 only
# overwrites keys present in the checkin, so omitting them would leave
# a dead decoder showing "Running" forever.
payload["secondary_sdr_running"] = None
payload["sdr_devices"] = None
payload["op25_sdr_serial"] = None
if secondary is not None:
payload["secondary_sdr_running"] = [r["mode"] for r in secondary.get("running", [])]
if secondary.get("devices") is not None:
+40 -1
View File
@@ -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):
+5 -2
View File
@@ -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()
+141 -32
View File
@@ -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}
File diff suppressed because it is too large Load Diff
+14 -25
View File
@@ -124,7 +124,7 @@
}
.softkey:last-child { border-right: none; }
/* Overlay UI (Volume, Hold) */
/* Overlay UI (Volume, status) */
.overlay {
position: absolute;
top: 50%; left: 50%;
@@ -153,7 +153,6 @@
<ul>
<li><span>Up Arrow</span> Volume Up</li>
<li><span>Down Arrow</span> Volume Down</li>
<li><span>Enter / H</span> Toggle Hold (UI mock)</li>
<li><span>P</span> Play/Pause Stream</li>
<li><span>S</span> Select System</li>
<li><span>R</span> Revert Override</li>
@@ -171,15 +170,14 @@
<h3 style="color: var(--dept-color); margin-bottom: 6px; font-size: 13px;">Or Enter Freq Manually:</h3>
<div style="display: grid; grid-template-columns: 1fr 1fr; gap: 5px; font-size: 11px; margin-bottom: 8px;">
<div>
<label>Control Freq (MHz/Hz):</label>
<label>Freq (MHz) — P25 control channel or analog:</label>
<input type="text" id="manual-freq" style="background:#222; color:#fff; border:1px solid #555; padding:3px; width:100%; border-radius:2px;">
</div>
<div>
<label>System Type:</label>
<select id="manual-type" style="background:#222; color:#fff; border:1px solid #555; padding:3px; width:100%; border-radius:2px;">
<option value="P25">P25</option>
<option value="DMR">DMR</option>
<option value="NBFM">NBFM</option>
<option value="NBFM">Analog FM</option>
</select>
</div>
</div>
@@ -208,11 +206,11 @@
<div class="softkeys">
<div class="softkey">VOL (+/-)</div>
<div class="softkey">HOLD (H)</div>
<div class="softkey">SYSTEM (S)</div>
<div class="softkey">PLAY (P)</div>
</div>
<div class="overlay" id="overlay">HOLD</div>
<div class="overlay" id="overlay"></div>
<!-- Audio element for local playback -->
<audio id="audio-player" autoplay></audio>
@@ -231,7 +229,6 @@
const elOverlay = document.getElementById('overlay');
let currentIcecastUrl = null;
let isHolding = false;
let volume = 1.0;
// Clock updater
@@ -309,7 +306,7 @@
const container = document.getElementById('system-list');
container.innerHTML = '';
if (systems.length === 0) {
container.innerHTML = '<div style="padding: 10px; color: #888;">No cached systems found.</div>';
container.innerHTML = '<div style="padding: 10px; color: #888;">No server systems saved yet (they arrive once the node is approved). Enter a frequency below.</div>';
return;
}
systems.forEach((sys, index) => {
@@ -373,25 +370,20 @@
}
closeSystemModal();
showOverlay("APPLYING MANUAL", 3000);
const manualConfig = {
system_id: "manual-override",
name: `Freq ${freqVal} (${typeVal})`,
type: typeVal,
config: {
control_channels: [freqVal],
talkgroups: []
}
};
// Same validated path as the dashboard's "Listen locally" (MHz range check).
const mode = typeVal === 'NBFM' ? 'analog' : 'p25';
const freqs = freqVal.split(/[\s,;]+/).filter(Boolean).map(Number);
try {
const res = await fetch('/api/config/override', {
const res = await fetch('/api/listen', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ system_config: manualConfig })
body: JSON.stringify({ mode, frequencies: freqs })
});
if (res.ok) {
showOverlay('MANUAL OVERRIDE OK', 1500);
showOverlay('MANUAL OK', 1500);
} else {
showOverlay('MANUAL OVERRIDE FAIL', 2000);
const d = await res.json().catch(() => ({}));
showOverlay((d.detail || 'MANUAL FAILED').slice(0, 60), 4000);
}
} catch (err) {
showOverlay('ERROR APPLYING', 2000);
@@ -444,9 +436,6 @@
audioPlayer.volume = volume;
elVol.textContent = `VOL: ${Math.round(volume * 100)}%`;
showOverlay(`VOL: ${Math.round(volume * 100)}%`);
} else if (e.key === 'Enter' || e.key === 'h' || e.key === 'H') {
isHolding = !isHolding;
showOverlay(isHolding ? 'HOLD ON' : 'HOLD OFF');
} else if (e.key === 'p' || e.key === 'P') {
if (audioPlayer.paused) {
audioPlayer.play();
+248
View File
@@ -0,0 +1,248 @@
"""
Local dashboard: the health checklist (/api/health) and choosing what to
listen to without C2 (/api/listen, /api/config/system, /api/config/revert).
Same standalone-app pattern as test_auth_endpoints.py — the real routers,
none of app.main's lifespan. OP25/MQTT are mocked at the client boundary.
"""
import base64
import time
from unittest.mock import AsyncMock, patch
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from app.config import settings
from app.internal import config_manager, credentials, health
from app.internal.op25_client import parse_terminal_messages
from app.routers import api, ui
@pytest.fixture(autouse=True)
def isolated(tmp_path, monkeypatch):
monkeypatch.setattr(credentials, "_CREDS_FILE", tmp_path / "credentials.json")
monkeypatch.setattr(credentials, "_api_key", None)
monkeypatch.setattr(credentials, "_auth_salt", None)
monkeypatch.setattr(credentials, "_session_secret", None)
monkeypatch.setattr(config_manager, "_CONFIG_FILE", tmp_path / "node_config.json")
monkeypatch.setattr(settings, "dashboard_username", "tester")
monkeypatch.setattr(settings, "dashboard_password", "s3cret-pass")
yield
@pytest.fixture
def client():
app = FastAPI()
app.include_router(api.router)
app.include_router(ui.router)
auth = base64.b64encode(b"tester:s3cret-pass").decode()
with TestClient(app, headers={"Authorization": f"Basic {auth}"}) as c:
yield c
@pytest.fixture
def op25():
"""Mock the OP25 side of an apply: generate, stop, start, checkin."""
with patch("app.main._generate_op25_config", AsyncMock(return_value=True)) as gen, \
patch("app.routers.api.op25_client.stop", AsyncMock(return_value=True)) as stop, \
patch("app.routers.api.op25_client.start_with_reason", AsyncMock(return_value=None)) as start, \
patch("app.routers.api.asyncio.sleep", AsyncMock()), \
patch("app.routers.api.mqtt_manager.publish_checkin", AsyncMock()) as checkin:
yield {"gen": gen, "stop": stop, "start": start, "checkin": checkin}
# ---------------------------------------------------------------------------
# Health checklist — pure function of facts
# ---------------------------------------------------------------------------
def _facts(**over):
now = 1_000_000.0
f = {
"now": now, "node_id": "toughny", "default_password": False,
"c2_url": "https://api.example", "has_api_key": True, "mqtt_connected": True,
"mqtt_broker": "mqtt.example", "mqtt_port": 8883, "sdr_count": 1,
"source": {"name": "County", "type": "P25", "system_id": "s1", "local": False},
"op25_status": "running",
"console": {"last_poll_epoch": now - 1, "channels": [], "trunk": [{"last_tsbk": now - 1, "rxchan": 851012500}],
"last_heard": None},
"audio_capturing": True, "last_voice_epoch": now - 30, "icecast_live": True, "icecast_port": 8000,
}
f.update(over)
return f
def _by_id(checks):
return {c["id"]: c for c in checks}
def test_healthy_node_is_all_ok():
checks = health.build_checks(_facts())
assert all(c["state"] == "ok" for c in checks), [c for c in checks if c["state"] != "ok"]
assert "851.0125 MHz" in _by_id(checks)["decode"]["summary"]
def test_unapproved_node_explains_wait_and_mqtt_refusal():
"""Toughny's state: C2 set, no API key, broker refuses, nothing chosen."""
c = _by_id(health.build_checks(_facts(has_api_key=False, mqtt_connected=False, source=None,
op25_status="stopped")))
assert c["c2"]["state"] == "warn" and "approve" in c["c2"]["action"]
assert "toughny" in c["c2"]["action"]
assert c["mqtt"]["state"] == "warn"
assert c["source"]["state"] == "warn" and "Listen locally" in c["source"]["action"]
assert c["op25"]["state"] == "off"
assert c["decode"]["state"] == "off"
def test_key_on_disk_but_mqtt_down_says_restart():
c = _by_id(health.build_checks(_facts(mqtt_connected=False)))
assert c["mqtt"]["state"] == "error"
assert "restart the edge-node container" in c["mqtt"]["action"]
def test_standalone_node_is_not_an_error():
c = _by_id(health.build_checks(_facts(c2_url=None, has_api_key=False, mqtt_connected=False)))
assert c["c2"]["state"] == "off"
assert c["mqtt"]["state"] == "off"
def test_no_sdr_and_crashed_op25():
c = _by_id(health.build_checks(_facts(sdr_count=0, op25_status="crashed")))
assert c["sdr"]["state"] == "error"
assert c["op25"]["state"] == "error" and "log" in c["op25"]["action"]
def test_running_but_no_control_channel():
now = 1_000_000.0
c = _by_id(health.build_checks(_facts(console={"last_poll_epoch": now, "channels": [],
"trunk": [{"last_tsbk": 0}], "last_heard": None})))
assert c["decode"]["state"] == "error"
def test_analog_needs_no_control_channel():
now = 1_000_000.0
c = _by_id(health.build_checks(_facts(
source={"name": "Fire", "type": "NBFM", "system_id": "local-analog", "local": True},
console={"last_poll_epoch": now, "channels": [{"freq": 154430000}], "trunk": [], "last_heard": None},
)))
assert c["decode"]["state"] == "ok" and "154.4300 MHz" in c["decode"]["summary"]
assert "not recorded" in c["decode"]["action"]
assert "this dashboard" in c["source"]["summary"]
def test_op25_unreachable_and_icecast_down():
c = _by_id(health.build_checks(_facts(op25_status=None, icecast_live=None)))
assert c["op25"]["state"] == "error"
assert c["stream"]["state"] == "error"
def test_default_password_warns():
assert _by_id(health.build_checks(_facts(default_password=True)))["password"]["state"] == "warn"
def test_trunk_update_is_parsed():
update = parse_terminal_messages([
{"json_type": "trunk_update", "0": {"last_tsbk": 5.0, "rxchan": 851012500}, "nac": 0},
{"json_type": "channel_update", "channels": ["0"], "0": {"freq": 1, "tgid": None}},
])
assert update.trunk == [{"last_tsbk": 5.0, "rxchan": 851012500}]
assert len(update.channels) == 1
def test_health_route_returns_checks(client):
facts = _facts(now=time.time())
with patch("app.internal.health.gather_facts", AsyncMock(return_value=facts)):
r = client.get("/api/health")
assert r.status_code == 200
assert [c["id"] for c in r.json()["checks"]][:3] == ["password", "c2", "mqtt"]
# ---------------------------------------------------------------------------
# Choosing what to listen to
# ---------------------------------------------------------------------------
def test_listen_p25_saves_override_and_restarts_op25(client, op25):
r = client.post("/api/listen", json={
"mode": "p25", "frequencies": [851.0125, 851.5125], "talkgroups": [{"id": 1001, "name": "Fire"}],
})
assert r.status_code == 200, r.text
cfg = config_manager.load_node_config()
assert cfg.override_config.type == "P25"
assert cfg.override_config.config["control_channels"] == [851.0125, 851.5125]
assert cfg.override_config.config["talkgroups"] == [{"id": 1001, "name": "Fire"}]
op25["gen"].assert_awaited_once()
op25["start"].assert_awaited_once()
op25["checkin"].assert_not_awaited() # MQTT not connected in tests
def test_listen_analog_single_frequency(client, op25):
r = client.post("/api/listen", json={"mode": "analog", "frequencies": [154.43], "name": "Fire"})
assert r.status_code == 200, r.text
cfg = config_manager.load_node_config().override_config
assert (cfg.type, cfg.name, cfg.config["control_channels"]) == ("NBFM", "Fire", [154.43])
@pytest.mark.parametrize("body,needle", [
({"mode": "p25", "frequencies": []}, "at least one"),
({"mode": "p25", "frequencies": [851012500]}, "MHz"),
({"mode": "analog", "frequencies": [154.43, 155.1]}, "exactly one"),
({"mode": "system", "system_id": "nope"}, "saved list"),
])
def test_listen_rejects_bad_input_with_a_readable_reason(client, op25, body, needle):
r = client.post("/api/listen", json=body)
assert r.status_code in (400, 404)
assert needle in r.json()["detail"]
op25["gen"].assert_not_awaited()
def test_listen_reports_op25_start_failure(client, op25):
op25["start"].return_value = "No OP25 config yet — choose a system or frequency first."
r = client.post("/api/listen", json={"mode": "analog", "frequencies": [154.43]})
assert r.status_code == 502
assert "did not start" in r.json()["detail"]
def test_dmr_is_refused_up_front(client, op25):
r = client.post("/api/config/override", json={"system_config": {
"system_id": "x", "name": "x", "type": "DMR", "config": {"control_channels": [450.1]},
}})
assert r.status_code == 400 and "DMR" in r.json()["detail"]
op25["gen"].assert_not_awaited()
def test_config_system_goes_through_op25_generation(client, op25, tmp_path, monkeypatch):
"""It used to dump the raw blob into active.cfg.json — not an OP25 config."""
monkeypatch.setattr(settings, "config_path", str(tmp_path))
r = client.post("/api/config/system", json={
"system_id": "s1", "name": "County", "type": "P25", "config": {"control_channels": [851.0125]},
})
assert r.status_code == 200, r.text
op25["gen"].assert_awaited_once()
assert not (tmp_path / "active.cfg.json").exists()
cfg = config_manager.load_node_config()
assert cfg.configured and cfg.system_config.name == "County"
def test_revert_without_c2_system_stops_op25(client, op25):
client.post("/api/listen", json={"mode": "analog", "frequencies": [154.43]})
op25["stop"].reset_mock()
r = client.post("/api/config/revert")
assert r.status_code == 200
assert config_manager.load_node_config().override_config is None
op25["stop"].assert_awaited_once()
def test_status_exposes_stream_parts_and_source(client, op25):
client.post("/api/listen", json={"mode": "analog", "frequencies": [154.43]})
with patch("app.routers.api.op25_client.status", AsyncMock(return_value={"status": "running"})):
d = client.get("/api/status").json()
assert d["icecast_port"] == settings.icecast_port and d["icecast_mount"] == settings.icecast_mount
assert d["local_choice"] is True and d["has_c2_system"] is False
assert d["system_type"] == "NBFM"
def test_dashboard_page_has_no_dead_api_instruction(client):
client.post("/login", data={"username": "tester", "password": "s3cret-pass"}, follow_redirects=False)
html = client.get("/").text
assert "POST /api/config/system" not in html
assert "/api/listen" in html and "/api/health" in html
+14 -6
View File
@@ -52,12 +52,16 @@ class IcecastConfig(BaseModel):
class ConfigGenerator(BaseModel):
type: DecodeMode
systemName: str
# P25: control channel list (Hz). NBFM: channels[0] is the one analog
# frequency to listen to (Hz).
channels: List[Union[int, str]]
tags: Optional[List[TalkgroupTag]]
whitelist: Optional[List[int]]
icecastConfig: Optional[IcecastConfig]
tags: Optional[List[TalkgroupTag]] = None
whitelist: Optional[List[int]] = None
icecastConfig: Optional[IcecastConfig] = None
hardware_preset: str = "rtl-sdr-v3"
ppm_override: Optional[float] = None # when set, overrides the preset's default PPM
# NBFM only: power squelch threshold in dB (op25_nbfm default is -60).
nbfm_squelch: Optional[float] = None
class DemodType(str, Enum):
CQPSK = "cqpsk"
@@ -69,15 +73,16 @@ class FilterType(str, Enum):
class ChannelConfig(BaseModel):
name: str
trunking_sysname: Optional[str]
trunking_sysname: Optional[str] = None
enable_analog: str
meta_stream_name: str
demod_type: DemodType
filter_type: FilterType
device: Optional[str] = "sdr"
cqpsk_tracking: Optional[bool] = None
frequency: Optional[float] = None
nbfmSquelch: Optional[float] = None
frequency: Optional[int] = None
# Key name op25_nbfm.py actually reads (from_dict(config, 'nbfm_squelch_threshold', -60)).
nbfm_squelch_threshold: Optional[float] = None
destination: Optional[str] = "udp://127.0.0.1:23456"
tracking_threshold: Optional[int] = 120
tracking_feedback: Optional[float] = 0.75
@@ -93,6 +98,9 @@ class DeviceConfig(BaseModel):
gains: Optional[str] = "lna:39"
gain_mode: Optional[bool] = False
name: Optional[str] = "sdr"
# Centre frequency (Hz). Unset for trunked P25, where tk_p25 tunes the
# device; required for a fixed analog channel (tunable=False).
frequency: Optional[int] = None
offset: Optional[int] = 0
ppm: Optional[float] = 0.0
rate: Optional[int] = 1920000
+125 -20
View File
@@ -3,7 +3,11 @@ import subprocess
import os
import re
import signal
import sys
import json
import threading
from collections import deque
from typing import Optional
from models import ConfigGenerator, DecodeMode, ChannelConfig, DeviceConfig, TrunkingConfig, TrunkingChannelConfig, TerminalConfig, MetadataConfig, MetadataStreamConfig, HARDWARE_PRESETS
from internal.logger import create_logger
from internal.op25_config_utls import save_talkgroup_tags, save_whitelist, del_none_in_dict, get_current_system_from_config
@@ -14,6 +18,16 @@ LOGGER = create_logger(__name__)
OP25_PATH = "/op25/op25/gr-op25_repeater/apps/"
OP25_SCRIPT = "run_multi-rx_service.sh"
_PGID_FILE = "/tmp/op25.pgid"
_ACTIVE_CFG = "/configs/active.cfg.json"
# Analog: park the dongle's centre this far from the channel so the channel
# does not sit on the RTL-SDR's DC spike. Well inside the 1.92 MS/s window.
ANALOG_CENTER_OFFSET_HZ = 250_000
# Last lines multi_rx/liquidsoap printed — the only place the reason OP25
# died (SDR busy, bad config, no lock) is visible without SSH + docker logs.
_log: deque = deque(maxlen=300)
_proc: Optional[subprocess.Popen] = None
def _save_pgid(pgid: int) -> None:
@@ -28,15 +42,60 @@ def _read_pgid():
return None
def _multi_rx_alive(pgid: int) -> bool:
"""True only if a live multi_rx.py process is in OP25's process group.
node-26#12: killpg(pgid, 0) succeeds while the wrapper shell or
liquidsoap outlives a dead multi_rx (and even for a zombie leader), so
the old check reported a dead scanner as "running".
"""
try:
pids = [d for d in os.listdir("/proc") if d.isdigit()]
except OSError:
return False
for pid in pids:
try:
with open(f"/proc/{pid}/stat") as f:
fields = f.read().rsplit(")", 1)[1].split()
state, pgrp = fields[0], int(fields[2])
if pgrp != pgid or state == "Z":
continue
with open(f"/proc/{pid}/cmdline", "rb") as f:
if b"multi_rx.py" in f.read():
return True
except (OSError, IndexError, ValueError):
continue
return False
def _reap() -> None:
if _proc is not None:
_proc.poll()
def _is_running() -> bool:
pgid = _read_pgid()
if pgid is None:
return False
_reap()
return _multi_rx_alive(pgid)
def _kill_group(pgid: int) -> None:
try:
os.killpg(pgid, 0)
return True
os.killpg(pgid, signal.SIGTERM)
except OSError:
return False
pass
def _pump(proc: subprocess.Popen) -> None:
"""Mirror OP25's output to the container log and keep the tail for /log."""
for line in proc.stdout:
sys.stdout.write(line)
sys.stdout.flush()
_log.append(line.rstrip("\n"))
rc = proc.wait()
_log.append(f"[OP25 exited with code {rc}]")
def create_op25_router():
@@ -44,11 +103,28 @@ def create_op25_router():
@router.post("/start")
async def start_op25():
global _proc
if _is_running():
return {"status": "OP25 already running"}
if not os.path.exists(_ACTIVE_CFG):
raise HTTPException(
status_code=409,
detail="No OP25 config yet — choose a system or frequency first.",
)
stale = _read_pgid()
if stale is not None:
# multi_rx died but liquidsoap/the wrapper may still hold the
# Icecast mount — clear the old group before starting a new one.
_kill_group(stale)
try:
proc = subprocess.Popen(os.path.join(OP25_PATH, OP25_SCRIPT), shell=True, preexec_fn=os.setsid, cwd=OP25_PATH)
_log.clear()
proc = subprocess.Popen(
os.path.join(OP25_PATH, OP25_SCRIPT), shell=True, preexec_fn=os.setsid, cwd=OP25_PATH,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1,
)
_proc = proc
_save_pgid(proc.pid)
threading.Thread(target=_pump, args=(proc,), daemon=True).start()
LOGGER.debug(proc)
return {"status": "OP25 started"}
except Exception as e:
@@ -57,10 +133,10 @@ def create_op25_router():
@router.post("/stop")
async def stop_op25():
pgid = _read_pgid()
if pgid is None or not _is_running():
if pgid is None:
return {"status": "OP25 is not running"}
try:
os.killpg(pgid, signal.SIGTERM)
_kill_group(pgid)
os.remove(_PGID_FILE)
return {"status": "OP25 stopped"}
except Exception as e:
@@ -68,7 +144,18 @@ def create_op25_router():
@router.get("/status")
async def get_status():
return {"status": "running" if _is_running() else "stopped"}
if _is_running():
return {"status": "running"}
# Started (pgid file present, never stopped via the API) but multi_rx
# is gone: it crashed. The tail of /log says why.
if _read_pgid() is not None:
return {"status": "crashed", "configured": os.path.exists(_ACTIVE_CFG)}
return {"status": "stopped", "configured": os.path.exists(_ACTIVE_CFG)}
@router.get("/log")
async def get_log(lines: int = 80):
lines = max(1, min(lines, _log.maxlen))
return {"lines": list(_log)[-lines:]}
@router.get("/devices")
async def list_sdr_devices():
@@ -104,7 +191,7 @@ def create_op25_router():
meta_stream_name="stream_0"
)]
devices = [DeviceConfig(gains=preset["gains"], ppm=ppm)]
save_talkgroup_tags(generator.tags)
save_talkgroup_tags(generator.tags or [])
save_whitelist(generator.whitelist)
has_talkgroups = bool(generator.whitelist)
trunking = TrunkingConfig(
@@ -142,30 +229,48 @@ def create_op25_router():
"terminal": terminal.dict()
}
elif generator.type == DecodeMode.ANALOG:
generator = generator.config
if not generator.channels:
raise HTTPException(status_code=400, detail="Analog needs a frequency.")
preset = HARDWARE_PRESETS.get(generator.hardware_preset, HARDWARE_PRESETS["other"])
ppm = generator.ppm_override if generator.ppm_override is not None else preset["ppm"]
freq = int(float(generator.channels[0]))
# One fixed conventional channel: no trunking section, so
# multi_rx adds its default tk_p25 with no chans and reports
# the channel as `conventional` on the HTTP terminal.
channels = [ChannelConfig(
channelName=generator.systemName,
enableAnalog="on",
demodType="fsk4",
frequency=generator.frequency,
filterType="widepulse",
nbfmSquelch=generator.nbfmSquelch
name=generator.systemName,
trunking_sysname="",
enable_analog="on",
demod_type="fsk4",
filter_type="rc",
frequency=freq,
meta_stream_name="",
nbfm_squelch_threshold=generator.nbfm_squelch,
)]
devices = [DeviceConfig(gain="LNA:32")]
devices = [DeviceConfig(
gains=preset["gains"], ppm=ppm, tunable=False,
frequency=freq + ANALOG_CENTER_OFFSET_HZ,
)]
if generator.icecastConfig:
generate_liquid_script(generator.icecastConfig)
config_dict = {
"channels": [channel.dict() for channel in channels],
"devices": [device.dict() for device in devices]
"devices": [device.dict() for device in devices],
"terminal": TerminalConfig().dict(),
}
else:
raise HTTPException(status_code=400, detail="Invalid configuration type. Must be 'p25' or 'nbfm'.")
raise HTTPException(status_code=400, detail="Unsupported system type. Must be 'P25' or 'NBFM' (DMR is not implemented).")
with open('/configs/active.cfg.json', 'w') as f:
with open(_ACTIVE_CFG, 'w') as f:
json.dump(del_none_in_dict(config_dict), f, indent=2)
return {"message": "Config exported to '/configs/active.cfg.json'"}
return {"message": f"Config exported to '{_ACTIVE_CFG}'"}
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
+9
View File
@@ -20,3 +20,12 @@ LIQ_PID=$! # Store the PID of the op25.liq process
# The trap command ensures that SIGTERM is passed to the background jobs.
trap "kill $MULTI_RX_PID $LIQ_PID" SIGTERM SIGINT
wait $MULTI_RX_PID
RC=$?
# multi_rx is gone (crashed, or SDR could not be opened). Take liquidsoap down
# with it so nothing in this process group outlives the decoder — a leftover
# liquidsoap kept /op25/status reporting "running" (node-26#12) and held the
# Icecast mount against the next start.
echo "multi_rx.py exited (code $RC) — stopping liquidsoap."
kill $LIQ_PID 2>/dev/null
exit $RC