diff --git a/src/adn_server/application/ident_use_cases.py b/src/adn_server/application/ident_use_cases.py index 043e404..73d5866 100644 --- a/src/adn_server/application/ident_use_cases.py +++ b/src/adn_server/application/ident_use_cases.py @@ -31,6 +31,7 @@ import time from typing import Any, Callable from ..domain import HBPF_SLT_VTERM, bytes_3, int_id +from ..domain.hbp_protocol import normalize_fixed_width_ascii from .server_voice import server_voice_rf_src_bytes logger = logging.getLogger(__name__) @@ -97,8 +98,7 @@ class IdentUseCases: for _peerid in peers: peer_cfg = peers.get(_peerid, {}) if isinstance(peer_cfg, dict) and peer_cfg.get("CALLSIGN"): - cs = peer_cfg["CALLSIGN"] - _callsign = cs.decode("utf-8", errors="replace") if isinstance(cs, bytes) else cs + _callsign = normalize_fixed_width_ascii(peer_cfg["CALLSIGN"]) break if not _callsign: logger.debug("(IDENT) %s System has no peers or no recorded callsign, skipping", system) diff --git a/src/adn_server/application/routing/downlink.py b/src/adn_server/application/routing/downlink.py index e8e90fc..0fd9e84 100644 --- a/src/adn_server/application/routing/downlink.py +++ b/src/adn_server/application/routing/downlink.py @@ -41,6 +41,8 @@ from .helpers import ( parse_dmrd_burst_fields, parse_dmrd_route_fields, peer_downlink_voice_slot, + peer_dynamic_tg_active_on_both_slots, + peer_dynamic_tg_active_on_slot, peer_is_simplex, peer_options_static_tg_slot, peer_receives_group_tgid, @@ -165,23 +167,28 @@ def peer_hangtime_voice_slots( return {int(rf_slot), int(wire_slot), int(listen_slot)} -def peer_slot_blocks_downlink( +def peer_slot_block_reason( ctx: DownlinkContext, peer_id: bytes, peer: dict[str, Any], packet: bytes, *, pkt_time: float | None = None, -) -> bool: - """P2/P3: True when slot busy or GROUP_HANGTIME blocks this downlink.""" +) -> str | None: + """P2/P3: reason this downlink is slot-busy/hangtime blocked, or None if not. + + Same logic as ``peer_slot_blocks_downlink`` (which just checks ``is not + None``) -- kept as one function so the diagnostic reason can never drift + from the actual accept/reject decision. + """ if packet[:4] != b"DMRD": - return False + return None parsed = parse_dmrd_route_fields(packet) if parsed is None: - return False + return None wire_slot, tgid, call_type = parsed if call_type not in ("group", "vcsbk"): - return False + return None pk = bytes_4(int_id(peer_id)) burst = parse_dmrd_burst_fields(packet) stream_id = burst[3] if burst is not None else b"" @@ -218,22 +225,22 @@ def peer_slot_blocks_downlink( and int_id(slot_st.get("RX_TGID", b"")) == active_tg ) if not rx_listening: - return True + return f"VTERM: other stream TG {active_tg} still active on this peer" active = per_slot.get(int(listen_slot)) if not isinstance(active, dict): for row in per_slot.values(): if not isinstance(row, dict): continue if int(row.get("tgid", 0) or 0) == int_id(dst_id): - return False - return False + return None + return None if active.get("stream_id") == stream_id: - return False + return None if int(active.get("tgid", 0) or 0) == int_id(dst_id): - return False - return True + return None + return f"VTERM: listen slot busy with TG {int(active.get('tgid', 0) or 0)}" if not ctx.per_peer_contention(): - return False + return None hang = float(ctx.sys_cfg.get("GROUP_HANGTIME", 0) or 0) now = time.time() if pkt_time is None else float(pkt_time) seed_hangtime_for_stale_ingress_voice_slots(ctx, peer_id, pkt_time=now) @@ -241,7 +248,7 @@ def peer_slot_blocks_downlink( if hang > 0: for hang_row in ctx.peer_voice_hangtime.get(pk, {}).values(): if _peer_transmit_hangtime_blocks(hang_row, incoming_tgid_b, now, hang): - return True + return "GROUP_HANGTIME: recent local transmit still hanging on this peer" peer_slots = ctx.peer_voice_slots.get(pk) voice_slots = peer_hangtime_voice_slots( peer, wire_slot, tgid, ctx.sys_cfg, peer_id=peer_id, @@ -263,8 +270,20 @@ def peer_slot_blocks_downlink( voice_slot=voice_slot, sys_cfg=ctx.sys_cfg, ): - return True - return False + return f"slot {voice_slot} busy or in GROUP_HANGTIME" + return None + + +def peer_slot_blocks_downlink( + ctx: DownlinkContext, + peer_id: bytes, + peer: dict[str, Any], + packet: bytes, + *, + pkt_time: float | None = None, +) -> bool: + """P2/P3: True when slot busy or GROUP_HANGTIME blocks this downlink.""" + return peer_slot_block_reason(ctx, peer_id, peer, packet, pkt_time=pkt_time) is not None def remap_dmrd_for_peer( @@ -656,17 +675,38 @@ def iter_downlink_voice_slots( peer: dict[str, Any], wire_slot: int, tgid: int, + sys_cfg: dict[str, Any] | None = None, + *, + peer_id: bytes | None = None, ) -> list[int]: """Voice slot(s) to deliver a group downlink to this peer. Trusts ``peer_listen_slots`` -- it already collapses simplex peers/bridges to one slot via ``peer_is_simplex``, and only returns more than one slot for a peer confirmed duplex-capable with the TG on both TS1 and TS2, which - should genuinely receive on both (see peer_listen_slots docstring). + should genuinely receive on both (see peer_listen_slots docstring). Static + vs dynamic makes no difference here -- a TG independently activated on + both slots (SINGLE=0 keyed, or SINGLE=1 exclusive session per slot) gets + the same dual-slot treatment via peer_dynamic_tg_active_on_both_slots. + + Mixed case: static on exactly one slot *and* dynamically active on the + other (e.g. TS1 static, TS2 keyed dynamically) must also deliver to both + -- checked via peer_dynamic_tg_active_on_slot on whichever slot the + static list didn't already claim. """ listen = peer_listen_slots(peer, tgid) + if len(listen) > 1: + return listen + if len(listen) == 1 and not peer_is_simplex(peer): + static_slot = listen[0] + other_slot = 2 if static_slot == 1 else 1 + if peer_dynamic_tg_active_on_slot(peer, tgid, other_slot, sys_cfg, peer_id=peer_id): + return sorted({static_slot, other_slot}) + return listen + if peer_dynamic_tg_active_on_both_slots(peer, tgid, sys_cfg, peer_id=peer_id): + return [1, 2] if listen: return listen if peer_receives_group_tgid(peer, wire_slot, tgid): - return [peer_downlink_voice_slot(peer, wire_slot, tgid)] + return [peer_downlink_voice_slot(peer, wire_slot, tgid, sys_cfg, peer_id=peer_id)] return [int(wire_slot)] diff --git a/src/adn_server/application/routing/helpers.py b/src/adn_server/application/routing/helpers.py index 8fb6c8c..cda5025 100644 --- a/src/adn_server/application/routing/helpers.py +++ b/src/adn_server/application/routing/helpers.py @@ -1208,6 +1208,27 @@ def _peer_ua_multi_store(sys_cfg: dict[str, Any]) -> dict[bytes, dict[int, set[i return store +def _peer_static_tg_blocks_slot(peer: dict[str, Any], slot: int, tgid: int) -> bool: + """Does this peer's static OPTIONS already cover ``tgid`` for this exact + slot? Simplex peers have one real RF path regardless of nominal TS1/TS2, + so any static match blocks (matches peer_receives_group_tgid's + either-slot check); duplex peers are checked per-slot, since a static + match on one slot must not block genuinely independent dynamic activity + on the *other* slot (e.g. TG static on TS2, this same peer separately + keying up the same TG on TS1).""" + from adn_server.application.report.payloads import parse_peer_options_static + + ts1, ts2 = parse_peer_options_static(peer.get("OPTIONS")) + tg = str(tgid) + if peer_is_simplex(peer): + return tg in ts1 or tg in ts2 + if int(slot) == 1: + return tg in ts1 + if int(slot) == 2: + return tg in ts2 + return False + + def register_peer_ua_multi_tg( peer: dict[str, Any], peer_id: bytes, @@ -1221,7 +1242,7 @@ def register_peer_ua_multi_tg( tgid_i = int(tgid) if not is_ua_session_tgid(tgid_i): return - if peer_receives_group_tgid(peer, slot, tgid_i): + if _peer_static_tg_blocks_slot(peer, slot, tgid_i): return pk = bytes_4(int_id(peer_id)) per_peer = _peer_ua_multi_store(sys_cfg).setdefault(pk, {}) @@ -1259,6 +1280,50 @@ def peer_owns_multi_dynamic_ua( return False +def peer_dynamic_tg_active_on_slot( + peer: dict[str, Any], + tgid: int, + slot: int, + sys_cfg: dict[str, Any] | None, + *, + peer_id: bytes | None = None, +) -> bool: + """True when a *dynamic* (non-static OPTIONS) TG is active on this + specific slot for this peer right now. Covers SINGLE=1 (independent + exclusive session per slot) and SINGLE=0 (independently keyed multi-TG + set per slot).""" + if not sys_cfg or peer_id is None: + return False + tgid_i = int(tgid) + if peer_single_mode(peer, sys_cfg): + locked = peer_single_exclusive_tgid(peer, slot, sys_cfg, peer_id=peer_id) + return locked is not None and locked == tgid_i + store = sys_cfg.get("_PEER_UA_MULTI_TGS") + if not isinstance(store, dict): + return False + per_peer = store.get(bytes_4(int_id(peer_id))) + if not isinstance(per_peer, dict): + return False + slot_set = per_peer.get(slot) + return isinstance(slot_set, set) and tgid_i in slot_set + + +def peer_dynamic_tg_active_on_both_slots( + peer: dict[str, Any], + tgid: int, + sys_cfg: dict[str, Any] | None, + *, + peer_id: bytes | None = None, +) -> bool: + """True when a *dynamic* (non-static OPTIONS) TG is active on both slot 1 + and slot 2 for this peer right now -- static-or-dynamic makes no + difference to whether a duplex peer should get the call on both slots.""" + return ( + peer_dynamic_tg_active_on_slot(peer, tgid, 1, sys_cfg, peer_id=peer_id) + and peer_dynamic_tg_active_on_slot(peer, tgid, 2, sys_cfg, peer_id=peer_id) + ) + + def register_peer_ua_session( peer: dict[str, Any], peer_id: bytes, diff --git a/src/adn_server/application/routing/subscription_table.py b/src/adn_server/application/routing/subscription_table.py index 6eb4b64..2c76473 100644 --- a/src/adn_server/application/routing/subscription_table.py +++ b/src/adn_server/application/routing/subscription_table.py @@ -51,6 +51,7 @@ from typing import Any from ...domain import bytes_3, bytes_4, int_id from ...domain.config_coerce import coerce_bool, parse_options_single from ...domain.dynamic_tg import DynamicTgEntry +from ...domain.hbp_protocol import normalize_fixed_width_ascii from ..proxy.deployment import is_proxy_inject_only logger = logging.getLogger(__name__) @@ -805,10 +806,7 @@ class SubscriptionTableMixin: parts.append(f"peers_connected={len(connected)}") if connected: def _cs(c): - v = c.get("CALLSIGN") or b"" - if isinstance(v, bytes): - return v.decode("utf8", errors="replace").strip() or "?" - return str(v).strip() or "?" + return normalize_fixed_width_ascii(c.get("CALLSIGN")) or "?" parts.append( "peers=[%s]" % ", ".join("%s/%s" % (p.get("RADIO_ID", "?"), _cs(p)) for p in connected[:10]) diff --git a/src/adn_server/infrastructure/twisted_adapters/udp_hbp.py b/src/adn_server/infrastructure/twisted_adapters/udp_hbp.py index d9e23ce..6a1e196 100644 --- a/src/adn_server/infrastructure/twisted_adapters/udp_hbp.py +++ b/src/adn_server/infrastructure/twisted_adapters/udp_hbp.py @@ -45,6 +45,7 @@ from ...application.routing.downlink import ( normalize_ua_voice_slot, peer_accepts_dmra, peer_accepts_group_dmrd_packet, + peer_slot_block_reason, remap_dmrd_for_peer, track_peer_group_dmrd, ) @@ -354,7 +355,7 @@ class HBPProtocol(DatagramProtocol): if not isinstance(peer, dict) or call_type not in ("group", "vcsbk"): self.send_peer(_peer, _packet) continue - voice_slots = iter_downlink_voice_slots(peer, wire_slot, tgid) + voice_slots = iter_downlink_voice_slots(peer, wire_slot, tgid, self._config, peer_id=_peer) if not voice_slots: voice_slots = [ peer_downlink_voice_slot( @@ -752,16 +753,28 @@ class HBPProtocol(DatagramProtocol): stream_id = route_pkt[16:20] if len(route_pkt) >= 20 else b"" return pk, stream_id - def _log_downlink_drop_once(self, peer_id: bytes, route_pkt: bytes) -> None: + def _log_downlink_drop_once( + self, peer_id: bytes, route_pkt: bytes, peer: dict[str, Any] | None, + ) -> None: key = self._downlink_drop_key(peer_id, route_pkt) if key in self._downlink_drop_logged: return self._downlink_drop_logged.add(key) + # By the time send_peer reaches this call, _peer_should_receive_dmrd + # already passed (it returns silently, without logging, on its own + # earlier check) -- so the only thing left that could have rejected + # it here is slot-busy/GROUP_HANGTIME. peer_slot_block_reason shares + # its logic with the actual accept/reject check (peer_slot_blocks_downlink), + # so this can't drift from the real decision. + reason = "unknown" + if isinstance(peer, dict): + reason = peer_slot_block_reason(self._downlink_ctx(), peer_id, peer, route_pkt) or "unknown" logger.info( - "(%s) Downlink dropped for peer %s TG %s (OPTIONS filter, slot busy, or GROUP_HANGTIME)", + "(%s) Downlink dropped for peer %s TG %s (%s)", self._system, int_id(peer_id), int_id(route_pkt[8:11]), + reason, ) def send_peer(self, _peer: bytes, _packet: bytes, *, _skip_dual_expand: bool = False) -> None: @@ -770,20 +783,34 @@ class HBPProtocol(DatagramProtocol): return peer = self._peers.get(_peer) route_pkt = _packet - if peer is not None: + if peer is not None and not _skip_dual_expand: + # Caller already remapped to the intended slot via + # iter_downlink_voice_slots' explicit per-slot loop -- re-running + # the single-slot remap here would collapse a dual-slot delivery + # (e.g. a dynamically-locked TG active on both slots) back onto + # whichever slot that single-slot decision happens to prefer. route_pkt = remap_dmrd_for_peer( _packet, peer, self._config, peer_id=_peer, ) if not self._peer_would_accept_group_dmrd( _peer, _packet if not _skip_dual_expand else route_pkt, routed=_skip_dual_expand, ): - self._log_downlink_drop_once(_peer, route_pkt) + self._log_downlink_drop_once(_peer, route_pkt, peer) return self._downlink_drop_logged.discard(self._downlink_drop_key(_peer, route_pkt)) _packet = b"".join([route_pkt[:11], _peer, route_pkt[15:]]) ctx = self._downlink_ctx() if isinstance(peer, dict): - track_peer_group_dmrd(ctx, _peer, _packet, peer, from_ingress=False) + # Pass the packet's own (already-decided) slot explicitly -- + # track_peer_group_dmrd's own voice_slot=None fallback re-derives + # it via peer_downlink_voice_slot, which always prefers slot 1 + # when a dynamic TG is locked on both slots, mistracking state + # for whichever delivery actually used slot 2. + _route_parsed = parse_dmrd_route_fields(_packet) + track_peer_group_dmrd( + ctx, _peer, _packet, peer, from_ingress=False, + voice_slot=_route_parsed[0] if _route_parsed else None, + ) self.transport.write(_packet, self._peers[_peer]["SOCKADDR"]) def _ta_buffer_enabled(self) -> bool: @@ -1367,8 +1394,23 @@ class HBPProtocol(DatagramProtocol): _pvt_targets = None if _pvt_targets is None: for _peer in self._iter_downlink_peers(_repeat_pkt): - if _peer != _peer_id: - self.send_peer(_peer, _repeat_pkt) + if _peer == _peer_id: + continue + _repeat_peer_obj = self._peers.get(_peer) + if _call_type in ("group", "vcsbk") and isinstance(_repeat_peer_obj, dict): + _repeat_voice_slots = iter_downlink_voice_slots( + _repeat_peer_obj, _slot, int_id(_dst_id), + self._config, peer_id=_peer, + ) + if len(_repeat_voice_slots) > 1: + for _repeat_vs in _repeat_voice_slots: + _repeat_remapped = remap_dmrd_for_peer( + _repeat_pkt, _repeat_peer_obj, self._config, + peer_id=_peer, voice_slot=_repeat_vs, + ) + self.send_peer(_peer, _repeat_remapped, _skip_dual_expand=True) + continue + self.send_peer(_peer, _repeat_pkt) else: for _peer in _pvt_targets: if _peer != _peer_id: diff --git a/tests/application/test_peer_single_downlink.py b/tests/application/test_peer_single_downlink.py index 654e591..7ae0534 100644 --- a/tests/application/test_peer_single_downlink.py +++ b/tests/application/test_peer_single_downlink.py @@ -320,6 +320,32 @@ def test_peer_without_static_tgs_receives_nothing_until_dynamic() -> None: ) +def test_multi_zero_static_on_one_slot_still_tracks_dynamic_on_other() -> None: + """Real bug: TG static on TS2, peer separately keys the SAME TG on TS1 -- + the static match on TS2 must not block persisting TS1 as dynamic too. + register_peer_ua_multi_tg's guard used to be slot-blind (peer_receives_group_tgid + ignores its slot argument and checks either static list), so this dynamic + activation was silently never recorded.""" + peer = {"OPTIONS": b"TS2=71442;SINGLE=0;"} + sys_cfg = _sys_cfg() + peer_id = _peer_id() + register_peer_ua_multi_tg(peer, peer_id, 1, 71442, sys_cfg) + pk = bytes_4(int_id(peer_id)) + assert 71442 in sys_cfg["_PEER_UA_MULTI_TGS"][pk][1] + + +def test_multi_zero_simplex_static_on_either_slot_still_blocks_dynamic() -> None: + """Simplex peer: one real RF path regardless of nominal TS1/TS2, so a + static match on either slot still blocks dynamic tracking (unlike + duplex, which is tracked independently per slot).""" + peer = {"OPTIONS": b"TS2=71442;SINGLE=0;", "RF_MODE": "simplex"} + sys_cfg = _sys_cfg() + peer_id = _peer_id() + register_peer_ua_multi_tg(peer, peer_id, 1, 71442, sys_cfg) + pk = bytes_4(int_id(peer_id)) + assert pk not in sys_cfg.get("_PEER_UA_MULTI_TGS", {}) + + def test_single_zero_dynamic_heard_when_both_peers_keyed() -> None: """SINGLE=0: HS1 and HS2 both keyed 7304 → each hears the other's TX on 7304.""" peer_a = {"OPTIONS": b"TS2=730,7305;SINGLE=0;"} diff --git a/tests/infrastructure/test_hbp_repeat_group_dual_slot.py b/tests/infrastructure/test_hbp_repeat_group_dual_slot.py new file mode 100644 index 0000000..b172a0d --- /dev/null +++ b/tests/infrastructure/test_hbp_repeat_group_dual_slot.py @@ -0,0 +1,180 @@ +# ADN DMR Peer Server - tests infrastructure hbp repeat group dual slot +# +# Copyright (C) 2026 Rodrigo Pérez, CE5RPY +# +############################################################################### +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software Foundation, +# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA +############################################################################### + +"""REPEAT path through real HBPProtocol: peer subscribed to a TG on both +static OPTIONS slots must get the group call on both, regardless of the +source peer's own wire slot. + +Regression: the intra-MASTER REPEAT loop (_master_datagram_received) sends +the raw incoming packet unchanged via send_peer(), never going through +iter_downlink_voice_slots()/peer_listen_slots() at all -- unlike send_peers() +(exercised by tests/infrastructure/test_peer_downlink_fanout.py), which +already trusted iter_downlink_voice_slots(). A peer-to-peer call on the same +MASTER (the common case) goes through REPEAT, not send_peers(), so the +TS1+TS2 dual-slot fix landed there without covering the path real traffic +actually uses.""" + +from __future__ import annotations + +from tests.harness.deterministic import DeterministicScenario, PacketSpec, parse_dmr_fields +from tests.support.hbp_repeat_stack import build_hbp_repeat_stack + +from adn_server.domain import bytes_4 +from adn_server.infrastructure.hbp_constants import DMRD + +_TG = 730 +_PEER_TX = bytes_4(730039110) +_PEER_RX = bytes_4(730039101) +_ADDR_TX = ("10.0.0.1", 62001) +_ADDR_RX = ("10.0.0.2", 62002) + + +def _group_spec(slot: int) -> PacketSpec: + return PacketSpec( + peer_id=int.from_bytes(_PEER_TX, "big"), + rf_src=7300391, + dst_id=_TG, + slot=slot, + call_type="group", + stream_id=0xA1B2C3D4, + payload=b"\x00" * 33, + ) + + +def test_group_call_on_both_static_slots_repeats_to_both() -> None: + stack = build_hbp_repeat_stack(talker_alias=False) + # SINGLE_MODE defaults to True in the generic test scenario config, which + # pulls in the (unrelated) dynamic-session machinery -- pin it explicitly + # to False here so this test only exercises the static-OPTIONS path. + stack.config["SYSTEMS"][stack.system_name]["SINGLE_MODE"] = False + # No RX_FREQ/TX_FREQ/SLOTS given -> derive_peer_rf_mode defaults to duplex + # (matches a real MMDVM_HS_Dual_Hat hotspot, which reported duplex here). + stack.register_peer(_PEER_TX, _ADDR_TX, options=f"TS2={_TG};") + stack.register_peer(_PEER_RX, _ADDR_RX, options=f"TS1={_TG};TS2={_TG};") + + base = _group_spec(slot=2) + stack.inject_spec(DeterministicScenario.voice_head_spec(base), _ADDR_TX) + stack.transport.clear() + stack.inject_spec( + DeterministicScenario.voice_burst_spec(base, seq=1, dtype_vseq=1), _ADDR_TX, + ) + + downlink = [pkt for pkt, addr in stack.transport.sent if addr == _ADDR_RX and pkt[:4] == DMRD] + slots = sorted(parse_dmr_fields(pkt)["slot"] for pkt in downlink) + assert slots == [1, 2], ( + "peer with TG on both static slots must be repeated on both, " + f"regardless of the source's own wire slot -- got {slots}" + ) + + +def test_group_call_on_one_static_slot_still_repeats_once() -> None: + stack = build_hbp_repeat_stack(talker_alias=False) + stack.config["SYSTEMS"][stack.system_name]["SINGLE_MODE"] = False + stack.register_peer(_PEER_TX, _ADDR_TX, options=f"TS2={_TG};") + stack.register_peer(_PEER_RX, _ADDR_RX, options=f"TS1={_TG};") + + base = _group_spec(slot=2) + stack.inject_spec(DeterministicScenario.voice_head_spec(base), _ADDR_TX) + stack.transport.clear() + stack.inject_spec( + DeterministicScenario.voice_burst_spec(base, seq=1, dtype_vseq=1), _ADDR_TX, + ) + + downlink = [pkt for pkt, addr in stack.transport.sent if addr == _ADDR_RX and pkt[:4] == DMRD] + slots = sorted(parse_dmr_fields(pkt)["slot"] for pkt in downlink) + assert slots == [1] + + +def test_group_call_on_dynamic_tg_keyed_both_slots_repeats_to_both() -> None: + """SINGLE=0 hotspot that has independently keyed the same dynamic + (non-static) TG on both slots -- static vs dynamic makes no difference to + whether a duplex peer should get the call on both.""" + stack = build_hbp_repeat_stack(talker_alias=False) + stack.config["SYSTEMS"][stack.system_name]["SINGLE_MODE"] = False + stack.register_peer(_PEER_TX, _ADDR_TX, options=f"TS2={_TG};") + stack.register_peer(_PEER_RX, _ADDR_RX, options="") + rx_pk = bytes_4(730039101) + stack.config["SYSTEMS"][stack.system_name].setdefault("_PEER_UA_MULTI_TGS", {})[rx_pk] = { + 1: {_TG}, 2: {_TG}, + } + + base = _group_spec(slot=2) + stack.inject_spec(DeterministicScenario.voice_head_spec(base), _ADDR_TX) + stack.transport.clear() + stack.inject_spec( + DeterministicScenario.voice_burst_spec(base, seq=1, dtype_vseq=1), _ADDR_TX, + ) + + downlink = [pkt for pkt, addr in stack.transport.sent if addr == _ADDR_RX and pkt[:4] == DMRD] + slots = sorted(parse_dmr_fields(pkt)["slot"] for pkt in downlink) + assert slots == [1, 2] + + +def test_group_call_static_on_one_slot_dynamic_on_other_repeats_to_both() -> None: + """Real-world case: TG static on TS1, independently keyed dynamically + (SINGLE=0) on TS2 (e.g. via peer_dynamic_tgs DB restore) -- must repeat + to both, not just the statically-configured slot.""" + stack = build_hbp_repeat_stack(talker_alias=False) + stack.config["SYSTEMS"][stack.system_name]["SINGLE_MODE"] = False + stack.register_peer(_PEER_TX, _ADDR_TX, options=f"TS2={_TG};") + stack.register_peer(_PEER_RX, _ADDR_RX, options=f"TS1={_TG};") + rx_pk = bytes_4(730039101) + stack.config["SYSTEMS"][stack.system_name].setdefault("_PEER_UA_MULTI_TGS", {})[rx_pk] = { + 2: {_TG}, + } + + base = _group_spec(slot=2) + stack.inject_spec(DeterministicScenario.voice_head_spec(base), _ADDR_TX) + stack.transport.clear() + stack.inject_spec( + DeterministicScenario.voice_burst_spec(base, seq=1, dtype_vseq=1), _ADDR_TX, + ) + + downlink = [pkt for pkt, addr in stack.transport.sent if addr == _ADDR_RX and pkt[:4] == DMRD] + slots = sorted(parse_dmr_fields(pkt)["slot"] for pkt in downlink) + assert slots == [1, 2] + + +def test_group_call_on_dynamic_tg_single_mode_stays_exclusive_to_one_slot() -> None: + """SINGLE=1 ("one exclusive dynamic TG per hotspot, either RF slot; new + local TX replaces all others" -- register_peer_ua_session) cannot have + the same TG genuinely active on both slots at once: registering it on + slot 2 clears slot 1's session. This is intentional exclusivity, not a + duplex/simultaneous-dual-slot case like static OPTIONS or SINGLE=0 -- it + must keep collapsing to one slot.""" + stack = build_hbp_repeat_stack(talker_alias=False) + stack.config["SYSTEMS"][stack.system_name]["SINGLE_MODE"] = False + stack.register_peer(_PEER_TX, _ADDR_TX, options=f"TS2={_TG};") + stack.register_peer(_PEER_RX, _ADDR_RX, options="SINGLE=1;") + rx_pk = bytes_4(730039101) + stack.config["SYSTEMS"][stack.system_name].setdefault("_PEER_UA_SESSIONS", {})[rx_pk] = { + 2: {"tgid": _TG, "expires": 0, "source": "local"}, + } + + base = _group_spec(slot=2) + stack.inject_spec(DeterministicScenario.voice_head_spec(base), _ADDR_TX) + stack.transport.clear() + stack.inject_spec( + DeterministicScenario.voice_burst_spec(base, seq=1, dtype_vseq=1), _ADDR_TX, + ) + + downlink = [pkt for pkt, addr in stack.transport.sent if addr == _ADDR_RX and pkt[:4] == DMRD] + slots = sorted(parse_dmr_fields(pkt)["slot"] for pkt in downlink) + assert slots == [2]