server-26/drb-c2-core/app/internal/system_config.py

"""
Radio-system shape checks for the `systems` collection (node-26#24).

A system's `config` blob is pushed to edge nodes verbatim, and what a node does
with it depends on the system's kind:

    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?}, ...]
    Analog            type "NBFM"; config.channels = [{freq, name}] (one), or
                      the pre-#24 config.control_channels[0]

`mode` absent means trunked, so every system saved before this existed is
valid and unchanged — there is no migration. What this rejects is the mistake
that left the White Plains node silent: voice frequencies entered as a trunked
system's control channels can't be caught (they are just numbers), but a
conventional system can no longer carry control channels, and a conventional
channel list must be decodable by one node.

Synthetic talkgroup ids: the edge node labels a conventional call with the
channel's frequency in 100 Hz units (460.400 MHz -> 4604000) when its link
control carries no distinctive talkgroup (edge drb-edge-node
internal/conventional.py). channel_tgid() is the same function, so C2 can name
those calls after the channel.
"""
from typing import Any, Dict, List, Optional

TRUNKED, CONVENTIONAL = "trunked", "conventional"
P25_MODES = (TRUNKED, CONVENTIONAL)

# Mirrors the node: one audio pipeline carries two channels, one RTL-SDR's
# usable passband is ~1.6 MHz.
MAX_CONVENTIONAL_CHANNELS = 2
MAX_CONVENTIONAL_SPAN_MHZ = 1.5
MIN_MHZ, MAX_MHZ = 24.0, 1766.0


class SystemConfigError(ValueError):
    pass


def channel_tgid(freq_mhz: Any) -> Optional[int]:
    try:
        f = float(freq_mhz)
    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 p25_mode(config: Dict[str, Any]) -> str:
    return str((config or {}).get("mode") or TRUNKED).lower()


def _nac(value: Any, where: str) -> Optional[str]:
    if value is None or (isinstance(value, str) and not value.strip()):
        return None
    text = str(value).strip().lower()
    if isinstance(value, int) and not isinstance(value, bool):
        n = value
    else:
        text = text[2:] if text.startswith("0x") else text
        try:
            n = int(text, 16)
        except ValueError:
            raise SystemConfigError(f"{where}: NAC {value!r} isn't hex (e.g. 293).")
    if not 0 < n <= 0xFFF:
        raise SystemConfigError(f"{where}: NAC {value!r} is outside 001-FFF.")
    return f"{n:03X}"


def _channels(config: Dict[str, Any], *, allow_nac: bool) -> List[Dict[str, Any]]:
    raw = config.get("channels")
    if not isinstance(raw, list) or not raw:
        raise SystemConfigError("Add at least one channel frequency.")
    out = []
    for i, ch in enumerate(raw, 1):
        if not isinstance(ch, dict):
            raise SystemConfigError(f"Channel {i} must be an object with a freq.")
        try:
            freq = float(ch.get("freq"))
        except (TypeError, ValueError):
            raise SystemConfigError(f"Channel {i} needs a frequency in MHz.")
        if not MIN_MHZ <= freq <= MAX_MHZ:
            raise SystemConfigError(
                f"Channel {i}: {freq:g} isn't a frequency in MHz an RTL-SDR can tune ({MIN_MHZ:g}-{MAX_MHZ:g})."
            )
        clean = {**ch, "freq": freq, "name": str(ch.get("name") or "").strip()}
        if allow_nac:
            nac = _nac(ch.get("nac"), f"Channel {i}")
            if nac is None:
                clean.pop("nac", None)
            else:
                clean["nac"] = nac
        out.append(clean)
    return out


def normalize(system_type: str, config: Dict[str, Any]) -> Dict[str, Any]:
    """Validate and return the config to store. Raises SystemConfigError."""
    config = dict(config or {})
    t = str(system_type or "").upper()

    if t == "P25":
        mode = p25_mode(config)
        if mode not in P25_MODES:
            raise SystemConfigError(f"P25 mode must be 'trunked' or 'conventional', not {config.get('mode')!r}.")
        if mode == TRUNKED:
            return config   # untouched: exactly what was accepted before #24
        if config.get("control_channels"):
            raise SystemConfigError(
                "A conventional system has no control channel — list its voice frequencies as channels instead."
            )
        config.pop("control_channels", None)
        config.pop("voice_channels", None)
        config["mode"] = CONVENTIONAL
        chans = _channels(config, allow_nac=True)
        if len(chans) > MAX_CONVENTIONAL_CHANNELS:
            raise SystemConfigError(
                f"A node decodes at most {MAX_CONVENTIONAL_CHANNELS} conventional channels on one SDR."
            )
        freqs = [c["freq"] for c in chans]
        if len(set(freqs)) != len(freqs):
            raise SystemConfigError("The same channel frequency is listed twice.")
        if max(freqs) - min(freqs) > MAX_CONVENTIONAL_SPAN_MHZ:
            raise SystemConfigError(
                f"These channels are {max(freqs) - min(freqs):.3f} MHz apart; one SDR covers about "
                f"{MAX_CONVENTIONAL_SPAN_MHZ:g} MHz."
            )
        config["channels"] = chans
        return config

    if t == "NBFM" and ("channels" in config or not config.get("control_channels")):
        chans = _channels(config, allow_nac=False)
        if len(chans) != 1:
            raise SystemConfigError("An analog system listens to exactly one frequency.")
        config["channels"] = chans
        # Nodes from before node-26#24 read the analog frequency from here.
        config["control_channels"] = [chans[0]["freq"]]
        return config

    return config


def conventional_channel_name(system_doc: Dict[str, Any], talkgroup_id: Any) -> Optional[str]:
    """Name of the conventional/analog channel whose synthetic id this is."""
    try:
        tgid = int(talkgroup_id)
    except (TypeError, ValueError):
        return None
    for ch in (system_doc.get("config") or {}).get("channels") or []:
        if isinstance(ch, dict) and channel_tgid(ch.get("freq")) == tgid:
            return ch.get("name") or None
    return None