Merge pull request 'P25 conventional decode + conventional-channel attribution (node-26#24, #20)' (#27) from feat/24-p25-conventional into main

Reviewed-on: #27

.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:

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

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,

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:
+            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,13 +617,14 @@ 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
+                if srcaddr is not None:
                     self._current_srcaddr = srcaddr
                 self._last_activity = now
                 self._refresh_meta_from_channel(channel)
@@ -658,12 +722,13 @@ 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
+                if srcaddr is not None:
                     self._current_srcaddr = srcaddr
                 self._refresh_meta_from_channel(channel)
             elif chan_tgid is not None:
@@ -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."""

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

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

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": {
+    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} if node_cfg.ppm_override is not None else {}),
-    }
+            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)
 
 

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,8 +233,11 @@ 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(".")
+        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}",
@@ -241,6 +252,43 @@ async def listen(body: ListenBody):
     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

drb-edge-node/app/templates/index.html

@@ -167,6 +167,7 @@
 
     <div class="tabs" role="radiogroup" aria-label="Source type">
       <label><input type="radio" name="mode" value="p25" checked> P25 trunked</label>
+      <label><input type="radio" name="mode" value="p25conv"> P25 conventional</label>
       <label><input type="radio" name="mode" value="analog"> Analog FM</label>
       <label><input type="radio" name="mode" value="system" id="mode-system"> Server's systems</label>
     </div>
@@ -188,6 +189,23 @@
       </div>
     </div>
 
+    <div class="pane" data-pane="p25conv" hidden>
+      <div class="field">
+        <label for="pc-freqs">Channel frequencies (MHz)</label>
+        <input type="text" id="pc-freqs" class="mono" placeholder="460.4000, 460.4250" inputmode="decimal">
+        <span class="hint">One or two voice channels, comma-separated, within about 1.5 MHz of each other. No control channel — RadioReference lists these as conventional frequencies with mode P25 (or "DE").</span>
+      </div>
+      <div class="field">
+        <label for="pc-name">Name <span class="hint">(optional)</span></label>
+        <input type="text" id="pc-name" placeholder="City PD">
+      </div>
+      <div class="field">
+        <label for="pc-nac">NAC <span class="hint">(optional, hex)</span></label>
+        <input type="text" id="pc-nac" class="mono" placeholder="293">
+        <span class="hint">RadioReference's NAC for these channels. Leave empty to accept any.</span>
+      </div>
+    </div>
+
     <div class="pane" data-pane="analog" hidden>
       <div class="field">
         <label for="an-freq">Frequency (MHz)</label>
@@ -307,7 +325,7 @@
       $('node-name').textContent = d.node_name || 'DRB Edge Node';
       $('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_type})` : 'Nothing chosen yet';
+      $('now-system').textContent = d.system_name ? `${d.system_name} (${d.system_kind || d.system_type})` : 'Nothing chosen yet';
       $('now-tg').innerHTML = d.active_tgid
         ? `<span class="live">${esc(d.active_tgid_name || 'Talkgroup')}</span> <span class="mono">TG ${esc(d.active_tgid)}</span>`
         : 'No call right now';
@@ -335,7 +353,7 @@
       actions.hidden = true;
       return;
     }
-    cur.innerHTML = `Now: <strong>${esc(d.system_name)}</strong> (${esc(d.system_type)}) — ` +
+    cur.innerHTML = `Now: <strong>${esc(d.system_name)}</strong> (${esc(d.system_kind || d.system_type)}) — ` +
       (d.local_choice ? 'chosen on this dashboard.' : 'assigned by the DRB server.');
     actions.hidden = !d.local_choice;
     $('revert-btn').textContent = d.has_c2_system ? "Go back to the server's system" : 'Clear and stop';
@@ -387,7 +405,9 @@
     try { systems = await api('/api/systems'); } catch (e) { systems = []; }
     const pick = $('sys-pick');
     const usable = systems.filter(s => String(s.type).toUpperCase() !== 'DMR');
-    pick.innerHTML = usable.map(s => `<option value="${esc(s.system_id)}">${esc(s.name)} (${esc(s.type)})</option>`).join('');
+    const kind = s => String(s.type).toUpperCase() === 'P25' && s.config && s.config.mode === 'conventional'
+      ? 'P25 conventional' : s.type;
+    pick.innerHTML = usable.map(s => `<option value="${esc(s.system_id)}">${esc(s.name)} (${esc(kind(s))})</option>`).join('');
     const none = usable.length === 0;
     $('mode-system').disabled = none;
     pick.disabled = none;
@@ -416,6 +436,10 @@
         tgs.push({ id: Number(mm[1]), name: mm[2] });
       }
       body = { mode: 'p25', frequencies: freqs, name: $('p25-name').value, talkgroups: tgs };
+    } else if (m === 'p25conv') {
+      const freqs = parseFreqs($('pc-freqs').value);
+      if (!freqs.length || freqs.length > 2 || freqs.some(isNaN)) { msg.className = 'msg err'; msg.textContent = 'Enter one or two channel frequencies in MHz, e.g. 460.4000, 460.4250'; return; }
+      body = { mode: 'p25conv', frequencies: freqs, name: $('pc-name').value, nac: $('pc-nac').value.trim() || null };
     } else {
       const freqs = parseFreqs($('an-freq').value);
       if (freqs.length !== 1 || isNaN(freqs[0])) { msg.className = 'msg err'; msg.textContent = 'Enter one frequency in MHz, e.g. 154.4300'; return; }

drb-edge-node/tests/test_conventional.py

@@ -0,0 +1,402 @@
+"""
+Conventional channels: P25 conventional (node-26#24) and analog (node-26#20).
+
+Covers the generate-config payload (radio_config), the console normalizer
+(conventional.py) and how the watcher attributes conventional audio. Channel
+rows are shaped like tk_drb_conv's / tk_p25's conventional `channel_update`
+entries; grants like tk_p25.log_call entries.
+"""
+import pytest
+from unittest.mock import AsyncMock
+
+from app.config import settings
+from app.internal import radio_config as rc
+from app.internal.conventional import ConventionalChannels, channel_tgid, NON_DISTINCT_TGIDS
+from app.internal.metadata_watcher import MetadataWatcher
+from tests.test_metadata_watcher import FakeAudio, FakeClock, tick, update
+
+PD1, PD3 = 460_400_000, 460_425_000
+
+
+def conv_row(msgqid=0, freq=PD1, name="PD1 Dispatch", tgid=None, srcaddr=0, voice=False,
+             encrypted=0, **extra):
+    row = {
+        "freq": freq, "name": name, "tgid": tgid, "tag": name, "srcaddr": srcaddr,
+        "encrypted": encrypted, "conventional": True, "msgqid": msgqid, "voice": voice,
+    }
+    row.update(extra)
+    return row
+
+
+def conv_grant(rcvr, tgid, at, freq=PD1, rid=4242):
+    return {"time": at, "sysid": 0, "rcvr": rcvr, "rcvrtag": "", "freq": freq, "slot": 0,
+            "prio": 0, "tgid": tgid, "tgtag": "", "rid": rid, "rtag": ""}
+
+
+# ---------------------------------------------------------------------------
+# radio_config: system document -> op25 generate-config payload
+# ---------------------------------------------------------------------------
+
+ICE = {"icecast_host": "localhost", "icecast_port": 8000, "icecast_mountpoint": "/n", "icecast_password": "p"}
+
+WHITE_PLAINS = {
+    "mode": "conventional",
+    "channels": [
+        {"freq": 460.4, "name": "PD1 Dispatch"},
+        {"freq": 460.425, "name": "PD3 Tac", "nac": "293"},
+    ],
+    "talkgroups": [],
+}
+
+
+def test_conventional_payload():
+    p = rc.build_generate_payload("P25", "White Plains", WHITE_PLAINS, ICE, "rtl-sdr-v3", None)
+    assert p["type"] == "P25" and p["p25_mode"] == "conventional"
+    assert p["channels"] == []    # nothing that could become a control channel
+    assert p["conv_channels"] == [
+        {"frequency": PD1, "name": "PD1 Dispatch", "nac": None},
+        {"frequency": PD3, "name": "PD3 Tac", "nac": 0x293},
+    ]
+    assert "ppm_override" not in p
+
+
+def test_trunked_payload_unchanged_when_mode_missing():
+    cfg = {"control_channels": [851.0125, 851.5125], "talkgroups": [{"id": 9048, "name": "Ossining PD"}]}
+    p = rc.build_generate_payload("P25", "County", cfg, ICE, "other", 1.5)
+    assert "p25_mode" not in p
+    assert p["channels"] == [851_012_500, 851_512_500]
+    assert p["tags"] == [{"talkgroup": "Ossining PD", "tagDec": 9048}]
+    assert p["whitelist"] == [9048]
+    assert p["ppm_override"] == 1.5
+
+
+@pytest.mark.parametrize("cfg", [
+    {"channels": [{"freq": 154.43, "name": "Fire"}]},
+    {"control_channels": [154.43]},                     # saved before #24
+])
+def test_analog_payload_reads_channels_or_legacy(cfg):
+    p = rc.build_generate_payload("NBFM", "Fire", cfg, ICE, "rtl-sdr-v3", None)
+    assert p["channels"] == [154_430_000]
+    assert "p25_mode" not in p
+
+
+def test_system_mode():
+    assert rc.system_mode("P25", {}) == rc.TRUNKED
+    assert rc.system_mode("P25", {"mode": "Conventional"}) == rc.CONVENTIONAL
+    assert rc.system_mode("NBFM", {"mode": "conventional"}) == rc.ANALOG
+    assert rc.system_mode("DMR", {}) == rc.DMR
+
+
+@pytest.mark.parametrize("raw, want", [("293", 0x293), ("0x293", 0x293), (0x293, 0x293), ("", None), (None, None)])
+def test_parse_nac(raw, want):
+    assert rc.parse_nac(raw) == want
+
+
+@pytest.mark.parametrize("raw", ["xyz", "1000", 0, True])
+def test_parse_nac_rejects(raw):
+    with pytest.raises(ValueError):
+        rc.parse_nac(raw)
+
+
+def test_conventional_channel_without_freq_rejected():
+    with pytest.raises(ValueError):
+        rc.conventional_channels({"channels": [{"name": "no freq"}]})
+
+
+# ---------------------------------------------------------------------------
+# Normalizer
+# ---------------------------------------------------------------------------
+
+def test_synthetic_tgid_is_frequency_in_100hz_and_clear_of_p25_space():
+    assert channel_tgid(PD1) == 4_604_000
+    assert channel_tgid(460.4) == 4_604_000
+    assert channel_tgid(460_406_250) != channel_tgid(460_412_500)   # 6.25 kHz steps stay distinct
+    assert channel_tgid(24_000_000) > 0xFFFF                        # lowest RTL-SDR frequency
+
+
+def test_non_distinct_link_control_tgid_becomes_the_channel():
+    conv = ConventionalChannels()
+    up = update(channels=[conv_row(0, PD1, tgid=1, srcaddr=7, voice=True),
+                          conv_row(1, PD3, "PD3 Tac", tgid=1, srcaddr=8, voice=True)])
+    conv.normalize(up)
+    assert [r["tgid"] for r in up.channels] == [4_604_000, 4_604_250]
+    assert [r["lc_tgid"] for r in up.channels] == [1, 1]
+    assert [r["tag"] for r in up.channels] == ["PD1 Dispatch", "PD3 Tac"]
+    assert all(r["drb_active"] for r in up.channels)
+    assert 1 in NON_DISTINCT_TGIDS and 0xFFFF in NON_DISTINCT_TGIDS
+
+
+def test_distinct_link_control_tgid_is_kept():
+    conv = ConventionalChannels()
+    up = update(channels=[conv_row(0, tgid=101, srcaddr=7)],
+                call_log=[conv_grant(0, 101, 1.0)])
+    conv.normalize(up)
+    assert up.channels[0]["tgid"] == 101
+    assert up.call_log[0]["tgid"] == 101
+    assert up.call_log[0]["tgtag"] == "PD1 Dispatch"
+
+
+def test_grants_are_mapped_per_receiver():
+    conv = ConventionalChannels()
+    up = update(channels=[conv_row(0, PD1), conv_row(1, PD3, "PD3 Tac")],
+                call_log=[conv_grant(1, 1, 1.0, freq=PD3)])
+    conv.normalize(up)
+    assert up.call_log[0]["tgid"] == 4_604_250
+    assert up.call_log[0]["tgtag"] == "PD3 Tac"
+
+
+def test_voice_without_link_control_is_still_a_talker():
+    conv = ConventionalChannels()
+    up = update(channels=[conv_row(0, voice=True)])
+    conv.normalize(up)
+    assert up.channels[0]["tgid"] == 4_604_000 and up.channels[0]["drb_active"]
+
+
+def test_idle_and_encrypted_channels_are_not_talkers():
+    conv = ConventionalChannels()
+    up = update(channels=[conv_row(0), conv_row(1, PD3, "PD3 Tac", tgid=1, voice=True, encrypted=1)])
+    conv.normalize(up)
+    assert [r["tgid"] for r in up.channels] == [None, None]
+
+
+def test_trunked_rows_untouched():
+    conv = ConventionalChannels()
+    row = {"freq": 851_000_000, "tgid": 9048, "srcaddr": 5, "tag": "PD", "msgqid": 0}
+    up = update(channels=[dict(row)], call_log=[conv_grant(0, 9048, 1.0)])
+    conv.normalize(up)
+    assert up.channels[0] == row
+    assert up.call_log[0]["tgid"] == 9048
+    assert conv.sole_channel() is None
+
+
+def test_sole_channel_only_with_exactly_one_conventional_channel():
+    conv = ConventionalChannels()
+    conv.normalize(update(channels=[conv_row(0, 154_430_000, "Fire")]))
+    assert conv.sole_channel() == {"tgid": 1_544_300, "name": "Fire", "freq": 154_430_000}
+    conv.normalize(update(channels=[conv_row(0), conv_row(1, PD3)]))
+    assert conv.sole_channel() is None
+    # A poll without channel_update does not forget the channel list.
+    conv.normalize(update())
+    assert conv.sole_channel() is None
+
+
+def test_encryption_report():
+    conv = ConventionalChannels()
+    conv.normalize(update(channels=[
+        conv_row(0, enc_calls=3, clear_calls=0, last_enc_time=50.0, last_enc_algid=0x84,
+                 last_enc_keyid=0x1234, nac=0x293),
+        conv_row(1, PD3, "PD3 Tac", enc_calls=1, clear_calls=5, last_enc_time=10.0, last_clear_time=20.0),
+    ]))
+    pd1, pd3 = conv.encryption_report()
+    assert pd1["encrypted"] is True and pd1["algid"] == 0x84 and pd1["keyid"] == 0x1234 and pd1["nac"] == 0x293
+    assert pd3["encrypted"] is False   # its most recent call was clear
+
+
+# ---------------------------------------------------------------------------
+# Watcher attribution
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def clock():
+    return FakeClock()
+
+
+@pytest.fixture
+def audio(clock):
+    return FakeAudio(clock)
+
+
+@pytest.fixture
+def hearing(clock, audio):
+    w = MetadataWatcher()
+    w._clock = clock
+    w.on_call_start = AsyncMock()
+    w.on_call_end = AsyncMock()
+    w.audio_activity = audio
+    return w
+
+
+async def _finish(hearing, clock, channels):
+    clock.advance(settings.call_silence_timeout + 0.5)
+    await tick(hearing, update(channels=channels))
+    return hearing.on_call_end.call_args[0][0]
+
+
+@pytest.mark.asyncio
+async def test_conventional_call_attributed_to_its_channel(hearing, clock, audio):
+    """White Plains: both channels carry the default TGID 1; the keyed one wins."""
+    audio.speak()
+    await tick(hearing, update(
+        channels=[conv_row(0, tgid=1, srcaddr=4242, voice=True), conv_row(1, PD3, "PD3 Tac")],
+        call_log=[conv_grant(0, 1, clock.now)],
+    ))
+    assert hearing.is_active and hearing.current_tgid == 4_604_000
+
+    ended = await _finish(hearing, clock, [conv_row(0), conv_row(1, PD3, "PD3 Tac")])
+    assert ended["attributed"] is True
+    assert (ended["tgid"], ended["tgid_name"], ended["srcaddr"]) == (4_604_000, "PD1 Dispatch", 4242)
+    assert hearing.unattributed_segments == 0
+
+
+@pytest.mark.asyncio
+async def test_other_channel_keying_splits(hearing, clock, audio):
+    audio.speak()
+    await tick(hearing, update(
+        channels=[conv_row(0, tgid=1, voice=True), conv_row(1, PD3, "PD3 Tac")],
+        call_log=[conv_grant(0, 1, clock.now)],
+    ))
+    clock.advance(0.5)
+    audio.speak()
+    await tick(hearing, update(
+        channels=[conv_row(0), conv_row(1, PD3, "PD3 Tac", tgid=1, voice=True)],
+        call_log=[conv_grant(1, 1, clock.now, freq=PD3)],
+    ))
+    first = hearing.on_call_end.call_args[0][0]
+    assert first["tgid"] == 4_604_000 and first["end_reason"] == "tgid_change"
+    assert hearing.current_tgid == 4_604_250
+
+
+@pytest.mark.asyncio
+async def test_single_analog_channel_audio_is_kept_not_orphaned(hearing, clock, audio):
+    """node-26#20: analog never reports a talker; the only channel is the answer."""
+    audio.speak()
+    rows = [conv_row(0, 154_430_000, "County Fire")]
+    await tick(hearing, update(channels=rows))
+    # Not labelled at open — only at close.
+    assert hearing.on_call_start.call_args[0][0]["tgid"] is None
+
+    ended = await _finish(hearing, clock, rows)
+    assert ended["attributed"] is True
+    assert (ended["tgid"], ended["tgid_name"]) == (1_544_300, "County Fire")
+    assert hearing.unattributed_segments == 0
+
+
+@pytest.mark.asyncio
+async def test_two_channels_with_no_evidence_is_still_orphan(hearing, clock, audio):
+    audio.speak()
+    rows = [conv_row(0), conv_row(1, PD3, "PD3 Tac")]
+    await tick(hearing, update(channels=rows))
+    ended = await _finish(hearing, clock, rows)
+    assert ended["attributed"] is False
+    assert hearing.unattributed_segments == 1
+
+
+@pytest.mark.asyncio
+async def test_late_real_talkgroup_on_single_channel_does_not_split(hearing, clock, audio):
+    """Audio first, link-control grant a poll later: adopt it, don't split."""
+    audio.speak()
+    await tick(hearing, update(channels=[conv_row(0)]))
+    clock.advance(0.5)
+    audio.speak()
+    await tick(hearing, update(channels=[conv_row(0, tgid=101, srcaddr=9, voice=True)],
+                               call_log=[conv_grant(0, 101, clock.now - 0.3)]))
+    assert hearing.current_tgid == 101
+    hearing.on_call_end.assert_not_called()
+    ended = await _finish(hearing, clock, [conv_row(0)])
+    assert ended["tgid"] == 101 and ended["end_reason"] == "audio_silence"
+
+
+@pytest.mark.asyncio
+async def test_encrypted_channel_does_not_label_clear_audio(hearing, clock, audio):
+    """PD3 is keyed encrypted (muted) while PD1 talks in the clear."""
+    audio.speak()
+    await tick(hearing, update(channels=[
+        conv_row(0, tgid=1, voice=True),
+        conv_row(1, PD3, "PD3 Tac", tgid=1, voice=True, encrypted=1, algid=0x84),
+    ]))
+    ended = await _finish(hearing, clock, [conv_row(0), conv_row(1, PD3, "PD3 Tac")])
+    assert ended["tgid"] == 4_604_000
+
+
+@pytest.mark.asyncio
+async def test_conventional_channels_property(hearing, clock):
+    await tick(hearing, update(channels=[conv_row(0, encrypted=1, voice=True, algid=0x84, keyid=7)]))
+    (only,) = hearing.conventional_channels
+    assert only["encrypted"] and only["encrypted_now"] and only["algid"] == 0x84
+
+
+# drb-audio-review findings on #24: a header frame (voice, no link control yet)
+# must not label the call with the synthetic id and then split against the
+# real talkgroup a poll later; nor may a reply's header split it back.
+
+@pytest.mark.asyncio
+async def test_header_before_link_control_does_not_split(hearing, clock, audio):
+    audio.speak()
+    await tick(hearing, update(channels=[conv_row(0, voice=True)]))       # HDU only
+    clock.advance(0.5)
+    audio.speak()
+    await tick(hearing, update(channels=[conv_row(0, tgid=101, srcaddr=9, voice=True)],
+                               call_log=[conv_grant(0, 101, clock.now - 0.2)]))
+    hearing.on_call_end.assert_not_called()
+    assert hearing.current_tgid == 101
+    ended = await _finish(hearing, clock, [conv_row(0)])
+    assert ended["tgid"] == 101 and ended["end_reason"] == "audio_silence"
+    assert hearing.on_call_end.call_count == 1
+
+
+@pytest.mark.asyncio
+async def test_reply_header_inside_conversation_does_not_ping_pong(hearing, clock, audio):
+    audio.speak()
+    await tick(hearing, update(channels=[conv_row(0, tgid=101, srcaddr=9, voice=True)],
+                               call_log=[conv_grant(0, 101, clock.now)]))
+    clock.advance(0.5)
+    await tick(hearing, update(channels=[conv_row(0)]))                   # TDU
+    clock.advance(0.5)
+    audio.speak()
+    await tick(hearing, update(channels=[conv_row(0, voice=True)]))       # reply's HDU
+    clock.advance(0.5)
+    audio.speak()
+    await tick(hearing, update(channels=[conv_row(0, tgid=101, srcaddr=12, voice=True)],
+                               call_log=[conv_grant(0, 101, clock.now)]))
+    hearing.on_call_end.assert_not_called()
+    ended = await _finish(hearing, clock, [conv_row(0)])
+    assert ended["tgid"] == 101 and hearing.on_call_end.call_count == 1
+
+
+@pytest.mark.asyncio
+async def test_never_decoded_link_control_then_real_tg_relabels(hearing, clock, audio):
+    """First-ever call on a channel: synthetic label, then the real TG adopts it."""
+    audio.speak()
+    await tick(hearing, update(channels=[conv_row(0, voice=True)]))
+    clock.advance(0.5)
+    audio.speak()
+    await tick(hearing, update(channels=[conv_row(0, voice=True)]))
+    assert hearing.current_tgid == 4_604_000
+    clock.advance(0.5)
+    audio.speak()
+    await tick(hearing, update(channels=[conv_row(0, tgid=101, voice=True)],
+                               call_log=[conv_grant(0, 101, clock.now)]))
+    hearing.on_call_end.assert_not_called()
+    assert hearing.current_tgid == 101
+
+
+def test_voice_without_link_control_uses_channels_last_talkgroup():
+    conv = ConventionalChannels()
+    conv.normalize(update(channels=[conv_row(0, tgid=101, voice=True)]))
+    up = update(channels=[conv_row(0, voice=True)])
+    conv.normalize(up)
+    assert up.channels[0]["tgid"] == 101
+
+
+def test_same_channel_relabel_is_conventional_only():
+    conv = ConventionalChannels()
+    assert conv.same_channel_relabel(4_604_000, PD1, 101, PD1) is None    # no conventional rows known
+    conv.normalize(update(channels=[conv_row(0), conv_row(1, PD3)]))
+    assert conv.same_channel_relabel(4_604_000, PD1, 101, PD1) == "adopt"
+    assert conv.same_channel_relabel(101, PD1, 4_604_000, PD1) == "keep"
+    assert conv.same_channel_relabel(4_604_000, PD1, 4_604_250, PD3) is None   # other channel: split
+
+
+def test_remembered_talkgroup_expires():
+    conv = ConventionalChannels()
+    t = [1000.0]
+    conv._clock = lambda: t[0]
+    conv.normalize(update(channels=[conv_row(0, tgid=101, voice=True)]))
+    t[0] += 10
+    up = update(channels=[conv_row(0, voice=True)])
+    conv.normalize(up)
+    assert up.channels[0]["tgid"] == 101
+    t[0] += 3600
+    up = update(channels=[conv_row(0, voice=True)])
+    conv.normalize(up)
+    assert up.channels[0]["tgid"] == 4_604_000

drb-edge-node/tests/test_local_dashboard.py

@@ -126,10 +126,48 @@ def test_analog_needs_no_control_channel():
         console={"last_poll_epoch": now, "channels": [{"freq": 154430000}], "trunk": [], "last_heard": None},
     )))
     assert c["decode"]["state"] == "ok" and "154.4300 MHz" in c["decode"]["summary"]
-    assert "not recorded" in c["decode"]["action"]
+    # node-26#20: analog calls are recorded now (single-channel attribution).
+    assert c["decode"]["action"] == ""
     assert "this dashboard" in c["source"]["summary"]
 
 
+_CONV_SRC = {"name": "White Plains", "type": "P25", "mode": "conventional", "system_id": "wp", "local": False}
+_PD1 = {"name": "PD1 Dispatch", "freq": 460400000, "nac": None, "encrypted": False, "algid": None, "keyid": None}
+_PD3 = {"name": "PD3 Tac", "freq": 460425000, "nac": None, "encrypted": False, "algid": None, "keyid": None}
+
+
+def _conv_facts(chans):
+    now = 1_000_000.0
+    return _facts(source=_CONV_SRC, conventional=chans,
+                  console={"last_poll_epoch": now, "channels": [], "trunk": [], "last_heard": None})
+
+
+def test_conventional_quiet_channels_are_ok_not_a_control_channel_error():
+    c = _by_id(health.build_checks(_conv_facts([_PD1, _PD3])))
+    assert c["decode"]["state"] == "ok"
+    assert "PD1 Dispatch (460.4000 MHz)" in c["decode"]["summary"]
+    assert "control channel" not in c["decode"]["summary"]
+    assert "P25 conventional" in c["source"]["summary"]
+
+
+def test_conventional_decoding_shows_nac():
+    c = _by_id(health.build_checks(_conv_facts([dict(_PD1, nac=0x293), _PD3])))
+    assert c["decode"]["state"] == "ok" and "NAC 0x293" in c["decode"]["summary"]
+
+
+def test_conventional_all_encrypted_is_an_error_naming_the_algorithm():
+    enc = [dict(_PD1, encrypted=True, algid=0x84, keyid=0x1234), dict(_PD3, encrypted=True, algid=0x84)]
+    c = _by_id(health.build_checks(_conv_facts(enc)))
+    assert c["decode"]["state"] == "error"
+    assert "AES-256" in c["decode"]["summary"] and "0x1234" in c["decode"]["summary"]
+    assert "different system" in c["decode"]["action"]
+
+
+def test_conventional_partly_encrypted_warns():
+    c = _by_id(health.build_checks(_conv_facts([dict(_PD1, encrypted=True, algid=0x81), _PD3])))
+    assert c["decode"]["state"] == "warn" and "PD1 Dispatch" in c["decode"]["summary"]
+
+
 def test_op25_unreachable_and_icecast_down():
     c = _by_id(health.build_checks(_facts(op25_status=None, icecast_live=None)))
     assert c["op25"]["state"] == "error"
@@ -182,7 +220,26 @@ def test_listen_analog_single_frequency(client, op25):
     assert (cfg.type, cfg.name, cfg.config["control_channels"]) == ("NBFM", "Fire", [154.43])
 
 
+def test_listen_p25_conventional(client, op25):
+    r = client.post("/api/listen", json={
+        "mode": "p25conv", "frequencies": [460.4, 460.425], "name": "White Plains", "nac": "0x293",
+    })
+    assert r.status_code == 200, r.text
+    cfg = config_manager.load_node_config().override_config
+    assert cfg.type == "P25" and cfg.config["mode"] == "conventional"
+    assert "control_channels" not in cfg.config
+    assert [c["freq"] for c in cfg.config["channels"]] == [460.4, 460.425]
+    assert {c["nac"] for c in cfg.config["channels"]} == {"293"}
+    op25["gen"].assert_awaited_once()
+    status = client.get("/api/status").json()
+    assert status["system_kind"] == "P25 conventional"
+
+
 @pytest.mark.parametrize("body,needle", [
+    ({"mode": "p25conv", "frequencies": [460.4, 460.425, 460.45]}, "at most 2"),
+    ({"mode": "p25conv", "frequencies": [460.4, 462.6]}, "apart"),
+    ({"mode": "p25conv", "frequencies": [460.4, 460.4]}, "twice"),
+    ({"mode": "p25conv", "frequencies": [460.4], "nac": "zz"}, "hex"),
     ({"mode": "p25", "frequencies": []}, "at least one"),
     ({"mode": "p25", "frequencies": [851012500]}, "MHz"),
     ({"mode": "analog", "frequencies": [154.43, 155.1]}, "exactly one"),

op25-container/Dockerfile

@@ -31,6 +31,10 @@ RUN ./install.sh -f
 COPY requirements.txt /tmp/requirements.txt
 RUN pip3 install --no-cache-dir -r /tmp/requirements.txt
 
+# DRB's trunking module for P25 conventional channels (node-26#24): reports
+# encryption + NAC and applies the NAC filter, which stock tk_p25 does not.
+COPY op25_addons/tk_drb_conv.py /op25/op25/gr-op25_repeater/apps/tk_drb_conv.py
+
 # Create the run_multi-rx_service.sh script
 COPY run_multi-rx_service.sh /op25/op25/gr-op25_repeater/apps/run_multi-rx_service.sh
 RUN chmod +x /op25/op25/gr-op25_repeater/apps/run_multi-rx_service.sh

op25-container/app/internal/conventional_config.py

@@ -0,0 +1,118 @@
+"""
+P25 conventional OP25 config (node-26#24).
+
+A conventional P25 channel is a fixed voice frequency with no control channel:
+radios key up on it directly, C4FM (fsk4), and identify themselves in the link
+control of each LDU1 (talkgroup + unit id) and the ESS of each LDU2
+(encryption algorithm + key id). Building a `tk_p25.py` trunking config for one
+— what "P25" always meant before — makes OP25 hunt forever for a control
+channel that does not exist, so nothing is ever decoded.
+
+Layout this produces, per boatbod OP25 gr310 `p25_conventional_example.json`:
+
+  * one non-tunable device parked so every channel sits inside its passband;
+  * one channel per frequency, `trunking_sysname: ""` (so the trunking module
+    treats it as conventional), `demod_type: fsk4`;
+  * trunking module `tk_drb_conv.py` with no chans — our thin subclass of
+    tk_p25.rx_ctl that reports encryption and the NAC for conventional
+    channels and applies the per-channel NAC filter (tk_p25 itself does
+    neither for FDMA conventional);
+  * channel N's audio to UDP 23456+N; with two channels audio.py runs in
+    two-channel mode (left = first channel, right = second).
+
+Pure: no file I/O, so it is unit-testable outside the container.
+"""
+from typing import Any, Dict, List, Tuple
+
+from models import ChannelConfig, ConventionalChannel, DeviceConfig, TerminalConfig
+
+# Two is what one audio.py instance can carry (its A and B sockets). A third
+# channel would need a second audio pipeline into PulseAudio.
+MAX_CONVENTIONAL_CHANNELS = 2
+
+AUDIO_UDP_BASE_PORT = 23456
+
+DEVICE_RATE = 1_920_000
+DEVICE_USABLE_BW_PCT = 0.85
+# Half the occupied bandwidth of a 12.5 kHz P25 channel.
+CHANNEL_HALF_BW_HZ = 6_250
+# Keep the RTL-SDR's DC spike off the channels. A single channel (or two very
+# close together) is offset from the centre; channels further apart than
+# 2x this already straddle the spike, so the centre goes between them.
+DC_SPIKE_OFFSET_HZ = 250_000
+DC_SPIKE_CLEARANCE_HZ = 50_000
+
+CONVENTIONAL_TRUNKING_MODULE = "tk_drb_conv.py"
+
+
+def device_center(freqs: List[int]) -> int:
+    lo, hi = min(freqs), max(freqs)
+    mid = (lo + hi) // 2
+    if (hi - lo) // 2 >= DC_SPIKE_CLEARANCE_HZ:
+        return mid
+    return mid + DC_SPIKE_OFFSET_HZ
+
+
+def validate_channels(channels: List[ConventionalChannel]) -> None:
+    """Raise ValueError with an operator-readable reason."""
+    if not channels:
+        raise ValueError("P25 conventional needs at least one channel frequency.")
+    if len(channels) > MAX_CONVENTIONAL_CHANNELS:
+        raise ValueError(
+            f"P25 conventional supports up to {MAX_CONVENTIONAL_CHANNELS} channels on one SDR "
+            f"(got {len(channels)})."
+        )
+    freqs = [c.frequency for c in channels]
+    if len(set(freqs)) != len(freqs):
+        raise ValueError("The same conventional frequency is listed twice.")
+    for c in channels:
+        if c.nac is not None and not (0 < c.nac <= 0xFFF):
+            raise ValueError(f"NAC {c.nac:#x} on {c.frequency / 1e6:.5f} MHz is outside 0x001-0xFFF.")
+    center = device_center(freqs)
+    half = DEVICE_RATE * DEVICE_USABLE_BW_PCT / 2
+    for f in freqs:
+        if abs(f - center) + CHANNEL_HALF_BW_HZ > half:
+            span = (max(freqs) - min(freqs)) / 1e6
+            raise ValueError(
+                f"These channels are {span:.3f} MHz apart — too far for one SDR "
+                f"(max ~{(2 * (half - CHANNEL_HALF_BW_HZ)) / 1e6:.2f} MHz)."
+            )
+
+
+def build_conventional_config(
+    system_name: str,
+    channels: List[ConventionalChannel],
+    gains: str,
+    ppm: float,
+) -> Tuple[Dict[str, Any], bool]:
+    """Return (OP25 config dict, two_channel_audio)."""
+    validate_channels(channels)
+    freqs = [c.frequency for c in channels]
+    center = device_center(freqs)
+
+    chans = []
+    for i, c in enumerate(channels):
+        chans.append(ChannelConfig(
+            name=c.name or f"{system_name} {c.frequency / 1e6:.4f}",
+            trunking_sysname="",
+            enable_analog="off",
+            demod_type="fsk4",
+            filter_type="rc",
+            frequency=c.frequency,
+            meta_stream_name="",
+            destination=f"udp://127.0.0.1:{AUDIO_UDP_BASE_PORT + i}",
+            nac=c.nac,
+            crypt_behavior=2,
+        ))
+
+    device = DeviceConfig(
+        gains=gains, ppm=ppm, tunable=False, frequency=center,
+        rate=DEVICE_RATE, usable_bw_pct=DEVICE_USABLE_BW_PCT,
+    )
+    config = {
+        "channels": [ch.model_dump() for ch in chans],
+        "devices": [device.model_dump()],
+        "trunking": {"module": CONVENTIONAL_TRUNKING_MODULE, "chans": []},
+        "terminal": TerminalConfig().model_dump(),
+    }
+    return config, len(chans) > 1

op25-container/app/internal/liquidsoap_config_utils.py

@@ -2,7 +2,26 @@ import os
 from internal.op25_liq_template import liquidsoap_config_template
 from models import IcecastConfig
 
-def generate_liquid_script(config: IcecastConfig):
+# P25 conventional with two channels (node-26#24): OP25 sends channel A's
+# audio to UDP 23456 and channel B's to 23457, and audio.py -2 puts them on the
+# left and right of the stereo stream. Every output below is mean(input) — mono
+# — so a lone talker on one side would come out 6 dB down; doubling first keeps
+# the common case (one channel keyed) at the same level as single-channel mode.
+# Liquidsoap mixes in float, so the doubling cannot clip before the output.
+_TWO_CHANNEL_AUDIO_ARGS = " -2"
+_TWO_CHANNEL_MIX = "input = amplify(2.0, input)  # two conventional channels, L=A R=B"
+
+
+def render_liquid_script(config: IcecastConfig, two_channel: bool = False) -> str:
+    content = liquidsoap_config_template
+    for key, value in config.model_dump().items():
+        content = content.replace(f"${{{key}}}", str(value))
+    content = content.replace("${audio_py_args}", _TWO_CHANNEL_AUDIO_ARGS if two_channel else "")
+    content = content.replace("${two_channel_mix}", _TWO_CHANNEL_MIX if two_channel else "")
+    return content
+
+
+def generate_liquid_script(config: IcecastConfig, two_channel: bool = False):
     """
     Generates the "*.liq" file that's run by OP25 on startup. 
 
@@ -13,13 +32,7 @@ def generate_liquid_script(config: IcecastConfig):
                           Keys should match the variable names in the template (e.g., 'icecast_host').
     """
     try:
-        content = liquidsoap_config_template
-        # Replace variables
-        for key, value in config.model_dump().items():
-            placeholder = f"${{{key}}}"
-            # Ensure the value is converted to string for replacement
-            content = content.replace(placeholder, str(value))
-            print(f"  - Replaced placeholder {placeholder}")
+        content = render_liquid_script(config, two_channel)
 
         # Write the processed content to the output path
         output_path = "/configs/op25.liq"

op25-container/app/internal/op25_liq_template.py

@@ -17,8 +17,9 @@ ICE_DESCRIPTION = "${icecast_description}"
 ICE_GENRE = "${icecast_genre}"
 # ==========================================================
 
-input = mksafe(input.external(buffer=0.25, channels=2, samplerate=8000, restart_on_error=false, "./audio.py -x 2.5 -s"))
+input = mksafe(input.external(buffer=0.25, channels=2, samplerate=8000, restart_on_error=false, "./audio.py -x 2.5 -s${audio_py_args}"))
 # Consider increasing the buffer value on slow systems such as RPi3. e.g. buffer=0.25
+${two_channel_mix}
 
 # Compression
 input = compress(input, attack = 2.0, gain = 0.0, knee = 13.0, ratio = 2.0, release = 12.3, threshold = -18.0)

op25-container/app/models.py

@@ -49,12 +49,28 @@ class IcecastConfig(BaseModel):
     icecast_genre: Optional[str] = "Public Safety"
 
 
+class P25Mode(str, Enum):
+    TRUNKED = "trunked"
+    CONVENTIONAL = "conventional"
+
+
+class ConventionalChannel(BaseModel):
+    """One fixed P25 conventional voice channel (node-26#24)."""
+    frequency: int                 # Hz
+    name: str = ""
+    nac: Optional[int] = None      # 0x001-0xFFF; None = accept any NAC
+
+
 class ConfigGenerator(BaseModel):
     type: DecodeMode
     systemName: str
-    # P25: control channel list (Hz). NBFM: channels[0] is the one analog
-    # frequency to listen to (Hz).
-    channels: List[Union[int, str]]
+    # P25 trunked: control channel list (Hz). NBFM: channels[0] is the one
+    # analog frequency to listen to (Hz). P25 conventional: unused — see
+    # conv_channels.
+    channels: List[Union[int, str]] = []
+    # P25 only. Missing = trunked, so every existing caller is unchanged.
+    p25_mode: P25Mode = P25Mode.TRUNKED
+    conv_channels: List[ConventionalChannel] = []
     tags: Optional[List[TalkgroupTag]] = None
     whitelist: Optional[List[int]] = None
     icecastConfig: Optional[IcecastConfig] = None
@@ -84,6 +100,13 @@ class ChannelConfig(BaseModel):
     # Key name op25_nbfm.py actually reads (from_dict(config, 'nbfm_squelch_threshold', -60)).
     nbfm_squelch_threshold: Optional[float] = None
     destination: Optional[str] = "udp://127.0.0.1:23456"
+    # P25 conventional only, read by tk_drb_conv.py (not by multi_rx): the NAC
+    # to lock the frame assembler to. multi_rx passes the whole channel dict
+    # through, so an extra key is harmless.
+    nac: Optional[int] = None
+    # Set on the channel itself for conventional (multi_rx forwards it to the
+    # frame assembler). 2 = drop encrypted voice instead of decoding noise.
+    crypt_behavior: Optional[int] = None
     tracking_threshold: Optional[int] = 120
     tracking_feedback: Optional[float] = 0.75
     excess_bw: Optional[float] = 0.2

op25-container/app/routers/op25_controller.py

@@ -8,7 +8,8 @@ import json
 import threading
 from collections import deque
 from typing import Optional
-from models import ConfigGenerator, DecodeMode, ChannelConfig, DeviceConfig, TrunkingConfig, TrunkingChannelConfig, TerminalConfig, MetadataConfig, MetadataStreamConfig, HARDWARE_PRESETS
+from models import ConfigGenerator, DecodeMode, P25Mode, ChannelConfig, DeviceConfig, TrunkingConfig, TrunkingChannelConfig, TerminalConfig, MetadataConfig, MetadataStreamConfig, HARDWARE_PRESETS
+from internal.conventional_config import build_conventional_config, CONVENTIONAL_TRUNKING_MODULE
 from internal.logger import create_logger
 from internal.op25_config_utls import save_talkgroup_tags, save_whitelist, del_none_in_dict, get_current_system_from_config
 from internal.liquidsoap_config_utils import generate_liquid_script
@@ -178,7 +179,20 @@ def create_op25_router():
     @router.post("/generate-config")
     async def generate_config(generator: ConfigGenerator):
         try:
-            if generator.type == DecodeMode.P25:
+            if generator.type == DecodeMode.P25 and generator.p25_mode == P25Mode.CONVENTIONAL:
+                # node-26#24: fixed voice channels, no control channel to hunt.
+                preset = HARDWARE_PRESETS.get(generator.hardware_preset, HARDWARE_PRESETS["other"])
+                ppm = generator.ppm_override if generator.ppm_override is not None else preset["ppm"]
+                try:
+                    config_dict, two_channel = build_conventional_config(
+                        generator.systemName, generator.conv_channels, preset["gains"], ppm,
+                    )
+                except ValueError as e:
+                    raise HTTPException(status_code=400, detail=str(e))
+                if generator.icecastConfig:
+                    generate_liquid_script(generator.icecastConfig, two_channel=two_channel)
+
+            elif generator.type == DecodeMode.P25:
                 preset = HARDWARE_PRESETS.get(generator.hardware_preset, HARDWARE_PRESETS["other"])
                 ppm = generator.ppm_override if generator.ppm_override is not None else preset["ppm"]
                 channels = [ChannelConfig(
@@ -236,9 +250,9 @@ def create_op25_router():
                 preset = HARDWARE_PRESETS.get(generator.hardware_preset, HARDWARE_PRESETS["other"])
                 ppm = generator.ppm_override if generator.ppm_override is not None else preset["ppm"]
                 freq = int(float(generator.channels[0]))
-                # One fixed conventional channel: no trunking section, so
-                # multi_rx adds its default tk_p25 with no chans and reports
-                # the channel as `conventional` on the HTTP terminal.
+                # One fixed conventional channel: a trunking module with no
+                # chans, so the channel is reported as `conventional` on the
+                # HTTP terminal.
                 channels = [ChannelConfig(
                     name=generator.systemName,
                     trunking_sysname="",
@@ -259,6 +273,11 @@ def create_op25_router():
                 config_dict = {
                     "channels": [channel.dict() for channel in channels],
                     "devices": [device.dict() for device in devices],
+                    # Same module as P25 conventional (node-26#24): it reports
+                    # the channel as `conventional` whatever tk_p25 version the
+                    # image was built with, which is what the edge node's
+                    # single-channel attribution keys on (node-26#20).
+                    "trunking": {"module": CONVENTIONAL_TRUNKING_MODULE, "chans": []},
                     "terminal": TerminalConfig().dict(),
                 }
 

op25-container/op25_addons/tk_drb_conv.py

@@ -0,0 +1,263 @@
+"""
+tk_drb_conv — DRB's trunking module for P25 CONVENTIONAL channels (node-26#24).
+
+Copied into OP25's apps directory by the Dockerfile and named as the trunking
+module by op25-container's conventional config, so multi_rx imports it in place
+of tk_p25. It is tk_p25.rx_ctl with the conventional-channel handling replaced;
+trunked receivers (none in a conventional config) still go to the base class.
+
+Why not stock tk_p25: for an FDMA conventional channel it reports the talkgroup
+and unit id from LDU1 link control, but
+
+  * never reports encryption — the encrypted/algid/keyid JSON OP25 sends after
+    every LDU2 carries no grpaddr, and tk_p25 only records `encrypted` when
+    grpaddr != 0 — so an encrypted channel looks exactly like a quiet one;
+  * never reports the NAC it is hearing, and has no per-channel NAC filter;
+  * has no "voice frames are arriving" signal of its own, so a transmission
+    whose link control is unreadable is invisible to the console.
+
+All three matter to the edge node: it attributes audio to a channel from what
+this module reports, and must tell the owner when a channel is encrypted rather
+than silently recording nothing.
+
+Each conventional channel row in `channel_update` keeps every key tk_p25 sends
+(freq, tgid, srcaddr, tag, encrypted, conventional, ...) and adds:
+
+    voice            True while voice frames are arriving (HDU/LDU1/LDU2 seen,
+                     no TDU or sync timeout since)
+    nac              last NAC heard on the channel (int) or None
+    nac_filter       the configured NAC filter (int) or None
+    algid, keyid     ALGID/key id of the current call (None when clear/idle)
+    enc_calls, clear_calls     calls heard since start, by encryption
+    last_enc_algid, last_enc_keyid, last_enc_time, last_clear_time
+
+Message types handled (lower 16 bits of msg.type(), P25 protocol only):
+    0          HDU                 -> voice start
+    3, 15      TDU / TDULC         -> transmission end
+    19         LDU1 link control   -> tgid, srcaddr, service options (E bit)
+    -3         LDU2 ESS JSON       -> encrypted / algid / keyid
+    -1         sync timeout        -> transmission end
+"""
+import ctypes
+import json
+import sys
+import time
+
+import tk_p25
+from helper_funcs import from_dict, get_ordinals
+from log_ts import log_ts
+
+ALGID_CLEAR = 0x80
+SVCOPT_ENCRYPTED = 0x40
+LCW_PROTECTED = 0x80
+LCO_GROUP_VOICE = 0x00
+DUID_HDU, DUID_TDU, DUID_TDULC, DUID_LCW = 0, 3, 15, 19
+M_TIMEOUT, M_JSON_DATA = -1, -3
+
+
+def _parse_nac(value):
+    if value is None or value == "":
+        return None
+    try:
+        nac = int(value, 16) if isinstance(value, str) else int(value)
+    except (TypeError, ValueError):
+        return None
+    return nac if 0 < nac <= 0xFFF else None
+
+
+class rx_ctl(tk_p25.rx_ctl):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.drb_conv = {}   # msgq_id -> state dict
+
+    def add_receiver(self, msgq_id, config, meta_q=None, freq=0):
+        super().add_receiver(msgq_id, config, meta_q=meta_q, freq=freq)
+        rcvr = self.receivers.get(msgq_id)
+        if rcvr is not None and rcvr.get('rx_rcvr') is not None:
+            return   # trunked receiver: base class owns it
+        self.drb_conv[msgq_id] = {
+            'name': from_dict(config, 'name', str(msgq_id)),
+            'freq': freq,
+            'nac_filter': _parse_nac(config.get('nac')),
+            'nac': None,
+            'voice': False,
+            'tgid': None,
+            'srcaddr': 0,
+            'svc_encrypted': False,
+            'ess_encrypted': False,
+            'algid': None,
+            'keyid': None,
+            'call_start': None,
+            'call_encrypted': False,
+            'logged_tgid': None,
+            'prev_type': None,
+            'enc_calls': 0,
+            'clear_calls': 0,
+            'last_enc_algid': None,
+            'last_enc_keyid': None,
+            'last_enc_time': None,
+            'last_clear_time': None,
+            'last_voice_time': None,
+        }
+
+    def post_init(self):
+        super().post_init()
+        for msgq_id, cs in self.drb_conv.items():
+            if cs['nac_filter'] is not None and self.fa_ctrl is not None:
+                self.fa_ctrl({'tuner': msgq_id, 'cmd': 'set_nac', 'nac': cs['nac_filter']})
+                sys.stderr.write("%s [%d] conventional NAC filter 0x%03x\n" % (log_ts.get(), msgq_id, cs['nac_filter']))
+
+    # ------------------------------------------------------------------
+
+    def _encrypted(self, cs):
+        return cs['svc_encrypted'] or cs['ess_encrypted']
+
+    def _voice_start(self, cs, now):
+        if not cs['voice']:
+            cs['voice'] = True
+            cs['call_start'] = now
+            cs['call_encrypted'] = False
+            cs['logged_tgid'] = None
+        cs['last_voice_time'] = now
+
+    def _voice_end(self, cs, now):
+        if cs['voice']:
+            if cs['call_encrypted']:
+                cs['enc_calls'] += 1
+                cs['last_enc_time'] = now
+            else:
+                cs['clear_calls'] += 1
+                cs['last_clear_time'] = now
+        cs['voice'] = False
+        cs['tgid'] = None
+        cs['srcaddr'] = 0
+        cs['svc_encrypted'] = False
+        cs['ess_encrypted'] = False
+        cs['algid'] = None
+        cs['keyid'] = None
+        cs['call_start'] = None
+        cs['logged_tgid'] = None
+
+    def _note_encryption(self, cs, now):
+        if self._encrypted(cs):
+            cs['call_encrypted'] = True
+            if cs['algid'] is not None:
+                cs['last_enc_algid'] = cs['algid']
+                cs['last_enc_keyid'] = cs['keyid']
+            cs['last_enc_time'] = now
+
+    def _maybe_log_call(self, msgq_id, cs):
+        # One call_log entry per (call, talkgroup), and none for an encrypted
+        # call: it will produce no audio, so a "grant" would only mislead the
+        # edge node's attribution.
+        if cs['tgid'] is None or self._encrypted(cs) or cs['logged_tgid'] == cs['tgid']:
+            return
+        cs['logged_tgid'] = cs['tgid']
+        try:
+            self.log_call(0, msgq_id, cs['freq'], 0, 0, cs['tgid'], cs['name'], cs['srcaddr'], '')
+        except Exception as e:   # older/newer tk_p25 log_call signature
+            sys.stderr.write("%s [%d] conventional call_log failed: %s\n" % (log_ts.get(), msgq_id, e))
+
+    def _process_conv(self, msgq_id, cs, m_type, msg, now):
+        if m_type == M_TIMEOUT or m_type in (DUID_TDU, DUID_TDULC):
+            self._voice_end(cs, now)
+        elif m_type == DUID_HDU:
+            s = msg.to_string()
+            cs['nac'] = get_ordinals(s[0:2]) if len(s) >= 2 else cs['nac']
+            self._voice_start(cs, now)
+        elif m_type == DUID_LCW:
+            s = msg.to_string()
+            if len(s) < 11:
+                return
+            cs['nac'] = get_ordinals(s[0:2])
+            if cs['prev_type'] in (DUID_TDU, DUID_TDULC):
+                return   # TDULC's link control follows its TDU: the call is over
+            lcw = s[2:]
+            self._voice_start(cs, now)
+            pb_sf_lco = get_ordinals(lcw[0:1])
+            if pb_sf_lco & LCW_PROTECTED:
+                # Link control itself is encrypted — the call certainly is.
+                cs['svc_encrypted'] = True
+            elif (pb_sf_lco & 0x3f) == LCO_GROUP_VOICE:
+                svcopts = get_ordinals(lcw[2:3])
+                ga = get_ordinals(lcw[4:6])
+                sa = get_ordinals(lcw[6:9])
+                cs['svc_encrypted'] = bool(svcopts & SVCOPT_ENCRYPTED)
+                if ga:
+                    cs['tgid'] = ga
+                if sa:
+                    cs['srcaddr'] = sa
+            self._note_encryption(cs, now)
+            self._maybe_log_call(msgq_id, cs)
+        elif m_type == M_JSON_DATA:
+            try:
+                js = json.loads(msg.to_string())
+            except (ValueError, TypeError):
+                return
+            if 'encrypted' in js:
+                self._voice_start(cs, now)
+                algid = from_dict(js, 'algid', None)
+                cs['ess_encrypted'] = bool(js.get('encrypted')) and algid != ALGID_CLEAR
+                cs['algid'] = algid if cs['ess_encrypted'] else None
+                cs['keyid'] = from_dict(js, 'keyid', None) if cs['ess_encrypted'] else None
+                self._note_encryption(cs, now)
+            grpaddr = from_dict(js, 'grpaddr', 0)
+            if grpaddr:
+                cs['tgid'] = grpaddr
+                cs['srcaddr'] = from_dict(js, 'srcaddr', 0) or cs['srcaddr']
+                self._maybe_log_call(msgq_id, cs)
+
+    def process_qmsg(self, msg):
+        m_proto = ctypes.c_int16(msg.type() >> 16).value
+        m_rxid = int(msg.arg1()) >> 1
+        if m_proto != 0 or m_rxid not in self.drb_conv:
+            return super().process_qmsg(msg)
+        m_type = ctypes.c_int16(msg.type() & 0xffff).value
+        cs = self.drb_conv[m_rxid]
+        try:
+            self._process_conv(m_rxid, cs, m_type, msg, time.time())
+        except Exception as e:   # never take multi_rx down over a bad frame
+            sys.stderr.write("%s [%d] conventional message type %d failed: %s\n" % (log_ts.get(), m_rxid, m_type, e))
+        cs['prev_type'] = m_type
+
+    def get_chan_status(self):
+        d = json.loads(super().get_chan_status())
+        ids = list(d.get('channels') or [])
+        for msgq_id, cs in self.drb_conv.items():
+            key = str(msgq_id)
+            row = d.get(key) or {}
+            row.update({
+                'freq': cs['freq'],
+                'name': cs['name'],
+                'system': cs['name'],
+                'tdma': 0,
+                'tgid': cs['tgid'],
+                'tag': cs['name'],
+                'srcaddr': cs['srcaddr'],
+                'srctag': '',
+                'svcopts': 0,
+                'encrypted': 1 if self._encrypted(cs) else 0,
+                'emergency': 0,
+                'hold_tgid': 0,
+                'mode': None,
+                'stream': '',
+                'msgqid': msgq_id,
+                'conventional': True,
+                'voice': cs['voice'],
+                'nac': cs['nac'],
+                'nac_filter': cs['nac_filter'],
+                'algid': cs['algid'],
+                'keyid': cs['keyid'],
+                'enc_calls': cs['enc_calls'],
+                'clear_calls': cs['clear_calls'],
+                'last_enc_algid': cs['last_enc_algid'],
+                'last_enc_keyid': cs['last_enc_keyid'],
+                'last_enc_time': cs['last_enc_time'],
+                'last_clear_time': cs['last_clear_time'],
+                'last_voice_time': cs['last_voice_time'],
+            })
+            d[key] = row
+            if key not in ids:
+                ids.append(key)
+        d['channels'] = ids
+        return json.dumps(d)

op25-container/tests/test_conventional.py

@@ -0,0 +1,305 @@
+"""
+P25 conventional (node-26#24): config generation, the liquidsoap two-channel
+variant, and the tk_drb_conv state machine — all without OP25/GNU Radio.
+
+tk_drb_conv imports tk_p25/helper_funcs/log_ts from OP25's apps dir; those are
+stubbed here with the minimum of tk_p25.rx_ctl it relies on. The message
+shapes fed in are the ones boatbod gr310 p25p1_fdma.cc sends: LCW = 2-byte NAC
++ 9 LC bytes (type 19), LDU2 ESS = JSON (type -3), TDU = 2-byte NAC (type 3/15),
+sync timeout = type -1.
+"""
+import json
+import os
+import sys
+import types
+
+import pytest
+
+HERE = os.path.dirname(__file__)
+sys.path.insert(0, os.path.join(HERE, "..", "app"))
+sys.path.insert(0, os.path.join(HERE, "..", "op25_addons"))
+
+from fastapi.testclient import TestClient  # noqa: E402
+from fastapi import FastAPI  # noqa: E402
+
+from models import ConventionalChannel  # noqa: E402
+from internal import conventional_config as cc  # noqa: E402
+from internal.liquidsoap_config_utils import render_liquid_script  # noqa: E402
+from models import IcecastConfig  # noqa: E402
+
+
+# ---------------------------------------------------------------------------
+# Config generation
+# ---------------------------------------------------------------------------
+
+WP = [
+    ConventionalChannel(frequency=460_400_000, name="PD1 Dispatch"),
+    ConventionalChannel(frequency=460_425_000, name="PD3 Tac", nac=0x293),
+]
+
+
+def test_white_plains_two_channels_one_sdr():
+    config, two = cc.build_conventional_config("White Plains", WP, "lna:34", 0.0)
+    assert two is True
+    assert "trunking" in config and config["trunking"]["module"] == "tk_drb_conv.py"
+    assert config["trunking"]["chans"] == []          # nothing for tk_p25 to hunt
+    a, b = config["channels"]
+    for ch in (a, b):
+        assert ch["demod_type"] == "fsk4"
+        assert ch["trunking_sysname"] == ""
+        assert ch["enable_analog"] == "off"
+        assert ch["crypt_behavior"] == 2
+        assert "control_channel_list" not in ch
+    assert (a["frequency"], b["frequency"]) == (460_400_000, 460_425_000)
+    assert a["destination"].endswith(":23456") and b["destination"].endswith(":23457")
+    assert a["name"] == "PD1 Dispatch" and b["nac"] == 0x293
+    dev = config["devices"][0]
+    assert dev["tunable"] is False
+    # 25 kHz apart: centre is offset off both channels (DC spike), and both
+    # still sit inside the usable passband.
+    half = dev["rate"] * dev["usable_bw_pct"] / 2
+    for f in (460_400_000, 460_425_000):
+        assert 100_000 < abs(f - dev["frequency"]) < half
+
+
+def test_single_channel_is_not_two_channel_audio():
+    config, two = cc.build_conventional_config("X", WP[:1], "lna:34", 0.0)
+    assert two is False
+    assert len(config["channels"]) == 1
+
+
+def test_wide_pair_centres_between_channels():
+    chans = [ConventionalChannel(frequency=453_000_000), ConventionalChannel(frequency=454_000_000)]
+    config, _ = cc.build_conventional_config("X", chans, "lna:34", 0.0)
+    assert config["devices"][0]["frequency"] == 453_500_000
+
+
+@pytest.mark.parametrize("chans, needle", [
+    ([], "at least one"),
+    ([ConventionalChannel(frequency=f) for f in (460_000_000, 460_025_000, 460_050_000)], "up to 2"),
+    ([ConventionalChannel(frequency=460_000_000), ConventionalChannel(frequency=462_000_000)], "too far"),
+    ([ConventionalChannel(frequency=460_000_000), ConventionalChannel(frequency=460_000_000)], "twice"),
+    ([ConventionalChannel(frequency=460_000_000, nac=0x1000)], "NAC"),
+])
+def test_rejects_bad_channel_sets(chans, needle):
+    with pytest.raises(ValueError, match=needle):
+        cc.build_conventional_config("X", chans, "lna:34", 0.0)
+
+
+ICE = IcecastConfig(icecast_host="localhost", icecast_port=8000, icecast_mountpoint="/n", icecast_password="p")
+
+
+def test_liquidsoap_single_channel_unchanged_command():
+    liq = render_liquid_script(ICE, two_channel=False)
+    assert '"./audio.py -x 2.5 -s"' in liq
+    assert "amplify" not in liq and "${" not in liq
+
+
+def test_liquidsoap_two_channel():
+    liq = render_liquid_script(ICE, two_channel=True)
+    assert '"./audio.py -x 2.5 -s -2"' in liq
+    assert "input = amplify(2.0, input)" in liq
+    assert "${" not in liq
+
+
+def test_generate_config_route(tmp_path, monkeypatch):
+    from routers import op25_controller
+    monkeypatch.setattr(op25_controller, "_ACTIVE_CFG", str(tmp_path / "active.cfg.json"))
+    written = {}
+    monkeypatch.setattr(op25_controller, "generate_liquid_script",
+                        lambda ice, two_channel=False: written.setdefault("two", two_channel))
+    monkeypatch.setattr(op25_controller, "save_talkgroup_tags", lambda tags: None)
+    monkeypatch.setattr(op25_controller, "save_whitelist", lambda wl: None)
+    app = FastAPI()
+    app.include_router(op25_controller.create_op25_router(), prefix="/op25")
+    client = TestClient(app)
+    ice = ICE.model_dump()
+    r = client.post("/op25/generate-config", json={
+        "type": "P25", "systemName": "White Plains", "p25_mode": "conventional",
+        "conv_channels": [{"frequency": 460400000, "name": "PD1"}, {"frequency": 460425000, "name": "PD3"}],
+        "icecastConfig": ice,
+    })
+    assert r.status_code == 200, r.text
+    cfg = json.loads((tmp_path / "active.cfg.json").read_text())
+    assert cfg["trunking"]["module"] == "tk_drb_conv.py"
+    assert [c["frequency"] for c in cfg["channels"]] == [460400000, 460425000]
+    assert written["two"] is True
+
+    # Trunked P25 without p25_mode is exactly what it was.
+    r = client.post("/op25/generate-config", json={
+        "type": "P25", "systemName": "County", "channels": [851012500], "icecastConfig": ice,
+    })
+    assert r.status_code == 200, r.text
+    cfg = json.loads((tmp_path / "active.cfg.json").read_text())
+    assert cfg["trunking"]["module"] == "tk_p25.py"
+    assert cfg["trunking"]["chans"][0]["control_channel_list"] == "851012500"
+    assert cfg["channels"][0]["demod_type"] == "cqpsk"
+
+    r = client.post("/op25/generate-config", json={
+        "type": "P25", "systemName": "X", "p25_mode": "conventional", "conv_channels": [],
+    })
+    assert r.status_code == 400
+
+    # Analog uses the same module, so the edge node sees a `conventional` row
+    # whatever tk_p25 the image was built with (node-26#20).
+    r = client.post("/op25/generate-config", json={
+        "type": "NBFM", "systemName": "Fire", "channels": [154430000], "icecastConfig": ice,
+    })
+    assert r.status_code == 200, r.text
+    cfg = json.loads((tmp_path / "active.cfg.json").read_text())
+    assert cfg["trunking"]["module"] == "tk_drb_conv.py"
+    assert cfg["channels"][0]["frequency"] == 154430000
+
+
+# ---------------------------------------------------------------------------
+# tk_drb_conv state machine
+# ---------------------------------------------------------------------------
+
+class _BaseRxCtl:
+    """The slice of tk_p25.rx_ctl that tk_drb_conv relies on."""
+
+    def __init__(self, debug=0, frequency_set=None, nbfm_ctrl=None, fa_ctrl=None, chans={}):
+        self.fa_ctrl = fa_ctrl
+        self.receivers = {}
+        self.call_log = []
+
+    def add_receiver(self, msgq_id, config, meta_q=None, freq=0):
+        self.receivers[msgq_id] = {'msgq_id': msgq_id, 'config': config, 'rx_rcvr': None,
+                                   'conv_state': {'tgid': None}}
+
+    def post_init(self):
+        pass
+
+    def process_qmsg(self, msg):
+        self.base_saw = msg
+
+    def log_call(self, sysid, rcvr, freq, slot, prio, tgid, tgtag, rid, rtag):
+        self.call_log.append({"rcvr": rcvr, "freq": freq, "tgid": tgid, "tgtag": tgtag, "rid": rid})
+
+    def get_chan_status(self):
+        return json.dumps({'json_type': 'channel_update', 'channels': []})
+
+
+def _stub_op25_modules():
+    tk = types.ModuleType("tk_p25")
+    tk.rx_ctl = _BaseRxCtl
+    hf = types.ModuleType("helper_funcs")
+
+    def from_dict(d, key, def_val):
+        return d[key] if key in d and d[key] != "" else def_val
+
+    def get_ordinals(s):
+        t = 0
+        for c in s:
+            t = (t << 8) + c
+        return t
+    hf.from_dict = from_dict
+    hf.get_ordinals = get_ordinals
+    lt = types.ModuleType("log_ts")
+    lt.log_ts = types.SimpleNamespace(get=lambda *a: "ts")
+    sys.modules.update({"tk_p25": tk, "helper_funcs": hf, "log_ts": lt})
+
+
+class _Msg:
+    def __init__(self, m_type, rxid, payload=b""):
+        self._type = m_type & 0xffff   # proto 0 in the upper 16 bits
+        self._arg1 = rxid << 1
+        self._payload = payload
+
+    def type(self):
+        return self._type
+
+    def arg1(self):
+        return self._arg1
+
+    def to_string(self):
+        return self._payload
+
+
+def _lcw(nac, tgid, src, svcopts=0, lco=0x00):
+    lc = bytes([lco, 0x00, svcopts, 0x00]) + tgid.to_bytes(2, "big") + src.to_bytes(3, "big")
+    return nac.to_bytes(2, "big") + lc
+
+
+def _ess(encrypted, algid, keyid):
+    return json.dumps({"encrypted": encrypted, "algid": algid, "keyid": keyid}).encode()
+
+
+@pytest.fixture
+def tk():
+    _stub_op25_modules()
+    sys.modules.pop("tk_drb_conv", None)
+    import tk_drb_conv
+    sent = []
+    rx = tk_drb_conv.rx_ctl(fa_ctrl=sent.append, chans=[])
+    rx.add_receiver(0, {"name": "PD1 Dispatch", "nac": 0x293}, freq=460_400_000)
+    rx.add_receiver(1, {"name": "PD3 Tac"}, freq=460_425_000)
+    rx.post_init()
+    rx.sent = sent
+    return rx
+
+
+def _rows(rx):
+    d = json.loads(rx.get_chan_status())
+    return {cid: d[cid] for cid in d["channels"]}
+
+
+def test_nac_filter_applied_only_where_configured(tk):
+    assert tk.sent == [{'tuner': 0, 'cmd': 'set_nac', 'nac': 0x293}]
+
+
+def test_clear_call_reports_tgid_unit_nac_and_logs_once(tk):
+    tk.process_qmsg(_Msg(0, 0, b"\x02\x93"))                       # HDU
+    tk.process_qmsg(_Msg(19, 0, _lcw(0x293, 101, 4242)))            # LDU1
+    tk.process_qmsg(_Msg(-3, 0, _ess(0, 0x80, 0)))                  # LDU2 clear
+    tk.process_qmsg(_Msg(19, 0, _lcw(0x293, 101, 4242)))
+    row = _rows(tk)["0"]
+    assert row["conventional"] and row["voice"]
+    assert (row["tgid"], row["srcaddr"], row["nac"], row["encrypted"]) == (101, 4242, 0x293, 0)
+    assert row["algid"] is None
+    assert [e["tgid"] for e in tk.call_log] == [101]
+    assert tk.call_log[0]["tgtag"] == "PD1 Dispatch"
+    # The other channel is idle.
+    assert _rows(tk)["1"]["voice"] is False
+
+    tk.process_qmsg(_Msg(3, 0, b"\x02\x93"))                         # TDU
+    row = _rows(tk)["0"]
+    assert row["voice"] is False and row["tgid"] is None
+    assert (row["clear_calls"], row["enc_calls"]) == (1, 0)
+
+
+def test_tdulc_link_control_does_not_reopen_the_call(tk):
+    tk.process_qmsg(_Msg(19, 1, _lcw(0x111, 7, 9)))
+    tk.process_qmsg(_Msg(15, 1, b"\x01\x11"))                        # TDULC
+    tk.process_qmsg(_Msg(19, 1, _lcw(0x111, 7, 9)))                 # its LC
+    assert _rows(tk)["1"]["voice"] is False
+
+
+def test_encrypted_call_is_reported_and_never_logged(tk):
+    tk.process_qmsg(_Msg(19, 1, _lcw(0x111, 7, 9, svcopts=0x40)))  # E bit
+    tk.process_qmsg(_Msg(-3, 1, _ess(1, 0x84, 0x1234)))
+    row = _rows(tk)["1"]
+    assert row["encrypted"] == 1 and row["algid"] == 0x84 and row["keyid"] == 0x1234
+    assert tk.call_log == []
+    tk.process_qmsg(_Msg(-1, 1))                                    # sync timeout
+    row = _rows(tk)["1"]
+    assert row["encrypted"] == 0 and row["voice"] is False
+    assert (row["enc_calls"], row["last_enc_algid"], row["last_enc_keyid"]) == (1, 0x84, 0x1234)
+
+
+def test_encryption_seen_only_in_ess_still_counts(tk):
+    tk.process_qmsg(_Msg(-3, 0, _ess(1, 0x81, 0x0001)))
+    assert _rows(tk)["0"]["encrypted"] == 1
+    tk.process_qmsg(_Msg(3, 0, b"\x02\x93"))
+    assert _rows(tk)["0"]["enc_calls"] == 1
+
+
+def test_protected_link_control_marks_encrypted(tk):
+    tk.process_qmsg(_Msg(19, 0, _lcw(0x293, 0, 0, lco=0x80)))
+    assert _rows(tk)["0"]["encrypted"] == 1
+
+
+def test_trunked_receivers_and_other_protocols_go_to_base(tk):
+    msg = _Msg(7, 5)
+    tk.process_qmsg(msg)
+    assert tk.base_saw is msg