From d07d312e9269e03c5a7355de2ac7aea5dd6dfb94 Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Sun, 4 Oct 2026 15:57:15 -0400 Subject: [PATCH] P25 conventional decode mode + conventional-channel attribution (node-26#24, #20) op25-container - generate-config takes p25_mode "conventional" + conv_channels [{frequency, name, nac}]. It builds fixed fsk4 channels (up to 2 on one non-tunable SDR, DC spike kept off the channels) with no control channel to hunt. Two channels send audio to UDP 23456/23457, and audio.py -2 + amplify(2) keeps levels. - tk_drb_conv.py (a tk_p25.rx_ctl subclass, copied into OP25 apps) reports voice / NAC / encryption (LC service options + LDU2 ESS algid/keyid) for conventional channels and applies the per-channel NAC filter, which stock tk_p25 does not. Analog uses it too. - Trunked P25 config is unchanged when p25_mode is absent. drb-edge-node - radio_config: system kind from type + config.mode (missing = trunked). - conventional.py normalizes the console: a distinctive LC TGID, else the channel's last TGID (30 s), else a synthetic per-channel id (freq/100 Hz). Encrypted rows are never talkers. With exactly one conventional channel, orphan audio is attributed to it at close, which fixes analog recording (#20). The synthetic id and the real TGID on one channel never split. - Health/dashboard: P25 conventional listen tab (/api/listen p25conv), and conventional decode + encryption checks. The checkin carries radio_channels. - Tests: edge 296 pass; op25-container 18 (new CI job). Reviewed by drb-audio-review (findings fixed; simultaneous two-channel mixdown filed as #26) and drb-qa-review. Not verified on SDR hardware. Co-Authored-By: Claude Opus 5.5 --- .gitea/workflows/ci.yml | 16 + drb-edge-node/app/internal/conventional.py | 243 +++++++++++ drb-edge-node/app/internal/health.py | 55 ++- .../app/internal/metadata_watcher.py | 103 ++++- drb-edge-node/app/internal/mqtt_manager.py | 5 + drb-edge-node/app/internal/radio_config.py | 129 ++++++ drb-edge-node/app/main.py | 48 +-- drb-edge-node/app/routers/api.py | 74 +++- drb-edge-node/app/templates/index.html | 30 +- drb-edge-node/tests/test_conventional.py | 402 ++++++++++++++++++ drb-edge-node/tests/test_local_dashboard.py | 59 ++- op25-container/Dockerfile | 4 + .../app/internal/conventional_config.py | 118 +++++ .../app/internal/liquidsoap_config_utils.py | 29 +- .../app/internal/op25_liq_template.py | 3 +- op25-container/app/models.py | 29 +- op25-container/app/routers/op25_controller.py | 29 +- op25-container/op25_addons/tk_drb_conv.py | 263 ++++++++++++ op25-container/tests/test_conventional.py | 305 +++++++++++++ 19 files changed, 1865 insertions(+), 79 deletions(-) create mode 100644 drb-edge-node/app/internal/conventional.py create mode 100644 drb-edge-node/app/internal/radio_config.py create mode 100644 drb-edge-node/tests/test_conventional.py create mode 100644 op25-container/app/internal/conventional_config.py create mode 100644 op25-container/op25_addons/tk_drb_conv.py create mode 100644 op25-container/tests/test_conventional.py diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml index a8fc6fe..c22f042 100644 --- a/.gitea/workflows/ci.yml +++ b/.gitea/workflows/ci.yml @@ -19,6 +19,22 @@ jobs: - name: Run tests run: python3 -m pytest tests -v + test-op25: + # Config generation + tk_drb_conv state machine only (node-26#24) — no + # GNU Radio here; tests/test_conventional.py stubs OP25's own modules. + runs-on: ubuntu-latest + defaults: + run: + working-directory: op25-container + steps: + - uses: actions/checkout@v4 + + - name: Install dependencies + run: python3 -m pip install --break-system-packages -r requirements.txt pytest httpx + + - name: Run tests + run: python3 -m pytest tests -v + lint: runs-on: ubuntu-latest steps: diff --git a/drb-edge-node/app/internal/conventional.py b/drb-edge-node/app/internal/conventional.py new file mode 100644 index 0000000..a7cdb71 --- /dev/null +++ b/drb-edge-node/app/internal/conventional.py @@ -0,0 +1,243 @@ +""" +Attribution for CONVENTIONAL channels — P25 conventional (node-26#24) and +analog NBFM (node-26#20) share this one path. + +metadata_watcher labels audio with whatever the OP25 console says was talking. +On a trunked system that is a talkgroup from a control-channel grant. On a +conventional channel there is no control channel, so: + + * P25 conventional: link control in each LDU1 usually carries the talkgroup + and unit id, and our tk_drb_conv module adds `voice` (frames arriving) and + encryption. The talkgroup is often the P25 default (1) or "everyone" + (0xFFFF) and identical on every channel, which says nothing about WHICH + channel was keyed, and sometimes it is missing altogether. + * Analog: OP25 reports the channel with tgid None and nothing else — the + console never says anyone is talking. + +So a conventional channel gets a stable SYNTHETIC talkgroup id derived from its +frequency (channel_tgid), used whenever link control gives nothing that +distinguishes the channel. A real, distinct link-control talkgroup is still +used when present. normalize() rewrites the console update so the rest of the +watcher sees ordinary "talkgroup X is active" rows and grants and needs no +conventional special cases, except one: with exactly ONE conventional channel +configured, every sound OP25 puts on drb_sink came from that channel, so audio +the console said nothing about is attributed to it (sole_channel) instead of +being discarded as orphan audio. With two channels there is no such certainty +and orphan audio is still discarded (DEFERRED.md, metadata_watcher row). + +Encrypted conventional traffic is never a talker: the decoder mutes it, so no +audio follows, and letting it compete for attribution could mislabel a clear +call on the other channel. It is reported instead (encryption_report). +""" +import time +from typing import Any, Callable, Dict, List, Optional, Tuple + +# P25 talkgroups that identify no particular group: TIA-102 default (1) and +# "everyone" (0xFFFF). 0 is "none" and already rejected by _as_int. +NON_DISTINCT_TGIDS = frozenset({1, 0xFFFF}) + +# Synthetic ids are frequency in 100 Hz units: 460.400 MHz -> 4_604_000. Any +# RTL-SDR frequency (>= 24 MHz) gives >= 240_000, clear of the 16-bit P25 +# talkgroup space, and 6.25 kHz channel steps stay distinct. C2 resolves the +# same number back to the channel name (server talkgroups.name_from_system). +SYNTHETIC_TGID_MIN = 240_000 + +# How long a channel's last real talkgroup is remembered for a transmission +# whose link control hasn't decoded (yet). Covers the back-and-forth of one +# conversation; past it, an undecoded call is labelled with the channel. +REMEMBER_TGID_SECONDS = 30.0 + + +def channel_tgid(freq: Any) -> Optional[int]: + """Stable synthetic talkgroup id for a conventional channel frequency (Hz or MHz).""" + try: + f = float(freq) + except (TypeError, ValueError): + return None + if f <= 0: + return None + hz = f * 1_000_000 if f < 1_000_000 else f + return int(round(hz / 100)) + + +def _as_int(value: Any) -> Optional[int]: + try: + n = int(value) + except (TypeError, ValueError): + return None + return n if n > 0 else None + + +def _truthy(value: Any) -> bool: + return bool(value) and value not in ("0", "false", "False") + + +ALGID_NAMES = { + 0x81: "DES-OFB", 0x83: "Triple DES", 0x84: "AES-256", 0x85: "AES-128", + 0x9F: "DES-XL", 0xA0: "DVI-XL", 0xA1: "DVP-XL", 0xAA: "ADP (RC4)", +} + + +def algid_name(algid: Any) -> str: + a = _as_int(algid) + if a is None: + return "unknown algorithm" + return f"{ALGID_NAMES.get(a, 'algorithm')} (ALGID 0x{a:02X})" + + +class ConventionalChannels: + """Per-poll normalizer plus what it has learned about the channels.""" + + def __init__(self) -> None: + # msgqid -> latest raw conventional row. channel_update is not on every + # poll, so this outlives a single update. + self._rows: Dict[str, Dict[str, Any]] = {} + self._known = False # have we seen a channel_update at all + # synthetic channel id -> the talkgroup that channel last carried. A + # transmission's header frame sets `voice` ~180 ms before its first + # link control names the talkgroup; without this, every such poll + # would label the call with the synthetic id and the real talkgroup + # a poll later would force a bogus split (drb-audio-review, #24). + self._last_identity: Dict[int, Tuple[int, float]] = {} + self._clock: Callable[[], float] = time.time + + # -- identity -------------------------------------------------------- + + @staticmethod + def _name(row: Dict[str, Any]) -> str: + name = str(row.get("name") or "").strip() + if name and not name.startswith("["): + return name + f = row.get("freq") + try: + return f"{float(f) / 1e6:.4f} MHz" + except (TypeError, ValueError): + return "Conventional" + + def identity(self, row: Dict[str, Any], lc_tgid: Optional[int]) -> Optional[int]: + """Distinctive link-control talkgroup; else (no link control decoded + yet) what this channel last carried; else the channel's synthetic id.""" + synthetic = channel_tgid(row.get("freq")) + now = self._clock() + if lc_tgid is not None and lc_tgid not in NON_DISTINCT_TGIDS: + if synthetic is not None: + self._last_identity[synthetic] = (lc_tgid, now) + return lc_tgid + remembered = self._last_identity.get(synthetic) if synthetic is not None else None + if lc_tgid is None and remembered and now - remembered[1] <= REMEMBER_TGID_SECONDS: + self._last_identity[synthetic] = (remembered[0], now) # still the same conversation + return remembered[0] + if synthetic is not None and lc_tgid is not None: + self._last_identity[synthetic] = (synthetic, now) # explicit default TG: channel identity + return synthetic + + # -- normalization --------------------------------------------------- + + def normalize(self, update: Any) -> Any: + """Rewrite conventional rows and grants in place; trunked ones untouched.""" + if update.channels: + self._known = True + self._rows = {} + for i, row in enumerate(update.channels): + if not _truthy(row.get("conventional")): + continue + key = str(row.get("msgqid", i)) + self._rows[key] = row + update.channels[i] = self._normalize_row(row) + + for i, entry in enumerate(update.call_log): + row = self._rows.get(str(entry.get("rcvr"))) + if row is None: + continue + lc = _as_int(entry.get("tgid")) + tgid = self.identity(row, lc) + fixed = dict(entry) + fixed["lc_tgid"] = lc + fixed["tgid"] = tgid + fixed["tgtag"] = self._label(row, lc, entry.get("tgtag")) + fixed["freq"] = entry.get("freq") or row.get("freq") + update.call_log[i] = fixed + return update + + def _label(self, row: Dict[str, Any], lc: Optional[int], tag: Any) -> str: + name = self._name(row) + tag = str(tag or "").strip() + if lc is not None and lc not in NON_DISTINCT_TGIDS and tag and tag not in ("Conventional", name): + return tag + return name + + def _normalize_row(self, row: Dict[str, Any]) -> Dict[str, Any]: + lc = _as_int(row.get("tgid")) + encrypted = _truthy(row.get("encrypted")) + # tgid is only ever set while a call is up (tk_p25 / tk_drb_conv clear + # it on TDU and sync timeout); `voice` is tk_drb_conv's own signal. + active = (lc is not None or _truthy(row.get("voice"))) and not encrypted + out = dict(row) + out["lc_tgid"] = lc + out["tgid"] = self.identity(row, lc) if active else None + out["tag"] = self._label(row, lc, row.get("tag")) + out["drb_conventional"] = True + out["drb_active"] = active + return out + + # -- what the watcher asks ------------------------------------------- + + def same_channel_relabel(self, current_tgid: Optional[int], current_freq: Any, + new_tgid: Optional[int], new_freq: Any) -> Optional[str]: + """Two labels for ONE conventional channel are not a talkgroup change. + + "adopt": the open segment holds the channel's synthetic id and the + console now names the real talkgroup — relabel, don't split. + "keep": the segment holds a real talkgroup and the console offers the + synthetic id (link control not decoded on this frame) — ignore. + None: not the same conventional channel; the normal split rules apply. + """ + if current_tgid is None or new_tgid is None or current_tgid == new_tgid: + return None + synthetic = channel_tgid(new_freq) + if synthetic is None or channel_tgid(current_freq) != synthetic: + return None + if not any(channel_tgid(r.get("freq")) == synthetic for r in self._rows.values()): + return None # trunked: freq says nothing about identity + if current_tgid == synthetic: + return "adopt" + if new_tgid == synthetic: + return "keep" + return None + + def sole_channel(self) -> Optional[Dict[str, Any]]: + """The one conventional channel this node listens to, if exactly one + (and it is not encrypted). None on any trunked or multi-channel setup.""" + if len(self._rows) != 1: + return None + row = next(iter(self._rows.values())) + tgid = channel_tgid(row.get("freq")) + if tgid is None: + return None + if _truthy(row.get("encrypted")): + return None + return {"tgid": tgid, "name": self._name(row), "freq": row.get("freq")} + + def encryption_report(self) -> List[Dict[str, Any]]: + """Per conventional channel: is it (or was its last call) encrypted.""" + out = [] + for row in self._rows.values(): + enc_calls = _as_int(row.get("enc_calls")) or 0 + clear_calls = _as_int(row.get("clear_calls")) or 0 + now_enc = _truthy(row.get("encrypted")) + last_enc = row.get("last_enc_time") or 0 + last_clear = row.get("last_clear_time") or 0 + out.append({ + "name": self._name(row), + "freq": row.get("freq"), + "nac": _as_int(row.get("nac")), + "encrypted_now": now_enc, + "enc_calls": enc_calls, + "clear_calls": clear_calls, + # The channel "is encrypted" when its most recent call was. + "encrypted": now_enc or (enc_calls > 0 and float(last_enc) >= float(last_clear)), + "algid": _as_int(row.get("algid")) or _as_int(row.get("last_enc_algid")), + "keyid": _as_int(row.get("keyid")) or _as_int(row.get("last_enc_keyid")), + "voice": _truthy(row.get("voice")), + }) + return out diff --git a/drb-edge-node/app/internal/health.py b/drb-edge-node/app/internal/health.py index 064222e..1268a67 100644 --- a/drb-edge-node/app/internal/health.py +++ b/drb-edge-node/app/internal/health.py @@ -56,6 +56,51 @@ def _ago(seconds: float) -> str: return f"{seconds // 3600} h ago" +def _channel_label(ch: Dict[str, Any]) -> str: + freq = _mhz(ch.get("freq")) + name = ch.get("name") or "" + if name and freq and freq not in name: + return f"{name} ({freq})" + return name or freq or "a channel" + + +def _conventional_decode_check(chans: List[Dict[str, Any]]) -> Dict[str, str]: + """P25 conventional (node-26#24): no control channel to lock on to, so the + proof of decoding is P25 frames (a NAC) heard on the channels — and the one + failure the owner can't fix is encryption.""" + if not chans: + return _check("decode", "Decoding", WARN, "Waiting for OP25 to report its channels…") + encrypted = [c for c in chans if c.get("encrypted")] + if encrypted: + from app.internal.conventional import algid_name + parts = [] + for c in encrypted: + key = f", key 0x{c['keyid']:04X}" if c.get("keyid") is not None else "" + parts.append(f"{_channel_label(c)}: {algid_name(c.get('algid'))}{key}") + # "Everything is encrypted" only when no clear call was ever heard — + # `encrypted` alone means a channel's most recent call was. + every = len(encrypted) == len(chans) and not any(c.get("clear_calls") for c in chans) + return _check( + "decode", "Decoding", ERROR if every else WARN, + ("Encrypted — " if every else "Partly encrypted — ") + "; ".join(parts) + + ". Encrypted calls can't be decoded, so nothing from " + + ("these channels" if every else "those channels") + " is recorded.", + "Nothing to fix on the node: the agency encrypts this traffic. " + + ("Point the node at a different system." if every else "Clear calls are still recorded."), + ) + heard = [c for c in chans if c.get("nac")] + listening = ", ".join(_channel_label(c) for c in chans) + if not heard: + return _check( + "decode", "Decoding", OK, + f"Listening on {listening} (P25 conventional) — no P25 traffic heard yet.", + "Normal on a quiet channel: conventional channels are silent until someone talks. If it stays " + "silent through a busy period, check the frequency, antenna and the SDR's PPM.", + ) + nacs = ", ".join(sorted({f"0x{c['nac']:03X}" for c in heard})) + return _check("decode", "Decoding", OK, f"Listening on {listening} (P25 conventional) — decoding, NAC {nacs}.") + + def active_config(node_cfg) -> Any: if node_cfg.override_system_id and node_cfg.override_config: return node_cfg.override_config @@ -144,7 +189,8 @@ def build_checks(f: Dict[str, Any]) -> List[Dict[str, str]]: )) 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}.")) + 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}.")) # 6. OP25 op25 = f.get("op25_status") @@ -170,6 +216,7 @@ def build_checks(f: Dict[str, Any]) -> List[Dict[str, str]]: 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") + is_conventional = bool(src) and src.get("mode") == "conventional" 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: @@ -182,8 +229,9 @@ def build_checks(f: Dict[str, Any]) -> List[Dict[str, str]]: 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).", )) + elif is_conventional: + checks.append(_conventional_decode_check(f.get("conventional") or [])) else: trunk = console.get("trunk") or [] tsbk = [float(t.get("last_tsbk") or 0) for t in trunk] @@ -261,6 +309,7 @@ async def gather_facts() -> Dict[str, Any]: 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.radio_config import system_mode from app.internal.secondary_sdr_client import secondary_sdr_client node_cfg = load_node_config() @@ -289,8 +338,10 @@ async def gather_facts() -> Dict[str, Any]: "sdr_count": sdr_count, "source": None if cfg is None else { "name": cfg.name, "type": cfg.type, "system_id": cfg.system_id, + "mode": system_mode(cfg.type, cfg.config), "local": bool(node_cfg.override_system_id and node_cfg.override_config), }, + "conventional": metadata_watcher.conventional_channels, "op25_status": (op25 or {}).get("status") if op25 is not None else None, "console": metadata_watcher.console_state, "audio_capturing": call_recorder.is_capturing, diff --git a/drb-edge-node/app/internal/metadata_watcher.py b/drb-edge-node/app/internal/metadata_watcher.py index 2718bdf..4a8e86d 100644 --- a/drb-edge-node/app/internal/metadata_watcher.py +++ b/drb-edge-node/app/internal/metadata_watcher.py @@ -36,6 +36,19 @@ have seen the whole window and ask the rolling console history "what was active during this audio, give or take a few seconds" — see _attribute and the ATTRIBUTION_* constants. +CONVENTIONAL CHANNELS (node-26#24 P25 conventional, node-26#20 analog). There +is no control channel, so internal/conventional.py rewrites each console update +first: a conventional channel that is keyed becomes an ordinary active row whose +talkgroup is the link-control TGID when it is distinctive, else a stable +synthetic id derived from the channel frequency. Such a row counts as a talker +without a radio id (`drb_active`) — link control does not always carry one and +analog never does. One more rule, applied only at CLOSE: with exactly one +conventional channel configured, audio the console said nothing about can only +have come from that channel, so it is attributed to it rather than discarded. +Not at open: on a channel whose link control carries a real talkgroup, the +grant may arrive a poll after the audio, and a premature synthetic label would +force a bogus split when it does. + ORPHAN AUDIO. If nothing in the console history overlaps the window, the audio is unattributed: Liquidsoap fallback, a test tone, stray noise, or a dropped `call_log`. Policy is DISCARD AND SHOUT — the recording is not uploaded and no @@ -74,6 +87,7 @@ from typing import Optional, Callable, Awaitable, Any, List, Dict from app.config import settings from app.internal.call_recorder import AudioActivity +from app.internal.conventional import ConventionalChannels from app.internal.op25_client import op25_client from app.internal.logger import logger @@ -177,6 +191,22 @@ def _as_float(value: Any) -> Optional[float]: return None +def _talker(channel: Dict[str, Any]) -> Optional[tuple]: + """(tgid, rid) if this channel row says someone is transmitting, else None. + + Trunked rows need both a talkgroup and a source address — an idle channel + can keep its last tgid. A normalized conventional row (`drb_active`) only + has a tgid while a call is up, and may legitimately have no radio id. + """ + tgid = _as_int(channel.get("tgid")) + if tgid is None: + return None + rid = _as_int(channel.get("srcaddr")) + if rid is None and not channel.get("drb_active"): + return None + return tgid, rid + + def _iso(epoch: Optional[float]) -> Optional[str]: if epoch is None: return None @@ -252,6 +282,10 @@ class MetadataWatcher: # Field-visible counter of discarded orphan audio. self._unattributed_segments: int = 0 + # Conventional-channel normalizer (node-26#24/#20); also remembers the + # channel list and their encryption state for the dashboard + checkin. + self._conventional = ConventionalChannels() + # Injectable for tests; production is always the host wall clock. self._clock: Callable[[], float] = time.time @@ -300,6 +334,7 @@ class MetadataWatcher: return self._last_ok_poll = now + update = self._conventional.normalize(update) self._record_console(update, now) self._note_terminal(update, now) @@ -341,10 +376,10 @@ class MetadataWatcher: self._console.append(event) for channel in update.channels: - tgid = _as_int(channel.get("tgid")) - srcaddr = _as_int(channel.get("srcaddr")) - if tgid is None or srcaddr is None: + talker = _talker(channel) + if talker is None: continue # idle channel says nothing about who is talking + tgid, srcaddr = talker self._console.append(ConsoleEvent( epoch=now, tgid=tgid, @@ -379,7 +414,7 @@ class MetadataWatcher: return True return False - def _attribute(self, start: float, end: float) -> Optional[Attribution]: + def _attribute(self, start: float, end: float, allow_sole_channel: bool = False) -> Optional[Attribution]: """ Resolve which talkgroup a stretch of audio belongs to. @@ -387,7 +422,8 @@ class MetadataWatcher: how well its console activity overlaps the audio itself, preferring real overlap over merely being nearby, and grants over channel rows. Returns None only when NOTHING was observed in that band at all — the - orphan-audio case. + orphan-audio case — unless allow_sole_channel and the node listens to + exactly one conventional channel (see module docstring). """ low = start - ATTRIBUTION_LOOKBACK_SECONDS high = end + ATTRIBUTION_LOOKAHEAD_SECONDS @@ -415,7 +451,14 @@ class MetadataWatcher: found.rid = event.rid # most recent wins if not candidates: - return None + sole = self._conventional.sole_channel() if allow_sole_channel else None + if sole is None: + return None + logger.info( + f"No console activity near {start:.3f}-{end:.3f}; attributing to the only " + f"conventional channel, {sole['name']} (tgid {sole['tgid']})." + ) + return Attribution(tgid=sole["tgid"], name=sole["name"], freq=sole["freq"]) best = max( candidates.values(), @@ -530,6 +573,11 @@ class MetadataWatcher: self._refresh_meta_from_log(entry) return + if self._same_channel(tgid, entry.get("freq"), entry.get("tgtag") or ""): + self._transmissions += 1 + self._refresh_meta_from_log(entry) + return + # FORCED SPLIT. Two talkgroups can be back to back with no silence # between them; pure audio segmentation would merge them into one file # under one label, which is exactly the kind of wrong that corrupts @@ -546,6 +594,21 @@ class MetadataWatcher: transmissions=1, ) + def _same_channel(self, tgid: int, freq: Any, name: str) -> bool: + """A conventional channel seen under its synthetic id and its real + talkgroup is one channel, not a split (see ConventionalChannels. + same_channel_relabel). Adopts the real talkgroup when offered.""" + decision = self._conventional.same_channel_relabel(self._current_tgid, self._current_freq, tgid, freq) + if decision == "adopt": + logger.info( + f"Segment {self._active_call_id}: link control names tgid {tgid} for the channel " + f"labelled {self._current_tgid} — relabelling, not splitting." + ) + self._current_tgid = tgid + if name: + self._current_tgid_name = name + return decision is not None + async def _scan_channels(self, channels: List[Dict[str, Any]], now: float) -> None: """Channel rows in audio mode: refresh metadata, catch an unlogged split.""" if self._active_call_id is None: @@ -554,14 +617,15 @@ class MetadataWatcher: active: List[Dict[str, Any]] = [] ours = False for channel in channels: - tgid = _as_int(channel.get("tgid")) - srcaddr = _as_int(channel.get("srcaddr")) - if tgid is None or srcaddr is None: + talker = _talker(channel) + if talker is None: continue + tgid, srcaddr = talker active.append(channel) - if tgid == self._current_tgid: + if tgid == self._current_tgid or self._same_channel(tgid, channel.get("freq"), channel.get("tag") or ""): ours = True - self._current_srcaddr = srcaddr + if srcaddr is not None: + self._current_srcaddr = srcaddr self._last_activity = now self._refresh_meta_from_channel(channel) @@ -658,13 +722,14 @@ class MetadataWatcher: foreign_active_tgid: Optional[int] = None for channel in channels: - srcaddr = _as_int(channel.get("srcaddr")) - chan_tgid = _as_int(channel.get("tgid")) - if srcaddr is None: + talker = _talker(channel) + if talker is None: continue + chan_tgid, srcaddr = talker if chan_tgid == self._current_tgid: tx_active = True - self._current_srcaddr = srcaddr + if srcaddr is not None: + self._current_srcaddr = srcaddr self._refresh_meta_from_channel(channel) elif chan_tgid is not None: foreign_active_tgid = chan_tgid @@ -868,7 +933,7 @@ class MetadataWatcher: two talkgroups' console activity overlaps one recording, i.e. the split logic should have fired and did not. """ - found = self._attribute(start, end) + found = self._attribute(start, end, allow_sole_channel=self._current_tgid is None) if found is None: return @@ -931,6 +996,12 @@ class MetadataWatcher: }, } + @property + def conventional_channels(self) -> List[Dict[str, Any]]: + """Per conventional channel: name, freq, NAC heard, encryption. Empty on + a trunked system. Read by /api/health and the checkin.""" + return self._conventional.encryption_report() + @property def unattributed_segments(self) -> int: """Orphan-audio segments discarded since start. Surfaced on /api/status.""" diff --git a/drb-edge-node/app/internal/mqtt_manager.py b/drb-edge-node/app/internal/mqtt_manager.py index 39d6332..9434212 100644 --- a/drb-edge-node/app/internal/mqtt_manager.py +++ b/drb-edge-node/app/internal/mqtt_manager.py @@ -327,6 +327,11 @@ class MQTTManager: "secondary_sdr_priority": config.secondary_sdr_priority, } payload["sdr_pins"] = config.sdr_pins + # Conventional channels (node-26#24): name/freq/NAC and whether the + # agency encrypts them — the one decode failure nobody can fix on the + # node, so C2 has to be told. Empty list on a trunked system. + from app.internal.metadata_watcher import metadata_watcher + payload["radio_channels"] = metadata_watcher.conventional_channels 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 diff --git a/drb-edge-node/app/internal/radio_config.py b/drb-edge-node/app/internal/radio_config.py new file mode 100644 index 0000000..08b853b --- /dev/null +++ b/drb-edge-node/app/internal/radio_config.py @@ -0,0 +1,129 @@ +""" +What kind of radio system a SystemConfig describes, and the op25-container +generate-config payload for it. + +C2 system documents (the `config` blob is passed through untouched): + + P25 trunked type "P25", config.mode absent or "trunked", + config.control_channels = [MHz, ...] + P25 conventional type "P25", config.mode "conventional", + config.channels = [{freq: MHz, name, nac?}, ...] (node-26#24) + Analog type "NBFM", config.channels = [{freq, name}] — or, from + before #24, config.control_channels[0] + +A missing mode is trunked, so every system saved before #24 behaves exactly as +it did. Pure functions; main._generate_op25_config does the I/O. +""" +from typing import Any, Dict, List, Optional + +TRUNKED, CONVENTIONAL, ANALOG, DMR = "trunked", "conventional", "analog", "dmr" + +# Mirrors op25-container's MAX_CONVENTIONAL_CHANNELS: one audio.py carries two. +MAX_CONVENTIONAL_CHANNELS = 2 + + +def to_hz(freq: Any) -> int: + """MHz floats (< 1e6) or Hz ints -> Hz.""" + f = float(freq) + return int(round(f * 1_000_000)) if f < 1_000_000 else int(f) + + +def parse_nac(value: Any) -> Optional[int]: + """NAC as typed by an admin ("293", "0x293") or stored as an int. Hex, like + RadioReference prints it. None when blank; ValueError when nonsense.""" + if value is None or (isinstance(value, str) and not value.strip()): + return None + if isinstance(value, bool): + raise ValueError(f"NAC {value!r} isn't a number.") + if isinstance(value, int): + nac = value + else: + text = str(value).strip().lower() + text = text[2:] if text.startswith("0x") else text + try: + nac = int(text, 16) + except ValueError: + raise ValueError(f"NAC {value!r} isn't hex (e.g. 293 or 0x293).") + if not 0 < nac <= 0xFFF: + raise ValueError(f"NAC {value!r} is outside 001-FFF.") + return nac + + +def system_mode(system_type: str, config: Dict[str, Any]) -> str: + t = str(system_type or "").upper() + if t in ("NBFM", "ANALOG"): + return ANALOG + if t == "DMR": + return DMR + mode = str((config or {}).get("mode") or TRUNKED).lower() + return CONVENTIONAL if mode == CONVENTIONAL else TRUNKED + + +def system_kind(system_type: str, config: Dict[str, Any]) -> str: + """Label for dashboards: "P25 trunked", "P25 conventional", "Analog FM", "DMR".""" + return { + TRUNKED: "P25 trunked", CONVENTIONAL: "P25 conventional", ANALOG: "Analog FM", DMR: "DMR", + }[system_mode(system_type, config)] + + +def conventional_channels(config: Dict[str, Any]) -> List[Dict[str, Any]]: + """config.channels -> [{frequency: Hz, name, nac}] (ValueError if malformed).""" + out = [] + for raw in (config or {}).get("channels") or []: + if not isinstance(raw, dict) or raw.get("freq") in (None, ""): + raise ValueError("Every conventional channel needs a frequency.") + out.append({ + "frequency": to_hz(raw["freq"]), + "name": str(raw.get("name") or "").strip(), + "nac": parse_nac(raw.get("nac")), + }) + return out + + +def analog_frequency(config: Dict[str, Any]) -> Optional[int]: + chans = (config or {}).get("channels") or [] + if chans and isinstance(chans[0], dict) and chans[0].get("freq") not in (None, ""): + return to_hz(chans[0]["freq"]) + legacy = (config or {}).get("control_channels") or [] + return to_hz(legacy[0]) if legacy else None + + +def build_generate_payload( + system_type: str, + name: str, + config: Dict[str, Any], + icecast: Dict[str, Any], + hardware_preset: str, + ppm_override: Optional[float], +) -> Dict[str, Any]: + """SystemConfig fields -> op25-container /generate-config body.""" + mode = system_mode(system_type, config) + payload: Dict[str, Any] = { + "type": system_type, + "systemName": name, + "icecastConfig": icecast, + "hardware_preset": hardware_preset, + } + if ppm_override is not None: + payload["ppm_override"] = ppm_override + + if mode == CONVENTIONAL: + payload["type"] = "P25" + payload["p25_mode"] = CONVENTIONAL + payload["channels"] = [] + payload["conv_channels"] = conventional_channels(config) + return payload + + if mode == ANALOG: + freq = analog_frequency(config) + payload["channels"] = [freq] if freq else [] + return payload + + talkgroups = (config or {}).get("talkgroups", []) or [] + payload["channels"] = [to_hz(ch) for ch in (config or {}).get("control_channels", []) or []] + payload["tags"] = [ + {"talkgroup": str(tg.get("name", "")), "tagDec": int(tg["id"])} + for tg in talkgroups if tg.get("id") is not None + ] + payload["whitelist"] = [int(tg["id"]) for tg in talkgroups if tg.get("id") is not None] + return payload diff --git a/drb-edge-node/app/main.py b/drb-edge-node/app/main.py index 6b5da86..f89c3f0 100644 --- a/drb-edge-node/app/main.py +++ b/drb-edge-node/app/main.py @@ -191,36 +191,30 @@ async def on_api_key(payload: dict): logger.info("Node API key received and saved.") -def _to_hz(freq) -> int: - """Convert a frequency to Hz. Accepts MHz floats (< 1e6) or Hz ints.""" - f = float(freq) - return int(f * 1_000_000) if f < 1_000_000 else int(f) - - async def _generate_op25_config(config: SystemConfig) -> bool: - """Translate a SystemConfig (Firestore format) into OP25 active.cfg.json + op25.liq.""" + """Translate a SystemConfig (Firestore format) into OP25 active.cfg.json + op25.liq. + + 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. + """ from app.internal.op25_client import op25_client + from app.internal.radio_config import build_generate_payload node_cfg = load_node_config() - raw = config.config - payload = { - "type": config.type, - "systemName": config.name, - "channels": [_to_hz(ch) for ch in raw.get("control_channels", [])], - "tags": [ - {"talkgroup": str(tg.get("name", "")), "tagDec": int(tg["id"])} - for tg in raw.get("talkgroups", []) - if tg.get("id") is not None - ], - "whitelist": [int(tg["id"]) for tg in raw.get("talkgroups", []) if tg.get("id") is not None], - "icecastConfig": { - "icecast_host": settings.icecast_host, - "icecast_port": settings.icecast_port, - "icecast_mountpoint": settings.icecast_mount, - "icecast_password": settings.icecast_source_password, - }, - "hardware_preset": node_cfg.hardware_preset, - **({"ppm_override": node_cfg.ppm_override} if node_cfg.ppm_override is not None else {}), - } + try: + payload = build_generate_payload( + config.type, config.name, config.config, + icecast={ + "icecast_host": settings.icecast_host, + "icecast_port": settings.icecast_port, + "icecast_mountpoint": settings.icecast_mount, + "icecast_password": settings.icecast_source_password, + }, + hardware_preset=node_cfg.hardware_preset, + ppm_override=node_cfg.ppm_override, + ) + except ValueError as e: + logger.error(f"System {config.name!r} has an unusable channel list: {e}") + return False return await op25_client.generate_config(payload) diff --git a/drb-edge-node/app/routers/api.py b/drb-edge-node/app/routers/api.py index bba4175..0b6ab34 100644 --- a/drb-edge-node/app/routers/api.py +++ b/drb-edge-node/app/routers/api.py @@ -37,9 +37,12 @@ async def get_status(): else node_cfg.system_config ) system_type = None + system_kind = None if active_config: + from app.internal.radio_config import system_kind as _kind system_name = active_config.name system_type = active_config.type + system_kind = _kind(active_config.type, active_config.config) if active_tgid: # Reverse engineer the TGID to talkgroup name based on active system object tgs = active_config.config.get("talkgroups", []) @@ -61,6 +64,8 @@ async def get_status(): "assigned_system_id": node_cfg.assigned_system_id, "system_name": system_name, "system_type": system_type, + # Human label that tells P25 trunked from P25 conventional (node-26#24). + "system_kind": system_kind, # 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), @@ -192,11 +197,14 @@ class Talkgroup(BaseModel): class ListenBody(BaseModel): """What to listen to, chosen on the local dashboard.""" - mode: Literal["system", "p25", "analog"] + mode: Literal["system", "p25", "p25conv", "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 + # MHz. p25 = control channels; p25conv = 1-2 voice channels (node-26#24); + # analog = exactly one. + frequencies: List[float] = [] talkgroups: List[Talkgroup] = [] # p25 only; empty = every talkgroup + nac: Optional[str] = None # p25conv only, hex; empty = any NAC # RTL-SDR (R820T/R828D) tuning range. @@ -225,22 +233,62 @@ async def listen(body: ListenBody): ) 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 [], - }, - ) + if body.mode == "p25conv": + config = _conventional_listen_config(body, freqs, first) + else: + is_p25 = body.mode == "p25" + 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} +# One SDR's usable passband, less a channel's width — op25-container makes the +# exact check; this one just turns the common mistake into a readable error. +_CONVENTIONAL_MAX_SPAN_MHZ = 1.5 + + +def _conventional_listen_config(body: ListenBody, freqs: List[float], first: str) -> SystemConfig: + from app.internal.radio_config import MAX_CONVENTIONAL_CHANNELS, parse_nac + if len(freqs) > MAX_CONVENTIONAL_CHANNELS: + raise HTTPException(400, f"P25 conventional listens to at most {MAX_CONVENTIONAL_CHANNELS} channels.") + if len(set(freqs)) != len(freqs): + raise HTTPException(400, "The same frequency is entered twice.") + if max(freqs) - min(freqs) > _CONVENTIONAL_MAX_SPAN_MHZ: + raise HTTPException( + 400, f"Those channels are {max(freqs) - min(freqs):.3f} MHz apart — one SDR covers about " + f"{_CONVENTIONAL_MAX_SPAN_MHZ:g} MHz. Pick channels closer together.", + ) + try: + nac = parse_nac(body.nac) + except ValueError as e: + raise HTTPException(400, str(e)) + label = (body.name or "").strip() or f"P25 conventional {first} MHz" + return SystemConfig( + system_id="local-p25conv", + name=label, + type="P25", + config={ + "mode": "conventional", + "channels": [ + {"freq": f, "name": f"{label} {f:.4f}" if len(freqs) > 1 else label, + **({"nac": f"{nac:03X}"} if nac is not None else {})} + for f in freqs + ], + "talkgroups": [], + }, + ) + + @router.get("/systems") async def get_systems(): from app.internal.system_cacher import load_cached_systems, fetch_and_cache_systems diff --git a/drb-edge-node/app/templates/index.html b/drb-edge-node/app/templates/index.html index 4121d20..8847e2d 100644 --- a/drb-edge-node/app/templates/index.html +++ b/drb-edge-node/app/templates/index.html @@ -167,6 +167,7 @@
+
@@ -188,6 +189,23 @@ + +