Implement live radio configuration updates and add unit tests for radio handling

repeater/config_manager.py

@@ -21,6 +21,129 @@ class ConfigManager:
         self.config_path = config_path
         self.config = config
         self.daemon = daemon_instance
+
+    def _get_live_radio_snapshot(self) -> Dict[str, Any]:
+        radio_cfg = self.config.get("radio", {}) or {}
+        return {
+            "frequency": int(radio_cfg.get("frequency", 0) or 0),
+            "bandwidth": int(radio_cfg.get("bandwidth", 0) or 0),
+            "spreading_factor": int(radio_cfg.get("spreading_factor", 0) or 0),
+            "coding_rate": int(radio_cfg.get("coding_rate", 0) or 0),
+            "tx_power": int(radio_cfg.get("tx_power", 0) or 0),
+        }
+
+    def _sync_repeater_handler_radio_config(self, radio_cfg: Dict[str, Any]) -> None:
+        repeater_handler = getattr(self.daemon, "repeater_handler", None)
+        if not repeater_handler or not hasattr(repeater_handler, "radio_config"):
+            return
+
+        if not isinstance(repeater_handler.radio_config, dict):
+            repeater_handler.radio_config = {}
+
+        repeater_handler.radio_config.update(
+            {
+                key: value
+                for key, value in radio_cfg.items()
+                if value not in (None, 0)
+            }
+        )
+
+    def _kiss_transport_restart_required(self) -> bool:
+        radio = getattr(self.daemon, "radio", None)
+        kiss_cfg = self.config.get("kiss", {}) or {}
+        if radio is None or not kiss_cfg:
+            return False
+
+        runtime_port = getattr(radio, "port", None)
+        runtime_baudrate = getattr(radio, "baudrate", None)
+
+        configured_port = kiss_cfg.get("port")
+        configured_baudrate = kiss_cfg.get("baud_rate")
+
+        if configured_port and runtime_port and str(configured_port) != str(runtime_port):
+            logger.info("KISS port change detected; service restart required")
+            return True
+
+        if configured_baudrate and runtime_baudrate and int(configured_baudrate) != int(runtime_baudrate):
+            logger.info("KISS baud rate change detected; service restart required")
+            return True
+
+        return False
+
+    def _apply_live_radio_config(self) -> bool:
+        radio = getattr(self.daemon, "radio", None)
+        if radio is None:
+            logger.warning("Radio not available for live update")
+            return False
+
+        radio_cfg = self._get_live_radio_snapshot()
+
+        try:
+            if hasattr(radio, "configure_radio"):
+                if hasattr(radio, "radio_config") and isinstance(radio.radio_config, dict):
+                    radio.radio_config.update(radio_cfg)
+
+                applied = radio.configure_radio(
+                    frequency=radio_cfg["frequency"],
+                    bandwidth=radio_cfg["bandwidth"],
+                    spreading_factor=radio_cfg["spreading_factor"],
+                    coding_rate=radio_cfg["coding_rate"],
+                )
+                if not applied:
+                    logger.warning("Live radio reconfiguration failed")
+                    return False
+            else:
+                current_frequency = getattr(radio, "frequency", None)
+                current_bandwidth = getattr(radio, "bandwidth", None)
+                current_spreading_factor = getattr(radio, "spreading_factor", None)
+                current_coding_rate = getattr(radio, "coding_rate", None)
+                current_tx_power = getattr(radio, "tx_power", None)
+
+                if (
+                    current_frequency != radio_cfg["frequency"]
+                    and hasattr(radio, "set_frequency")
+                    and not radio.set_frequency(radio_cfg["frequency"])
+                ):
+                    return False
+
+                if (
+                    current_tx_power != radio_cfg["tx_power"]
+                    and hasattr(radio, "set_tx_power")
+                    and not radio.set_tx_power(radio_cfg["tx_power"])
+                ):
+                    return False
+
+                coding_rate_changed = current_coding_rate != radio_cfg["coding_rate"]
+                if coding_rate_changed:
+                    setattr(radio, "coding_rate", radio_cfg["coding_rate"])
+
+                if current_spreading_factor != radio_cfg["spreading_factor"]:
+                    if not hasattr(radio, "set_spreading_factor"):
+                        return False
+                    if not radio.set_spreading_factor(radio_cfg["spreading_factor"]):
+                        return False
+
+                if current_bandwidth != radio_cfg["bandwidth"]:
+                    if not hasattr(radio, "set_bandwidth"):
+                        return False
+                    if not radio.set_bandwidth(radio_cfg["bandwidth"]):
+                        return False
+                elif coding_rate_changed:
+                    if hasattr(radio, "set_bandwidth"):
+                        if not radio.set_bandwidth(radio_cfg["bandwidth"]):
+                            return False
+                    elif hasattr(radio, "set_spreading_factor"):
+                        if not radio.set_spreading_factor(radio_cfg["spreading_factor"]):
+                            return False
+                    else:
+                        return False
+
+            self._sync_repeater_handler_radio_config(radio_cfg)
+            logger.info("Applied live radio configuration to running daemon")
+            return True
+        except Exception as e:
+            logger.error(f"Failed to apply live radio config: {e}", exc_info=True)
+            return False
     
     def save_to_file(self) -> bool:
         """
@@ -66,6 +189,7 @@ class ConfigManager:
         
         try:
             daemon_config = self.daemon.config
+            live_update_ok = True
             
             # Default sections to update if not specified
             if sections is None:
@@ -112,8 +236,18 @@ class ConfigManager:
                     path_hash_mode = 0
                 self.daemon.dispatcher.set_default_path_hash_mode(path_hash_mode)
                 logger.info(f"Reloaded path hash mode: mesh.path_hash_mode={path_hash_mode}")
+
+            if 'radio_type' in sections:
+                logger.info("radio_type change detected; service restart required")
+                live_update_ok = False
+
+            if 'kiss' in sections and self._kiss_transport_restart_required():
+                live_update_ok = False
+
+            if 'radio' in sections:
+                live_update_ok = self._apply_live_radio_config() and live_update_ok
             
-            return True
+            return live_update_ok
             
         except Exception as e:
             logger.error(f"Failed to live update daemon config: {e}", exc_info=True)
@@ -141,7 +275,7 @@ class ConfigManager:
                 - live_updated: bool - Whether daemon was live updated
                 - error: str (optional) - Error message if failed
         """
-        result = {
+        result: Dict[str, Any] = {
             "success": False,
             "saved": False,
             "live_updated": False

tests/test_config_manager.py

@@ -0,0 +1,130 @@
+from repeater.config_manager import ConfigManager
+
+
+class _DummyRepeaterHandler:
+    def __init__(self):
+        self.radio_config = {}
+
+
+class _DummySX1262Radio:
+    def __init__(self):
+        self.frequency = 868000000
+        self.bandwidth = 125000
+        self.spreading_factor = 7
+        self.coding_rate = 5
+        self.tx_power = 14
+        self.calls = []
+
+    def set_frequency(self, frequency):
+        self.calls.append(("set_frequency", frequency))
+        self.frequency = frequency
+        return True
+
+    def set_tx_power(self, power):
+        self.calls.append(("set_tx_power", power))
+        self.tx_power = power
+        return True
+
+    def set_spreading_factor(self, spreading_factor):
+        self.calls.append(("set_spreading_factor", spreading_factor))
+        self.spreading_factor = spreading_factor
+        return True
+
+    def set_bandwidth(self, bandwidth):
+        self.calls.append(("set_bandwidth", bandwidth))
+        self.bandwidth = bandwidth
+        return True
+
+
+class _DummyKissRadio:
+    def __init__(self):
+        self.radio_config = {
+            "frequency": 869618000,
+            "bandwidth": 62500,
+            "spreading_factor": 8,
+            "coding_rate": 8,
+            "tx_power": 20,
+        }
+        self.calls = []
+
+    def configure_radio(self, **kwargs):
+        self.calls.append(("configure_radio", kwargs))
+        self.frequency = kwargs["frequency"]
+        self.bandwidth = kwargs["bandwidth"]
+        self.spreading_factor = kwargs["spreading_factor"]
+        self.coding_rate = kwargs["coding_rate"]
+        self.tx_power = self.radio_config["tx_power"]
+        return True
+
+
+class _DummyDaemon:
+    def __init__(self, config, radio):
+        self.config = {
+            "radio": dict(config.get("radio", {})),
+            "kiss": dict(config.get("kiss", {})),
+        }
+        self.radio = radio
+        self.repeater_handler = _DummyRepeaterHandler()
+        self.advert_helper = None
+        self.dispatcher = None
+
+
+def test_live_update_daemon_applies_sx1262_radio_config():
+    config = {
+        "radio": {
+            "frequency": 915000000,
+            "bandwidth": 250000,
+            "spreading_factor": 10,
+            "coding_rate": 6,
+            "tx_power": 20,
+        }
+    }
+    radio = _DummySX1262Radio()
+    daemon = _DummyDaemon(config, radio)
+    manager = ConfigManager("/tmp/config.yaml", config, daemon)
+
+    assert manager.live_update_daemon(["radio"])
+
+    assert radio.calls == [
+        ("set_frequency", 915000000),
+        ("set_tx_power", 20),
+        ("set_spreading_factor", 10),
+        ("set_bandwidth", 250000),
+    ]
+    assert radio.coding_rate == 6
+    assert daemon.repeater_handler.radio_config == config["radio"]
+
+
+def test_live_update_daemon_applies_kiss_radio_config():
+    config = {
+        "radio": {
+            "frequency": 915500000,
+            "bandwidth": 125000,
+            "spreading_factor": 9,
+            "coding_rate": 7,
+            "tx_power": 22,
+        },
+        "kiss": {
+            "port": "/dev/ttyUSB0",
+            "baud_rate": 115200,
+        },
+    }
+    radio = _DummyKissRadio()
+    daemon = _DummyDaemon(config, radio)
+    manager = ConfigManager("/tmp/config.yaml", config, daemon)
+
+    assert manager.live_update_daemon(["radio"])
+
+    assert radio.calls == [
+        (
+            "configure_radio",
+            {
+                "frequency": 915500000,
+                "bandwidth": 125000,
+                "spreading_factor": 9,
+                "coding_rate": 7,
+            },
+        )
+    ]
+    assert radio.radio_config == config["radio"]
+    assert daemon.repeater_handler.radio_config == config["radio"]