From f657e61c58051f65185d3d555b581111d21d4d8d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Rodrigo=20P=C3=A9rez?= Date: Sat, 6 Jun 2026 19:46:08 -0400 Subject: [PATCH] refactor(v2): extract bridge helpers and timer mixin (V2-P0-004 chunk 1) Move shared TG/OBP helpers and legacy timer loops (rule_timer, trimmers, bridge_reset, bridge_debug) into application/bridge/; BridgeUseCases keeps the public facade via BridgeTimerMixin. --- src/adn_server/application/bridge/__init__.py | 1 + src/adn_server/application/bridge/helpers.py | 45 ++ src/adn_server/application/bridge/timers.py | 667 +++++++++++++++++ .../application/bridge_use_cases.py | 690 +----------------- 4 files changed, 718 insertions(+), 685 deletions(-) create mode 100644 src/adn_server/application/bridge/__init__.py create mode 100644 src/adn_server/application/bridge/helpers.py create mode 100644 src/adn_server/application/bridge/timers.py diff --git a/src/adn_server/application/bridge/__init__.py b/src/adn_server/application/bridge/__init__.py new file mode 100644 index 0000000..7fd36ef --- /dev/null +++ b/src/adn_server/application/bridge/__init__.py @@ -0,0 +1 @@ +# Bridge use-case submodules (V2-P0-004 split). Public API stays in bridge_use_cases. diff --git a/src/adn_server/application/bridge/helpers.py b/src/adn_server/application/bridge/helpers.py new file mode 100644 index 0000000..6e4211a --- /dev/null +++ b/src/adn_server/application/bridge/helpers.py @@ -0,0 +1,45 @@ +# ADN DMR Peer Server - bridge helpers (V2-P0-004) +# Copyright (C) 2026 Rodrigo Pérez, CE5RPY + +"""Shared bridge routing helpers (no Twisted).""" + +from __future__ import annotations + +from typing import Any + +from ...domain import bytes_3, int_id + +# Embedded LC codeword sits at bits 116:148 inside the 48-bit EMB field (108:156). +# Legacy bridge_master.py replaces dmrbits[116:148] on bursts B–E (dtype_vseq 1–4). +EMB_LC_SLICE = slice(116, 148) + + +def obp_target_bcsq_quenches_stream( + systems_cfg: dict[str, Any], target_name: str, dst_id_b: bytes, stream_id: bytes +) -> bool: + """True if target OBP config has _bcsq[tgid]==stream_id (bytes key or same int TG).""" + m = systems_cfg.get(target_name, {}).get("_bcsq") + if not isinstance(m, dict) or not m: + return False + tid = dst_id_b[:3] if isinstance(dst_id_b, bytes) and len(dst_id_b) >= 3 else bytes_3(int_id(dst_id_b)) + if m.get(tid) == stream_id: + return True + for k, v in m.items(): + if v != stream_id: + continue + try: + if isinstance(k, bytes) and len(k) >= 3 and int_id(k) == int_id(tid): + return True + except Exception: + continue + return False + + +def is_special_tg(bridge_key: str) -> bool: + """True if bridge is special TGID 9990-9999 (excluded from infinite timer).""" + if bridge_key and bridge_key[0:1] == "#": + return False + try: + return 9990 <= int(bridge_key) <= 9999 + except ValueError: + return False diff --git a/src/adn_server/application/bridge/timers.py b/src/adn_server/application/bridge/timers.py new file mode 100644 index 0000000..d71b0ba --- /dev/null +++ b/src/adn_server/application/bridge/timers.py @@ -0,0 +1,667 @@ +# ADN DMR Peer Server - bridge timer loops (V2-P0-004) +# Copyright (C) 2026 Rodrigo Pérez, CE5RPY + +"""Legacy bridge timer / trimmer loops (no Twisted imports).""" + +from __future__ import annotations + +import logging +import time +from collections import deque +from typing import Any + +from ...domain import HBPF_SLT_VTERM, bytes_3, int_id +from .helpers import is_special_tg + +logger = logging.getLogger(__name__) + + +class BridgeTimerMixin: + """rule_timer, stream_trimmer, bridge_reset, stat_trimmer, bridge_debug loops.""" + + def rule_timer_loop(self) -> None: + """Run one iteration of rule_timer_loop (legacy 52s LoopingCall). Activate/deactivate by timeout.""" + bridges = self._router.get_bridges() + systems_cfg = self._config.get("SYSTEMS", {}) + now = time.time() + remove_bridges: deque = deque() + _debug_msgs: list[str] = [] + + for bridge_key, entries in list(bridges.items()): + if bridge_key not in bridges: + continue + bridge_used = False + is_special_tg = is_special_tg(bridge_key) + + for sys_entry in entries: + system_name = sys_entry.get("SYSTEM", "") + sys_config = systems_cfg.get(system_name, {}) + is_single_mode = sys_config.get("SINGLE_MODE", False) + to_type = sys_entry.get("TO_TYPE", "") + active = sys_entry.get("ACTIVE", False) + timer = sys_entry.get("TIMER", 0.0) + is_dynamic = bridge_key[0:1] != "#" and to_type != "STAT" + is_obp = sys_config.get("MODE") == "OPENBRIDGE" + + if not is_single_mode and is_dynamic and not is_obp and not is_special_tg: + if to_type == "ON": + if active: + bridge_used = True + _debug_msgs.append('(ROUTER) Conference Bridge ACTIVE (INFINITE TIMER): System: %s Bridge: %s, TS: %s, TGID: %s' % (system_name, bridge_key, sys_entry.get("TS"), int_id(sys_entry.get("TGID", b"")))) + else: + _debug_msgs.append('(ROUTER) Conference Bridge INACTIVE (no change): System: %s Bridge: %s, TS: %s, TGID: %s' % (system_name, bridge_key, sys_entry.get("TS"), int_id(sys_entry.get("TGID", b"")))) + elif to_type == "OFF": + if not active: + sys_entry["ACTIVE"] = True + bridge_used = True + logger.info( + "(ROUTER) Conference Bridge ACTIVATED (NO TIMEOUT): System: %s, Bridge: %s, TS: %s, TGID: %s", + system_name, bridge_key, sys_entry.get("TS"), int_id(sys_entry.get("TGID", b"")) + ) + else: + bridge_used = True + _debug_msgs.append('(ROUTER) Conference Bridge ACTIVE (no change): System: %s Bridge: %s, TS: %s, TGID: %s' % (system_name, bridge_key, sys_entry.get("TS"), int_id(sys_entry.get("TGID", b"")))) + else: + if to_type == "ON": + if active: + bridge_used = True + if timer < now: + sys_entry["ACTIVE"] = False + if self._on_bridge_deactivated and bridge_key[:1] == "#": + self._on_bridge_deactivated(system_name) + logger.info( + "(ROUTER) Conference Bridge TIMEOUT: DEACTIVATE System: %s, Bridge: %s, TS: %s, TGID: %s", + system_name, bridge_key, sys_entry.get("TS"), int_id(sys_entry.get("TGID", b"")) + ) + else: + logger.info( + "(ROUTER) Conference Bridge ACTIVE (ON timer running): System: %s Bridge: %s, TS: %s, TGID: %s, Timeout in: %.2fs,", + system_name, bridge_key, sys_entry.get("TS"), int_id(sys_entry.get("TGID", b"")), timer - now + ) + elif not active: + _debug_msgs.append('(ROUTER) Conference Bridge INACTIVE (no change): System: %s Bridge: %s, TS: %s, TGID: %s' % (system_name, bridge_key, sys_entry.get("TS"), int_id(sys_entry.get("TGID", b"")))) + elif to_type == "OFF": + if not active: + if timer < now: + sys_entry["ACTIVE"] = True + bridge_used = True + logger.info( + "(ROUTER) Conference Bridge TIMEOUT: ACTIVATE System: %s, Bridge: %s, TS: %s, TGID: %s", + system_name, bridge_key, sys_entry.get("TS"), int_id(sys_entry.get("TGID", b"")) + ) + else: + bridge_used = True + logger.info( + "(ROUTER) Conference Bridge INACTIVE (OFF timer running): System: %s Bridge: %s, TS: %s, TGID: %s, Timeout in: %.2fs,", + system_name, bridge_key, sys_entry.get("TS"), int_id(sys_entry.get("TGID", b"")), timer - now + ) + elif active: + bridge_used = True + _debug_msgs.append('(ROUTER) Conference Bridge ACTIVE (no change): System: %s Bridge: %s, TS: %s, TGID: %s' % (system_name, bridge_key, sys_entry.get("TS"), int_id(sys_entry.get("TGID", b"")))) + else: + if not is_obp or (is_obp and to_type == "STAT"): + bridge_used = True + _debug_msgs.append('(ROUTER) Conference Bridge NO ACTION: System: %s, Bridge: %s, TS: %s, TGID: %s' % (system_name, bridge_key, sys_entry.get("TS"), int_id(sys_entry.get("TGID", b"")))) + + if not bridge_used: + remove_bridges.append(bridge_key) + + if _debug_msgs: + logger.debug('\n'.join(_debug_msgs)) + + for key in remove_bridges: + del bridges[key] + logger.debug("(ROUTER) Unused conference bridge %s removed", key) + + def bridge_debug_loop(self) -> None: + """Legacy bridgeDebug (bridge_master.py 487-543): remove invalid bridges, fix >1 active dial per MASTER.""" + logger.debug("(BRIDGEDEBUG) Running bridge debug") + bridges = self._router.get_bridges() + systems_cfg = self._config.get("SYSTEMS", {}) + now = time.time() + statroll = 0 + + # Kill off any bridges that should not exist, ever (legacy: 0-9, #0-#9) + for b in ("0", "1", "2", "3", "4", "5", "6", "7", "8", "9"): + bridges.pop(b, None) + bridges.pop("#" + b, None) + + for system in systems_cfg: + bridgeroll = 0 + dialroll = 0 + activeroll = 0 + for _bridge, entries in list(bridges.items()): + if _bridge not in bridges: + continue + for enabled_system in entries: + if enabled_system.get("SYSTEM") == system: + bridgeroll += 1 + if enabled_system.get("ACTIVE"): + if _bridge and _bridge[:1] == "#": + dialroll += 1 + activeroll += 1 + else: + activeroll += 1 + if enabled_system.get("TO_TYPE") == "STAT": + statroll += 1 + if bridgeroll: + logger.debug( + "(BRIDGEDEBUG) system %s has %s bridges of which %s are in an ACTIVE state", + system, bridgeroll, activeroll, + ) + if dialroll > 1 and systems_cfg.get(system, {}).get("MODE") == "MASTER": + logger.warning( + "(BRIDGEDEBUG) system %s has more than one active dial bridge (%s) - fixing", + system, dialroll, + ) + _tmout = float(systems_cfg.get(system, {}).get("DEFAULT_UA_TIMER", 10)) + times: dict[float, str] = {} + for _bridge, entries in list(bridges.items()): + if _bridge not in bridges: + continue + for enabled_system in entries: + if enabled_system.get("ACTIVE") and _bridge and _bridge[:1] == "#": + t = enabled_system.get("TIMER") + if isinstance(t, (int, float)): + times[t] = _bridge + for _bridge in set(times.values()): + if _bridge not in bridges: + continue + logger.warning("(BRIDGEDEBUG) deactivating system: %s for bridge: %s", system, _bridge) + try: + _setbridge = int(_bridge[1:]) if _bridge[:1] == "#" else int(_bridge) + except ValueError: + _setbridge = 9 + bridgetemp: list[dict[str, Any]] = [] + for bridgesystem in bridges.get(_bridge, []): + if bridgesystem.get("SYSTEM") == system and bridgesystem.get("TS") == 2: + bridgetemp.append({ + "SYSTEM": system, + "TS": 2, + "TGID": bytes_3(9), + "ACTIVE": False, + "TIMEOUT": _tmout * 60.0, + "TO_TYPE": "ON", + "OFF": [], + "ON": [bytes_3(_setbridge)], + "RESET": [], + "TIMER": now + _tmout * 60.0, + }) + else: + bridgetemp.append(bridgesystem) + bridges[_bridge] = bridgetemp + + logger.info("(BRIDGEDEBUG) The server currently has %s STATic bridges", statroll) + + def apply_in_band_signalling( + self, system_name: str, slot: int, dst_id: bytes, pkt_time: float + ) -> None: + """Legacy in-band signalling on voice terminator (bridge_master.py ~3447-3549). + + Reflector bridges (#xxx) are ONLY processed when dst TG is 9 (legacy ~3455). + De-activation distinguishes SINGLE_MODE True/False (legacy ~3484-3548). + """ + bridges = self._router.get_bridges() + systems_cfg = self._config.get("SYSTEMS", {}) + _dst_group = int_id(dst_id) + dst_id_b = dst_id if isinstance(dst_id, bytes) and len(dst_id) >= 3 else bytes_3(_dst_group) + + for _bridge, entries in list(bridges.items()): + if _bridge not in bridges: + continue + # Legacy bridge_master.py ~3455: reflector bridges only respond to TG9 + if _bridge[:1] == "#" and _dst_group != 9: + continue + for _system in entries: + if _system.get("SYSTEM") != system_name: + continue + _ts = _system.get("TS") + _tgid = _system.get("TGID") + tgid_match = _tgid == dst_id_b or (isinstance(_tgid, bytes) and isinstance(dst_id_b, bytes) and _tgid == dst_id_b) + if not tgid_match and _tgid is not None: + try: + tgid_match = int_id(_tgid) == _dst_group + except (TypeError, ValueError): + pass + # [1] TGID matches a rule source, reset its timer + if slot == _ts and tgid_match: + to_type = _system.get("TO_TYPE", "") + active = _system.get("ACTIVE", False) + timeout = _system.get("TIMEOUT") + timeout_sec = timeout if isinstance(timeout, (int, float)) else 0.0 + if (to_type == "ON" and active) or (to_type == "OFF" and not active): + if timeout_sec: + _system["TIMER"] = pkt_time + timeout_sec + logger.info("(%s) [1] Transmission match for Bridge: %s. Reset timeout to %s", system_name, _bridge, _system["TIMER"]) + + # [2-4] TGID matches an ACTIVATION trigger (dst_id in ON or RESET) + on_list = _system.get("ON") or [] + reset_list = _system.get("RESET") or [] + if slot == _ts and (dst_id_b in on_list or dst_id_b in reset_list or any(int_id(x) == _dst_group for x in on_list) or any(int_id(x) == _dst_group for x in reset_list)): + if dst_id_b in on_list or any(int_id(x) == _dst_group for x in on_list): + if not _system.get("ACTIVE"): + _system["ACTIVE"] = True + _system["TIMER"] = pkt_time + (float(_system.get("TIMEOUT") or 0) or 0) + logger.info("(%s) [2] Bridge: %s, connection changed to state: %s", system_name, _bridge, _system["ACTIVE"]) + if _system.get("TO_TYPE") == "OFF": + _system["TIMER"] = pkt_time + logger.info("(%s) [3] Bridge: %s set to \"OFF\" with an on timer rule: timeout timer cancelled", system_name, _bridge) + if _system.get("ACTIVE") and _system.get("TO_TYPE") == "ON" and _system.get("TIMEOUT"): + _system["TIMER"] = pkt_time + float(_system["TIMEOUT"]) + logger.info("(%s) [4] Bridge: %s, timeout timer reset to: %s", system_name, _bridge, _system["TIMER"] - pkt_time) + + # [5-8] TGID matches a DE-ACTIVATION trigger + # Legacy bridge_master.py ~3484-3548: SINGLE_MODE True vs False + sys_cfg = systems_cfg.get(system_name, {}) + is_single_mode = sys_cfg.get("MODE") == "MASTER" and sys_cfg.get("SINGLE_MODE", False) + + if is_single_mode: + # SINGLE_MODE=True: aggressive de-activation (legacy ~3484-3503) + off_list = _system.get("OFF") or [] + if slot == _ts and (dst_id_b in off_list or dst_id_b in reset_list or dst_id_b == bytes_3(4000) or dst_id_b != _tgid): + if dst_id_b in off_list or dst_id_b != _tgid or dst_id_b == bytes_3(4000): + if _system.get("ACTIVE"): + _system["ACTIVE"] = False + logger.info("(%s) [5] Bridge: %s, connection changed to state: %s", system_name, _bridge, _system["ACTIVE"]) + if _system.get("TO_TYPE") == "ON": + _system["TIMER"] = pkt_time + logger.info("(%s) [6] Bridge: %s set to \"OFF\" with an on timer rule: timeout timer cancelled", system_name, _bridge) + if not _system.get("ACTIVE") and _system.get("TO_TYPE") == "OFF" and _system.get("TIMEOUT"): + _system["TIMER"] = pkt_time + float(_system["TIMEOUT"]) + logger.info("(%s) [7] Bridge: %s, timeout timer reset to: %s", system_name, _bridge, _system["TIMER"] - pkt_time) + if _system.get("ACTIVE") and _system.get("TO_TYPE") == "ON" and dst_id_b in off_list: + _system["TIMER"] = pkt_time + logger.info("(%s) [8] Bridge: %s set to ON with and \"OFF\" timer rule: timeout timer cancelled", system_name, _bridge) + else: + # SINGLE_MODE=False: only de-activate on TG 4000 (legacy ~3504-3548) + if dst_id_b == bytes_3(4000) and slot == _ts: + is_static_tg = False + ts1_static = sys_cfg.get("TS1_STATIC") or "" + ts2_static = sys_cfg.get("TS2_STATIC") or "" + if ts1_static and slot == 1: + static_tgs = [int(tg) for tg in ts1_static.split(",") if tg.strip()] + if _dst_group in static_tgs: + is_static_tg = True + elif ts2_static and slot == 2: + static_tgs = [int(tg) for tg in ts2_static.split(",") if tg.strip()] + if _dst_group in static_tgs: + is_static_tg = True + + is_reflector = _bridge[:1] == "#" + off_list = _system.get("OFF") or [] + if dst_id_b in off_list or dst_id_b == bytes_3(4000) or (dst_id_b != _tgid and not is_static_tg and not is_reflector): + if _system.get("ACTIVE"): + _system["ACTIVE"] = False + logger.info("(%s) [5b] Bridge: %s, connection changed to state: %s (TG 4000 forced deactivation)", system_name, _bridge, _system["ACTIVE"]) + if _system.get("TO_TYPE") == "ON": + _system["TIMER"] = pkt_time + logger.info("(%s) [6b] Bridge: %s set to \"OFF\" with an on timer rule: timeout timer cancelled", system_name, _bridge) + if not _system.get("ACTIVE") and _system.get("TO_TYPE") == "OFF" and _system.get("TIMEOUT"): + _system["TIMER"] = pkt_time + float(_system["TIMEOUT"]) + logger.info("(%s) [7b] Bridge: %s, timeout timer reset to: %s", system_name, _bridge, _system["TIMER"] - pkt_time) + if _system.get("ACTIVE") and _system.get("TO_TYPE") == "ON" and dst_id_b in (off_list or []): + _system["TIMER"] = pkt_time + logger.info("(%s) [8b] Bridge: %s set to ON with and \"OFF\" timer rule: timeout timer cancelled", system_name, _bridge) + + def _obp_emit_end_tx_forward_leg( + self, + tgt_name: str, + stream_id: bytes, + tst: dict[str, Any], + now: float, + ) -> bool: + """Emit GROUP VOICE,END,TX for one to_target OBP forward leg (H_LC in STATUS). + + Same CSV shape as legacy bridge_master.py send_bridgeEvent on VTERM (~2039, ~2121). + Safe for legacy and v2 monitors (OPENBRIDGE STREAMS chip clear on END,TX). + """ + if not isinstance(tst, dict) or "H_LC" not in tst: + return False + if tst.get("_end_tx_sent"): + # Already cleared this leg's monitor TX chip; do not re-emit on repeated BCSQ. + return False + if not bool(self._config.get("REPORTS", {}).get("REPORT", True)): + return False + rfs = tst.get("RFS", b"\x00\x00\x00") + peer = tst.get("RX_PEER", b"\x00\x00\x00\x00") + tgid_b = tst.get("TGID", b"\x00\x00\x00") + start = tst.get("START", now) + duration = max(0.0, now - start) + if not self._send_bridge_event( + "GROUP VOICE,END,TX,{},{},{},{},{},{},{:.2f}".format( + tgt_name, + int_id(stream_id), + int_id(peer), + int_id(rfs), + 1, + int_id(tgid_b), + duration, + ) + ): + return False + tst["_end_tx_sent"] = True + return True + + def _obp_emit_end_tx_for_forward_legs(self, stream_id: bytes, source_system: str, now: float) -> None: + """Emit GROUP VOICE,END,TX for every OBP that still holds this stream as a to_target forward leg. + + On idle timeout the trimmer sends END,RX for the source only. VTERM may never arrive for + forwarded legs, so the monitor would otherwise keep stale TX chips on destination rows. + Forward legs are identified by STATUS[stream_id] containing H_LC (see to_target OPENBRIDGE). + """ + protocols = self._get_protocols() if self._get_protocols else {} + systems_cfg = self._config.get("SYSTEMS", {}) + for tgt_name, tgt_proto in (protocols or {}).items(): + if tgt_name == source_system: + continue + if systems_cfg.get(tgt_name, {}).get("MODE") != "OPENBRIDGE": + continue + tstatus = getattr(tgt_proto, "STATUS", None) + if not tstatus or stream_id not in tstatus: + continue + tst = tstatus[stream_id] + # Emit the monitor END,TX chip-clear but DO NOT pop STATUS: legacy keeps the + # forward-leg entry until the stream trimmer removes it by age (180s). Popping + # mid-stream caused the source/forward-leg entry to vanish and re-CALL-START churn. + self._obp_emit_end_tx_forward_leg(tgt_name, stream_id, tst, now) + + def on_obp_bcsq_received(self, system_name: str, tgid: bytes, stream_id: bytes) -> None: + """After valid BCSQ on this OBP leg: emit the monitor END,TX chip-clear only. + + Legacy parity (hblink.py ~629-639): BCSQ only records CONFIG['_bcsq'][tgid]=stream_id + (done inline in udp_hbp) and routerOBP.to_target then *skips* the quenched target for + that stream (see _obp_target_bcsq_quenches_stream). It never destroys stream state. + We additionally emit a one-shot END,TX so the monitor clears stale TX chips, but we do + NOT pop STATUS[stream_id]: popping mid-call made the same stream re-CALL-START over and + over (visible as BCSQ-storm churn), which flapped loop-control and broke OBP->HBP audio. + The forward-leg entry is removed by the stream trimmer on idle (legacy behaviour). + """ + protocols = self._get_protocols() if self._get_protocols else {} + tgt_proto = protocols.get(system_name) + if not tgt_proto: + return + if self._config.get("SYSTEMS", {}).get(system_name, {}).get("MODE") != "OPENBRIDGE": + return + tstatus = getattr(tgt_proto, "STATUS", None) + if not tstatus or stream_id not in tstatus: + return + tst = tstatus[stream_id] + if not isinstance(tst, dict) or "H_LC" not in tst: + return + if tst.get("TGID", b"\x00\x00\x00") != tgid: + return + self._obp_emit_end_tx_forward_leg(system_name, stream_id, tst, time.time()) + + def flush_monitor_events_for_system(self, system_name: str, protocol: Any) -> None: + if not self._config.get("REPORTS", {}).get("REPORT", True): + return + status = getattr(protocol, "STATUS", None) + if not isinstance(status, dict): + return + mode = self._config.get("SYSTEMS", {}).get(system_name, {}).get("MODE") + now = time.time() + + if mode == "OPENBRIDGE": + for stream_id, st in list(status.items()): + if not isinstance(stream_id, (bytes, bytearray)) or not isinstance(st, dict): + continue + trx = "TX" if "H_LC" in st else "RX" + start = st.get("START", now) + self._send_bridge_event( + "GROUP VOICE,END,{},{},{},{},{},{},{},{:.2f}".format( + trx, system_name, int_id(stream_id), + int_id(st.get("RX_PEER", b"\x00\x00\x00\x00")), + int_id(st.get("RFS", b"\x00\x00\x00")), 1, + int_id(st.get("TGID", b"\x00\x00\x00")), + max(0.0, now - start), + ) + ) + if trx == "RX": + self._obp_emit_end_tx_for_forward_legs(stream_id, system_name, now) + return + + for slot in (1, 2): + slot_st = status.get(slot) + if not isinstance(slot_st, dict): + continue + for trx, sid_key, peer_key, rfs_key, tgid_key, start_key, type_key in ( + ("RX", "RX_STREAM_ID", "RX_PEER", "RX_RFS", "RX_TGID", "RX_START", "RX_TYPE"), + ("TX", "TX_STREAM_ID", "TX_PEER", "TX_RFS", "TX_TGID", "TX_START", "TX_TYPE"), + ): + sid = slot_st.get(sid_key, b"\x00") + if sid in (b"\x00", b"") or slot_st.get(type_key) == HBPF_SLT_VTERM: + continue + self._send_bridge_event( + "GROUP VOICE,END,{},{},{},{},{},{},{},{:.2f}".format( + trx, system_name, int_id(sid), + int_id(slot_st.get(peer_key, b"\x00\x00\x00\x00")), + int_id(slot_st.get(rfs_key, b"\x00\x00\x00")), slot, + int_id(slot_st.get(tgid_key, b"\x00\x00\x00")), + max(0.0, now - slot_st.get(start_key, now)), + ) + ) + + def stream_trimmer_loop(self) -> None: + """Trim old stream state (legacy stream_trimmer_loop, 5s). RX/TX timeout per system/slot; OBP streams (legacy bridge.py 181-240).""" + logger.debug("(ROUTER) Trimming inactive stream IDs from system lists") + protocols = self._get_protocols() if self._get_protocols else {} + systems_cfg = self._config.get("SYSTEMS", {}) + now = time.time() + for system_name, protocol in protocols.items(): + if not getattr(protocol, "STATUS", None): + continue + # OBP: legacy bridge_master.stream_trimmer_loop:631-703 — two-stage lifecycle: + # Stage 1 (5s idle, no _to, no _fin): set _to=True, emit END,RX, continue. + # Stage 2 (180s idle): remove stream entry. + # + # Legacy parity: routerOBP.STATUS is a *flat* dict keyed only by stream_id + # (bridge_master.py:1911). The loop iterates `for stream_id in systems[s].STATUS:`, + # which automatically catches everything seeded by to_target HBP->OBP, + # sendDataToOBP, pvt_call_received and the OBP source path itself. We do + # not keep a parallel dict -- that was a divergence that produced leaks. + if systems_cfg.get(system_name, {}).get("MODE") == "OPENBRIDGE": + obp_status = getattr(protocol, "STATUS", None) + if isinstance(obp_status, dict) and obp_status: + to_remove: list[bytes] = [] + for stream_id, st in list(obp_status.items()): + if not isinstance(st, dict): + continue + last = st.get("LAST", 0) + # Stage 2: finished streams older than 180s → remove + if st.get("_fin") and last < now - 180: + to_remove.append(stream_id) + continue + # Stage 2: timed-out streams older than 180s → remove + if st.get("_to") and last < now - 180: + to_remove.append(stream_id) + continue + # Stage 1: 5s idle, not yet timed out → mark _to, emit END + if "_to" not in st and "_fin" not in st and last < now - 5: + rfs = st.get("RFS", b"\x00\x00\x00") + peer = st.get("RX_PEER", b"\x00\x00\x00\x00") + tgid = st.get("TGID", b"\x00\x00\x00") + start = st.get("START", now) + duration = max(0.0, last - start) + self._send_bridge_event( + "GROUP VOICE,END,RX,{},{},{},{},{},{},{:.2f}".format( + system_name, int_id(stream_id), int_id(peer), int_id(rfs), 1, int_id(tgid), duration + ) + ) + st["_to"] = True + # Legacy trimmer emits END,RX here only; forward legs waited ~180s. + # END,TX now (END_TX_FORWARD): same event as legacy VTERM path (~2039). + self._obp_emit_end_tx_for_forward_legs(stream_id, system_name, now) + continue + for stream_id in to_remove: + _syscfg = systems_cfg.get(system_name, {}) + _bmap = _syscfg.get("_bcsq") + if isinstance(_bmap, dict): + for _tgid_k, _sid in list(_bmap.items()): + if _sid == stream_id: + _bmap.pop(_tgid_k, None) + self._obp_emit_end_tx_for_forward_legs(stream_id, system_name, now) + obp_status.pop(stream_id, None) + continue + for slot in (1, 2): + _slot = protocol.STATUS.get(slot) + if not _slot: + continue + if _slot.get("RX_TYPE") != HBPF_SLT_VTERM and _slot.get("RX_TIME", 0) < now - 5: + _slot["RX_TYPE"] = HBPF_SLT_VTERM + logger.info( + "(%s) *TIME OUT* RX STREAM ID: %s SUB: %s TGID %s, TS %s, Duration: %.2f", + system_name, int_id(_slot.get("RX_STREAM_ID", b"")), int_id(_slot.get("RX_RFS", b"")), + int_id(_slot.get("RX_TGID", b"")), slot, _slot.get("RX_TIME", 0) - _slot.get("RX_START", 0), + ) + self._send_bridge_event( + "GROUP VOICE,END,RX,{},{},{},{},{},{},{:.2f}".format( + system_name, int_id(_slot.get("RX_STREAM_ID", b"")), int_id(_slot.get("RX_PEER", b"")), + int_id(_slot.get("RX_RFS", b"")), slot, int_id(_slot.get("RX_TGID", b"")), + _slot.get("RX_TIME", 0) - _slot.get("RX_START", 0), + ) + ) + if _slot.get("RX_TIME", 0) < now - 60: + _slot["RX_STREAM_ID"] = b"\x00" + if _slot.get("TX_TYPE") != HBPF_SLT_VTERM and _slot.get("TX_TIME", 0) < now - 5: + _slot["TX_TYPE"] = HBPF_SLT_VTERM + logger.debug( + "(%s) *TIME OUT* TX STREAM ID: %s SUB: %s TGID %s, TS %s, Duration: %.2f", + system_name, int_id(_slot.get("TX_STREAM_ID", b"")), int_id(_slot.get("TX_RFS", b"")), + int_id(_slot.get("TX_TGID", b"")), slot, _slot.get("TX_TIME", 0) - _slot.get("TX_START", 0), + ) + self._send_bridge_event( + "GROUP VOICE,END,TX,{},{},{},{},{},{},{:.2f}".format( + system_name, int_id(_slot.get("TX_STREAM_ID", b"")), int_id(_slot.get("TX_PEER", b"")), + int_id(_slot.get("TX_RFS", b"")), slot, int_id(_slot.get("TX_TGID", b"")), + _slot.get("TX_TIME", 0) - _slot.get("TX_START", 0), + ) + ) + # -- Intentional divergence from legacy -- + # Legacy bridge_master.py:602 only iterates `range(1,3)` in the HBP branch, + # so any STATUS[stream_id] entry seeded by voice_use_cases (announcements + # via sys_obj.STATUS[stream_id], cf. bridge_master.py:1156/1626) or by + # send_voice_packet against HBP MASTER targets is never freed in legacy. + # Here we defensively sweep any bytes-keyed entry with LAST > 180 s. + # This does not change the wire protocol; it only releases memory. + hbp_status = getattr(protocol, "STATUS", None) + if isinstance(hbp_status, dict): + for _k in list(hbp_status.keys()): + if isinstance(_k, (bytes, bytearray)): + _v = hbp_status.get(_k) + if isinstance(_v, dict) and _v.get("LAST", 0) < now - 180: + hbp_status.pop(_k, None) + if hasattr(protocol, "trim_dmra_streams"): + protocol.trim_dmra_streams() + + def bridge_reset_loop(self) -> None: + """Bridge reset iteration (legacy bridge_reset, 6s). Clear _reset and remove_bridge_system.""" + systems_cfg = self._config.get("SYSTEMS", {}) + for system_name in list(systems_cfg.keys()): + sys_cfg = systems_cfg.get(system_name, {}) + if sys_cfg.get("_reset"): + logger.info("(BRIDGERESET) Bridge reset for %s - no peers", system_name) + self.remove_bridge_system(system_name) + try: + del sys_cfg["_opt_key"] + except KeyError: + pass + try: + del sys_cfg["_options_static_apply_fp"] + except KeyError: + pass + self._restore_prohibited_static_bridge_legs(system_name) + sys_cfg["_reset"] = False + sys_cfg["_resetlog"] = False + + def _restore_prohibited_static_bridge_legs(self, system_name: str) -> None: + """After BRIDGERESET / peer RPTO: restore static TGs in prohibited_tgs (parity with _make_echo_bridges).""" + prohibited_tgs = (0, 1, 2, 3, 4, 5, 9, 9990, 9991, 9992, 9993, 9994, 9995, 9996, 9997, 9998, 9999) + sys_cfg = self._config.get("SYSTEMS", {}).get(system_name, {}) + if sys_cfg.get("MODE") != "MASTER" or not sys_cfg.get("ENABLED", True): + return + bridges = self._router.get_bridges() + now = time.time() + for ts, static_key, acl_key in ( + (1, "TS1_STATIC", "TG1_ACL"), + (2, "TS2_STATIC", "TG2_ACL"), + ): + for tg_s in str(sys_cfg.get(static_key) or "").split(","): + tg_s = tg_s.strip() + if not tg_s: + continue + try: + tg = int(tg_s) + except ValueError: + continue + if tg not in prohibited_tgs: + continue + if sys_cfg.get("USE_ACL") and not self.acl_check( + bytes_3(tg), sys_cfg.get(acl_key, (True, [])) + ): + continue + bridge_key = str(tg) + if bridge_key not in bridges: + continue + timeout_sec = (1.0 / 6.0) * 60.0 + leg: dict[str, Any] = { + "SYSTEM": system_name, + "TS": ts, + "TGID": bytes_3(tg), + "ACTIVE": True, + "TIMEOUT": timeout_sec, + "TO_TYPE": "NONE", + "ON": [], + "OFF": [], + "RESET": [], + "TIMER": now + timeout_sec, + } + for i, entry in enumerate(bridges[bridge_key]): + if entry.get("SYSTEM") == system_name and entry.get("TS") == ts: + if entry.get("ACTIVE") and entry.get("TO_TYPE") == "NONE": + break + bridges[bridge_key][i] = leg + logger.info( + "(ROUTER) Restored service bridge leg: %s bridge %s TS %s", + system_name, bridge_key, ts, + ) + break + else: + bridges[bridge_key].append(leg) + logger.info( + "(ROUTER) Re-added service bridge leg: %s bridge %s TS %s", + system_name, bridge_key, ts, + ) + + def _remove_bridge_system(self, system_name: str, bridges: dict[str, list[dict[str, Any]]]) -> None: + """Remove all bridge entries for system (legacy remove_bridge_system).""" + to_remove: list[str] = [] + for bridge_key, entries in list(bridges.items()): + if bridge_key not in bridges: + continue + new_entries = [e for e in entries if e.get("SYSTEM") != system_name] + if len(new_entries) < len(entries): + if new_entries: + bridges[bridge_key] = new_entries + else: + to_remove.append(bridge_key) + for key in to_remove: + del bridges[key] + + def stat_trimmer_loop(self) -> None: + """Trim STAT-only bridges with no ON/OFF in use (legacy statTrimmer, 303s).""" + logger.debug("(ROUTER) STAT trimmer loop started") + bridges = self._router.get_bridges() + remove_bridges: deque = deque() + for bridge_key, entries in list(bridges.items()): + if bridge_key not in bridges: + continue + has_stat = any(e.get("TO_TYPE") == "STAT" for e in entries) + in_use = any( + (e.get("TO_TYPE") == "ON" and e.get("ACTIVE")) or e.get("TO_TYPE") == "OFF" + for e in entries + ) + if has_stat and not in_use: + remove_bridges.append(bridge_key) + for key in remove_bridges: + del bridges[key] + logger.debug("(ROUTER) STAT bridge %s removed", key) diff --git a/src/adn_server/application/bridge_use_cases.py b/src/adn_server/application/bridge_use_cases.py index 4dbbe48..591dd1d 100644 --- a/src/adn_server/application/bridge_use_cases.py +++ b/src/adn_server/application/bridge_use_cases.py @@ -44,46 +44,13 @@ from ..domain import HBPF_DATA_SYNC, HBPF_SLT_VHEAD, HBPF_SLT_VTERM, STREAM_TO, from ..domain.talker_alias import DMRA_BLOCK_COUNT from .ports import BridgeRouter, DmrEmbeddedLcEncoder, TalkerAliasEmblcEncoder from .talker_alias_use_cases import TalkerAliasUseCases, passthrough_complete, talker_alias_settings +from .bridge.helpers import EMB_LC_SLICE, is_special_tg, obp_target_bcsq_quenches_stream +from .bridge.timers import BridgeTimerMixin logger = logging.getLogger(__name__) -# Embedded LC codeword sits at bits 116:148 inside the 48-bit EMB field (108:156). -# Legacy bridge_master.py replaces dmrbits[116:148] on bursts B–E (dtype_vseq 1–4). -_EMB_LC_SLICE = slice(116, 148) - -# Legacy loop-control loser sends BCSQ once when ENHANCED_OBP (_bcsq not in stream state). -def _obp_target_bcsq_quenches_stream( - systems_cfg: dict[str, Any], target_name: str, dst_id_b: bytes, stream_id: bytes -) -> bool: - """True if target OBP config has _bcsq[tgid]==stream_id (bytes key or same int TG).""" - m = systems_cfg.get(target_name, {}).get("_bcsq") - if not isinstance(m, dict) or not m: - return False - tid = dst_id_b[:3] if isinstance(dst_id_b, bytes) and len(dst_id_b) >= 3 else bytes_3(int_id(dst_id_b)) - if m.get(tid) == stream_id: - return True - for k, v in m.items(): - if v != stream_id: - continue - try: - if isinstance(k, bytes) and len(k) >= 3 and int_id(k) == int_id(tid): - return True - except Exception: - continue - return False - - -def _is_special_tg(bridge_key: str) -> bool: - """True if bridge is special TGID 9990-9999 (excluded from infinite timer).""" - if bridge_key and bridge_key[0:1] == "#": - return False - try: - return 9990 <= int(bridge_key) <= 9999 - except ValueError: - return False - -class BridgeUseCases: +class BridgeUseCases(BridgeTimerMixin): """Use cases for conference bridge state (BRIDGES).""" def __init__( @@ -431,7 +398,7 @@ class BridgeUseCases: frag = st[emb_key][dtype_vseq] if dtype_vseq == 4 and ta_emb is not None: st["TX_TA_ON"] = True - dmrbits[_EMB_LC_SLICE] = frag + dmrbits[EMB_LC_SLICE] = frag def _clear_talker_alias_embed(self, st: dict[str, Any]) -> None: st.pop("TX_TA_EMB", None) @@ -443,653 +410,6 @@ class BridgeUseCases: """Return current BRIDGES.""" return self._router.get_bridges() - def rule_timer_loop(self) -> None: - """Run one iteration of rule_timer_loop (legacy 52s LoopingCall). Activate/deactivate by timeout.""" - bridges = self._router.get_bridges() - systems_cfg = self._config.get("SYSTEMS", {}) - now = time.time() - remove_bridges: deque = deque() - _debug_msgs: list[str] = [] - - for bridge_key, entries in list(bridges.items()): - if bridge_key not in bridges: - continue - bridge_used = False - is_special_tg = _is_special_tg(bridge_key) - - for sys_entry in entries: - system_name = sys_entry.get("SYSTEM", "") - sys_config = systems_cfg.get(system_name, {}) - is_single_mode = sys_config.get("SINGLE_MODE", False) - to_type = sys_entry.get("TO_TYPE", "") - active = sys_entry.get("ACTIVE", False) - timer = sys_entry.get("TIMER", 0.0) - is_dynamic = bridge_key[0:1] != "#" and to_type != "STAT" - is_obp = sys_config.get("MODE") == "OPENBRIDGE" - - if not is_single_mode and is_dynamic and not is_obp and not is_special_tg: - if to_type == "ON": - if active: - bridge_used = True - _debug_msgs.append('(ROUTER) Conference Bridge ACTIVE (INFINITE TIMER): System: %s Bridge: %s, TS: %s, TGID: %s' % (system_name, bridge_key, sys_entry.get("TS"), int_id(sys_entry.get("TGID", b"")))) - else: - _debug_msgs.append('(ROUTER) Conference Bridge INACTIVE (no change): System: %s Bridge: %s, TS: %s, TGID: %s' % (system_name, bridge_key, sys_entry.get("TS"), int_id(sys_entry.get("TGID", b"")))) - elif to_type == "OFF": - if not active: - sys_entry["ACTIVE"] = True - bridge_used = True - logger.info( - "(ROUTER) Conference Bridge ACTIVATED (NO TIMEOUT): System: %s, Bridge: %s, TS: %s, TGID: %s", - system_name, bridge_key, sys_entry.get("TS"), int_id(sys_entry.get("TGID", b"")) - ) - else: - bridge_used = True - _debug_msgs.append('(ROUTER) Conference Bridge ACTIVE (no change): System: %s Bridge: %s, TS: %s, TGID: %s' % (system_name, bridge_key, sys_entry.get("TS"), int_id(sys_entry.get("TGID", b"")))) - else: - if to_type == "ON": - if active: - bridge_used = True - if timer < now: - sys_entry["ACTIVE"] = False - if self._on_bridge_deactivated and bridge_key[:1] == "#": - self._on_bridge_deactivated(system_name) - logger.info( - "(ROUTER) Conference Bridge TIMEOUT: DEACTIVATE System: %s, Bridge: %s, TS: %s, TGID: %s", - system_name, bridge_key, sys_entry.get("TS"), int_id(sys_entry.get("TGID", b"")) - ) - else: - logger.info( - "(ROUTER) Conference Bridge ACTIVE (ON timer running): System: %s Bridge: %s, TS: %s, TGID: %s, Timeout in: %.2fs,", - system_name, bridge_key, sys_entry.get("TS"), int_id(sys_entry.get("TGID", b"")), timer - now - ) - elif not active: - _debug_msgs.append('(ROUTER) Conference Bridge INACTIVE (no change): System: %s Bridge: %s, TS: %s, TGID: %s' % (system_name, bridge_key, sys_entry.get("TS"), int_id(sys_entry.get("TGID", b"")))) - elif to_type == "OFF": - if not active: - if timer < now: - sys_entry["ACTIVE"] = True - bridge_used = True - logger.info( - "(ROUTER) Conference Bridge TIMEOUT: ACTIVATE System: %s, Bridge: %s, TS: %s, TGID: %s", - system_name, bridge_key, sys_entry.get("TS"), int_id(sys_entry.get("TGID", b"")) - ) - else: - bridge_used = True - logger.info( - "(ROUTER) Conference Bridge INACTIVE (OFF timer running): System: %s Bridge: %s, TS: %s, TGID: %s, Timeout in: %.2fs,", - system_name, bridge_key, sys_entry.get("TS"), int_id(sys_entry.get("TGID", b"")), timer - now - ) - elif active: - bridge_used = True - _debug_msgs.append('(ROUTER) Conference Bridge ACTIVE (no change): System: %s Bridge: %s, TS: %s, TGID: %s' % (system_name, bridge_key, sys_entry.get("TS"), int_id(sys_entry.get("TGID", b"")))) - else: - if not is_obp or (is_obp and to_type == "STAT"): - bridge_used = True - _debug_msgs.append('(ROUTER) Conference Bridge NO ACTION: System: %s, Bridge: %s, TS: %s, TGID: %s' % (system_name, bridge_key, sys_entry.get("TS"), int_id(sys_entry.get("TGID", b"")))) - - if not bridge_used: - remove_bridges.append(bridge_key) - - if _debug_msgs: - logger.debug('\n'.join(_debug_msgs)) - - for key in remove_bridges: - del bridges[key] - logger.debug("(ROUTER) Unused conference bridge %s removed", key) - - def bridge_debug_loop(self) -> None: - """Legacy bridgeDebug (bridge_master.py 487-543): remove invalid bridges, fix >1 active dial per MASTER.""" - logger.debug("(BRIDGEDEBUG) Running bridge debug") - bridges = self._router.get_bridges() - systems_cfg = self._config.get("SYSTEMS", {}) - now = time.time() - statroll = 0 - - # Kill off any bridges that should not exist, ever (legacy: 0-9, #0-#9) - for b in ("0", "1", "2", "3", "4", "5", "6", "7", "8", "9"): - bridges.pop(b, None) - bridges.pop("#" + b, None) - - for system in systems_cfg: - bridgeroll = 0 - dialroll = 0 - activeroll = 0 - for _bridge, entries in list(bridges.items()): - if _bridge not in bridges: - continue - for enabled_system in entries: - if enabled_system.get("SYSTEM") == system: - bridgeroll += 1 - if enabled_system.get("ACTIVE"): - if _bridge and _bridge[:1] == "#": - dialroll += 1 - activeroll += 1 - else: - activeroll += 1 - if enabled_system.get("TO_TYPE") == "STAT": - statroll += 1 - if bridgeroll: - logger.debug( - "(BRIDGEDEBUG) system %s has %s bridges of which %s are in an ACTIVE state", - system, bridgeroll, activeroll, - ) - if dialroll > 1 and systems_cfg.get(system, {}).get("MODE") == "MASTER": - logger.warning( - "(BRIDGEDEBUG) system %s has more than one active dial bridge (%s) - fixing", - system, dialroll, - ) - _tmout = float(systems_cfg.get(system, {}).get("DEFAULT_UA_TIMER", 10)) - times: dict[float, str] = {} - for _bridge, entries in list(bridges.items()): - if _bridge not in bridges: - continue - for enabled_system in entries: - if enabled_system.get("ACTIVE") and _bridge and _bridge[:1] == "#": - t = enabled_system.get("TIMER") - if isinstance(t, (int, float)): - times[t] = _bridge - for _bridge in set(times.values()): - if _bridge not in bridges: - continue - logger.warning("(BRIDGEDEBUG) deactivating system: %s for bridge: %s", system, _bridge) - try: - _setbridge = int(_bridge[1:]) if _bridge[:1] == "#" else int(_bridge) - except ValueError: - _setbridge = 9 - bridgetemp: list[dict[str, Any]] = [] - for bridgesystem in bridges.get(_bridge, []): - if bridgesystem.get("SYSTEM") == system and bridgesystem.get("TS") == 2: - bridgetemp.append({ - "SYSTEM": system, - "TS": 2, - "TGID": bytes_3(9), - "ACTIVE": False, - "TIMEOUT": _tmout * 60.0, - "TO_TYPE": "ON", - "OFF": [], - "ON": [bytes_3(_setbridge)], - "RESET": [], - "TIMER": now + _tmout * 60.0, - }) - else: - bridgetemp.append(bridgesystem) - bridges[_bridge] = bridgetemp - - logger.info("(BRIDGEDEBUG) The server currently has %s STATic bridges", statroll) - - def apply_in_band_signalling( - self, system_name: str, slot: int, dst_id: bytes, pkt_time: float - ) -> None: - """Legacy in-band signalling on voice terminator (bridge_master.py ~3447-3549). - - Reflector bridges (#xxx) are ONLY processed when dst TG is 9 (legacy ~3455). - De-activation distinguishes SINGLE_MODE True/False (legacy ~3484-3548). - """ - bridges = self._router.get_bridges() - systems_cfg = self._config.get("SYSTEMS", {}) - _dst_group = int_id(dst_id) - dst_id_b = dst_id if isinstance(dst_id, bytes) and len(dst_id) >= 3 else bytes_3(_dst_group) - - for _bridge, entries in list(bridges.items()): - if _bridge not in bridges: - continue - # Legacy bridge_master.py ~3455: reflector bridges only respond to TG9 - if _bridge[:1] == "#" and _dst_group != 9: - continue - for _system in entries: - if _system.get("SYSTEM") != system_name: - continue - _ts = _system.get("TS") - _tgid = _system.get("TGID") - tgid_match = _tgid == dst_id_b or (isinstance(_tgid, bytes) and isinstance(dst_id_b, bytes) and _tgid == dst_id_b) - if not tgid_match and _tgid is not None: - try: - tgid_match = int_id(_tgid) == _dst_group - except (TypeError, ValueError): - pass - # [1] TGID matches a rule source, reset its timer - if slot == _ts and tgid_match: - to_type = _system.get("TO_TYPE", "") - active = _system.get("ACTIVE", False) - timeout = _system.get("TIMEOUT") - timeout_sec = timeout if isinstance(timeout, (int, float)) else 0.0 - if (to_type == "ON" and active) or (to_type == "OFF" and not active): - if timeout_sec: - _system["TIMER"] = pkt_time + timeout_sec - logger.info("(%s) [1] Transmission match for Bridge: %s. Reset timeout to %s", system_name, _bridge, _system["TIMER"]) - - # [2-4] TGID matches an ACTIVATION trigger (dst_id in ON or RESET) - on_list = _system.get("ON") or [] - reset_list = _system.get("RESET") or [] - if slot == _ts and (dst_id_b in on_list or dst_id_b in reset_list or any(int_id(x) == _dst_group for x in on_list) or any(int_id(x) == _dst_group for x in reset_list)): - if dst_id_b in on_list or any(int_id(x) == _dst_group for x in on_list): - if not _system.get("ACTIVE"): - _system["ACTIVE"] = True - _system["TIMER"] = pkt_time + (float(_system.get("TIMEOUT") or 0) or 0) - logger.info("(%s) [2] Bridge: %s, connection changed to state: %s", system_name, _bridge, _system["ACTIVE"]) - if _system.get("TO_TYPE") == "OFF": - _system["TIMER"] = pkt_time - logger.info("(%s) [3] Bridge: %s set to \"OFF\" with an on timer rule: timeout timer cancelled", system_name, _bridge) - if _system.get("ACTIVE") and _system.get("TO_TYPE") == "ON" and _system.get("TIMEOUT"): - _system["TIMER"] = pkt_time + float(_system["TIMEOUT"]) - logger.info("(%s) [4] Bridge: %s, timeout timer reset to: %s", system_name, _bridge, _system["TIMER"] - pkt_time) - - # [5-8] TGID matches a DE-ACTIVATION trigger - # Legacy bridge_master.py ~3484-3548: SINGLE_MODE True vs False - sys_cfg = systems_cfg.get(system_name, {}) - is_single_mode = sys_cfg.get("MODE") == "MASTER" and sys_cfg.get("SINGLE_MODE", False) - - if is_single_mode: - # SINGLE_MODE=True: aggressive de-activation (legacy ~3484-3503) - off_list = _system.get("OFF") or [] - if slot == _ts and (dst_id_b in off_list or dst_id_b in reset_list or dst_id_b == bytes_3(4000) or dst_id_b != _tgid): - if dst_id_b in off_list or dst_id_b != _tgid or dst_id_b == bytes_3(4000): - if _system.get("ACTIVE"): - _system["ACTIVE"] = False - logger.info("(%s) [5] Bridge: %s, connection changed to state: %s", system_name, _bridge, _system["ACTIVE"]) - if _system.get("TO_TYPE") == "ON": - _system["TIMER"] = pkt_time - logger.info("(%s) [6] Bridge: %s set to \"OFF\" with an on timer rule: timeout timer cancelled", system_name, _bridge) - if not _system.get("ACTIVE") and _system.get("TO_TYPE") == "OFF" and _system.get("TIMEOUT"): - _system["TIMER"] = pkt_time + float(_system["TIMEOUT"]) - logger.info("(%s) [7] Bridge: %s, timeout timer reset to: %s", system_name, _bridge, _system["TIMER"] - pkt_time) - if _system.get("ACTIVE") and _system.get("TO_TYPE") == "ON" and dst_id_b in off_list: - _system["TIMER"] = pkt_time - logger.info("(%s) [8] Bridge: %s set to ON with and \"OFF\" timer rule: timeout timer cancelled", system_name, _bridge) - else: - # SINGLE_MODE=False: only de-activate on TG 4000 (legacy ~3504-3548) - if dst_id_b == bytes_3(4000) and slot == _ts: - is_static_tg = False - ts1_static = sys_cfg.get("TS1_STATIC") or "" - ts2_static = sys_cfg.get("TS2_STATIC") or "" - if ts1_static and slot == 1: - static_tgs = [int(tg) for tg in ts1_static.split(",") if tg.strip()] - if _dst_group in static_tgs: - is_static_tg = True - elif ts2_static and slot == 2: - static_tgs = [int(tg) for tg in ts2_static.split(",") if tg.strip()] - if _dst_group in static_tgs: - is_static_tg = True - - is_reflector = _bridge[:1] == "#" - off_list = _system.get("OFF") or [] - if dst_id_b in off_list or dst_id_b == bytes_3(4000) or (dst_id_b != _tgid and not is_static_tg and not is_reflector): - if _system.get("ACTIVE"): - _system["ACTIVE"] = False - logger.info("(%s) [5b] Bridge: %s, connection changed to state: %s (TG 4000 forced deactivation)", system_name, _bridge, _system["ACTIVE"]) - if _system.get("TO_TYPE") == "ON": - _system["TIMER"] = pkt_time - logger.info("(%s) [6b] Bridge: %s set to \"OFF\" with an on timer rule: timeout timer cancelled", system_name, _bridge) - if not _system.get("ACTIVE") and _system.get("TO_TYPE") == "OFF" and _system.get("TIMEOUT"): - _system["TIMER"] = pkt_time + float(_system["TIMEOUT"]) - logger.info("(%s) [7b] Bridge: %s, timeout timer reset to: %s", system_name, _bridge, _system["TIMER"] - pkt_time) - if _system.get("ACTIVE") and _system.get("TO_TYPE") == "ON" and dst_id_b in (off_list or []): - _system["TIMER"] = pkt_time - logger.info("(%s) [8b] Bridge: %s set to ON with and \"OFF\" timer rule: timeout timer cancelled", system_name, _bridge) - - def _obp_emit_end_tx_forward_leg( - self, - tgt_name: str, - stream_id: bytes, - tst: dict[str, Any], - now: float, - ) -> bool: - """Emit GROUP VOICE,END,TX for one to_target OBP forward leg (H_LC in STATUS). - - Same CSV shape as legacy bridge_master.py send_bridgeEvent on VTERM (~2039, ~2121). - Safe for legacy and v2 monitors (OPENBRIDGE STREAMS chip clear on END,TX). - """ - if not isinstance(tst, dict) or "H_LC" not in tst: - return False - if tst.get("_end_tx_sent"): - # Already cleared this leg's monitor TX chip; do not re-emit on repeated BCSQ. - return False - if not bool(self._config.get("REPORTS", {}).get("REPORT", True)): - return False - rfs = tst.get("RFS", b"\x00\x00\x00") - peer = tst.get("RX_PEER", b"\x00\x00\x00\x00") - tgid_b = tst.get("TGID", b"\x00\x00\x00") - start = tst.get("START", now) - duration = max(0.0, now - start) - if not self._send_bridge_event( - "GROUP VOICE,END,TX,{},{},{},{},{},{},{:.2f}".format( - tgt_name, - int_id(stream_id), - int_id(peer), - int_id(rfs), - 1, - int_id(tgid_b), - duration, - ) - ): - return False - tst["_end_tx_sent"] = True - return True - - def _obp_emit_end_tx_for_forward_legs(self, stream_id: bytes, source_system: str, now: float) -> None: - """Emit GROUP VOICE,END,TX for every OBP that still holds this stream as a to_target forward leg. - - On idle timeout the trimmer sends END,RX for the source only. VTERM may never arrive for - forwarded legs, so the monitor would otherwise keep stale TX chips on destination rows. - Forward legs are identified by STATUS[stream_id] containing H_LC (see to_target OPENBRIDGE). - """ - protocols = self._get_protocols() if self._get_protocols else {} - systems_cfg = self._config.get("SYSTEMS", {}) - for tgt_name, tgt_proto in (protocols or {}).items(): - if tgt_name == source_system: - continue - if systems_cfg.get(tgt_name, {}).get("MODE") != "OPENBRIDGE": - continue - tstatus = getattr(tgt_proto, "STATUS", None) - if not tstatus or stream_id not in tstatus: - continue - tst = tstatus[stream_id] - # Emit the monitor END,TX chip-clear but DO NOT pop STATUS: legacy keeps the - # forward-leg entry until the stream trimmer removes it by age (180s). Popping - # mid-stream caused the source/forward-leg entry to vanish and re-CALL-START churn. - self._obp_emit_end_tx_forward_leg(tgt_name, stream_id, tst, now) - - def on_obp_bcsq_received(self, system_name: str, tgid: bytes, stream_id: bytes) -> None: - """After valid BCSQ on this OBP leg: emit the monitor END,TX chip-clear only. - - Legacy parity (hblink.py ~629-639): BCSQ only records CONFIG['_bcsq'][tgid]=stream_id - (done inline in udp_hbp) and routerOBP.to_target then *skips* the quenched target for - that stream (see _obp_target_bcsq_quenches_stream). It never destroys stream state. - We additionally emit a one-shot END,TX so the monitor clears stale TX chips, but we do - NOT pop STATUS[stream_id]: popping mid-call made the same stream re-CALL-START over and - over (visible as BCSQ-storm churn), which flapped loop-control and broke OBP->HBP audio. - The forward-leg entry is removed by the stream trimmer on idle (legacy behaviour). - """ - protocols = self._get_protocols() if self._get_protocols else {} - tgt_proto = protocols.get(system_name) - if not tgt_proto: - return - if self._config.get("SYSTEMS", {}).get(system_name, {}).get("MODE") != "OPENBRIDGE": - return - tstatus = getattr(tgt_proto, "STATUS", None) - if not tstatus or stream_id not in tstatus: - return - tst = tstatus[stream_id] - if not isinstance(tst, dict) or "H_LC" not in tst: - return - if tst.get("TGID", b"\x00\x00\x00") != tgid: - return - self._obp_emit_end_tx_forward_leg(system_name, stream_id, tst, time.time()) - - def flush_monitor_events_for_system(self, system_name: str, protocol: Any) -> None: - if not self._config.get("REPORTS", {}).get("REPORT", True): - return - status = getattr(protocol, "STATUS", None) - if not isinstance(status, dict): - return - mode = self._config.get("SYSTEMS", {}).get(system_name, {}).get("MODE") - now = time.time() - - if mode == "OPENBRIDGE": - for stream_id, st in list(status.items()): - if not isinstance(stream_id, (bytes, bytearray)) or not isinstance(st, dict): - continue - trx = "TX" if "H_LC" in st else "RX" - start = st.get("START", now) - self._send_bridge_event( - "GROUP VOICE,END,{},{},{},{},{},{},{},{:.2f}".format( - trx, system_name, int_id(stream_id), - int_id(st.get("RX_PEER", b"\x00\x00\x00\x00")), - int_id(st.get("RFS", b"\x00\x00\x00")), 1, - int_id(st.get("TGID", b"\x00\x00\x00")), - max(0.0, now - start), - ) - ) - if trx == "RX": - self._obp_emit_end_tx_for_forward_legs(stream_id, system_name, now) - return - - for slot in (1, 2): - slot_st = status.get(slot) - if not isinstance(slot_st, dict): - continue - for trx, sid_key, peer_key, rfs_key, tgid_key, start_key, type_key in ( - ("RX", "RX_STREAM_ID", "RX_PEER", "RX_RFS", "RX_TGID", "RX_START", "RX_TYPE"), - ("TX", "TX_STREAM_ID", "TX_PEER", "TX_RFS", "TX_TGID", "TX_START", "TX_TYPE"), - ): - sid = slot_st.get(sid_key, b"\x00") - if sid in (b"\x00", b"") or slot_st.get(type_key) == HBPF_SLT_VTERM: - continue - self._send_bridge_event( - "GROUP VOICE,END,{},{},{},{},{},{},{},{:.2f}".format( - trx, system_name, int_id(sid), - int_id(slot_st.get(peer_key, b"\x00\x00\x00\x00")), - int_id(slot_st.get(rfs_key, b"\x00\x00\x00")), slot, - int_id(slot_st.get(tgid_key, b"\x00\x00\x00")), - max(0.0, now - slot_st.get(start_key, now)), - ) - ) - - def stream_trimmer_loop(self) -> None: - """Trim old stream state (legacy stream_trimmer_loop, 5s). RX/TX timeout per system/slot; OBP streams (legacy bridge.py 181-240).""" - logger.debug("(ROUTER) Trimming inactive stream IDs from system lists") - protocols = self._get_protocols() if self._get_protocols else {} - systems_cfg = self._config.get("SYSTEMS", {}) - now = time.time() - for system_name, protocol in protocols.items(): - if not getattr(protocol, "STATUS", None): - continue - # OBP: legacy bridge_master.stream_trimmer_loop:631-703 — two-stage lifecycle: - # Stage 1 (5s idle, no _to, no _fin): set _to=True, emit END,RX, continue. - # Stage 2 (180s idle): remove stream entry. - # - # Legacy parity: routerOBP.STATUS is a *flat* dict keyed only by stream_id - # (bridge_master.py:1911). The loop iterates `for stream_id in systems[s].STATUS:`, - # which automatically catches everything seeded by to_target HBP->OBP, - # sendDataToOBP, pvt_call_received and the OBP source path itself. We do - # not keep a parallel dict -- that was a divergence that produced leaks. - if systems_cfg.get(system_name, {}).get("MODE") == "OPENBRIDGE": - obp_status = getattr(protocol, "STATUS", None) - if isinstance(obp_status, dict) and obp_status: - to_remove: list[bytes] = [] - for stream_id, st in list(obp_status.items()): - if not isinstance(st, dict): - continue - last = st.get("LAST", 0) - # Stage 2: finished streams older than 180s → remove - if st.get("_fin") and last < now - 180: - to_remove.append(stream_id) - continue - # Stage 2: timed-out streams older than 180s → remove - if st.get("_to") and last < now - 180: - to_remove.append(stream_id) - continue - # Stage 1: 5s idle, not yet timed out → mark _to, emit END - if "_to" not in st and "_fin" not in st and last < now - 5: - rfs = st.get("RFS", b"\x00\x00\x00") - peer = st.get("RX_PEER", b"\x00\x00\x00\x00") - tgid = st.get("TGID", b"\x00\x00\x00") - start = st.get("START", now) - duration = max(0.0, last - start) - self._send_bridge_event( - "GROUP VOICE,END,RX,{},{},{},{},{},{},{:.2f}".format( - system_name, int_id(stream_id), int_id(peer), int_id(rfs), 1, int_id(tgid), duration - ) - ) - st["_to"] = True - # Legacy trimmer emits END,RX here only; forward legs waited ~180s. - # END,TX now (END_TX_FORWARD): same event as legacy VTERM path (~2039). - self._obp_emit_end_tx_for_forward_legs(stream_id, system_name, now) - continue - for stream_id in to_remove: - _syscfg = systems_cfg.get(system_name, {}) - _bmap = _syscfg.get("_bcsq") - if isinstance(_bmap, dict): - for _tgid_k, _sid in list(_bmap.items()): - if _sid == stream_id: - _bmap.pop(_tgid_k, None) - self._obp_emit_end_tx_for_forward_legs(stream_id, system_name, now) - obp_status.pop(stream_id, None) - continue - for slot in (1, 2): - _slot = protocol.STATUS.get(slot) - if not _slot: - continue - if _slot.get("RX_TYPE") != HBPF_SLT_VTERM and _slot.get("RX_TIME", 0) < now - 5: - _slot["RX_TYPE"] = HBPF_SLT_VTERM - logger.info( - "(%s) *TIME OUT* RX STREAM ID: %s SUB: %s TGID %s, TS %s, Duration: %.2f", - system_name, int_id(_slot.get("RX_STREAM_ID", b"")), int_id(_slot.get("RX_RFS", b"")), - int_id(_slot.get("RX_TGID", b"")), slot, _slot.get("RX_TIME", 0) - _slot.get("RX_START", 0), - ) - self._send_bridge_event( - "GROUP VOICE,END,RX,{},{},{},{},{},{},{:.2f}".format( - system_name, int_id(_slot.get("RX_STREAM_ID", b"")), int_id(_slot.get("RX_PEER", b"")), - int_id(_slot.get("RX_RFS", b"")), slot, int_id(_slot.get("RX_TGID", b"")), - _slot.get("RX_TIME", 0) - _slot.get("RX_START", 0), - ) - ) - if _slot.get("RX_TIME", 0) < now - 60: - _slot["RX_STREAM_ID"] = b"\x00" - if _slot.get("TX_TYPE") != HBPF_SLT_VTERM and _slot.get("TX_TIME", 0) < now - 5: - _slot["TX_TYPE"] = HBPF_SLT_VTERM - logger.debug( - "(%s) *TIME OUT* TX STREAM ID: %s SUB: %s TGID %s, TS %s, Duration: %.2f", - system_name, int_id(_slot.get("TX_STREAM_ID", b"")), int_id(_slot.get("TX_RFS", b"")), - int_id(_slot.get("TX_TGID", b"")), slot, _slot.get("TX_TIME", 0) - _slot.get("TX_START", 0), - ) - self._send_bridge_event( - "GROUP VOICE,END,TX,{},{},{},{},{},{},{:.2f}".format( - system_name, int_id(_slot.get("TX_STREAM_ID", b"")), int_id(_slot.get("TX_PEER", b"")), - int_id(_slot.get("TX_RFS", b"")), slot, int_id(_slot.get("TX_TGID", b"")), - _slot.get("TX_TIME", 0) - _slot.get("TX_START", 0), - ) - ) - # -- Intentional divergence from legacy -- - # Legacy bridge_master.py:602 only iterates `range(1,3)` in the HBP branch, - # so any STATUS[stream_id] entry seeded by voice_use_cases (announcements - # via sys_obj.STATUS[stream_id], cf. bridge_master.py:1156/1626) or by - # send_voice_packet against HBP MASTER targets is never freed in legacy. - # Here we defensively sweep any bytes-keyed entry with LAST > 180 s. - # This does not change the wire protocol; it only releases memory. - hbp_status = getattr(protocol, "STATUS", None) - if isinstance(hbp_status, dict): - for _k in list(hbp_status.keys()): - if isinstance(_k, (bytes, bytearray)): - _v = hbp_status.get(_k) - if isinstance(_v, dict) and _v.get("LAST", 0) < now - 180: - hbp_status.pop(_k, None) - if hasattr(protocol, "trim_dmra_streams"): - protocol.trim_dmra_streams() - - def bridge_reset_loop(self) -> None: - """Bridge reset iteration (legacy bridge_reset, 6s). Clear _reset and remove_bridge_system.""" - systems_cfg = self._config.get("SYSTEMS", {}) - for system_name in list(systems_cfg.keys()): - sys_cfg = systems_cfg.get(system_name, {}) - if sys_cfg.get("_reset"): - logger.info("(BRIDGERESET) Bridge reset for %s - no peers", system_name) - self.remove_bridge_system(system_name) - try: - del sys_cfg["_opt_key"] - except KeyError: - pass - try: - del sys_cfg["_options_static_apply_fp"] - except KeyError: - pass - self._restore_prohibited_static_bridge_legs(system_name) - sys_cfg["_reset"] = False - sys_cfg["_resetlog"] = False - - def _restore_prohibited_static_bridge_legs(self, system_name: str) -> None: - """After BRIDGERESET / peer RPTO: restore static TGs in prohibited_tgs (parity with _make_echo_bridges).""" - prohibited_tgs = (0, 1, 2, 3, 4, 5, 9, 9990, 9991, 9992, 9993, 9994, 9995, 9996, 9997, 9998, 9999) - sys_cfg = self._config.get("SYSTEMS", {}).get(system_name, {}) - if sys_cfg.get("MODE") != "MASTER" or not sys_cfg.get("ENABLED", True): - return - bridges = self._router.get_bridges() - now = time.time() - for ts, static_key, acl_key in ( - (1, "TS1_STATIC", "TG1_ACL"), - (2, "TS2_STATIC", "TG2_ACL"), - ): - for tg_s in str(sys_cfg.get(static_key) or "").split(","): - tg_s = tg_s.strip() - if not tg_s: - continue - try: - tg = int(tg_s) - except ValueError: - continue - if tg not in prohibited_tgs: - continue - if sys_cfg.get("USE_ACL") and not self.acl_check( - bytes_3(tg), sys_cfg.get(acl_key, (True, [])) - ): - continue - bridge_key = str(tg) - if bridge_key not in bridges: - continue - timeout_sec = (1.0 / 6.0) * 60.0 - leg: dict[str, Any] = { - "SYSTEM": system_name, - "TS": ts, - "TGID": bytes_3(tg), - "ACTIVE": True, - "TIMEOUT": timeout_sec, - "TO_TYPE": "NONE", - "ON": [], - "OFF": [], - "RESET": [], - "TIMER": now + timeout_sec, - } - for i, entry in enumerate(bridges[bridge_key]): - if entry.get("SYSTEM") == system_name and entry.get("TS") == ts: - if entry.get("ACTIVE") and entry.get("TO_TYPE") == "NONE": - break - bridges[bridge_key][i] = leg - logger.info( - "(ROUTER) Restored service bridge leg: %s bridge %s TS %s", - system_name, bridge_key, ts, - ) - break - else: - bridges[bridge_key].append(leg) - logger.info( - "(ROUTER) Re-added service bridge leg: %s bridge %s TS %s", - system_name, bridge_key, ts, - ) - - def _remove_bridge_system(self, system_name: str, bridges: dict[str, list[dict[str, Any]]]) -> None: - """Remove all bridge entries for system (legacy remove_bridge_system).""" - to_remove: list[str] = [] - for bridge_key, entries in list(bridges.items()): - if bridge_key not in bridges: - continue - new_entries = [e for e in entries if e.get("SYSTEM") != system_name] - if len(new_entries) < len(entries): - if new_entries: - bridges[bridge_key] = new_entries - else: - to_remove.append(bridge_key) - for key in to_remove: - del bridges[key] - - def stat_trimmer_loop(self) -> None: - """Trim STAT-only bridges with no ON/OFF in use (legacy statTrimmer, 303s).""" - logger.debug("(ROUTER) STAT trimmer loop started") - bridges = self._router.get_bridges() - remove_bridges: deque = deque() - for bridge_key, entries in list(bridges.items()): - if bridge_key not in bridges: - continue - has_stat = any(e.get("TO_TYPE") == "STAT" for e in entries) - in_use = any( - (e.get("TO_TYPE") == "ON" and e.get("ACTIVE")) or e.get("TO_TYPE") == "OFF" - for e in entries - ) - if has_stat and not in_use: - remove_bridges.append(bridge_key) - for key in remove_bridges: - del bridges[key] - logger.debug("(ROUTER) STAT bridge %s removed", key) - def acl_check(self, id_val: bytes | int, acl: tuple[bool, list[tuple[int, int]]]) -> bool: """Check ID against ACL. Legacy acl_check.""" return self._router.acl_check(id_val, acl) @@ -2604,7 +1924,7 @@ class BridgeUseCases: if isinstance(target_tgid, int): target_tgid = bytes_3(target_tgid) # If target has quenched us, don't send (~1856-1859). - if _obp_target_bcsq_quenches_stream(systems_cfg, entry["SYSTEM"], dst_id_b, stream_id): + if obp_target_bcsq_quenches_stream(systems_cfg, entry["SYSTEM"], dst_id_b, stream_id): continue # If target has missed keepalives (ENHANCED_OBP), don't send (~1861-1863) if _target_system.get("ENHANCED_OBP") and (