diff --git a/drb-edge-node/app/internal/call_recorder.py b/drb-edge-node/app/internal/call_recorder.py index ff80551..b2826fd 100644 --- a/drb-edge-node/app/internal/call_recorder.py +++ b/drb-edge-node/app/internal/call_recorder.py @@ -785,6 +785,7 @@ class CallRecorder: system_id: Optional[str] = None, audio_start_epoch: Optional[float] = None, audio_end_epoch: Optional[float] = None, + radios: Optional[str] = None, ) -> Optional[str]: """ Upload once, inline. Returns the storage URL on success. @@ -802,6 +803,7 @@ class CallRecorder: meta = UploadMeta( call_id=call_id, talkgroup_id=talkgroup_id, talkgroup_name=talkgroup_name, system_id=system_id, audio_start_epoch=audio_start_epoch, audio_end_epoch=audio_end_epoch, + radios=radios, ) result = await self._post_inline(file_path, meta) self.spool.last_result = {"call_id": call_id, "kind": result.kind, "status": result.status, diff --git a/drb-edge-node/app/internal/metadata_watcher.py b/drb-edge-node/app/internal/metadata_watcher.py index 4a8e86d..700628e 100644 --- a/drb-edge-node/app/internal/metadata_watcher.py +++ b/drb-edge-node/app/internal/metadata_watcher.py @@ -132,6 +132,11 @@ ATTRIBUTION_LOOKAHEAD_SECONDS = 2.0 CONSOLE_HISTORY_SECONDS = 180.0 CONSOLE_HISTORY_MAX = 1200 +# Sightings of the same radio closer together than this are one transmission. +# The radio ID shows up in only ~44% of 0.5 s polls during a burst (server-26#304 +# P0 probe, 2026-11-08), so gaps of a few polls inside one key-up are normal. +TX_MERGE_GAP_SECONDS = 2.0 + # How far back to look for a duplicate before appending a grant. OP25's call_log # deque drains on read so repeats should not happen, but a re-delivered entry # would otherwise inflate the transmission count and the attribution score. @@ -207,6 +212,21 @@ def _talker(channel: Dict[str, Any]) -> Optional[tuple]: return tgid, rid +def rebase_radios(radios: List[Dict[str, Any]], shift: float, span: float) -> List[Dict[str, Any]]: + """Move call-relative radio offsets into another frame (the trimmed audio file). + + `shift` is call start minus the new origin; offsets land clamped to [0, span] + so a talker before the file's first sample sits at 0, not before it. + """ + span = max(0.0, span) + return [ + {**r, + "start": round(min(max(0.0, r["start"] + shift), span), 2), + "end": round(min(max(0.0, r["end"] + shift), span), 2)} + for r in radios + ] + + def _iso(epoch: Optional[float]) -> Optional[str]: if epoch is None: return None @@ -254,6 +274,10 @@ class MetadataWatcher: self._current_srcaddr: Optional[int] = None self._started_at: Optional[float] = None # audio onset, or grant epoch in fallback mode self._transmissions: int = 0 + # Every radio heard on the open segment, in order: [rid, first_epoch, + # last_epoch]. srcaddr above keeps only the latest one; one recording + # often holds a dispatcher and a unit (server-26#304). + self._tx: List[List[Any]] = [] # True when the open segment is governed by audio, False for the # console fallback. Fixed at open so capture flapping cannot switch the # rules underneath a live segment. @@ -559,6 +583,7 @@ class MetadataWatcher: self._current_tgid_name = entry.get("tgtag") or "" self._current_freq = entry.get("freq") self._current_srcaddr = _as_int(entry.get("rid")) + self._note_tx(self._current_srcaddr, started_at) self._transmissions += 1 logger.info( f"Late attribution: segment {self._active_call_id} adopted tgid {tgid} from a grant " @@ -571,11 +596,13 @@ class MetadataWatcher: # of a single conversation lands in one file. self._transmissions += 1 self._refresh_meta_from_log(entry) + self._note_tx(_as_int(entry.get("rid")), started_at) return if self._same_channel(tgid, entry.get("freq"), entry.get("tgtag") or ""): self._transmissions += 1 self._refresh_meta_from_log(entry) + self._note_tx(_as_int(entry.get("rid")), started_at) return # FORCED SPLIT. Two talkgroups can be back to back with no silence @@ -626,6 +653,7 @@ class MetadataWatcher: ours = True if srcaddr is not None: self._current_srcaddr = srcaddr + self._note_tx(srcaddr, now) self._last_activity = now self._refresh_meta_from_channel(channel) @@ -638,6 +666,7 @@ class MetadataWatcher: self._current_tgid_name = active[0].get("tag") or "" self._current_freq = active[0].get("freq") self._current_srcaddr = _as_int(active[0].get("srcaddr")) + self._note_tx(self._current_srcaddr, now) logger.info(f"Late attribution: segment {self._active_call_id} adopted tgid {tgid} from channel state.") return @@ -709,6 +738,7 @@ class MetadataWatcher: self._last_tx_end = None self._last_activity = now self._refresh_meta_from_log(entry) + self._note_tx(_as_int(entry.get("rid")), started_at) return await self._close_segment(started_at + _tail_pad(), reason="tgid_change") @@ -730,6 +760,7 @@ class MetadataWatcher: tx_active = True if srcaddr is not None: self._current_srcaddr = srcaddr + self._note_tx(srcaddr, now) self._refresh_meta_from_channel(channel) elif chan_tgid is not None: foreign_active_tgid = chan_tgid @@ -792,6 +823,28 @@ class MetadataWatcher: # Segment open / close # ------------------------------------------------------------------ + def _note_tx(self, rid: Optional[int], epoch: float) -> None: + """Record a sighting of `rid` talking on the open segment.""" + if rid is None or self._active_call_id is None: + return + last = self._tx[-1] if self._tx else None + if last is not None and last[0] == rid and epoch - last[2] <= TX_MERGE_GAP_SECONDS: + last[2] = max(last[2], epoch) + return + self._tx.append([rid, epoch, epoch]) + + def _radios(self, start: float, end: float) -> List[Dict[str, Any]]: + """The segment's transmissions as offsets (seconds) from its start, clamped to it.""" + span = max(0.0, end - start) + return [ + { + "radio": rid, + "start": round(min(max(0.0, first - start), span), 2), + "end": round(min(max(0.0, last - start), span), 2), + } + for rid, first, last in self._tx + ] + def _refresh_meta_from_log(self, entry: Dict[str, Any]) -> None: if not self._current_tgid_name: self._current_tgid_name = entry.get("tgtag") or "" @@ -826,6 +879,8 @@ class MetadataWatcher: self._started_at = started_at self._transmissions = max(1, transmissions) self._audio_driven = audio_driven + self._tx = [] + self._note_tx(srcaddr, started_at) # Console fallback assumes the transmission is still up; it learns # otherwise from the next channel scan. @@ -876,6 +931,7 @@ class MetadataWatcher: "ended_at": _iso(end_epoch), "ended_at_epoch": end_epoch, "transmissions": self._transmissions, + "radios": self._radios(started_at if started_at is not None else end_epoch, end_epoch), "end_reason": reason, "attributed": attributed, "driver": "audio" if self._audio_driven else "console", @@ -910,6 +966,7 @@ class MetadataWatcher: self._current_srcaddr = None self._started_at = None self._transmissions = 0 + self._tx = [] self._tx_active = False self._last_tx_end = None self._audio_driven = False @@ -949,6 +1006,10 @@ class MetadataWatcher: self._current_freq = found.freq if found.rid is not None and self._current_srcaddr is None: self._current_srcaddr = found.rid + if not self._tx: + # Only as the sole entry: the window's latest radio has no + # timing of its own, so it must not reorder real sightings. + self._note_tx(found.rid, start) self._transmissions = max(self._transmissions, found.grants) return diff --git a/drb-edge-node/app/internal/upload_spool.py b/drb-edge-node/app/internal/upload_spool.py index 351f68d..a2999a4 100644 --- a/drb-edge-node/app/internal/upload_spool.py +++ b/drb-edge-node/app/internal/upload_spool.py @@ -71,6 +71,10 @@ class UploadMeta: system_id: Optional[str] = None audio_start_epoch: Optional[float] = None audio_end_epoch: Optional[float] = None + # JSON list of {radio, start, end} — every radio heard on the call + # (server-26#304). Sent here as well as on call_end because C2 starts the + # pipeline from /upload, before call_end arrives. + radios: Optional[str] = None @dataclass @@ -108,6 +112,8 @@ def _form(meta: UploadMeta) -> Dict[str, str]: form["audio_start_epoch"] = f"{meta.audio_start_epoch:.3f}" if meta.audio_end_epoch is not None: form["audio_end_epoch"] = f"{meta.audio_end_epoch:.3f}" + if meta.radios: + form["radios"] = meta.radios return form diff --git a/drb-edge-node/app/main.py b/drb-edge-node/app/main.py index beca2b7..2f9acf9 100644 --- a/drb-edge-node/app/main.py +++ b/drb-edge-node/app/main.py @@ -1,4 +1,5 @@ import asyncio +import json from contextlib import asynccontextmanager from datetime import datetime, timezone from typing import Optional @@ -9,7 +10,7 @@ from app.models import SystemConfig, normalize_sdr_pins, normalize_secondary_pri from app.internal.logger import logger from app.internal.mqtt_manager import mqtt_manager from app.internal import credentials -from app.internal.metadata_watcher import metadata_watcher +from app.internal.metadata_watcher import metadata_watcher, rebase_radios from app.internal.call_recorder import call_recorder from app.internal.discord_radio import radio_bot from app.internal.config_manager import ( @@ -93,6 +94,15 @@ async def on_call_end(data: dict): data["audio_end_at"] = _iso(recording.audio_end_epoch) data["audio_start_epoch"] = recording.audio_start_epoch data["audio_end_epoch"] = recording.audio_end_epoch + # Radio offsets were measured from the call start; C2 maps an offset + # to wall clock as audio_start_epoch + t (above), so put them in the + # audio file's frame (server-26#304). + if data.get("radios") and data.get("started_at_epoch") is not None: + data["radios"] = rebase_radios( + data["radios"], + shift=data["started_at_epoch"] - recording.audio_start_epoch, + span=recording.audio_end_epoch - recording.audio_start_epoch, + ) data["audio_lead_trimmed"] = round(recording.lead_trimmed, 3) data["audio_tail_trimmed"] = round(recording.tail_trimmed, 3) if recording.clamped_seconds: @@ -108,6 +118,7 @@ async def on_call_end(data: dict): system_id=node_cfg.assigned_system_id, audio_start_epoch=recording.audio_start_epoch, audio_end_epoch=recording.audio_end_epoch, + radios=json.dumps(data["radios"]) if data.get("radios") else None, ) if audio_url: data["audio_url"] = audio_url diff --git a/drb-edge-node/tests/test_metadata_watcher.py b/drb-edge-node/tests/test_metadata_watcher.py index e240fce..18be1e8 100644 --- a/drb-edge-node/tests/test_metadata_watcher.py +++ b/drb-edge-node/tests/test_metadata_watcher.py @@ -27,6 +27,7 @@ from app.internal.metadata_watcher import ( MAX_SEGMENT_SECONDS, MetadataWatcher, OP25_OFFLINE_GRACE, + rebase_radios, ) from app.internal.op25_client import TerminalUpdate, parse_terminal_messages @@ -1147,3 +1148,86 @@ async def test_console_history_is_bounded(hearing, clock, audio): clock.advance(CONSOLE_HISTORY_SECONDS + 1) await tick(hearing, update()) assert all(e.epoch >= clock.now - CONSOLE_HISTORY_SECONDS for e in hearing._console) + + +# --------------------------------------------------------------------------- +# Every radio on the call (server-26#304) — not just the last srcaddr +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_call_end_lists_every_radio_in_order(watcher, clock): + """Dispatcher, unit, dispatcher on one talkgroup: three transmissions.""" + start = clock.now + await tick(watcher, update(call_log=[grant(1234, start, rid=555)], + channels=[channel(tgid=1234, srcaddr=555)])) + clock.advance(1.0) + await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=555)])) + clock.advance(1.0) + await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=777)])) + clock.advance(0.5) + await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=777)])) + clock.advance(1.0) + await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=555)])) + clock.advance(0.5) + await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)])) + clock.advance(settings.call_idle_timeout + 0.5) + await tick(watcher, update(channels=[channel()])) + + payload = watcher.on_call_end.call_args[0][0] + assert [r["radio"] for r in payload["radios"]] == [555, 777, 555] + assert payload["radios"][0] == {"radio": 555, "start": 0.0, "end": 1.0} + assert payload["radios"][1] == {"radio": 777, "start": 2.0, "end": 2.5} + # srcaddr keeps its old meaning (the latest talker) for existing readers. + assert payload["srcaddr"] == 555 + + +@pytest.mark.asyncio +async def test_missed_polls_inside_one_keyup_stay_one_transmission(watcher, clock): + """The radio ID drops out of some polls mid-transmission; that is not a new key-up.""" + await tick(watcher, update(call_log=[grant(1234, clock.now, rid=555)], + channels=[channel(tgid=1234, srcaddr=555)])) + clock.advance(0.5) + await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)])) + clock.advance(1.0) + await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=555)])) + clock.advance(0.5) + await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)])) + clock.advance(settings.call_idle_timeout + 0.5) + await tick(watcher, update(channels=[channel()])) + + radios = watcher.on_call_end.call_args[0][0]["radios"] + assert radios == [{"radio": 555, "start": 0.0, "end": 1.5}] + + +@pytest.mark.asyncio +async def test_radio_list_resets_between_calls(watcher, clock): + await tick(watcher, update(call_log=[grant(1234, clock.now, rid=555)])) + clock.advance(0.2) + await tick(watcher, update(call_log=[grant(4321, clock.now, rid=888)])) + + first = watcher.on_call_end.call_args[0][0] + assert [r["radio"] for r in first["radios"]] == [555] + assert watcher._tx == [[888, clock.now, clock.now]] + + +@pytest.mark.asyncio +async def test_audio_mode_records_a_radio_seen_only_in_a_grant(hearing, clock, audio): + """A short key-up between channel polls is only visible as a call_log grant.""" + audio.speak() + await tick(hearing, update(call_log=[grant(1234, clock.now, rid=555)], + channels=[channel(tgid=1234, srcaddr=555)])) + clock.advance(1.0) + audio.speak() + await tick(hearing, update(call_log=[grant(1234, clock.now, rid=777)])) + clock.advance(settings.call_silence_timeout + 0.5) + await tick(hearing, update(channels=[channel()])) + + radios = hearing.on_call_end.call_args[0][0]["radios"] + assert [r["radio"] for r in radios] == [555, 777] + + +def test_rebase_radios_moves_offsets_into_the_audio_frame(): + """The file starts 0.25 s of pre-roll before the call, then 1.0 s of lead was trimmed.""" + radios = [{"radio": 555, "start": 0.0, "end": 2.0}, {"radio": 777, "start": 3.0, "end": 9.0}] + out = rebase_radios(radios, shift=-0.75, span=6.0) + assert out == [{"radio": 555, "start": 0.0, "end": 1.25}, {"radio": 777, "start": 2.25, "end": 6.0}] diff --git a/drb-edge-node/tests/test_upload_spool.py b/drb-edge-node/tests/test_upload_spool.py index da518f1..ed6a72c 100644 --- a/drb-edge-node/tests/test_upload_spool.py +++ b/drb-edge-node/tests/test_upload_spool.py @@ -228,3 +228,10 @@ async def test_wake_makes_everything_due_now(tmp_path): sp.add(_recording(tmp_path), UploadMeta(call_id="c1"), UploadResult(RETRY, status=401)) sp.wake() assert await sp.drain_once() == 1 + + +def test_form_carries_the_radio_list_and_omits_it_when_empty(): + """C2 starts correlation from /upload, so the radio list rides on it (server-26#304).""" + radios = json.dumps([{"radio": 555, "start": 0.0, "end": 1.0}]) + assert upload_spool._form(UploadMeta(call_id="c1", radios=radios))["radios"] == radios + assert "radios" not in upload_spool._form(UploadMeta(call_id="c1"))