fix: per-hotspot voice CTABLE targeting (2.1.1)

monitor/src/adn_monitor/application/monitor_controller.py

@@ -640,10 +640,13 @@ def _handle_brdg_event_parts(
             log_message = _format_log_message_unknown_end(_now, parts, alias_svc)
         else:
             log_message = f"{_now[10:19]} Unknown voice bridge log message ({parts[0]})."
-        state.LOGBUF.append(log_message)
-        logger.info("(VOICE) %s", log_message)
-        if broadcast:
-            broadcast.broadcast("l" + log_message, "log")
+        from .rts_update import voice_event_skip_master_downlink_log
+
+        if not voice_event_skip_master_downlink_log(parts, state.CTABLE):
+            state.LOGBUF.append(log_message)
+            logger.info("(VOICE) %s", log_message)
+            if broadcast:
+                broadcast.broadcast("l" + log_message, "log")
     elif parts[0] == "UNIT DATA HEADER" and parts[2] != "TX" and parts[5] not in config_global.get("OPB_FILTER", []):
         _u_ts = time.time()
         _u_utc = format_utc_naive_datetime(_u_ts)

monitor/src/adn_monitor/application/rts_update.py

@@ -31,6 +31,7 @@
 from .alias_service import AliasService
 from .monitor_controller import MonitorState
 from .tgstats import (
+    _active_tgid_from_peer_ts,
     _apply_multi_mode_chips,
     _is_echo_service_live_tgid,
     _is_service_voice_tgid,
@@ -121,6 +122,130 @@ def _static_tg_slot_for_peer(peer_row: dict, destination: int, event_slot: int)
     return event_slot
 
 
+def _is_master_peer_row(system: str) -> bool:
+    """Inject-proxy CTABLE row (``SYSTEM-3``): one hotspot radio per upstream slot."""
+    if "-" not in system:
+        return False
+    base, suffix = system.rsplit("-", 1)
+    return bool(base) and suffix.isdigit()
+
+
+def voice_event_skip_master_downlink_log(parts: list[str], ctable: dict) -> bool:
+    """True when a MASTER downlink leg should update chips only (no duplicate log line).
+
+    The server announces one call; remapped ``SYSTEM-N`` TX fan-out must not emit
+    one log row per connected hotspot.
+    """
+    if len(parts) < 4 or parts[1] not in ("START", "END") or parts[2] != "TX":
+        return False
+    system = parts[3]
+    if system in ctable.get("OPENBRIDGES", {}):
+        return False
+    if _is_master_peer_row(system):
+        return True
+    return system in ctable.get("MASTERS", {})
+
+
+def _peer_static_lists_include_tg(peer_row: dict, destination: int) -> bool:
+    tg = str(destination)
+    ts1 = [str(x).strip() for x in (peer_row.get("TS1_STATIC") or []) if str(x).strip()]
+    ts2 = [str(x).strip() for x in (peer_row.get("TS2_STATIC") or []) if str(x).strip()]
+    return tg in ts1 or tg in ts2
+
+
+def _peer_slot_busy_other_tg(peer_ts: dict, destination: int) -> bool:
+    active = _active_tgid_from_peer_ts(peer_ts)
+    return active is not None and int(active) != int(destination)
+
+
+def _peer_row_shows_destination(peer_row: dict, destination: int) -> bool:
+    for slot in (1, 2):
+        peer_ts = peer_row.get(slot)
+        if isinstance(peer_ts, dict) and _active_tgid_from_peer_ts(peer_ts) == destination:
+            return True
+    return False
+
+
+def _voice_event_target_peers(
+    system: str,
+    peers: dict,
+    *,
+    action: str,
+    trx: str,
+    source_peer: int,
+    source_sub: int,
+    call_type: str,
+    event_slot: int,
+    destination: int,
+) -> list[tuple[int | bytes, dict]]:
+    """Peers whose CTABLE chips this voice event should touch."""
+    if action == "END":
+        if trx == "RX":
+            peer_id, peer_row = _resolve_master_peer(peers, source_peer)
+            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)
+                for peer_key, peer_row in peers.items()
+                if isinstance(peer_row, dict) and _peer_row_shows_destination(peer_row, destination)
+            ]
+        if _is_echo_service_live_tgid(destination):
+            svc_id, svc_row = _resolve_master_peer(peers, destination)
+            if svc_row is not None and svc_id is not None:
+                return [(svc_id, svc_row)]
+            sub_id, sub_row = _resolve_master_peer(peers, source_sub)
+            if sub_row is not None and sub_id is not None:
+                return [(sub_id, sub_row)]
+            return []
+        return [
+            (peer_key, peer_row)
+            for peer_key, peer_row in peers.items()
+            if isinstance(peer_row, dict)
+            and (
+                _peer_row_shows_destination(peer_row, destination)
+                or (
+                    not _peer_keys_equal(source_peer, peer_key)
+                    and _peer_static_lists_include_tg(peer_row, destination)
+                )
+            )
+        ]
+
+    if trx == "RX":
+        peer_id, peer_row = _resolve_master_peer(peers, source_peer)
+        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)
+            for peer_key, peer_row in peers.items()
+            if isinstance(peer_row, dict)
+        ]
+
+    if _is_echo_service_live_tgid(destination):
+        svc_id, svc_row = _resolve_master_peer(peers, destination)
+        if svc_row is not None and svc_id is not None:
+            return [(svc_id, svc_row)]
+        sub_id, sub_row = _resolve_master_peer(peers, source_sub)
+        if sub_row is not None and sub_id is not None:
+            return [(sub_id, sub_row)]
+        return []
+
+    targets: list[tuple[int | bytes, dict]] = []
+    for peer_key, peer_row in peers.items():
+        if not isinstance(peer_row, dict):
+            continue
+        if _peer_keys_equal(source_peer, peer_key):
+            continue
+        if not _peer_static_lists_include_tg(peer_row, destination):
+            continue
+        targets.append((peer_key, peer_row))
+    return targets
+
+
 def _peer_display_slot(
     peer_row: dict,
     peer_key,
@@ -205,8 +330,17 @@ def rts_update_impl(
                 _apply_voice_single_ts(
                     state, ctable, system, time_slot, destination, source_peer, trx=trx
                 )
-        for peer in ctable["MASTERS"][system]["PEERS"]:
-            peer_row = ctable["MASTERS"][system]["PEERS"][peer]
+        for peer, peer_row in _voice_event_target_peers(
+            system,
+            ctable["MASTERS"][system]["PEERS"],
+            action=action,
+            trx=trx,
+            source_peer=source_peer,
+            source_sub=source_sub,
+            call_type=call_type,
+            event_slot=time_slot,
+            destination=destination,
+        ):
             display_slot = _peer_display_slot(
                 peer_row,
                 peer,
@@ -219,6 +353,12 @@ def rts_update_impl(
             crxstatus = "RX" if _peer_keys_equal(source_peer, peer) else "TX"
             peer_ts = peer_row[display_slot]
             if action == "START":
+                if (
+                    peer_ts.get("TS")
+                    and not _is_echo_service_live_tgid(destination)
+                    and _peer_slot_busy_other_tg(peer_ts, destination)
+                ):
+                    continue
                 # Local PTT (TRX=RX / red): do not replace with another peer's downlink (TRX=TX / green).
                 if (
                     peer_ts.get("TS")

monitor/tests/test_rts_display_slot.py

@@ -247,6 +247,80 @@ def test_echo_tx_downlink_uses_wire_slot_not_static_map() -> None:
     assert peer[2]["TS"] is False
 
 
+def test_rx_start_only_updates_transmitting_peer_on_aggregate_master() -> None:
+    """RX must not mark every peer with the same static TG as TX (green)."""
+    state = MonitorState()
+    state.CTABLE = {
+        "MASTERS": {
+            "SYSTEM": {
+                "PEERS": {
+                    730001: {
+                        "TS1_STATIC": [],
+                        "TS2_STATIC": ["7144"],
+                        1: {"TS": False, "TRX": ""},
+                        2: {"TS": False, "TRX": ""},
+                    },
+                    730002: {
+                        "TS1_STATIC": [],
+                        "TS2_STATIC": ["7144"],
+                        1: {"TS": False, "TRX": ""},
+                        2: {"TS": False, "TRX": ""},
+                    },
+                }
+            }
+        },
+        "PEERS": {},
+        "OPENBRIDGES": {},
+    }
+    alias = _alias()
+    rts_update_impl(
+        "GROUP VOICE,START,RX,SYSTEM,1,730002,730002,2,7144".split(","),
+        state,
+        alias,
+        lambda: "12:00",
+    )
+    assert state.CTABLE["MASTERS"]["SYSTEM"]["PEERS"][730002][2]["TS"] is True
+    assert state.CTABLE["MASTERS"]["SYSTEM"]["PEERS"][730002][2]["TRX"] == "RX"
+    assert state.CTABLE["MASTERS"]["SYSTEM"]["PEERS"][730001][2]["TS"] is False
+
+
+def test_tx_downlink_blocked_when_peer_slot_busy_other_tg() -> None:
+    """While QSO on TG 7141, downlink START for TG 71442 must not light the chip."""
+    state = MonitorState()
+    state.CTABLE = {
+        "MASTERS": {
+            "SYSTEM-2": {
+                "PEERS": {
+                    714002301: {
+                        "TS1_STATIC": [],
+                        "TS2_STATIC": ["7141", "71442"],
+                        1: {"TS": False, "TRX": ""},
+                        2: {"TS": False, "TRX": ""},
+                    }
+                }
+            }
+        },
+        "PEERS": {},
+        "OPENBRIDGES": {},
+    }
+    alias = _alias()
+    rts_update_impl(
+        "GROUP VOICE,START,RX,SYSTEM-2,1,714002301,714002301,2,7141".split(","),
+        state,
+        alias,
+        lambda: "12:00",
+    )
+    rts_update_impl(
+        "GROUP VOICE,START,TX,SYSTEM-2,2,730002,730002,2,71442".split(","),
+        state,
+        alias,
+        lambda: "12:01",
+    )
+    peer = state.CTABLE["MASTERS"]["SYSTEM-2"]["PEERS"][714002301]
+    assert peer[2]["TRX"] == "RX"
+    assert "7141" in peer[2]["TG"]
+
+
 def test_companion_tx_does_not_replace_own_active_qso_on_other_tg() -> None:
     """While TX on TG 7144 (RX chip), companion TX for another TG must not overwrite the slot."""
     state = MonitorState()

monitor/tests/test_voice_log_dedupe.py

@@ -0,0 +1,25 @@
+# ADN Monitor - voice log dedupe tests
+#
+# Copyright (C) 2026  Rodrigo Pérez, CE5RPY <ce5rpy@qmd.cl>
+
+from __future__ import annotations
+
+from adn_monitor.application.rts_update import voice_event_skip_master_downlink_log
+
+
+def test_skip_master_peer_row_tx_start_log() -> None:
+    ctable = {"MASTERS": {"SYSTEM-2": {"PEERS": {}}}, "OPENBRIDGES": {}}
+    parts = "GROUP VOICE,START,TX,SYSTEM-2,1,730002,730002,2,71442".split(",")
+    assert voice_event_skip_master_downlink_log(parts, ctable)
+
+
+def test_keep_obp_rx_start_log() -> None:
+    ctable = {"MASTERS": {}, "OPENBRIDGES": {"OBP-CL": {}}}
+    parts = "GROUP VOICE,START,RX,OBP-CL,1,7140023,7140023,1,71442".split(",")
+    assert not voice_event_skip_master_downlink_log(parts, ctable)
+
+
+def test_keep_master_rx_start_log() -> None:
+    ctable = {"MASTERS": {"SYSTEM-2": {"PEERS": {}}}, "OPENBRIDGES": {}}
+    parts = "GROUP VOICE,START,RX,SYSTEM-2,1,714002301,714002301,2,7141".split(",")
+    assert not voice_event_skip_master_downlink_log(parts, ctable)