diff --git a/drb-edge-node/app/config.py b/drb-edge-node/app/config.py index 1db01b5..eff9557 100644 --- a/drb-edge-node/app/config.py +++ b/drb-edge-node/app/config.py @@ -1,5 +1,5 @@ from pydantic_settings import BaseSettings -from typing import Optional +from typing import Literal, Optional class Settings(BaseSettings): @@ -170,6 +170,46 @@ class Settings(BaseSettings): # 120 s; a failure here just lands in the spool. upload_inline_timeout_seconds: float = 30.0 + # ------------------------------------------------------------------ + # Recorder backend (node-26#45) + # + # op25 OP25 decodes, Liquidsoap mixes into PulseAudio, and + # call_recorder/metadata_watcher cut calls out of that + # one stream. The default; every existing node. + # trunk-recorder Trunk Recorder owns the SDR and records every grant at + # once, with call boundaries from the decoder. Its + # uploadScript drops each finished call in tr_inbox_dir + # and app/internal/trunk_recorder.py uploads it through + # the same spool. OP25 is never started (it would fight + # TR for the dongle) and the PulseAudio recorder stays + # off, so a call is never uploaded twice. Needs the + # compose profile too: COMPOSE_PROFILES=trunk-recorder. + # ------------------------------------------------------------------ + recorder_backend: Literal["op25", "trunk-recorder"] = "op25" + # Shared with the trunk-recorder container (mounted there as /app): + # config.json, the CSVs, its logs, and the supervisor's status file. + tr_config_dir: Optional[str] = None # default /trunk-recorder + # Where TR's uploadScript hands over finished calls (default + # /tr-inbox). Same bind mount in both containers. + tr_inbox_dir: Optional[str] = None + # Source (SDR) settings for the generated config. 2.4 Msps is the most an + # RTL-SDR does without dropping samples; ~2 MHz of it is usable. + tr_sample_rate: int = 2_400_000 + tr_gain: float = 34.0 # dB; matches op25-container's rtl-sdr-v3 preset (lna:34) + tr_center_hz: Optional[int] = None # override the computed centre frequency + # Simultaneous voice channels on a trunked system. Each costs CPU (on a + # Pi 4 roughly 4-6 is the ceiling); conventional channels get one each. + tr_digital_recorders: int = 4 + # TR deletes calls shorter than this. Matches the key-up noise the old + # recorder's trim + all-silence check threw away. + tr_min_duration: float = 1.0 + # Conventional squelch, absolute dB (TR has no noise squelch). Used when + # the system config gives no squelch_threshold. Needs hand tuning per site. + tr_squelch_db: float = -50.0 + # simplestream UDP port for live audio (TR -> edge-node -> PulseAudio -> + # Discord). Local only. + tr_live_port: int = 9123 + # ------------------------------------------------------------------ # Local dashboard / API authentication # diff --git a/drb-edge-node/app/internal/health.py b/drb-edge-node/app/internal/health.py index 66ea358..361aa28 100644 --- a/drb-edge-node/app/internal/health.py +++ b/drb-edge-node/app/internal/health.py @@ -8,6 +8,10 @@ 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) +With RECORDER_BACKEND=trunk-recorder (node-26#45) the last four are replaced +by Trunk Recorder's: supervisor state, control-channel decode rate (from its +log), recordings handed over, and the live feed to Discord. + `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. @@ -208,6 +212,10 @@ def build_checks(f: Dict[str, Any]) -> List[Dict[str, str]]: kind = "P25 conventional" if src.get("mode") == "conventional" else src["type"] checks.append(_check("source", "What to listen to", OK, f"{src['name']} ({kind}) — {origin}.")) + if f.get("backend") == "trunk-recorder": + checks.extend(_trunk_recorder_checks(f, src, now)) + return checks + # 6. OP25 op25 = f.get("op25_status") running = op25 == "running" @@ -302,6 +310,104 @@ def build_checks(f: Dict[str, Any]) -> List[Dict[str, str]]: return checks +def _trunk_recorder_checks(f: Dict[str, Any], src: Optional[Dict[str, Any]], now: float) -> List[Dict[str, str]]: + """Checks 6-9 when Trunk Recorder is the recorder (node-26#45).""" + tr = f.get("tr") or {} + checks: List[Dict[str, str]] = [] + status = tr.get("status") + state = (status or {}).get("state") + running = False + label = "Radio recorder (Trunk Recorder)" + if tr.get("config_error"): + checks.append(_check( + "recorder", label, ERROR, f"Couldn't build its config — {tr['config_error']}", + "Fix the system's channel list (or TR_* settings in .env); the node retries on the next config push.", + )) + elif status is None: + checks.append(_check( + "recorder", label, ERROR, "The trunk-recorder service isn't running.", + "Set COMPOSE_PROFILES=trunk-recorder in the node's .env, then run `docker compose up -d`.", + )) + elif status.get("stale"): + checks.append(_check( + "recorder", label, ERROR, f"Its supervisor stopped reporting {_ago(status.get('age_seconds') or 0)}.", + "Check the trunk-recorder container: `docker compose logs trunk-recorder`.", + )) + elif state == "running": + running = True + up = now - float(status.get("started_at") or now) + restarts = int(status.get("restarts") or 0) + checks.append(_check( + "recorder", label, OK if restarts < 3 else WARN, + f"Running for {_ago(up).replace(' ago', '')}" + + (f" — restarted {restarts} times since the container started." if restarts else "."), + "" if restarts < 3 else "Frequent restarts: `docker compose logs trunk-recorder` says why.", + )) + elif state == "crashed": + checks.append(_check( + "recorder", label, ERROR, f"Exited (code {status.get('last_exit')}) — restarting.", + "Usually the SDR is unplugged or held by another program. `docker compose logs trunk-recorder`.", + )) + elif state == "stopped": + checks.append(_check("recorder", label, WARN, "Stopped from the dashboard.", "Press Start.")) + elif not src: + checks.append(_check("recorder", label, OFF, "Waiting — choose what to listen to first.")) + else: + checks.append(_check("recorder", label, WARN, "Waiting for its config…")) + + mode = (src or {}).get("mode") + if not running: + checks.append(_check("decode", "Decoding", OFF, "Starts when Trunk Recorder runs.")) + elif mode == "trunked": + rate = tr.get("decode_rate") + age = tr.get("log_age_seconds") + if rate is None or age is None or age > 30: + checks.append(_check( + "decode", "Decoding", WARN, "No control-channel report from Trunk Recorder yet.", + "Give it 30 seconds after a start; if it stays, `docker compose logs trunk-recorder`.", + )) + elif rate > 0: + checks.append(_check("decode", "Decoding", OK, f"Decoding the control channel ({rate:g} messages/sec).")) + else: + checks.append(_check( + "decode", "Decoding", ERROR, "Not decoding the control channel (0 messages/sec).", + "Usually a wrong control-channel frequency, a weak signal/antenna, or SDR frequency error (PPM).", + )) + else: + checks.append(_check( + "decode", "Decoding", OK, "Listening (conventional) — channels record when their squelch opens.", + "If calls never arrive through a busy period, the squelch (TR_SQUELCH_DB or the system's " + "squelch_threshold) may be set too high.", + )) + + last = tr.get("last_call_epoch") + backlog = int(tr.get("inbox_backlog") or 0) + if not running and not last: + checks.append(_check("audio", "Recordings", OFF, "Nothing recorded yet.")) + elif backlog > 20: + checks.append(_check( + "audio", "Recordings", WARN, f"{backlog} finished calls waiting to be sent.", + "Normal for a minute after the server link comes back; if it keeps growing, check the C2 link.", + )) + elif last: + extra = f", {tr['encrypted_skipped']} encrypted skipped" if tr.get("encrypted_skipped") else "" + checks.append(_check( + "audio", "Recordings", OK, + f"Last call {_ago(now - last)} ({tr.get('calls_ingested', 0)} since start{extra}).", + )) + else: + checks.append(_check("audio", "Recordings", OK, "Recorder up — no call yet.")) + + live = tr.get("live") or {} + heard = live.get("last_packet_epoch") + checks.append(_check( + "stream", "Live audio", OK if heard else OFF, + ("Feeding Discord — last audio " + _ago(now - heard) + "." if heard else "Feeds Discord once a call is heard.") + + " The local browser stream isn't available with Trunk Recorder.", + )) + 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" @@ -342,8 +448,11 @@ async def gather_facts() -> Dict[str, Any]: sdr_count = op25_devs["count"] activity = call_recorder.audio_activity() + from app.internal import trunk_recorder as tr return { "now": time.time(), + "backend": settings.recorder_backend, + "tr": tr.trunk_recorder.health_facts() if tr.enabled() else None, "node_id": settings.node_id, "default_password": auth.is_using_default_password(), "c2_url": settings.c2_url, diff --git a/drb-edge-node/app/internal/mqtt_manager.py b/drb-edge-node/app/internal/mqtt_manager.py index 576e993..a802535 100644 --- a/drb-edge-node/app/internal/mqtt_manager.py +++ b/drb-edge-node/app/internal/mqtt_manager.py @@ -419,6 +419,14 @@ class MQTTManager: last = spool["last_result"] or {} payload["last_upload_status"] = last.get("status") payload["last_upload_kind"] = last.get("kind") + # node-26#45: which recorder this node runs; C2 ignores unknown keys. + payload["recorder_backend"] = settings.recorder_backend + from app.internal import trunk_recorder as tr + if tr.enabled(): + facts = tr.trunk_recorder.health_facts() + payload["tr_state"] = (facts["status"] or {}).get("state") + payload["tr_decode_rate"] = facts["decode_rate"] + payload["tr_inbox_backlog"] = facts["inbox_backlog"] self._publish(self._t_checkin, payload, qos=1) def _publish(self, topic: str, payload: dict, qos: int = 0, retain: bool = False): diff --git a/drb-edge-node/app/internal/radio_backend.py b/drb-edge-node/app/internal/radio_backend.py new file mode 100644 index 0000000..1bf9b84 --- /dev/null +++ b/drb-edge-node/app/internal/radio_backend.py @@ -0,0 +1,55 @@ +""" +Start/stop "the radio" without every caller knowing which recorder backend is +active (node-26#45). Every path that used to call op25_client.start/stop — C2 +config pushes, the op25_restart command, the dashboard, SDR re-pinning — goes +through here. + +With RECORDER_BACKEND=trunk-recorder, OP25 is never started: both open the +same RTL-SDR, and OP25 holding it would starve Trunk Recorder. start() becomes +a restart request to trunk-recorder/run.sh instead. +""" +import time +from typing import Optional + +from app.internal import trunk_recorder as tr +from app.internal.logger import logger +from app.internal.op25_client import op25_client + + +def name() -> str: + return "Trunk Recorder" if tr.enabled() else "OP25" + + +async def start(wait: bool = True) -> bool: + return await start_with_reason(wait) is None + + +async def start_with_reason(wait: bool = True) -> Optional[str]: + """None on success, else a human-readable reason. + + For Trunk Recorder, `wait` blocks until run.sh reports TR restarted (up + to trunk_recorder.RESTART_WAIT_SECONDS), like op25_client.start returns + only once OP25 has started — callers hand the spare dongles out next. + """ + if not tr.enabled(): + return await op25_client.start_with_reason() + if tr.trunk_recorder.config_error: + return tr.trunk_recorder.config_error + requested = time.time() + if not tr.trunk_recorder.request_restart(): + return f"Couldn't write to {tr.config_dir()} to restart Trunk Recorder." + if not wait: + return None + err = await tr.trunk_recorder.wait_started_since(requested) + if err: + logger.error(err) + return err + + +async def stop() -> bool: + if not tr.enabled(): + return await op25_client.stop() + ok = tr.trunk_recorder.request_stop() + if ok: + logger.info("Trunk Recorder stop requested.") + return ok diff --git a/drb-edge-node/app/internal/sdr_settings.py b/drb-edge-node/app/internal/sdr_settings.py index 8d58369..464ac63 100644 --- a/drb-edge-node/app/internal/sdr_settings.py +++ b/drb-edge-node/app/internal/sdr_settings.py @@ -83,7 +83,18 @@ async def set_sdr_settings(priority: Optional[Iterable[str]] = None, cfg.sdr_pins = normalize_sdr_pins(pins) save_node_config(cfg) - if cfg.sdr_pins.get("op25") != old_op25: + if cfg.sdr_pins.get("op25") != old_op25 and settings.recorder_backend == "trunk-recorder": + # Trunk Recorder takes the "op25" dongle (node-26#45): rewrite its + # config with the new serial; run.sh restarts it on the change. + from app.internal.health import active_config + from app.main import _generate_op25_config + await secondary_sdr_client.apply([], {}, []) + current = active_config(cfg) + if current is not None and await _generate_op25_config(current): + from app.internal import radio_backend + await radio_backend.start() # waits until TR holds the new dongle + logger.info(f"Trunk Recorder moved to SDR {cfg.sdr_pins.get('op25') or 'rtl=0 (first dongle)'}") + elif cfg.sdr_pins.get("op25") != old_op25: from app.internal.op25_client import op25_client # Free the new dongle first if a decoder holds it, then move OP25. await secondary_sdr_client.apply([], {}, []) diff --git a/drb-edge-node/app/internal/tr_calls.py b/drb-edge-node/app/internal/tr_calls.py new file mode 100644 index 0000000..2cf3e68 --- /dev/null +++ b/drb-edge-node/app/internal/tr_calls.py @@ -0,0 +1,210 @@ +""" +Trunk Recorder call JSON -> what DRB sends for one call (node-26#45). + +TR writes one JSON per call next to its WAV and passes both to uploadScript. +The shape is `create_call_json` in trunk-recorder/call_concluder/ +call_concluder.cc at v5.2.1 (keys, in order: freq, freq_error, signal, noise, +source_num, recorder_num, tdma_slot, phase2_tdma, start_time, stop_time, +start_time_ms, stop_time_ms, emergency, priority, mode, duplex, encrypted, +call_length, call_length_ms, talkgroup, talkgroup_tag, talkgroup_description, +talkgroup_group_tag, talkgroup_group, color_code, audio_type, short_name, +patched_talkgroups?, freqList[], srcList[]). Older TR releases omit the *_ms +keys, so both are read. + +Pure: trunk_recorder.py does the I/O. +""" +import json +import uuid +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + +DRIVER = "trunk-recorder" +# Fixed namespace so a call's id is a pure function of its JSON: re-reading the +# same file after a crash yields the same call_id, so C2 overwrites rather than +# duplicates (trunk_recorder.py relies on this to be restart-safe). +CALL_ID_NAMESPACE = uuid.UUID("5d0c7f4e-2b8f-4e0e-9a59-45c0de7c0045") + + +def _iso(epoch: float) -> str: + return datetime.fromtimestamp(epoch, timezone.utc).isoformat() + + +def _int(value: Any) -> Optional[int]: + try: + return int(value) + except (TypeError, ValueError): + return None + + +def _epoch(data: Dict[str, Any], key: str) -> Optional[float]: + ms = data.get(f"{key}_ms") + if isinstance(ms, (int, float)) and ms > 0: + return ms / 1000.0 + s = data.get(key) + if isinstance(s, (int, float)) and s > 0: + return float(s) + return None + + +@dataclass +class TrCall: + call_id: str + tgid: int + tgid_name: str + freq: Optional[int] + start_epoch: float + stop_epoch: float + units: List[int] = field(default_factory=list) + emergency: bool = False + encrypted: bool = False + src_list: List[Dict[str, Any]] = field(default_factory=list) + freq_list: List[Dict[str, Any]] = field(default_factory=list) + call_length: Optional[float] = None + signal: Optional[int] = None + noise: Optional[int] = None + freq_error: Optional[int] = None + decode_errors: Optional[int] = None + spikes: Optional[int] = None + patched_talkgroups: List[int] = field(default_factory=list) + short_name: str = "" + + @property + def srcaddr(self) -> Optional[int]: + return self.units[0] if self.units else None + + def start_payload(self) -> Dict[str, Any]: + """MQTT call_start — the keys main.on_call_start publishes, which C2's + _on_call_start reads (tgid, tgid_name, freq, srcaddr, started_at).""" + return { + "call_id": self.call_id, + "tgid": self.tgid, + "tgid_name": self.tgid_name, + "freq": self.freq, + "srcaddr": self.srcaddr, + "started_at": _iso(self.start_epoch), + "started_at_epoch": self.start_epoch, + "attributed": True, + "driver": DRIVER, + } + + def audio_end_epoch(self, wav_seconds: Optional[float] = None) -> float: + """Wall clock at the END of the audio file. + + NOT stop_time: TR joins a call's transmissions back to back and drops + the gaps between them, so the WAV is shorter than stop - start by the + sum of the gaps. start + WAV length is what the file's last sample + maps to only if there were no gaps; the exact per-transmission map is + srcList/freqList (time = wall clock, pos = offset in the file), sent + as src_list / freq_list. + """ + if wav_seconds is None: + return self.stop_epoch + return self.start_epoch + wav_seconds + + def end_payload(self, wav_seconds: Optional[float] = None) -> Dict[str, Any]: + """MQTT call_end. TR cut the call on decoder boundaries with no + trimming, so the audio starts at the call start; see audio_end_epoch + for why it does not end at the call end.""" + audio_end = self.audio_end_epoch(wav_seconds) + out = self.start_payload() + out.update({ + "ended_at": _iso(self.stop_epoch), + "ended_at_epoch": self.stop_epoch, + "audio_start_at": _iso(self.start_epoch), + "audio_end_at": _iso(audio_end), + "audio_start_epoch": self.start_epoch, + "audio_end_epoch": audio_end, + "units": self.units, + "emergency": self.emergency, + "encrypted": self.encrypted, + }) + return out + + def upload_extra(self) -> Dict[str, str]: + """Extra /upload form fields. C2 declares none of these yet, so FastAPI + ignores them (same as audio_start_epoch today); they are here for when + it does.""" + extra = { + "recorder_backend": DRIVER, + "units": ",".join(str(u) for u in self.units), + "src_list": json.dumps(self.src_list, separators=(",", ":")), + "freq_list": json.dumps(self.freq_list, separators=(",", ":")), + "emergency": "1" if self.emergency else "0", + "encrypted": "1" if self.encrypted else "0", + "started_at_epoch": f"{self.start_epoch:.3f}", + "ended_at_epoch": f"{self.stop_epoch:.3f}", + } + optional = { + "freq": self.freq, "call_length": self.call_length, "signal": self.signal, "noise": self.noise, + "freq_error": self.freq_error, "decode_errors": self.decode_errors, "spikes": self.spikes, + } + extra.update({k: str(v) for k, v in optional.items() if v is not None}) + if self.patched_talkgroups: + extra["patched_talkgroups"] = ",".join(str(t) for t in self.patched_talkgroups) + return extra + + +def parse_call(data: Dict[str, Any], node_id: str, file_stem: str = "") -> TrCall: + """One TR call JSON -> TrCall. ValueError if it has no talkgroup or start.""" + tgid = _int(data.get("talkgroup")) + if not tgid or tgid <= 0: + raise ValueError(f"call JSON has no talkgroup ({data.get('talkgroup')!r})") + start = _epoch(data, "start_time") + if start is None: + raise ValueError("call JSON has no start_time") + stop = _epoch(data, "stop_time") + length = data.get("call_length_ms") + length_s = (length / 1000.0) if isinstance(length, (int, float)) and length > 0 else None + if length_s is None and isinstance(data.get("call_length"), (int, float)): + length_s = float(data["call_length"]) + if stop is None or stop < start: + stop = start + (length_s or 0.0) + + src_list: List[Dict[str, Any]] = [] + units: List[int] = [] + unit_emergency = False + for s in data.get("srcList") or []: + if not isinstance(s, dict): + continue + src = _int(s.get("src")) + entry = {"src": src, "time": s.get("time"), "pos": s.get("pos"), + "emergency": bool(_int(s.get("emergency")))} + if s.get("tag"): + entry["tag"] = s.get("tag") + src_list.append(entry) + unit_emergency = unit_emergency or entry["emergency"] + # TR reports 0 / -1 when the unit id wasn't decoded. + if src and src > 0 and src not in units: + units.append(src) + + freq_list = [f for f in data.get("freqList") or [] if isinstance(f, dict)] + errors = sum(_int(f.get("error_count")) or 0 for f in freq_list) if freq_list else None + spikes = sum(_int(f.get("spike_count")) or 0 for f in freq_list) if freq_list else None + + freq = _int(data.get("freq")) + short_name = str(data.get("short_name") or "") + key = f"{node_id}|{short_name}|{tgid}|{start:.3f}|{freq}|{file_stem}" + return TrCall( + call_id=str(uuid.uuid5(CALL_ID_NAMESPACE, key)), + tgid=tgid, + tgid_name=str(data.get("talkgroup_tag") or data.get("talkgroup_description") or ""), + freq=freq, + start_epoch=start, + stop_epoch=stop, + units=units, + emergency=bool(_int(data.get("emergency"))) or unit_emergency, + encrypted=bool(_int(data.get("encrypted"))), + src_list=src_list, + freq_list=[{"freq": _int(f.get("freq")), "time": f.get("time"), "pos": f.get("pos"), "len": f.get("len"), + "error_count": _int(f.get("error_count")), "spike_count": _int(f.get("spike_count"))} + for f in freq_list], + call_length=round(length_s, 3) if length_s is not None else None, + signal=_int(data.get("signal")), + noise=_int(data.get("noise")), + freq_error=_int(data.get("freq_error")), + decode_errors=errors, + spikes=spikes, + patched_talkgroups=[t for t in (_int(x) for x in data.get("patched_talkgroups") or []) if t], + short_name=short_name, + ) diff --git a/drb-edge-node/app/internal/tr_config.py b/drb-edge-node/app/internal/tr_config.py new file mode 100644 index 0000000..84f3fd2 --- /dev/null +++ b/drb-edge-node/app/internal/tr_config.py @@ -0,0 +1,261 @@ +""" +Trunk Recorder config.json (+ CSVs) from the C2 system config (node-26#45). + +Same input as radio_config.build_generate_payload — the SystemConfig C2 pushes — +so a node can switch backends without anyone re-entering its system: + + P25 trunked -> system type "p25", control_channels, talkgroupsFile + P25 conventional -> "conventionalP25", channelFile + Analog -> "conventional", channelFile, squelch + +Schema per Trunk Recorder v5.2.1 docs/CONFIGURE.md. Pure functions: the I/O is +in trunk_recorder.py. Unverified against a radio — see the notes inline. + +Conventional channels are numbered with conventional.channel_tgid (frequency in +100 Hz units), the same synthetic talkgroup the OP25 path reports, so C2 names +a TR call and an OP25 call on the same channel identically. +""" +import csv +import io +from typing import Any, Dict, List, Optional, Tuple + +from app.internal.conventional import channel_tgid +from app.internal.radio_config import ( + ANALOG, CONVENTIONAL, DMR, TRUNKED, analog_frequency, conventional_channels, system_mode, to_hz, +) + +# Paths INSIDE the trunk-recorder container (see docker-compose.yml). +TR_APP_DIR = "/app" # tr_config_dir is mounted here +TR_UPLOAD_SCRIPT = "/drb/drb-upload.sh" # ./trunk-recorder, read-only +TR_CAPTURE_DIR = "/recordings/tr-capture" # ./recordings +TR_TEMP_DIR = "/dev/shm" # TR's recommended tmpfs (shm_size in compose) +TALKGROUPS_FILE = "talkgroups.csv" +CHANNELS_FILE = "channels.csv" +SHORT_NAME = "drb" + +# TR splits a call longer than this; mirrors call_recorder.MAX_RECORDING_SECONDS. +MAX_DURATION_SECONDS = 600 +# Seconds of control-channel silence before TR closes a call (TR's default). +CALL_TIMEOUT_SECONDS = 3 + +# Of an RTL-SDR's sample rate, the share that is usable passband (anti-alias +# roll-off eats the edges). Channels must sit inside +-(rate * share / 2). +USABLE_BANDWIDTH_SHARE = 0.8 +# The RTL-SDR's DC spike sits on the centre frequency; keep channels off it. +DC_SPIKE_GUARD_HZ = 25_000 +CENTER_NUDGES_HZ = (0, 150_000, -150_000, 300_000, -300_000) + + +def _fits(center: int, freqs: List[int], half_band: float) -> bool: + return all(abs(f - center) <= half_band for f in freqs) + + +def choose_center(freqs: List[int], rate: int, override: Optional[int] = None) -> int: + """A centre frequency that puts every channel in the passband and none on + the DC spike. ValueError if they cannot all fit on one SDR.""" + if not freqs: + raise ValueError("No frequencies to tune to.") + half = rate * USABLE_BANDWIDTH_SHARE / 2 + if override: + if not _fits(override, freqs, half): + raise ValueError(f"TR_CENTER_HZ {override} leaves a channel outside the SDR's {2 * half / 1e6:.1f} MHz.") + return int(override) + lo, hi = min(freqs), max(freqs) + if hi - lo > 2 * half: + raise ValueError( + f"Channels span {(hi - lo) / 1e6:.3f} MHz, more than one SDR at {rate / 1e6:.1f} Msps covers " + f"({2 * half / 1e6:.2f} MHz). Trim the channel list or raise TR_SAMPLE_RATE." + ) + mid = int(round((lo + hi) / 2)) + for nudge in CENTER_NUDGES_HZ: + c = mid + nudge + if _fits(c, freqs, half) and all(abs(f - c) >= DC_SPIKE_GUARD_HZ for f in freqs): + return c + return mid # channels packed around the middle: accept the spike over not fitting + + +def _csv(header: List[str], rows: List[List[Any]]) -> str: + buf = io.StringIO() + w = csv.writer(buf, lineterminator="\n") + w.writerow(header) + w.writerows(rows) + return buf.getvalue() + + +def talkgroups_csv(talkgroups: List[Dict[str, Any]]) -> str: + """TR talkgroupsFile: header row required, "Decimal" first (CONFIGURE.md).""" + rows = [] + for tg in talkgroups: + if tg.get("id") is None: + continue + name = str(tg.get("name") or "").strip() + rows.append([int(tg["id"]), "D", name, name, "", ""]) + return _csv(["Decimal", "Mode", "Alpha Tag", "Description", "Tag", "Category"], rows) + + +def channels_csv(channels: List[Dict[str, Any]], squelch: Optional[float] = None) -> str: + """TR channelFile: header required, "TG Number" first; Frequency in Hz + (no decimal point = Hz). Per-channel Squelch overrides the system's.""" + header = ["TG Number", "Frequency", "Alpha Tag", "Description"] + if squelch is not None: + header.append("Squelch") + rows = [] + for ch in channels: + name = ch.get("name") or f"{ch['frequency'] / 1e6:.4f} MHz" + row = [channel_tgid(ch["frequency"]), ch["frequency"], name, name] + if squelch is not None: + row.append(squelch) + rows.append(row) + return _csv(header, rows) + + +def _analog_channels(config: Dict[str, Any]) -> List[Dict[str, Any]]: + """Every analog channel the config lists (OP25 only ever used the first).""" + out = [] + for raw in (config or {}).get("channels") or []: + if isinstance(raw, dict) and raw.get("freq") not in (None, ""): + out.append({"frequency": to_hz(raw["freq"]), "name": str(raw.get("name") or "").strip()}) + if not out: + freq = analog_frequency(config) + if freq: + out.append({"frequency": freq, "name": ""}) + return out + + +def _analog_squelch(config: Dict[str, Any], default_db: float) -> float: + """TR's squelch is an absolute power level — OP25's "power" mode. Its + squelch_threshold carries over; noise/voice squelch_db has no TR + equivalent, so those fall back to the default (which needs site tuning).""" + value = (config or {}).get("squelch_threshold") + if value is None or (isinstance(value, str) and not value.strip()): + return float(default_db) + try: + return float(value) + except (TypeError, ValueError): + raise ValueError(f"squelch_threshold {value!r} isn't a number.") + + +def build_tr_config( + system_type: str, + name: str, + config: Dict[str, Any], + *, + device_serial: Optional[str], + ppm: Optional[float], + sample_rate: int, + gain: float, + digital_recorders: int, + min_duration: float, + squelch_db: float, + live_port: int, + center_override: Optional[int] = None, +) -> Tuple[Dict[str, Any], Dict[str, str]]: + """SystemConfig fields -> (config.json dict, {filename: contents} to write beside it). + + ValueError when the system can't be expressed (DMR, no channels, too wide). + """ + config = config or {} + mode = system_mode(system_type, config) + files: Dict[str, str] = {} + # Kept per system, not globally: audioArchive/callLog false makes TR delete + # the wav/json after uploadScript returns, which is why drb-upload.sh + # copies them out before it exits. + system: Dict[str, Any] = { + "shortName": SHORT_NAME, + "uploadScript": TR_UPLOAD_SCRIPT, + "audioArchive": False, + "callLog": False, + # WAV, not TR's 32k AAC: the recorder's own rule is one lossless copy + # for Whisper (call_recorder AUDIO_* notes). 8 kHz mono is ~1 MB/min. + "compressWav": False, + "minDuration": float(min_duration), + "maxDuration": MAX_DURATION_SECONDS, + "hideEncrypted": False, + } + source: Dict[str, Any] = { + "driver": "osmosdr", + # By serial, like OP25 (sdr_settings.pin_op25_device): "rtl=0" means + # whichever dongle enumerated first, which a decoder can take. + "device": f"rtl={device_serial}" if device_serial else "rtl=0", + "rate": int(sample_rate), + "gain": float(gain), + "ppm": float(ppm or 0.0), + } + + if mode == DMR: + raise ValueError("DMR isn't supported by the edge node.") + + if mode == TRUNKED: + ccs = [to_hz(c) for c in config.get("control_channels") or []] + if not ccs: + raise ValueError(f"System {name!r} has no control channels.") + talkgroups = [tg for tg in config.get("talkgroups") or [] if tg.get("id") is not None] + system.update({ + "type": "p25", + "control_channels": ccs, + # TR's default; "fsk4" is accepted too. Overridable from C2. + "modulation": str(config.get("modulation") or "qpsk"), + }) + # Same rule as OP25's whitelist: a talkgroup list means record only + # those; no list means record everything. + if talkgroups: + files[TALKGROUPS_FILE] = talkgroups_csv(talkgroups) + system["talkgroupsFile"] = f"{TR_APP_DIR}/{TALKGROUPS_FILE}" + system["recordUnknown"] = False + else: + system["recordUnknown"] = True + # Voice channels are wherever the system grants them; only the control + # channels are known here, so the SDR is centred on those. A system + # whose voice channels sit outside that ~2 MHz loses those calls (TR + # logs "no source covering" for each) — needs a second source. + freqs = ccs + source["digitalRecorders"] = int(digital_recorders) + elif mode == CONVENTIONAL: + chans = conventional_channels(config) + if not chans: + raise ValueError(f"System {name!r} has no conventional channels.") + # No NAC filter: TR's conventionalP25 records whatever it decodes, so a + # mistyped NAC can no longer silently drop every call (node-26#33). + system.update({"type": "conventionalP25", "channelFile": f"{TR_APP_DIR}/{CHANNELS_FILE}", + "squelch": float(squelch_db)}) + files[CHANNELS_FILE] = channels_csv(chans) + freqs = [c["frequency"] for c in chans] + else: # ANALOG + chans = _analog_channels(config) + if not chans: + raise ValueError(f"System {name!r} has no analog frequency.") + squelch = _analog_squelch(config, squelch_db) + system.update({"type": "conventional", "channelFile": f"{TR_APP_DIR}/{CHANNELS_FILE}", + "squelch": squelch}) + files[CHANNELS_FILE] = channels_csv(chans) + freqs = [c["frequency"] for c in chans] + + source["center"] = choose_center(freqs, int(sample_rate), center_override) + + tr = { + "ver": 2, + "sources": [source], + "systems": [system], + "tempDir": TR_TEMP_DIR, + "captureDir": TR_CAPTURE_DIR, + "callTimeout": CALL_TIMEOUT_SECONDS, + # File log too (stdout still goes to docker logs): trunk_recorder.py + # reads the control-channel decode rate from it for the health check. + "logFile": True, + "logDir": f"{TR_APP_DIR}/logs", + "logColor": False, + # Live audio for Discord: simplestream sends every call's PCM to the + # edge-node, which follows one talkgroup at a time into PulseAudio. + "audioStreaming": True, + "plugins": [{ + "name": "simplestream", + "library": "libsimplestream.so", + "streams": [{"TGID": 0, "address": "127.0.0.1", "port": int(live_port), "sendTGID": True}], + }], + } + return tr, files + + +def live_sample_rate(system_type: str, config: Dict[str, Any]) -> int: + """simplestream sends analog audio at 16 kHz and digital at 8 kHz (Plugins.md).""" + return 16000 if system_mode(system_type, config or {}) == ANALOG else 8000 diff --git a/drb-edge-node/app/internal/trunk_recorder.py b/drb-edge-node/app/internal/trunk_recorder.py new file mode 100644 index 0000000..795bc91 --- /dev/null +++ b/drb-edge-node/app/internal/trunk_recorder.py @@ -0,0 +1,575 @@ +""" +Trunk Recorder backend, edge-node side (node-26#45). Active only when +RECORDER_BACKEND=trunk-recorder. + + C2 SystemConfig --apply_config--> /config.json (+ CSVs) + trunk-recorder/run.sh restarts TR on change + TR finishes a call --drb-upload.sh--> /.wav, then .json + ingest loop --> MQTT call_start --> /upload (call_recorder.upload_recording, + i.e. the node-26#44 spool) --> MQTT call_end + TR simplestream (UDP) --LiveAudioBridge--> pacat --> drb_sink --> Discord + +Why an inbox directory and not an HTTP endpoint: TR deletes the WAV/JSON as +soon as uploadScript returns (audioArchive/callLog false), so the script must +copy the files out synchronously anyway, and a file on the host disk survives +an edge-node restart where a POST to a down edge-node would not. The JSON is +moved in LAST, so its presence means the WAV is complete. + +Restart safety: a call_id is a pure function of the JSON (tr_calls.parse_call), +and the WAV is renamed to `_.wav` (the name the spool +recovers a call_id from) before anything is sent. A crash at any step re-runs +the same call with the same id, so C2 overwrites instead of duplicating; a JSON +whose WAV is already gone was handed to the uploader and is just removed. +""" +import asyncio +import json +import math +import os +import re +import time +import wave +from datetime import datetime +from pathlib import Path +from typing import Any, Dict, List, Optional, Set + +from app.config import settings +from app.internal.logger import logger +from app.internal.tr_calls import TrCall, parse_call +from app.models import SystemConfig + +CONFIG_FILE = "config.json" +STATUS_FILE = "drb-status.json" # written by trunk-recorder/run.sh every few seconds +RESTART_REQUEST = "restart-request" # run.sh restarts TR when this appears +STOP_REQUEST = "stop-request" # run.sh keeps TR stopped while this exists + +INBOX_POLL_SECONDS = 1.0 +INGEST_CONCURRENCY = 4 +# Hand-off is wav-then-json; a JSON whose WAV never shows is dropped after this. +ORPHAN_JSON_SECONDS = 60.0 +# C2 creates the call doc from MQTT call_start, and only that message carries +# the talkgroup and start time. With the broker down, call_start is dropped +# (mqtt_manager buffers only call_end), so a call waits in the inbox this long +# for MQTT to come back before it is sent without one. +MQTT_WAIT_SECONDS = 600.0 +# call_start must be written before /upload stamps audio_gcs_uri: C2's +# _on_call_start writes the doc with merge=False and would erase it +# (server-26#303; drop this lead once C2 makes call_start create-only). +CALL_START_LEAD_SECONDS = 1.0 +# A WAV with no JSON this old is a hand-off drb-upload.sh never finished. +STRAY_WAV_SECONDS = 3600.0 +# How long radio_backend waits for run.sh to (re)start TR after a request: it +# polls every 5 s and may spend 15 s stopping the old process. +RESTART_WAIT_SECONDS = 30.0 +RESTART_POLL_SECONDS = 1.0 +# The name process() renames a claimed WAV to; never swept (its JSON is what +# drives it, and os.replace keeps TR's old mtime). +CANONICAL_WAV_RE = re.compile(r"^\d{8}_\d{6}_[0-9a-f-]{36}\.wav$") +# run.sh rewrites its status every 5 s; past this the supervisor is gone. +STATUS_STALE_SECONDS = 20.0 +LOG_TAIL_BYTES = 64 * 1024 +# TR logs this for every trunked system every few seconds (main.cpp +# check_message_count). Exact wording unverified against a live log. +DECODE_RATE_RE = re.compile(r"Decode Rate:\s*([0-9.]+)\s*/\s*sec", re.IGNORECASE) + + +def enabled() -> bool: + return settings.recorder_backend == "trunk-recorder" + + +def config_dir() -> Path: + return Path(settings.tr_config_dir or (Path(settings.config_path) / "trunk-recorder")) + + +def inbox_dir() -> Path: + return Path(settings.tr_inbox_dir or (Path(settings.recordings_path) / "tr-inbox")) + + +def _atomic_write(path: Path, text: str) -> None: + tmp = path.with_name(path.name + ".tmp") + tmp.write_text(text) + os.replace(tmp, path) + + +def wav_seconds(path: Path) -> Optional[float]: + """Length of a WAV from its header; None if it can't be read.""" + try: + with wave.open(str(path), "rb") as w: + rate = w.getframerate() + return w.getnframes() / rate if rate else None + except (OSError, EOFError, wave.Error): + return None + + +def canonical_wav_name(call: TrCall) -> str: + stamp = datetime.fromtimestamp(call.start_epoch).strftime("%Y%m%d_%H%M%S") + return f"{stamp}_{call.call_id}.wav" + + +class TrunkRecorder: + def __init__(self): + self.config_error: Optional[str] = None + self.calls_ingested = 0 + self.encrypted_skipped = 0 + self.last_call_epoch: Optional[float] = None + self.last_call: Optional[Dict[str, Any]] = None + self._task: Optional[asyncio.Task] = None + self._inflight: Set[str] = set() + self._sem = asyncio.Semaphore(INGEST_CONCURRENCY) + self.live = LiveAudioBridge() + + # ------------------------------------------------------------------ + # Config + supervisor requests + # ------------------------------------------------------------------ + + async def apply_config(self, config: SystemConfig) -> bool: + """Write TR's config for `config`. run.sh restarts TR when config.json + changes; a restart request covers a CSV-only change.""" + from app.internal.config_manager import load_node_config + from app.internal.sdr_settings import op25_serial + from app.internal.secondary_sdr_client import secondary_sdr_client + from app.internal.tr_config import build_tr_config, live_sample_rate + + node_cfg = load_node_config() + # TR takes the dongle OP25 would have (pin or first detected), so the + # secondary decoders' reservation logic is unchanged. + devs = await secondary_sdr_client.devices() + serial = op25_serial(node_cfg, devs) + if serial and devs and sum(d["serial"] == serial for d in devs) > 1: + serial = None # two dongles share it: by-serial would be ambiguous + try: + tr, files = build_tr_config( + config.type, config.name, config.config, + device_serial=serial, ppm=node_cfg.ppm_override, + sample_rate=settings.tr_sample_rate, gain=settings.tr_gain, + digital_recorders=settings.tr_digital_recorders, min_duration=settings.tr_min_duration, + squelch_db=settings.tr_squelch_db, live_port=settings.tr_live_port, + center_override=settings.tr_center_hz, + ) + except ValueError as e: + self.config_error = f"System {config.name!r}: {e}" + logger.error(f"Trunk Recorder config not written — {self.config_error}") + return False + try: + d = config_dir() + d.mkdir(parents=True, exist_ok=True) + (d / "logs").mkdir(exist_ok=True) + # CSVs first: run.sh restarts on config.json, which must never + # point at a file that isn't there yet. + for name, text in files.items(): + _atomic_write(d / name, text) + _atomic_write(d / CONFIG_FILE, json.dumps(tr, indent=2)) + except OSError as e: + self.config_error = f"Couldn't write {config_dir()}: {e}" + logger.error(self.config_error) + return False + self.config_error = None + self.live.sample_rate = live_sample_rate(config.type, config.config) + logger.info(f"Trunk Recorder config written for {config.name} (SDR {serial or 'rtl=0'}).") + return True + + def request_restart(self) -> bool: + try: + d = config_dir() + d.mkdir(parents=True, exist_ok=True) + (d / STOP_REQUEST).unlink(missing_ok=True) + (d / RESTART_REQUEST).write_text(str(time.time())) + return True + except OSError as e: + logger.error(f"Couldn't ask Trunk Recorder to restart: {e}") + return False + + async def wait_started_since(self, since: float, timeout: Optional[float] = None) -> Optional[str]: + """Wait until run.sh reports TR running with started_at >= `since`. + + Callers hand the spare dongles to the secondary decoders right after a + start; returning before TR has actually reopened its SDR would let a + decoder take it. None when started, else why not. + """ + timeout = RESTART_WAIT_SECONDS if timeout is None else timeout + deadline = time.time() + timeout + while True: + st = self.supervisor_status() + if st and not st["stale"] and st.get("state") == "running" \ + and float(st.get("started_at") or 0) >= math.floor(since): + return None + if time.time() >= deadline: + break + await asyncio.sleep(RESTART_POLL_SECONDS) + if st is None: + return "The trunk-recorder service isn't running (is COMPOSE_PROFILES=trunk-recorder set?)." + return f"Trunk Recorder didn't restart within {timeout:.0f}s (state: {st.get('state')})." + + def request_stop(self) -> bool: + try: + d = config_dir() + d.mkdir(parents=True, exist_ok=True) + (d / STOP_REQUEST).write_text(str(time.time())) + return True + except OSError as e: + logger.error(f"Couldn't ask Trunk Recorder to stop: {e}") + return False + + # ------------------------------------------------------------------ + # Health + # ------------------------------------------------------------------ + + def supervisor_status(self, now: Optional[float] = None) -> Optional[Dict[str, Any]]: + now = time.time() if now is None else now + try: + data = json.loads((config_dir() / STATUS_FILE).read_text()) + except (OSError, ValueError): + return None + data["age_seconds"] = round(now - float(data.get("epoch") or 0), 1) + data["stale"] = data["age_seconds"] > STATUS_STALE_SECONDS + return data + + def log_facts(self, now: Optional[float] = None) -> Dict[str, Any]: + now = time.time() if now is None else now + logs = config_dir() / "logs" + try: + files = [p for p in logs.iterdir() if p.is_file()] + except OSError: + files = [] + if not files: + return {"log_age_seconds": None, "decode_rate": None} + newest = max(files, key=lambda p: p.stat().st_mtime) + age = now - newest.stat().st_mtime + rate = None + try: + with open(newest, "rb") as f: + f.seek(max(0, newest.stat().st_size - LOG_TAIL_BYTES)) + tail = f.read().decode(errors="replace") + found = DECODE_RATE_RE.findall(tail) + rate = float(found[-1]) if found else None + except OSError: + pass + return {"log_age_seconds": round(age, 1), "decode_rate": rate} + + def log_tail(self, lines: int = 80) -> List[str]: + """Last lines of TR's newest log file, for the dashboard's log panel.""" + try: + files = [p for p in (config_dir() / "logs").iterdir() if p.is_file()] + if not files: + return [] + newest = max(files, key=lambda p: p.stat().st_mtime) + with open(newest, "rb") as f: + f.seek(max(0, newest.stat().st_size - LOG_TAIL_BYTES)) + text = f.read().decode(errors="replace") + except OSError: + return [] + return text.splitlines()[-max(1, min(int(lines), 500)):] + + def sweep_stray_wavs(self, now: Optional[float] = None) -> int: + """Delete WAVs whose JSON never arrived (drb-upload.sh died between + its two moves). Canonical names (already claimed by a JSON) + are never touched, so only true strays go.""" + now = time.time() if now is None else now + removed = 0 + try: + jsons = {p.stem for p in inbox_dir().glob("*.json")} + for wav in inbox_dir().glob("*.wav"): + if wav.stem in jsons or CANONICAL_WAV_RE.match(wav.name) \ + or now - wav.stat().st_mtime < STRAY_WAV_SECONDS: + continue + logger.error(f"Trunk Recorder WAV {wav.name} has had no call JSON for an hour — deleting it.") + wav.unlink(missing_ok=True) + removed += 1 + except OSError: + pass + return removed + + def inbox_backlog(self) -> int: + try: + return sum(1 for _ in inbox_dir().glob("*.json")) + except OSError: + return 0 + + def health_facts(self) -> Dict[str, Any]: + return { + "status": self.supervisor_status(), + **self.log_facts(), + "config_error": self.config_error, + "inbox_backlog": self.inbox_backlog(), + "calls_ingested": self.calls_ingested, + "encrypted_skipped": self.encrypted_skipped, + "last_call_epoch": self.last_call_epoch, + "last_call": self.last_call, + "live": self.live.stats(), + } + + # ------------------------------------------------------------------ + # Ingest + # ------------------------------------------------------------------ + + async def start(self) -> None: + inbox_dir().mkdir(parents=True, exist_ok=True) + if self._task is None or self._task.done(): + self._task = asyncio.create_task(self._run()) + await self.live.start() + + async def stop(self) -> None: + if self._task: + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + self._task = None + await self.live.stop() + + async def _run(self) -> None: + backlog = self.inbox_backlog() + if backlog: + logger.info(f"Trunk Recorder inbox has {backlog} call(s) from before this start — sending.") + last_sweep = 0.0 + while True: + try: + await self.scan_once() + if time.time() - last_sweep > 300: + last_sweep = time.time() + self.sweep_stray_wavs() + except asyncio.CancelledError: + raise + except Exception as e: + logger.error(f"Trunk Recorder inbox pass failed: {e}") + await asyncio.sleep(INBOX_POLL_SECONDS) + + async def scan_once(self, wait: bool = False) -> int: + """Start a task per ready call. Returns how many were started; with + `wait`, also waits for them (tests).""" + from app.internal.mqtt_manager import mqtt_manager + try: + ready = sorted(inbox_dir().glob("*.json"), key=lambda p: p.stat().st_mtime) + except OSError: + return 0 + started: List[asyncio.Task] = [] + now = time.time() + for path in ready: + if path.name in self._inflight: + continue + try: + age = now - path.stat().st_mtime + except OSError: + continue + if not mqtt_manager.is_connected and age < MQTT_WAIT_SECONDS: + continue + self._inflight.add(path.name) + started.append(asyncio.create_task(self._guarded(path))) + if wait and started: + await asyncio.gather(*started) + return len(started) + + async def _guarded(self, path: Path) -> None: + try: + async with self._sem: + await self.process(path) + except Exception as e: + logger.error(f"Trunk Recorder call {path.name} failed: {e}") + finally: + self._inflight.discard(path.name) + + async def process(self, json_path: Path) -> None: + from app.internal.call_recorder import call_recorder + from app.internal.config_manager import load_node_config + from app.internal.mqtt_manager import mqtt_manager + + try: + data = json.loads(json_path.read_text()) + call = parse_call(data, settings.node_id, json_path.stem) + except (OSError, ValueError) as e: + logger.error(f"Trunk Recorder call {json_path.name} is unreadable ({e}) — dropping it.") + json_path.with_suffix(".wav").unlink(missing_ok=True) + json_path.unlink(missing_ok=True) + return + + raw_wav = json_path.with_suffix(".wav") + wav = json_path.parent / canonical_wav_name(call) + if raw_wav.exists(): + os.replace(raw_wav, wav) + if not wav.exists(): + age = time.time() - json_path.stat().st_mtime + if age > ORPHAN_JSON_SECONDS: + # Either TR's hand-off broke, or a previous pass already handed + # this WAV to the uploader (which deletes or spools it) and + # died before removing the JSON. Either way nothing to send. + # Still close it in C2: if a previous pass uploaded and died + # before this point, the doc would otherwise stay "active". + # _on_call_end merges, so a repeat is harmless. + logger.warning(f"Trunk Recorder call {call.call_id}: no audio beside {json_path.name} — removing it.") + await mqtt_manager.publish_metadata("call_end", {**call.end_payload(), "audio_skipped": "no_audio"}) + json_path.unlink(missing_ok=True) + return + + await mqtt_manager.publish_metadata("call_start", call.start_payload()) + seconds = wav_seconds(wav) + end = call.end_payload(seconds) + + if call.encrypted: + # Encrypted P25 voice decodes to noise: nothing for Whisper. The + # call is still reported so C2 sees the activity. + wav.unlink(missing_ok=True) + end["audio_skipped"] = "encrypted" + self.encrypted_skipped += 1 + elif settings.c2_url: + await asyncio.sleep(CALL_START_LEAD_SECONDS) + node_cfg = load_node_config() + url = await call_recorder.upload_recording( + wav, call.call_id, + talkgroup_id=call.tgid, talkgroup_name=call.tgid_name or None, + system_id=node_cfg.assigned_system_id, + audio_start_epoch=call.start_epoch, audio_end_epoch=call.audio_end_epoch(seconds), + extra=call.upload_extra(), + ) + if url: + end["audio_url"] = url + else: + kind = (call_recorder.spool.last_result or {}).get("kind") + end["audio_upload"] = {"retry": "spooled", "lost": "lost"}.get(kind, "rejected") + else: + wav.unlink(missing_ok=True) + + await mqtt_manager.publish_metadata("call_end", end) + json_path.unlink(missing_ok=True) + self.calls_ingested += 1 + self.last_call_epoch = time.time() + self.last_call = {"call_id": call.call_id, "tgid": call.tgid, "tgid_name": call.tgid_name, + "units": call.units, "duration": round(call.stop_epoch - call.start_epoch, 1), + "encrypted": call.encrypted, "emergency": call.emergency} + logger.info( + f"TR call {call.call_id}: TG {call.tgid} {call.tgid_name!r}, " + f"{call.stop_epoch - call.start_epoch:.1f}s, units {call.units or '-'}" + + (" [EMERGENCY]" if call.emergency else "") + (" [encrypted]" if call.encrypted else "") + ) + + +class LiveAudioBridge: + """TR simplestream (UDP, s16le mono, 4-byte little-endian TGID prefix) -> + `pacat` into drb_sink, so Discord keeps reading drb_sink.monitor unchanged. + + TR records every talkgroup at once and streams them all to one port; + interleaving two calls' packets would garble both. So this follows ONE + talkgroup — the first heard — until it has been quiet for HOLD_SECONDS, + like the one-voice-channel scanner OP25 was. Recording is unaffected: + every call is still recorded and uploaded in full. + """ + + HOLD_SECONDS = 2.0 + RESPAWN_SECONDS = 5.0 + MAX_BUFFERED_BYTES = 64 * 1024 # ~4 s at 8 kHz; beyond this pacat is stuck, drop + + def __init__(self): + self.sample_rate = 8000 + self.current_tgid: Optional[int] = None + self.last_packet_epoch: Optional[float] = None + self._last_current_epoch = 0.0 + self.packets = 0 + self.dropped = 0 + self._transport = None + self._proc: Optional[asyncio.subprocess.Process] = None + self._proc_rate: Optional[int] = None + self._spawn_after = 0.0 + self._spawning = False + + def stats(self) -> Dict[str, Any]: + return {"current_tgid": self.current_tgid, "last_packet_epoch": self.last_packet_epoch, + "packets": self.packets, "dropped": self.dropped, + "pacat_running": self._proc is not None and self._proc.returncode is None} + + async def start(self) -> None: + loop = asyncio.get_running_loop() + bridge = self + + class _Proto(asyncio.DatagramProtocol): + def datagram_received(self, data, addr): + bridge.feed(data) + + try: + self._transport, _ = await loop.create_datagram_endpoint( + _Proto, local_addr=("127.0.0.1", settings.tr_live_port), + ) + logger.info(f"Trunk Recorder live audio: listening on 127.0.0.1:{settings.tr_live_port}.") + except OSError as e: + logger.error(f"Trunk Recorder live audio disabled — can't bind UDP {settings.tr_live_port}: {e}") + + async def stop(self) -> None: + if self._transport: + self._transport.close() + self._transport = None + if self._proc and self._proc.returncode is None: + self._proc.terminate() + self._proc = None + + def select(self, tgid: int, now: float) -> bool: + """Follow-one-talkgroup policy. True if this packet should be played.""" + if self.current_tgid is None or now - self._last_current_epoch > self.HOLD_SECONDS: + self.current_tgid = tgid + if tgid != self.current_tgid: + return False + self._last_current_epoch = now + return True + + def feed(self, data: bytes, now: Optional[float] = None) -> Optional[bytes]: + """One datagram. Returns the PCM that was (or would be) played.""" + if len(data) <= 4: + return None + now = time.time() if now is None else now + tgid = int.from_bytes(data[:4], "little") + self.packets += 1 + self.last_packet_epoch = now + if not self.select(tgid, now): + return None + pcm = data[4:] + self._write(pcm) + return pcm + + def _write(self, pcm: bytes) -> None: + proc = self._proc + if proc is None or proc.returncode is not None: + self._ensure_pacat() + return + if self._proc_rate != self.sample_rate: + # A config change, not a failure: no respawn throttle. + self._spawn_after = 0.0 + self._ensure_pacat() + return + stdin = proc.stdin + if stdin is None or stdin.is_closing(): + return + try: + if stdin.transport.get_write_buffer_size() > self.MAX_BUFFERED_BYTES: + self.dropped += 1 + return + stdin.write(pcm) + except (BrokenPipeError, ConnectionResetError, RuntimeError): + self._proc = None # pacat died; respawned on the next packet + + def _ensure_pacat(self) -> None: + if self._spawning or time.time() < self._spawn_after: + return + self._spawning = True + self._spawn_after = time.time() + self.RESPAWN_SECONDS + try: + asyncio.get_running_loop().create_task(self._spawn()) + except RuntimeError: + self._spawning = False + + async def _spawn(self) -> None: + try: + if self._proc and self._proc.returncode is None: + self._proc.terminate() + sink = settings.pulse_source.removesuffix(".monitor") + self._proc = await asyncio.create_subprocess_exec( + "pacat", "--playback", "--raw", "--format=s16le", f"--rate={self.sample_rate}", + "--channels=1", f"--device={sink}", "--latency-msec=250", "--client-name=drb-trunk-recorder", + stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.DEVNULL, + stderr=asyncio.subprocess.DEVNULL, + ) + self._proc_rate = self.sample_rate + except Exception as e: + logger.error(f"Trunk Recorder live audio: pacat failed to start ({e}); retrying.") + self._proc = None + finally: + self._spawning = False + + +trunk_recorder = TrunkRecorder() diff --git a/drb-edge-node/app/main.py b/drb-edge-node/app/main.py index beca2b7..e42f3e0 100644 --- a/drb-edge-node/app/main.py +++ b/drb-edge-node/app/main.py @@ -168,10 +168,12 @@ async def on_command(payload: dict): except ValueError as e: logger.error(f"Rejected SDR settings from C2: {e}") elif action == "op25_restart": - from app.internal.op25_client import op25_client - await op25_client.stop() + # Restarts whichever recorder is active (node-26#45): C2 still calls + # it "Restart OP25". + from app.internal import radio_backend + await radio_backend.stop() await asyncio.sleep(2) - await op25_client.start() + await radio_backend.start() elif action == "node_update": # node-26#15: the container cannot (and must not) update itself. It # drops a sentinel in /configs; the host's drb-update.path unit @@ -201,7 +203,14 @@ async def _generate_op25_config(config: SystemConfig) -> bool: Trunked / conventional / analog are told apart in radio_config (node-26#24); a system saved before that has no `mode` and is trunked, as it always was. + + With RECORDER_BACKEND=trunk-recorder this writes Trunk Recorder's + config.json instead (node-26#45) — the one place every caller (C2 push, + dashboard, startup resume) goes through. """ + from app.internal import trunk_recorder as tr + if tr.enabled(): + return await tr.trunk_recorder.apply_config(config) from app.internal.op25_client import op25_client from app.internal.radio_config import build_generate_payload node_cfg = load_node_config() @@ -264,19 +273,20 @@ async def on_config_push(payload: dict): logger.error(f"Ignoring invalid sdr_pins in config push: {e}") save_node_config(node_cfg) - from app.internal.op25_client import op25_client + from app.internal import radio_backend if not await _generate_op25_config(config): - logger.error(f"Failed to generate OP25 config for {config.name}") + logger.error(f"Failed to generate {radio_backend.name()} config for {config.name}") return # OP25's (pinned) dongle may be one a decoder holds right now: free the # spares, restart OP25 on its own dongle, then hand the rest back out. + # (Trunk Recorder takes the same dongle — node-26#45.) from app.internal.secondary_sdr_client import secondary_sdr_client from app.internal.sdr_settings import apply_secondaries await secondary_sdr_client.apply([], {}, []) - await op25_client.stop() + await radio_backend.stop() await asyncio.sleep(2) - await op25_client.start() + await radio_backend.start() logger.info(f"Config push applied: {config.name}") await apply_secondaries() @@ -309,8 +319,21 @@ async def lifespan(app: FastAPI): # Start services (radio_bot starts when C2's retained discord/desired arrives) await mqtt_manager.connect() - await metadata_watcher.start() - await call_recorder.start() # persistent PulseAudio ring buffer + from app.internal import radio_backend + from app.internal import trunk_recorder as tr + if tr.enabled(): + # node-26#45: Trunk Recorder records; the OP25 watcher and the + # PulseAudio recorder stay off so no call is uploaded twice. Only the + # upload spool runs. OP25 is stopped in case it was running before the + # backend was switched — it would hold the SDR Trunk Recorder needs. + logger.info("Recorder backend: Trunk Recorder (OP25 and the PulseAudio recorder are off).") + from app.internal.op25_client import op25_client + await op25_client.stop() + await call_recorder.start(capture=False) + await tr.trunk_recorder.start() + else: + await metadata_watcher.start() + await call_recorder.start() # persistent PulseAudio ring buffer # Start system caching in background from app.internal.system_cacher import fetch_and_cache_systems @@ -330,13 +353,14 @@ async def lifespan(app: FastAPI): # 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 has a system to listen to — waiting for OP25 API then generating config.") + logger.info(f"Node has a system to listen to — generating the {radio_backend.name()} config.") for attempt in range(10): if await _generate_op25_config(active_config): - await op25_client.start() + # No wait at startup: the trunk-recorder container may still be + # coming up, and the app must start serving regardless. + await radio_backend.start(wait=False) break - logger.warning(f"OP25 not ready yet (attempt {attempt + 1}/10), retrying in 3s…") + logger.warning(f"{radio_backend.name()} not ready yet (attempt {attempt + 1}/10), retrying in 3s…") await asyncio.sleep(3) # After op25 has claimed its dongle, so the decoders only get the spares. @@ -355,6 +379,7 @@ async def lifespan(app: FastAPI): heartbeat_task.cancel() telemetry_task.cancel() await metadata_watcher.stop() + await tr.trunk_recorder.stop() await call_recorder.stop() await radio_bot.stop() await mqtt_manager.disconnect() diff --git a/drb-edge-node/app/routers/api.py b/drb-edge-node/app/routers/api.py index f0a8fbd..021cc15 100644 --- a/drb-edge-node/app/routers/api.py +++ b/drb-edge-node/app/routers/api.py @@ -97,9 +97,18 @@ async def get_status(): "icecast_port": settings.icecast_port, "icecast_mount": settings.icecast_mount, "op25": op25_status, + # Which recorder is live (node-26#45) and, for Trunk Recorder, its + # supervisor state / decode rate / handed-over calls. + "recorder_backend": settings.recorder_backend, + "trunk_recorder": _tr_facts(), } +def _tr_facts(): + from app.internal import trunk_recorder as tr + return tr.trunk_recorder.health_facts() if tr.enabled() else None + + @router.get("/health") async def get_health(): """Plain-language setup/status checklist for the dashboard (see internal/health.py).""" @@ -110,7 +119,9 @@ async def get_health(): @router.post("/op25/start") async def start_op25(): - err = await op25_client.start_with_reason() + # The active recorder (node-26#45) — the dashboard's Start button. + from app.internal import radio_backend + err = await radio_backend.start_with_reason() if err: raise HTTPException(502, err) return {"ok": True} @@ -118,14 +129,18 @@ async def start_op25(): @router.post("/op25/stop") async def stop_op25(): - ok = await op25_client.stop() + from app.internal import radio_backend + ok = await radio_backend.stop() if not ok: - raise HTTPException(502, "The op25 container did not answer the stop request.") + raise HTTPException(502, f"{radio_backend.name()} did not accept the stop request.") return {"ok": True} @router.get("/op25/log") async def op25_log(lines: int = 80): + from app.internal import trunk_recorder as tr + if tr.enabled(): + return {"lines": tr.trunk_recorder.log_tail(lines)} out = await op25_client.log(lines) if out is None: raise HTTPException(502, "The op25 container isn't answering.") @@ -138,17 +153,22 @@ async def get_config(): async def _restart_op25_with(config: SystemConfig) -> None: - """Generate OP25's config for `config` and restart OP25 on it.""" + """Generate the active recorder's config for `config` and restart it on it.""" from app.main import _generate_op25_config + from app.internal import radio_backend if not await _generate_op25_config(config): + from app.internal import trunk_recorder as tr + if tr.enabled(): + detail = tr.trunk_recorder.config_error or f"Trunk Recorder config for {config.name} failed." + raise HTTPException(502, detail) raise HTTPException( 502, f"The op25 container did not accept the config for {config.name} — is it running?", ) - await op25_client.stop() + await radio_backend.stop() await asyncio.sleep(2) - err = await op25_client.start_with_reason() + err = await radio_backend.start_with_reason() if err: - raise HTTPException(502, f"Config saved, but OP25 did not start: {err}") + raise HTTPException(502, f"Config saved, but {radio_backend.name()} did not start: {err}") async def _apply_override(config: SystemConfig) -> None: @@ -350,7 +370,8 @@ async def revert_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() + from app.internal import radio_backend + await radio_backend.stop() if mqtt_manager.is_connected: await mqtt_manager.publish_checkin() diff --git a/drb-edge-node/app/templates/index.html b/drb-edge-node/app/templates/index.html index 39e2333..04bdbed 100644 --- a/drb-edge-node/app/templates/index.html +++ b/drb-edge-node/app/templates/index.html @@ -142,6 +142,7 @@
Listening to
—
+
Recorder
—
Talkgroup
—
Last heard
—
Stream link
—
@@ -318,7 +319,8 @@ ? `TG ${esc(heard.tgid)}${heard.name ? ' ' + esc(heard.name) : ''}` + ` · ${ago(heard.epoch)}${heard.freq ? ` · ${esc(mhz(heard.freq))}` : ''}` : 'Nothing heard since the node started.'; - const op = d.checks.find(c => c.id === 'op25'); + // 'recorder' is the Trunk Recorder backend's check (node-26#45). + const op = d.checks.find(c => c.id === 'op25' || c.id === 'recorder'); const src = d.checks.find(c => c.id === 'source'); const running = op && op.state === 'ok'; $('op25-start').disabled = running || (src && src.state !== 'ok'); @@ -340,6 +342,15 @@ $('node-id').textContent = d.node_id || ''; document.title = `${d.node_name || 'DRB Edge Node'} — DRB`; $('now-system').textContent = d.system_name ? `${d.system_name} (${d.system_kind || d.system_type})` : 'Nothing chosen yet'; + const isTR = d.recorder_backend === 'trunk-recorder'; + const trState = isTR && d.trunk_recorder && d.trunk_recorder.status ? d.trunk_recorder.status.state : null; + $('now-backend').textContent = isTR + ? `Trunk Recorder${trState ? ` (${trState.replace(/_/g, ' ')})` : ' (not running)'} — every talkgroup recorded at once` + : 'OP25 — one call at a time'; + recorderName = isTR ? 'Trunk Recorder' : 'OP25'; + $('op25-start').textContent = `Start ${recorderName}`; + $('op25-stop').textContent = `Stop ${recorderName}`; + if ($('op25-log').hidden) $('log-toggle').textContent = `Show ${recorderName} log`; $('now-tg').innerHTML = d.active_tgid ? `${esc(d.active_tgid_name || 'Talkgroup')} TG ${esc(d.active_tgid)}` : 'No call right now'; @@ -374,7 +385,8 @@ $('ack-btn').hidden = !(d.local_choice && d.has_c2_system && d.enforce_override_timeout); } - // ── OP25 controls + log ──────────────────────────────────────────────── + // ── OP25 / Trunk Recorder controls + log ─────────────────────────────── + let recorderName = 'OP25'; async function op25(action) { $('op25-msg').className = 'msg'; $('op25-msg').textContent = action === 'start' ? 'Starting…' : 'Stopping…'; @@ -391,19 +403,19 @@ const pre = $('op25-log'); try { const d = await api('/api/op25/log?lines=120'); - pre.textContent = d.lines.length ? d.lines.join('\n') : 'OP25 has printed nothing since it last started.'; + pre.textContent = d.lines.length ? d.lines.join('\n') : `${recorderName} has printed nothing since it last started.`; } catch (e) { pre.textContent = `Couldn't load the log: ${e.message}`; } pre.hidden = false; - $('log-toggle').textContent = 'Hide OP25 log'; + $('log-toggle').textContent = `Hide ${recorderName} log`; $('log-toggle').setAttribute('aria-expanded', 'true'); } function toggleLog() { const pre = $('op25-log'); if (pre.hidden) return loadLog(); pre.hidden = true; - $('log-toggle').textContent = 'Show OP25 log'; + $('log-toggle').textContent = `Show ${recorderName} log`; $('log-toggle').setAttribute('aria-expanded', 'false'); } diff --git a/drb-edge-node/tests/test_trunk_recorder.py b/drb-edge-node/tests/test_trunk_recorder.py new file mode 100644 index 0000000..6c750e3 --- /dev/null +++ b/drb-edge-node/tests/test_trunk_recorder.py @@ -0,0 +1,494 @@ +""" +Trunk Recorder backend (node-26#45): config generation from the C2 system +config, the TR call JSON -> MQTT + /upload mapping, the inbox hand-off, the +live-audio follow policy and the health checklist. + +No radio here: nothing in this file has run against a real Trunk Recorder. +The sample call JSON below is built from `create_call_json` in +trunk-recorder/call_concluder/call_concluder.cc at tag v5.2.1 (key names and +order as written there); the values are invented. +""" +import json +from pathlib import Path + +import pytest + +from app.config import settings +from app.internal import health, radio_backend, tr_config, upload_spool +from app.internal import trunk_recorder as trmod +from app.internal.conventional import channel_tgid +from app.internal.tr_calls import parse_call +from app.internal.upload_spool import UploadMeta + +# Shape: call_concluder.cc v5.2.1 create_call_json. Values invented. +SAMPLE_CALL = { + "freq": 851012500, + "freq_error": -112, + "signal": -48, + "noise": -98, + "source_num": 0, + "recorder_num": 2, + "tdma_slot": 0, + "phase2_tdma": 0, + "start_time": 1793145600, + "stop_time": 1793145612, + "start_time_ms": 1793145600250, + "stop_time_ms": 1793145612730, + "emergency": 0, + "priority": 1, + "mode": 0, + "duplex": 0, + "encrypted": 0, + "call_length": 12, + "call_length_ms": 12480, + "talkgroup": 101, + "talkgroup_tag": "PD Dispatch", + "talkgroup_description": "Police Dispatch", + "talkgroup_group_tag": "Law Dispatch", + "talkgroup_group": "Police", + "color_code": -1, + "audio_type": "digital", + "short_name": "drb", + "freqList": [ + {"freq": 851012500, "time": 1793145600, "pos": 0.0, "len": 5.2, "error_count": 3, "spike_count": 1}, + {"freq": 851012500, "time": 1793145606, "pos": 5.2, "len": 7.28, "error_count": 2, "spike_count": 0}, + ], + "srcList": [ + {"src": 1234567, "time": 1793145600, "pos": 0.0, "emergency": 0, "signal_system": "", "tag": "Car 12", + "tag_ota": ""}, + {"src": 7654321, "time": 1793145606, "pos": 5.2, "emergency": 1, "signal_system": "", "tag": "", + "tag_ota": ""}, + {"src": 1234567, "time": 1793145610, "pos": 9.9, "emergency": 0, "signal_system": "", "tag": "Car 12", + "tag_ota": ""}, + {"src": -1, "time": 1793145611, "pos": 11.0, "emergency": 0, "signal_system": "", "tag": "", "tag_ota": ""}, + ], +} + +TR_KW = dict(device_serial="00000101", ppm=1.5, sample_rate=2_400_000, gain=34.0, digital_recorders=4, + min_duration=1.0, squelch_db=-50.0, live_port=9123) + + +# --------------------------------------------------------------------------- +# Config generation +# --------------------------------------------------------------------------- + +def test_trunked_config_from_c2_system(): + cfg = {"control_channels": [851.0125, 851.2625], "talkgroups": [{"id": 101, "name": "PD Dispatch, Main"}, + {"id": 202, "name": "Fire"}]} + tr, files = tr_config.build_tr_config("P25", "County", cfg, **TR_KW) + assert tr["ver"] == 2 + src, sys_ = tr["sources"][0], tr["systems"][0] + assert src["driver"] == "osmosdr" and src["device"] == "rtl=00000101" + assert src["ppm"] == 1.5 and src["gain"] == 34.0 and src["rate"] == 2_400_000 + assert src["digitalRecorders"] == 4 + assert all(abs(f - src["center"]) <= 960_000 for f in (851_012_500, 851_262_500)) + assert all(abs(f - src["center"]) >= tr_config.DC_SPIKE_GUARD_HZ for f in (851_012_500, 851_262_500)) + assert sys_["type"] == "p25" and sys_["control_channels"] == [851_012_500, 851_262_500] + assert sys_["talkgroupsFile"] == "/app/talkgroups.csv" and sys_["recordUnknown"] is False + assert sys_["uploadScript"] == "/drb/drb-upload.sh" + assert sys_["audioArchive"] is False and sys_["callLog"] is False and sys_["compressWav"] is False + assert tr["tempDir"] == "/dev/shm" and tr["audioStreaming"] is True + stream = tr["plugins"][0]["streams"][0] + assert tr["plugins"][0]["library"] == "libsimplestream.so" + assert stream == {"TGID": 0, "address": "127.0.0.1", "port": 9123, "sendTGID": True} + lines = files["talkgroups.csv"].splitlines() + assert lines[0].startswith("Decimal,") # TR: header required, Decimal first + assert lines[1] == '101,D,"PD Dispatch, Main","PD Dispatch, Main",,' + + +def test_trunked_without_talkgroups_records_everything(): + tr, files = tr_config.build_tr_config("P25", "County", {"control_channels": [851.0125]}, **TR_KW) + assert tr["systems"][0]["recordUnknown"] is True + assert "talkgroupsFile" not in tr["systems"][0] and not files + + +def test_conventional_p25_uses_synthetic_tgids(): + cfg = {"mode": "conventional", "channels": [{"freq": 460.400, "name": "PD1", "nac": "293"}, + {"freq": 460.425, "name": "PD3"}]} + tr, files = tr_config.build_tr_config("P25", "WPPD", cfg, **TR_KW) + sys_ = tr["systems"][0] + assert sys_["type"] == "conventionalP25" and sys_["channelFile"] == "/app/channels.csv" + assert "digitalRecorders" not in tr["sources"][0] + rows = files["channels.csv"].splitlines() + assert rows[0].startswith("TG Number,Frequency") + # Same id the OP25 path reports for the channel, so C2 names both alike. + assert rows[1] == f"{channel_tgid(460_400_000)},460400000,PD1,PD1" + + +def test_analog_squelch_threshold_carries_over_else_default(): + cfg = {"channels": [{"freq": 154.25, "name": "FD"}], "squelch_mode": "power", "squelch_threshold": -62} + tr, files = tr_config.build_tr_config("NBFM", "FD", cfg, **TR_KW) + assert tr["systems"][0]["type"] == "conventional" and tr["systems"][0]["squelch"] == -62.0 + tr, _ = tr_config.build_tr_config("NBFM", "FD", {"channels": [{"freq": 154.25}]}, **TR_KW) + assert tr["systems"][0]["squelch"] == -50.0 + # Single channel: centre moved off it (DC spike). + assert abs(tr["sources"][0]["center"] - 154_250_000) >= tr_config.DC_SPIKE_GUARD_HZ + assert tr_config.live_sample_rate("NBFM", {}) == 16000 and tr_config.live_sample_rate("P25", {}) == 8000 + + +def test_legacy_analog_and_unpinned_sdr(): + kw = {**TR_KW, "device_serial": None} + tr, _ = tr_config.build_tr_config("NBFM", "Old", {"control_channels": [155.1]}, **kw) + assert tr["sources"][0]["device"] == "rtl=0" + assert "155100000" in _["channels.csv"] + + +@pytest.mark.parametrize("stype,cfg", [ + ("DMR", {"control_channels": [450.0]}), + ("P25", {"control_channels": []}), + ("P25", {"control_channels": [851.0, 855.0]}), # 4 MHz: wider than one RTL-SDR + ("P25", {"mode": "conventional", "channels": []}), +]) +def test_unbuildable_systems_raise(stype, cfg): + with pytest.raises(ValueError): + tr_config.build_tr_config(stype, "x", cfg, **TR_KW) + + +def test_center_override_is_validated(): + assert tr_config.choose_center([851_000_000], 2_400_000, override=851_500_000) == 851_500_000 + with pytest.raises(ValueError): + tr_config.choose_center([851_000_000], 2_400_000, override=853_000_000) + + +# --------------------------------------------------------------------------- +# Call JSON -> MQTT + /upload +# --------------------------------------------------------------------------- + +def test_parse_call_maps_tr_json(): + call = parse_call(SAMPLE_CALL, "node-001", "101-1793145600.250_851012500-call_7") + assert call.tgid == 101 and call.tgid_name == "PD Dispatch" and call.freq == 851_012_500 + assert call.start_epoch == pytest.approx(1793145600.25) and call.stop_epoch == pytest.approx(1793145612.73) + assert call.units == [1234567, 7654321] # unique, in order, undecoded (-1) dropped + assert call.emergency is True # a unit's emergency flag counts + assert call.encrypted is False + assert call.decode_errors == 5 and call.spikes == 1 and call.call_length == pytest.approx(12.48) + # Deterministic: same JSON -> same id (restart safety), different node -> different id. + assert parse_call(SAMPLE_CALL, "node-001", "101-1793145600.250_851012500-call_7").call_id == call.call_id + assert parse_call(SAMPLE_CALL, "node-002", "101-1793145600.250_851012500-call_7").call_id != call.call_id + + start = call.start_payload() + assert start["tgid"] == 101 and start["tgid_name"] == "PD Dispatch" and start["srcaddr"] == 1234567 + assert start["attributed"] is True and start["driver"] == "trunk-recorder" + assert start["started_at"].startswith("2026-") + end = call.end_payload() + assert end["ended_at_epoch"] == pytest.approx(1793145612.73) + assert end["audio_start_epoch"] == call.start_epoch and end["units"] == [1234567, 7654321] + assert end["emergency"] is True and end["encrypted"] is False + + +def test_upload_form_carries_tr_fields(): + call = parse_call(SAMPLE_CALL, "node-001") + meta = UploadMeta(call_id=call.call_id, talkgroup_id=call.tgid, talkgroup_name=call.tgid_name, + system_id="sys1", audio_start_epoch=call.start_epoch, audio_end_epoch=call.stop_epoch, + extra=call.upload_extra()) + form = upload_spool._form(meta) + # Fields C2's /upload declares today: + assert form["call_id"] == call.call_id and form["talkgroup_id"] == "101" + assert form["talkgroup_name"] == "PD Dispatch" and form["system_id"] == "sys1" + # Extra fields (ignored by C2 until it declares them): + assert form["units"] == "1234567,7654321" and form["emergency"] == "1" and form["encrypted"] == "0" + assert form["freq"] == "851012500" and form["recorder_backend"] == "trunk-recorder" + assert json.loads(form["src_list"])[1] == {"src": 7654321, "time": 1793145606, "pos": 5.2, "emergency": True} + # Extra can never overwrite a core field. + meta.extra = {"call_id": "evil"} + assert upload_spool._form(meta)["call_id"] == call.call_id + + +def test_parse_call_old_tr_without_ms_and_bad_json(): + old = {k: v for k, v in SAMPLE_CALL.items() if not k.endswith("_ms")} + call = parse_call(old, "n") + assert call.start_epoch == 1793145600.0 and call.stop_epoch == 1793145612.0 + with pytest.raises(ValueError): + parse_call({**SAMPLE_CALL, "talkgroup": 0}, "n") + with pytest.raises(ValueError): + parse_call({k: v for k, v in old.items() if k != "start_time"}, "n") + + +def test_mime_follows_suffix(): + assert upload_spool.mime_for(Path("a.wav")) == "audio/wav" + assert upload_spool.mime_for(Path("a.flac")) == "audio/flac" + + +# --------------------------------------------------------------------------- +# Inbox hand-off +# --------------------------------------------------------------------------- + +class FakeMQTT: + is_connected = True + + def __init__(self): + self.events = [] + + async def publish_metadata(self, event, data): + self.events.append((event, dict(data))) + + +class FakeRecorder: + def __init__(self, url="gs://b/calls/x.wav"): + self.url = url + self.uploads = [] + + class _Spool: + last_result = {"kind": "retry"} + self.spool = _Spool() + + async def upload_recording(self, path, call_id, **kw): + assert path.exists() + self.uploads.append((path.name, call_id, kw)) + path.unlink() + return self.url + + +@pytest.fixture +def tr_env(tmp_path, monkeypatch): + monkeypatch.setattr(settings, "recorder_backend", "trunk-recorder") + monkeypatch.setattr(settings, "tr_inbox_dir", str(tmp_path / "inbox")) + monkeypatch.setattr(settings, "tr_config_dir", str(tmp_path / "tr")) + monkeypatch.setattr(settings, "c2_url", "https://c2.example") + monkeypatch.setattr(trmod, "CALL_START_LEAD_SECONDS", 0) + mqtt, rec = FakeMQTT(), FakeRecorder() + import app.internal.call_recorder as cr + import app.internal.mqtt_manager as mm + monkeypatch.setattr(mm, "mqtt_manager", mqtt) + monkeypatch.setattr(cr, "call_recorder", rec) + (tmp_path / "inbox").mkdir() + return tmp_path / "inbox", mqtt, rec + + +def _wav(path: Path, seconds: float = 2.0, rate: int = 8000) -> None: + import wave + with wave.open(str(path), "wb") as w: + w.setnchannels(1) + w.setsampwidth(2) + w.setframerate(rate) + w.writeframes(b"\x00\x00" * int(seconds * rate)) + + +def _drop(inbox: Path, data=SAMPLE_CALL, stem="101-1793145600.250_851012500-call_7", wav=True, seconds=11.68): + if wav: + _wav(inbox / f"{stem}.wav", seconds) + (inbox / f"{stem}.json").write_text(json.dumps(data)) + + +async def test_inbox_call_is_published_and_uploaded_once(tr_env): + inbox, mqtt, rec = tr_env + _drop(inbox) + t = trmod.TrunkRecorder() + assert await t.scan_once(wait=True) == 1 + assert [e for e, _ in mqtt.events] == ["call_start", "call_end"] + start, end = mqtt.events[0][1], mqtt.events[1][1] + assert len(rec.uploads) == 1 + name, call_id, kw = rec.uploads[0] + assert call_id == start["call_id"] == end["call_id"] + # Renamed to the spool's recoverable _