fix: keep YAML SINGLE_MODE on multi-peer masters for dynamic TGs

Peer OPTIONS SINGLE=1 was overwriting system-wide SINGLE_MODE, which
forced single-TG in-band deactivation and broke multi-dynamic (SINGLE=0)
and OBP bridge legs. Per-peer SINGLE still applies to downlink.

Also coerce SINGLE_MODE booleans from YAML/OPTIONS (true/True/0/1) and
apply DEFAULT_UA_TIMER / SINGLE_MODE defaults when missing from OPTIONS.
pull/11/head
Rodrigo Pérez 3 months ago
parent 8c9c3121cb
commit 7c1ec8cba0

@ -108,6 +108,7 @@ SYSTEMS:
TGID_TS1_ACL: PERMIT:ALL
TGID_TS2_ACL: PERMIT:ALL
DEFAULT_UA_TIMER: 60
# false = multiple user-activated dynamic TGs per hotspot (SINGLE=0); true = one at a time
SINGLE_MODE: false
VOICE_IDENT: false
TS1_STATIC: ""

@ -124,10 +124,17 @@ def resolve_peer_single_and_timer(
sys_cfg: dict[str, Any],
) -> tuple[bool, float]:
"""Use OPTIONS ``SINGLE``/``TIMER`` when present; else YAML ``SINGLE_MODE``/``DEFAULT_UA_TIMER``."""
from adn_server.domain.config_coerce import coerce_bool, parse_options_single
if "SINGLE" in fields:
single = str(fields["SINGLE"]).strip() == "1"
parsed_single = parse_options_single(fields["SINGLE"])
single = (
parsed_single
if parsed_single is not None
else coerce_bool(sys_cfg.get("SINGLE_MODE", False))
)
else:
single = bool(sys_cfg.get("SINGLE_MODE", False))
single = coerce_bool(sys_cfg.get("SINGLE_MODE", False))
if "TIMER" in fields:
try:
timer = float(fields["TIMER"])

@ -49,7 +49,9 @@ import time
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 ..proxy.deployment import is_proxy_inject_only
logger = logging.getLogger(__name__)
@ -140,7 +142,7 @@ class SubscriptionTableMixin:
"""Legacy make_static_tg: ensure bridge for tg exists and set system/ts to ACTIVE/OFF."""
from ..subscription.subscription_table_ops import make_static_tg_store
single_mode = bool(
single_mode = coerce_bool(
self._config.get("SYSTEMS", {}).get(system, {}).get("SINGLE_MODE", False)
)
make_static_tg_store(
@ -475,6 +477,22 @@ class SubscriptionTableMixin:
return False
return True
def _should_apply_system_single_from_options(self, system_name: str) -> bool:
"""Whether peer OPTIONS may overwrite system ``SINGLE_MODE`` (legacy single-hotspot only).
Inject-only proxy and multi-peer masters keep YAML ``SINGLE_MODE`` for bridge
timers / in-band signalling; per-peer ``SINGLE`` still applies to downlink via
``peer_single_mode()``.
"""
sys_cfg = self._config.get("SYSTEMS", {}).get(system_name, {})
if is_proxy_inject_only(self._config, system_name):
return False
try:
max_peers = int(sys_cfg.get("MAX_PEERS", 1))
except (TypeError, ValueError):
max_peers = 1
return max_peers <= 1
def _apply_master_runtime_options(self, system_name: str, _options: dict[str, Any]) -> None:
"""Apply SINGLE/TIMER/VOICE/LANG from peer OPTIONS over YAML defaults (legacy options_config).
@ -501,9 +519,11 @@ class SubscriptionTableMixin:
if "LANG" in _options and _options["LANG"] != sys_cfg.get("ANNOUNCEMENT_LANGUAGE"):
sys_cfg["ANNOUNCEMENT_LANGUAGE"] = _options["LANG"]
logger.debug("(OPTIONS) %s - Setting voice language to %s", system_name, sys_cfg["ANNOUNCEMENT_LANGUAGE"])
if "SINGLE" in _options and (sys_cfg.get("SINGLE_MODE") != bool(int(_options["SINGLE"]))):
sys_cfg["SINGLE_MODE"] = bool(int(_options["SINGLE"]))
logger.info("(OPTIONS) %s - Setting SINGLE_MODE to %s", system_name, sys_cfg["SINGLE_MODE"])
if "SINGLE" in _options and self._should_apply_system_single_from_options(system_name):
new_single = parse_options_single(_options["SINGLE"])
if new_single is not None and coerce_bool(sys_cfg.get("SINGLE_MODE", False)) != new_single:
sys_cfg["SINGLE_MODE"] = new_single
logger.info("(OPTIONS) %s - Setting SINGLE_MODE to %s", system_name, sys_cfg["SINGLE_MODE"])
# TIMER is per-peer: applied via make_static_tg for that peer's static TGs only.
def options_config_for_system(

@ -27,6 +27,7 @@ from typing import Any
from adn_server.application.ports import SubscriptionStore
from adn_server.application.routing.helpers import is_special_tg
from adn_server.domain.config_coerce import coerce_bool
from adn_server.application.subscription.routing_table_export import _legacy_to_type
from adn_server.application.subscription.trigger_bytes import dst_in_triggers
from adn_server.domain import bytes_3, int_id
@ -115,7 +116,9 @@ def apply_in_band_signalling_store(
)
sys_cfg = systems_cfg.get(system_name, {})
is_single_mode = sys_cfg.get("MODE") == "MASTER" and sys_cfg.get("SINGLE_MODE", False)
is_single_mode = sys_cfg.get("MODE") == "MASTER" and coerce_bool(
sys_cfg.get("SINGLE_MODE", False)
)
off_list = sub.triggers.off
if is_single_mode:

@ -57,7 +57,9 @@ def apply_rule_timer_store(
for sub in entries:
system_name = sub.system.value
sys_config = systems_cfg.get(system_name, {})
is_single_mode = sys_config.get("SINGLE_MODE", False)
from adn_server.domain.config_coerce import coerce_bool
is_single_mode = coerce_bool(sys_config.get("SINGLE_MODE", False))
to_type = _legacy_to_type(sub)
active = sub.is_active()
timer = float(sub.state.timer_expires_at or 0.0)

@ -0,0 +1,62 @@
# ADN DMR Peer Server - config scalar coercion
#
# Copyright (C) 2026 Rodrigo Pérez, CE5RPY <ce5rpy@qmd.cl>
#
###############################################################################
# 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
###############################################################################
"""Coerce YAML / OPTIONS scalars (bool, SINGLE) to runtime types."""
from __future__ import annotations
from typing import Any
_TRUTHY = frozenset({"1", "true", "yes", "on"})
_FALSY = frozenset({"0", "false", "no", "off", ""})
def coerce_bool(value: Any, *, default: bool = False) -> bool:
"""Parse bool from YAML bool, 0/1, or common string forms (case-insensitive)."""
if isinstance(value, bool):
return value
if value is None:
return default
if isinstance(value, (int, float)) and not isinstance(value, bool):
return bool(int(value))
if isinstance(value, str):
normalized = value.strip().lower()
if normalized in _TRUTHY:
return True
if normalized in _FALSY:
return False
return bool(value)
def parse_options_single(value: Any) -> bool | None:
"""Parse OPTIONS ``SINGLE`` (0/1/true/false). Returns None when unrecognised."""
if value is None:
return None
if isinstance(value, bool):
return value
text = str(value).strip().lower()
if text in _TRUTHY:
return True
if text in _FALSY:
return False
try:
return bool(int(text))
except (TypeError, ValueError):
return None

@ -111,6 +111,9 @@ class YamlConfigLoader:
"SELF_SERVICE": data.get("SELF_SERVICE", {}),
}
apply_proxy_env_overrides(config)
from .config_normalizer import normalize_config_scalars
normalize_config_scalars(config)
# Ensure REPORT_CLIENTS is list
if "REPORT_CLIENTS" in config["REPORTS"] and isinstance(config["REPORTS"]["REPORT_CLIENTS"], str):
config["REPORTS"]["REPORT_CLIENTS"] = [

@ -31,6 +31,29 @@ import logging
import socket
import time
from adn_server.domain.config_coerce import coerce_bool
def normalize_config_scalars(config: dict) -> None:
"""Apply YAML defaults and coerce bool-like strings before validation/runtime."""
for sys_cfg in config.get("SYSTEMS", {}).values():
if not isinstance(sys_cfg, dict):
continue
if sys_cfg.get("MODE") != "MASTER":
continue
if "SINGLE_MODE" in sys_cfg:
sys_cfg["SINGLE_MODE"] = coerce_bool(sys_cfg["SINGLE_MODE"])
else:
sys_cfg.setdefault("SINGLE_MODE", False)
timer = sys_cfg.get("DEFAULT_UA_TIMER")
if timer is None or timer == "":
sys_cfg["DEFAULT_UA_TIMER"] = 60
else:
try:
sys_cfg["DEFAULT_UA_TIMER"] = int(timer)
except (TypeError, ValueError):
pass
def expand_generator(config: dict, logger: logging.Logger) -> None:
"""Replace MASTER systems with GENERATOR > 1 by SYSTEM-0, SYSTEM-1, ... (legacy generator)."""
@ -71,6 +94,7 @@ def expand_generator(config: dict, logger: logging.Logger) -> None:
def ensure_system_runtime_config(config: dict) -> None:
"""Ensure MASTER has PEERS and PEER has STATS (legacy config.py runtime state)."""
normalize_config_scalars(config)
for name, sys_cfg in config.get("SYSTEMS", {}).items():
if sys_cfg.get("MODE") == "MASTER":
sys_cfg.setdefault("PEERS", {})

@ -0,0 +1,57 @@
# ADN DMR Peer Server - tests domain config coerce
#
# Copyright (C) 2026 Rodrigo Pérez, CE5RPY <ce5rpy@qmd.cl>
#
###############################################################################
# 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
###############################################################################
from __future__ import annotations
from adn_server.application.report.payloads import resolve_peer_single_and_timer
from adn_server.domain.config_coerce import coerce_bool, parse_options_single
def test_coerce_bool_accepts_common_string_forms() -> None:
assert coerce_bool(True) is True
assert coerce_bool(False) is False
assert coerce_bool("true") is True
assert coerce_bool("True") is True
assert coerce_bool("TRUE") is True
assert coerce_bool("false") is False
assert coerce_bool("False") is False
assert coerce_bool("0") is False
assert coerce_bool("1") is True
def test_parse_options_single_accepts_true_false_and_digits() -> None:
assert parse_options_single("0") is False
assert parse_options_single("1") is True
assert parse_options_single("true") is True
assert parse_options_single("True") is True
assert parse_options_single("false") is False
def test_resolve_peer_single_uses_yaml_when_options_omit_single() -> None:
yaml_cfg = {"SINGLE_MODE": "false", "DEFAULT_UA_TIMER": 15}
single, timer = resolve_peer_single_and_timer({"TS2_STATIC": "730444"}, yaml_cfg)
assert single is False
assert timer == 15.0
def test_resolve_peer_single_true_string_not_only_digit_one() -> None:
yaml_cfg = {"SINGLE_MODE": False, "DEFAULT_UA_TIMER": 10}
single, _ = resolve_peer_single_and_timer({"SINGLE": "true"}, yaml_cfg)
assert single is True

@ -40,10 +40,12 @@ from adn_server.domain import bytes_3, bytes_4
def _proxy_system_scenario(
*,
single_mode_yaml: bool = False,
max_peers: int = 50,
) -> DeterministicScenario:
config = DeterministicScenario().config
sys_cfg = config["SYSTEMS"]["MASTER-A"]
sys_cfg["SINGLE_MODE"] = single_mode_yaml
sys_cfg["MAX_PEERS"] = max_peers
sys_cfg["DEFAULT_UA_TIMER"] = 60
sys_cfg.pop("OPTIONS", None)
sys_cfg["TS1_STATIC"] = ""
@ -73,11 +75,21 @@ def test_rpto_single_and_timer_override_yaml() -> None:
)
sys_cfg = scenario.config["SYSTEMS"]["MASTER-A"]
assert sys_cfg["SINGLE_MODE"] is True
assert sys_cfg["SINGLE_MODE"] is False
assert scenario.routing.routing_table_for_report()["730444"][0]["TIMEOUT"] == 300.0
assert scenario.routing.routing_table_for_report()["52090"][0]["TIMEOUT"] == 300.0
def test_legacy_single_peer_master_applies_system_single_from_options() -> None:
"""GENERATOR-style master (MAX_PEERS=1) may still mirror OPTIONS into SINGLE_MODE."""
scenario = _proxy_system_scenario(single_mode_yaml=False, max_peers=1)
scenario.routing.options_config_for_system(
"MASTER-A",
peer_options=b"TS2=730444;SINGLE=1;TIMER=5;",
)
assert scenario.config["SYSTEMS"]["MASTER-A"]["SINGLE_MODE"] is True
def test_options_config_reads_connected_peer_without_yaml_options() -> None:
scenario = _proxy_system_scenario(single_mode_yaml=False)
bridges = active_routing_table(730444, (("MASTER-A", 2), ("MASTER-B", 2)))
@ -85,11 +97,11 @@ def test_options_config_reads_connected_peer_without_yaml_options() -> None:
scenario.routing.options_config_for_system("MASTER-A")
assert scenario.config["SYSTEMS"]["MASTER-A"]["SINGLE_MODE"] is True
assert scenario.config["SYSTEMS"]["MASTER-A"]["SINGLE_MODE"] is False
def test_single_mode_deactivates_other_static_tg_after_rpto() -> None:
scenario = _proxy_system_scenario(single_mode_yaml=False)
scenario = _proxy_system_scenario(single_mode_yaml=False, max_peers=1)
bridges = {
"730444": [
{

Loading…
Cancel
Save

Powered by TurnKey Linux.