diff --git a/monitor/src/adn_monitor/application/monitor_controller.py b/monitor/src/adn_monitor/application/monitor_controller.py index 8e21972..6ffd235 100644 --- a/monitor/src/adn_monitor/application/monitor_controller.py +++ b/monitor/src/adn_monitor/application/monitor_controller.py @@ -562,7 +562,13 @@ def _handle_brdg_event_parts( return Success(None) logger.debug("(REPORT) bridge event parts[0]=%s parts[1]=%s parts[2]=%s", parts[0], parts[1], parts[2]) alias_repo.resolve_subscriber_sync(int(parts[6])) - alias_repo.resolve_talkgroup_sync(int(parts[8])) + if parts[0] == "PRIVATE VOICE": + # Private call destinations are subscriber ids, not talkgroups -- without + # this, alias_short(destination) in rts_update always misses the cache and + # falls back to the raw id, even when the subscriber is registered. + alias_repo.resolve_subscriber_sync(int(parts[8])) + else: + alias_repo.resolve_talkgroup_sync(int(parts[8])) if parts[0] in ("GROUP VOICE", "PRIVATE VOICE"): _event_ts = time.time() rts_update(parts, state, alias_svc, time_str) diff --git a/monitor/src/adn_monitor/application/report_mapper.py b/monitor/src/adn_monitor/application/report_mapper.py index 840a7be..b4bf70a 100644 --- a/monitor/src/adn_monitor/application/report_mapper.py +++ b/monitor/src/adn_monitor/application/report_mapper.py @@ -357,7 +357,10 @@ def voice_event_to_csv_parts(voice: dict[str, Any]) -> list[str] | None: dur = voice.get("duration_s") if dur is not None: parts.append(f"{float(dur):.2f}") - parts.append("1" if voice.get("is_announcement") else "0") + if family == "PRIVATE" and direction == "TX" and voice.get("dest_peer_id") is not None: + parts.append(str(int(voice["dest_peer_id"]))) + else: + parts.append("1" if voice.get("is_announcement") else "0") return parts diff --git a/monitor/src/adn_monitor/application/rts_update.py b/monitor/src/adn_monitor/application/rts_update.py index 6b65e7e..3fa0d81 100644 --- a/monitor/src/adn_monitor/application/rts_update.py +++ b/monitor/src/adn_monitor/application/rts_update.py @@ -176,6 +176,7 @@ def _voice_event_target_peers( call_type: str, event_slot: int, destination: int, + dest_peer_id: int | None = None, ) -> list[tuple[int | bytes, dict]]: """Peers whose CTABLE chips this voice event should touch.""" if action == "END": @@ -184,6 +185,14 @@ def _voice_event_target_peers( if peer_row is not None and peer_id is not None: return [(peer_id, peer_row)] return [] + if call_type == "PRIVATE VOICE" and dest_peer_id is not None: + # Private calls have no static TG list to match against (the destination + # is a subscriber ID, not a talkgroup) -- go straight to the exact hotspot + # SUB_MAP said the destination was last heard on. + peer_id, peer_row = _resolve_master_peer(peers, dest_peer_id) + if peer_row is not None and peer_id is not None: + return [(peer_id, peer_row)] + return [] if _is_master_peer_row(system): return [ (peer_key, peer_row) @@ -217,6 +226,12 @@ def _voice_event_target_peers( return [(peer_id, peer_row)] return [] + if call_type == "PRIVATE VOICE" and dest_peer_id is not None: + peer_id, peer_row = _resolve_master_peer(peers, dest_peer_id) + if peer_row is not None and peer_id is not None: + return [(peer_id, peer_row)] + return [] + if _is_master_peer_row(system): return [ (peer_key, peer_row) @@ -289,12 +304,30 @@ def rts_update_impl( time_slot = int(p[7]) destination = int(p[8]) is_announcement = len(p) > 9 and p[-1] == "1" + dest_peer_id: int | None = None + if call_type == "PRIVATE VOICE" and trx == "TX": + _pvt_base_len = 10 if action == "END" else 9 + if len(p) > _pvt_base_len: + try: + dest_peer_id = int(p[-1]) + except ValueError: + dest_peer_id = None timeout = time.time() ctable = state.CTABLE sub_short = alias_svc.alias_short(source_sub) sub_call = alias_svc.alias_call(source_sub) - tg_dest = f"TG {destination}    {alias_svc.alias_tgid(destination)}" - tg_short = f"TG {destination}" + if call_type == "PRIVATE VOICE": + # The destination is a subscriber id, not a talkgroup -- resolve it the same + # way as the source (alias_short), not via alias_tgid. Same "name (id)" shape + # as the SUB field for display consistency; _active_tgid_from_peer_ts prefers + # the id inside "(...)" specifically so a callsign's own embedded digit (most + # ham callsigns have one, e.g. "CE5RPY") is never mistaken for the real id. + dest_short = alias_svc.alias_short(destination) + tg_dest = f"{dest_short} ({destination})" if dest_short and dest_short != str(destination) else str(destination) + tg_short = tg_dest + else: + tg_dest = f"TG {destination}    {alias_svc.alias_tgid(destination)}" + tg_short = f"TG {destination}" # INGRESS: register UA/SINGLE owner early; skip live TRX chips / Active QSO row. if call_type == "GROUP VOICE" and action == "INGRESS": @@ -340,6 +373,7 @@ def rts_update_impl( call_type=call_type, event_slot=time_slot, destination=destination, + dest_peer_id=dest_peer_id, ): display_slot = _peer_display_slot( peer_row, diff --git a/monitor/src/adn_monitor/application/tgstats.py b/monitor/src/adn_monitor/application/tgstats.py index 4c3eeba..46f136a 100644 --- a/monitor/src/adn_monitor/application/tgstats.py +++ b/monitor/src/adn_monitor/application/tgstats.py @@ -246,13 +246,22 @@ def _session_key(master_name: str, peer_id: int, slot: int) -> tuple[str, int, i def _active_tgid_from_peer_ts(peer_ts: dict) -> int | None: - """TG number from a live CTABLE timeslot row (DEST / TG fields).""" + """TG/destination id from a live CTABLE timeslot row (DEST / TG fields). + + A resolved display name can contain digits of its own (most ham callsigns do, + e.g. "CE5RPY") -- prefer the id in "(...)" (the convention every display format + here uses, e.g. "CE5RPY, Rodrigo (7300391)") over a bare first-digit-run match, + so a callsign's embedded digit is never mistaken for the real id. + """ if not peer_ts.get("TS"): return None raw = peer_ts.get("TG") or peer_ts.get("DEST") or "" if isinstance(raw, int): return raw text = str(raw).replace(" ", " ") + paren = re.search(r"\((\d+)\)", text) + if paren: + return int(paren.group(1)) match = re.search(r"\d+", text) return int(match.group()) if match else None @@ -302,6 +311,11 @@ def prune_voice_ts_not_in_static( tg_str = str(tgid) if _is_echo_service_live_tgid(tgid): continue + if peer_ts.get("TYPE") == "PRIVATE VOICE": + # Private call destinations are subscriber IDs, never a hotspot's static + # TG -- this check would otherwise wipe the chip on the very next + # build_tgstats pass (fires after every voice event, START included). + continue if tg_str in allowed: continue key = _session_key(master_name, peer_id, slot) diff --git a/monitor/src/adn_monitor/infrastructure/repositories/alias_repository.py b/monitor/src/adn_monitor/infrastructure/repositories/alias_repository.py index 33075b3..74685b2 100644 --- a/monitor/src/adn_monitor/infrastructure/repositories/alias_repository.py +++ b/monitor/src/adn_monitor/infrastructure/repositories/alias_repository.py @@ -212,7 +212,7 @@ def _load_talkgroup_sync(self, tg_id: int) -> None: with self._sync_pool.connection() as conn: cur = conn.cursor() cur.execute( - "SELECT id, name FROM talkgroup_ids WHERE id = %s", + "SELECT id, callsign FROM talkgroup_ids WHERE id = %s", (tg_id,), ) row = cur.fetchone() diff --git a/monitor/tests/test_report_mapper.py b/monitor/tests/test_report_mapper.py index 92617e4..bd076c1 100644 --- a/monitor/tests/test_report_mapper.py +++ b/monitor/tests/test_report_mapper.py @@ -248,6 +248,46 @@ def test_voice_event_to_csv_parts_end_with_duration(): assert parts[-1] == "0" +def test_voice_event_private_tx_appends_dest_peer_id_not_announcement_flag(): + """Regression: TX-direction PRIVATE VOICE events must reconstruct the trailing + dest_peer_id field (the one hotspot that's actually receiving), not the generic + is_announcement "0"/"1" placeholder every other event uses in that position.""" + voice = { + "type": "voice_event", + "call_family": "PRIVATE", + "phase": "START", + "direction": "TX", + "system": "SYSTEM-1", + "stream_id": 1610544978, + "peer_id": 730039110, + "src_id": 7300391, + "slot": 2, + "dst_id": 7300392, + "dest_peer_id": 730039101, + "is_announcement": False, + } + parts = voice_event_to_csv_parts(voice) + assert parts[-1] == "730039101" + + +def test_voice_event_private_rx_still_appends_announcement_flag(): + voice = { + "type": "voice_event", + "call_family": "PRIVATE", + "phase": "START", + "direction": "RX", + "system": "SYSTEM-1", + "stream_id": 1610544978, + "peer_id": 730039110, + "src_id": 7300391, + "slot": 2, + "dst_id": 7300392, + "is_announcement": False, + } + parts = voice_event_to_csv_parts(voice) + assert parts[-1] == "0" + + def test_voice_event_unit_data_header(): voice = { "type": "voice_event", diff --git a/monitor/tests/test_rts_display_slot.py b/monitor/tests/test_rts_display_slot.py index 4ca1c88..d85a789 100644 --- a/monitor/tests/test_rts_display_slot.py +++ b/monitor/tests/test_rts_display_slot.py @@ -26,7 +26,10 @@ from adn_monitor.application.monitor_controller import MonitorState from adn_monitor.application.rts_update import rts_update_impl -from adn_monitor.application.tgstats import prune_voice_ts_not_in_static +from adn_monitor.application.tgstats import ( + _active_tgid_from_peer_ts, + prune_voice_ts_not_in_static, +) def _state_with_peer( @@ -128,6 +131,130 @@ def test_start_without_trailing_field_defaults_announcement_false() -> None: assert peer[2]["ANNOUNCEMENT"] is False +def _state_with_two_peers(*, source_peer: int, dest_peer: int) -> MonitorState: + """Two hotspots on the same MASTER -- private call from source_peer to dest_peer.""" + state = MonitorState() + state.CTABLE = { + "MASTERS": { + "SYSTEM": { + "PEERS": { + source_peer: { + "TS1_STATIC": [], "TS2_STATIC": [], + 1: {"TS": False, "TRX": ""}, 2: {"TS": False, "TRX": ""}, + }, + dest_peer: { + "TS1_STATIC": [], "TS2_STATIC": [], + 1: {"TS": False, "TRX": ""}, 2: {"TS": False, "TRX": ""}, + }, + } + } + }, + "PEERS": {}, + "OPENBRIDGES": {}, + } + return state + + +def test_private_voice_rx_start_marks_transmitting_peer() -> None: + state = _state_with_two_peers(source_peer=730039110, dest_peer=730039101) + rts_update_impl( + "PRIVATE VOICE,START,RX,SYSTEM,1,730039110,7300391,2,7300392".split(","), + state, + _alias(), + lambda: "12:00", + ) + peer = state.CTABLE["MASTERS"]["SYSTEM"]["PEERS"][730039110] + assert peer[2]["TS"] is True + assert peer[2]["TRX"] == "RX" + + +def test_private_voice_start_tx_with_dest_peer_id_marks_receiving_peer() -> None: + """Regression: monitor never showed who a private call was delivered to -- the TX + event now carries the destination hotspot's own peer id (from SUB_MAP), letting it + be matched directly instead of by static TG list (private destinations aren't TGs).""" + state = _state_with_two_peers(source_peer=730039110, dest_peer=730039101) + rts_update_impl( + "PRIVATE VOICE,START,TX,SYSTEM,1,730039110,7300391,2,7300392,730039101".split(","), + state, + _alias(), + lambda: "12:00", + ) + dest = state.CTABLE["MASTERS"]["SYSTEM"]["PEERS"][730039101] + assert dest[2]["TS"] is True + assert dest[2]["TRX"] == "TX" + source = state.CTABLE["MASTERS"]["SYSTEM"]["PEERS"][730039110] + assert source[2]["TS"] is False + + +def test_private_voice_start_tx_without_dest_peer_id_touches_nothing() -> None: + """Backward compat: an old-format event (no trailing peer id) must not crash and + must not fall back to matching by static TG list (meaningless for a subscriber id).""" + state = _state_with_two_peers(source_peer=730039110, dest_peer=730039101) + rts_update_impl( + "PRIVATE VOICE,START,TX,SYSTEM,1,730039110,7300391,2,7300392".split(","), + state, + _alias(), + lambda: "12:00", + ) + for peer in state.CTABLE["MASTERS"]["SYSTEM"]["PEERS"].values(): + assert peer[2]["TS"] is False + + +def test_private_voice_end_tx_with_dest_peer_id_clears_receiving_peer() -> None: + state = _state_with_two_peers(source_peer=730039110, dest_peer=730039101) + rts_update_impl( + "PRIVATE VOICE,START,TX,SYSTEM,1,730039110,7300391,2,7300392,730039101".split(","), + state, + _alias(), + lambda: "12:00", + ) + dest = state.CTABLE["MASTERS"]["SYSTEM"]["PEERS"][730039101] + assert dest[2]["TS"] is True + rts_update_impl( + "PRIVATE VOICE,END,TX,SYSTEM,1,730039110,7300391,2,7300392,1.23,730039101".split(","), + state, + _alias(), + lambda: "12:00", + ) + assert dest[2]["TS"] is False + + +def test_active_tgid_prefers_parenthesized_id_over_first_digit_run() -> None: + assert _active_tgid_from_peer_ts({"TS": True, "TG": "CE5RPY, Rodrigo (7300391)"}) == 7300391 + + +def test_active_tgid_falls_back_to_first_digit_run_without_parens() -> None: + """Group-call format ("TG 7300391 Name") has no parens -- unaffected.""" + assert _active_tgid_from_peer_ts({"TS": True, "TG": "TG 7300391    Some TG"}) == 7300391 + + +def test_private_voice_end_clears_when_dest_callsign_contains_a_digit() -> None: + """Regression: a real ham callsign like "CE5RPY" contains a digit of its own. + _active_tgid_from_peer_ts must extract the id from "(...)", not just the first + digit run in the field -- otherwise a resolved callsign's embedded digit would + be matched instead of the real id, and the chip would never clear.""" + alias = MagicMock() + alias.alias_short.side_effect = lambda dmr_id: "CE5RPY, Rodrigo" if dmr_id == 7300391 else str(dmr_id) + alias.alias_call.return_value = "CE5RPY" + state = _state_with_two_peers(source_peer=730039101, dest_peer=730039110) + rts_update_impl( + "PRIVATE VOICE,START,TX,SYSTEM,1,730039101,7300392,2,7300391,730039110".split(","), + state, + alias, + lambda: "12:00", + ) + dest = state.CTABLE["MASTERS"]["SYSTEM"]["PEERS"][730039110] + assert dest[2]["TS"] is True + assert dest[2]["TG"] == "CE5RPY, Rodrigo (7300391)" + rts_update_impl( + "PRIVATE VOICE,END,TX,SYSTEM,1,730039101,7300392,2,7300391,1.23,730039110".split(","), + state, + alias, + lambda: "12:00", + ) + assert dest[2]["TS"] is False + + def test_end_clears_cross_slot_when_static_tg_removed() -> None: """END must clear the slot lit at START even if OPTIONS no longer map the TG there.""" state = _state_with_peer(peer_id=730001, ts1_static=["52090"], ts2_static=[]) @@ -178,6 +305,38 @@ def test_prune_preserves_echo_9990_live_trx_chip() -> None: assert peer[2]["TRX"] == "RX" +def test_prune_preserves_private_voice_chip_with_no_static_tg_match() -> None: + """Regression: build_tgstats fires after every voice event, including PRIVATE VOICE + START -- a private call's destination is a subscriber id and will never be in any + peer's static TG list, so the un-exempted prune wiped the chip almost immediately.""" + state = _state_with_peer(peer_id=730039110, ts1_static=[], ts2_static=["7304"]) + peer = state.CTABLE["MASTERS"]["SYSTEM"]["PEERS"][730039110] + peer[2]["TS"] = True + peer[2]["TYPE"] = "PRIVATE VOICE" + peer[2]["TRX"] = "RX" + peer[2]["TG"] = "TG 7300392" + peer[2]["DEST"] = "TG 7300392" + prune_voice_ts_not_in_static(state, "SYSTEM", 730039110, peer) + assert peer[2]["TS"] is True + assert peer[2]["TRX"] == "RX" + + +def test_rts_update_then_prune_keeps_private_voice_tx_chip_visible() -> None: + """End-to-end: START,TX event followed by the build_tgstats prune it always + triggers must not erase the just-set receiving chip.""" + state = _state_with_two_peers(source_peer=730039110, dest_peer=730039101) + rts_update_impl( + "PRIVATE VOICE,START,TX,SYSTEM,1,730039110,7300391,2,7300392,730039101".split(","), + state, + _alias(), + lambda: "12:00", + ) + dest = state.CTABLE["MASTERS"]["SYSTEM"]["PEERS"][730039101] + prune_voice_ts_not_in_static(state, "SYSTEM", 730039101, dest) + assert dest[2]["TS"] is True + assert dest[2]["TRX"] == "TX" + + def test_echo_9990_rx_tx_live_chips() -> None: state = _state_with_peer(peer_id=730039101, ts1_static=[], ts2_static=[]) alias = _alias()