P25 conventional decode mode + conventional-channel attribution (node-26#24, #20)
CI / lint (push) Successful in 6s
CI / test-op25 (push) Successful in 13s
CI / lint (pull_request) Successful in 7s
CI / test-op25 (pull_request) Successful in 13s
CI / test (push) Successful in 44s
CI / test (pull_request) Successful in 42s

op25-container
- generate-config takes p25_mode "conventional" + conv_channels [{frequency,
  name, nac}]. It builds fixed fsk4 channels (up to 2 on one non-tunable SDR,
  DC spike kept off the channels) with no control channel to hunt. Two
  channels send audio to UDP 23456/23457, and audio.py -2 + amplify(2) keeps
  levels.
- tk_drb_conv.py (a tk_p25.rx_ctl subclass, copied into OP25 apps) reports
  voice / NAC / encryption (LC service options + LDU2 ESS algid/keyid) for
  conventional channels and applies the per-channel NAC filter, which stock
  tk_p25 does not. Analog uses it too.
- Trunked P25 config is unchanged when p25_mode is absent.

drb-edge-node
- radio_config: system kind from type + config.mode (missing = trunked).
- conventional.py normalizes the console: a distinctive LC TGID, else the
  channel's last TGID (30 s), else a synthetic per-channel id (freq/100 Hz).
  Encrypted rows are never talkers. With exactly one conventional channel,
  orphan audio is attributed to it at close, which fixes analog recording
  (#20). The synthetic id and the real TGID on one channel never split.
- Health/dashboard: P25 conventional listen tab (/api/listen p25conv), and
  conventional decode + encryption checks. The checkin carries radio_channels.
- Tests: edge 296 pass; op25-container 18 (new CI job). Reviewed by
  drb-audio-review (findings fixed; simultaneous two-channel mixdown filed
  as #26) and drb-qa-review.

Not verified on SDR hardware.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Logan Cusano
2026-10-04 15:57:15 -04:00
co-authored by Claude Opus 5.5
parent 1e1dab66b1
commit d07d312e92
19 changed files with 1865 additions and 79 deletions
+129
View File
@@ -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