From 5cdde85d55a748bde22804219e188011993b19e4 Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Sun, 8 Nov 2026 23:00:40 -0500 Subject: [PATCH] Calls: store every radio heard on a call; call_end writes srcaddr (server-26#304) P1 of the radio-ID plan. The node now reports `radios`, one {radio, start, end} per transmission, on /upload and on call_end. Both go through app/internal/radios.py into `radios` + `radio_ids` on the call doc. /upload writes them before the pipeline starts, because call_end arrives after it. call_end also writes srcaddr now. Only call_start used to, and the node usually learns the talker after call_start, so most calls kept None and the correlator's srcaddr match rarely fired. /upload fills srcaddr from the latest radio when the doc has none. Older nodes send neither field; nothing is blanked. Co-Authored-By: Claude Opus 5.5 --- drb-c2-core/app/internal/mqtt_handler.py | 7 ++ drb-c2-core/app/internal/radios.py | 64 +++++++++++++++ drb-c2-core/app/routers/upload.py | 10 +++ drb-c2-core/tests/test_radios.py | 100 +++++++++++++++++++++++ 4 files changed, 181 insertions(+) create mode 100644 drb-c2-core/app/internal/radios.py create mode 100644 drb-c2-core/tests/test_radios.py diff --git a/drb-c2-core/app/internal/mqtt_handler.py b/drb-c2-core/app/internal/mqtt_handler.py index 052e0d9..a85bcd9 100644 --- a/drb-c2-core/app/internal/mqtt_handler.py +++ b/drb-c2-core/app/internal/mqtt_handler.py @@ -6,6 +6,7 @@ import paho.mqtt.client as mqtt from app.config import settings from app.internal.logger import logger from app.internal import firestore as fstore +from app.internal import radios from app.internal import talkgroups from app.internal.tenancy import FOUNDING_ORG_ID @@ -307,6 +308,12 @@ class MQTTHandler: updates["org_id"] = node["org_id"] if payload.get("audio_url"): updates["audio_url"] = payload["audio_url"] + # The node usually learns the talker after call_start went out; only + # call_start used to write srcaddr, so most calls kept None + # (server-26#304). + if payload.get("srcaddr") is not None: + updates["srcaddr"] = payload["srcaddr"] + updates.update(radios.call_fields(payload.get("radios"))) await fstore.doc_set("calls", call_id, updates) logger.info(f"Call end: {call_id}") diff --git a/drb-c2-core/app/internal/radios.py b/drb-c2-core/app/internal/radios.py new file mode 100644 index 0000000..f0233bb --- /dev/null +++ b/drb-c2-core/app/internal/radios.py @@ -0,0 +1,64 @@ +""" +Every radio heard on a call, as the edge node reports it (server-26#304). + +The node sends `radios`: a list of {radio, start, end}, one entry per +transmission, offsets in seconds into the uploaded audio file, so a talker's +wall-clock time is audio_start_epoch + offset (the same mapping the node uses +for the audio itself). It arrives twice — on +/upload (as a JSON string form field, because the pipeline starts there) and +on the MQTT call_end — and both go through `call_fields` so the call doc gets +the same shape either way. + +`radio_ids` is the distinct IDs in first-heard order, kept alongside so a +correlator or query can ask "was radio X on this call" without walking the list. +""" +import json +from typing import Any, Dict, List, Optional + +from app.internal.logger import logger + +# A 600 s recording polled every 0.5 s cannot honestly hold more than this. +MAX_TRANSMISSIONS = 200 + + +def _num(value: Any) -> Optional[float]: + try: + number = float(value) + except (TypeError, ValueError): + return None + return number if number >= 0 else None + + +def parse(value: Any) -> List[Dict[str, Any]]: + """The node's radio list, validated. Accepts a list or its JSON string; junk entries are dropped.""" + if value is None or value == "": + return [] + if isinstance(value, str): + try: + value = json.loads(value) + except ValueError: + logger.warning(f"radios: not JSON, ignored ({value[:80]!r})") + return [] + if not isinstance(value, list): + return [] + out: List[Dict[str, Any]] = [] + for item in value[:MAX_TRANSMISSIONS]: + if not isinstance(item, dict): + continue + try: + radio = int(item.get("radio")) + except (TypeError, ValueError): + continue + start, end = _num(item.get("start")), _num(item.get("end")) + if radio <= 0 or start is None or end is None: + continue + out.append({"radio": radio, "start": start, "end": max(start, end)}) + return out + + +def call_fields(value: Any) -> Dict[str, Any]: + """Call-doc fields for a radio list; empty when the node sent none (older nodes, analog).""" + radios = parse(value) + if not radios: + return {} + return {"radios": radios, "radio_ids": list(dict.fromkeys(r["radio"] for r in radios))} diff --git a/drb-c2-core/app/routers/upload.py b/drb-c2-core/app/routers/upload.py index 63cfeac..101f858 100644 --- a/drb-c2-core/app/routers/upload.py +++ b/drb-c2-core/app/routers/upload.py @@ -5,6 +5,7 @@ from fastapi import APIRouter, BackgroundTasks, UploadFile, File, Form, HTTPExce from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from app.internal.storage import upload_audio from app.internal import dedup +from app.internal import radios from app.internal import firestore as fstore from app.internal import clock from app.internal.logger import logger @@ -24,6 +25,7 @@ async def upload_call_audio( talkgroup_id: Optional[int] = Form(None), talkgroup_name: Optional[str] = Form(None), system_id: Optional[str] = Form(None), + radios_json: Optional[str] = Form(None, alias="radios"), credentials: Optional[HTTPAuthorizationCredentials] = Security(_bearer), ): """ @@ -101,6 +103,14 @@ async def upload_call_audio( updates = {"audio_gcs_uri": gcs_uri} if node and node.get("org_id"): updates["org_id"] = node["org_id"] + # Every radio heard on the call (server-26#304). Written here, not + # only on call_end, because the pipeline below starts before the + # node publishes call_end. + radio_fields = radios.call_fields(radios_json) + updates.update(radio_fields) + if radio_fields and existing.get("srcaddr") is None: + # Same meaning call_end gives it: the latest talker. + updates["srcaddr"] = radio_fields["radios"][-1]["radio"] await fstore.doc_set("calls", call_id, updates) except Exception as e: logger.warning(f"Could not update call {call_id} with audio_gcs_uri: {e}") diff --git a/drb-c2-core/tests/test_radios.py b/drb-c2-core/tests/test_radios.py new file mode 100644 index 0000000..7cf53bf --- /dev/null +++ b/drb-c2-core/tests/test_radios.py @@ -0,0 +1,100 @@ +""" +Every radio on a call reaches the call doc (server-26#304) — from /upload, +which starts the pipeline, and from call_end, which used to drop srcaddr. +""" +import json +from unittest.mock import AsyncMock, patch + +import pytest +from fastapi.testclient import TestClient + +from app.internal import radios +from app.internal.mqtt_handler import MQTTHandler +from app.main import app +from app.routers import upload + +LIST = [{"radio": 555, "start": 0.0, "end": 1.0}, {"radio": 777, "start": 2.0, "end": 2.5}, + {"radio": 555, "start": 3.5, "end": 4.0}] + + +def test_parse_accepts_json_and_drops_junk(): + raw = json.dumps(LIST + [{"radio": 0, "start": 1, "end": 2}, {"radio": "x"}, "junk", + {"radio": 9, "start": -1, "end": 2}, {"radio": 8, "start": 5, "end": 4}]) + assert radios.parse(raw) == LIST + [{"radio": 8, "start": 5.0, "end": 5.0}] + assert radios.parse("not json") == [] + assert radios.parse({"radio": 1}) == [] + assert radios.parse(None) == [] + + +def test_call_fields_keeps_distinct_ids_in_first_heard_order(): + assert radios.call_fields(LIST)["radio_ids"] == [555, 777] + assert radios.call_fields([]) == {} + + +@pytest.mark.asyncio +async def test_call_end_writes_srcaddr_and_radios(): + """call_start used to be the only writer of srcaddr; the talker is usually learned later.""" + handler = MQTTHandler() + with patch("app.internal.mqtt_handler.fstore") as fs: + fs.doc_get_cached = AsyncMock(return_value=None) + fs.doc_set = AsyncMock() + await handler._on_call_end("node-01", {"call_id": "c1", "srcaddr": 555, "radios": LIST}) + updates = fs.doc_set.call_args[0][2] + assert updates["srcaddr"] == 555 + assert updates["radios"] == LIST + assert updates["radio_ids"] == [555, 777] + + +@pytest.mark.asyncio +async def test_call_end_without_radios_leaves_them_alone(): + """An older node sends neither field; a merge write must not blank what /upload stored.""" + handler = MQTTHandler() + with patch("app.internal.mqtt_handler.fstore") as fs: + fs.doc_get_cached = AsyncMock(return_value=None) + fs.doc_set = AsyncMock() + await handler._on_call_end("node-01", {"call_id": "c1", "srcaddr": None}) + updates = fs.doc_set.call_args[0][2] + assert "srcaddr" not in updates and "radios" not in updates + + +def _upload(existing_call, form_radios): + store = { + "node_keys": {"api_key": "k"}, + "nodes": {"approval_status": "approved", "org_id": "org"}, + "calls": existing_call, + } + writes = [] + + async def doc_get(c, i): + return store.get(c) + + async def doc_set(c, i, data, merge=True): + writes.append((c, data)) + + data = {"call_id": "c1", "node_id": "n1"} + if form_radios is not None: + data["radios"] = form_radios + with patch.object(upload.fstore, "doc_get", doc_get), \ + patch.object(upload.fstore, "doc_set", doc_set), \ + patch.object(upload, "upload_audio", AsyncMock(return_value="gs://b/c1.mp3")), \ + patch.object(upload.dedup, "find_duplicate_of", AsyncMock(return_value="other")): + resp = TestClient(app).post( + "/upload", data=data, + files={"file": ("a.mp3", b"\x00\x01", "audio/mpeg")}, + headers={"Authorization": "Bearer k"}, + ) + assert resp.status_code == 200 + return writes[0][1] + + +def test_upload_stores_the_radio_list_before_the_pipeline_runs(): + first = _upload({}, json.dumps(LIST)) + assert first["radios"] == LIST + assert first["radio_ids"] == [555, 777] + # call_start had no talker yet: take the latest one, as call_end would. + assert first["srcaddr"] == 555 + + +def test_upload_keeps_an_existing_srcaddr_and_tolerates_no_list(): + assert "srcaddr" not in _upload({"srcaddr": 999}, json.dumps(LIST)) + assert "radios" not in _upload({}, None)