diff --git a/repeater/config.py b/repeater/config.py index ecb9350..e4f1857 100644 --- a/repeater/config.py +++ b/repeater/config.py @@ -507,6 +507,33 @@ def _load_or_create_identity_key(path: Optional[str] = None) -> bytes: return key +# MeshCore's private LoRa sync word, as the USB/TCP modem firmware expects it. +MESHCORE_SYNC_WORD = 0x12 + + +def _checked_modem_sync_word(sync_word: int, board_config: dict, radio_type: str) -> int: + """The sync word to hand a USB/TCP modem, warning when it is not MeshCore's. + + These are the drivers that actually push the value to the radio, and a + modem on another sync word connects, answers pings and transmits, yet hears + no MeshCore traffic -- a silent failure. A radios[] entry inherits the + top-level ``radio.sync_word``, which the SX1262 driver ignores, so a stray + top-level value is harmless until a modem entry picks it up. + """ + if sync_word != MESHCORE_SYNC_WORD: + radio_id = board_config.get("_radio_id") + logger.warning( + "%s sync_word is 0x%X, not MeshCore's 0x%02X: the modem will not hear " + "MeshCore traffic. Set sync_word: 0x%02X in its radio section%s.", + f"Radio {radio_id!r} ({radio_type})" if radio_id else f"{radio_type} radio", + sync_word, + MESHCORE_SYNC_WORD, + MESHCORE_SYNC_WORD, + " (radios[] entries inherit the top-level radio.sync_word)" if radio_id else "", + ) + return sync_word + + def get_radio_for_board(board_config: dict): board_config = normalize_modem_config(board_config) @@ -752,7 +779,13 @@ def get_radio_for_board(board_config: dict): spreading_factor=int(radio_cfg.get("spreading_factor", 8)), coding_rate=int(radio_cfg.get("coding_rate", 8)), tx_power=int(radio_cfg.get("tx_power", 22)), - sync_word=_parse_int(radio_cfg.get("sync_word", 0x12), default=0x12), + sync_word=_checked_modem_sync_word( + _parse_int( + radio_cfg.get("sync_word", MESHCORE_SYNC_WORD), default=MESHCORE_SYNC_WORD + ), + board_config, + radio_type, + ), preamble_length=int(radio_cfg.get("preamble_length", 16)), lbt_enabled=bool(tcp_cfg.get("lbt_enabled", True)), lbt_max_attempts=int(tcp_cfg.get("lbt_max_attempts", 5)), @@ -794,7 +827,13 @@ def get_radio_for_board(board_config: dict): spreading_factor=int(radio_cfg.get("spreading_factor", 8)), coding_rate=int(radio_cfg.get("coding_rate", 8)), tx_power=int(radio_cfg.get("tx_power", 22)), - sync_word=_parse_int(radio_cfg.get("sync_word", 0x12), default=0x12), + sync_word=_checked_modem_sync_word( + _parse_int( + radio_cfg.get("sync_word", MESHCORE_SYNC_WORD), default=MESHCORE_SYNC_WORD + ), + board_config, + radio_type, + ), preamble_length=int(radio_cfg.get("preamble_length", 16)), lbt_enabled=bool(usb_cfg.get("lbt_enabled", True)), lbt_max_attempts=int(usb_cfg.get("lbt_max_attempts", 5)), diff --git a/tests/test_radio_config.py b/tests/test_radio_config.py index 7f9c404..9e61f46 100644 --- a/tests/test_radio_config.py +++ b/tests/test_radio_config.py @@ -283,3 +283,85 @@ def test_get_radio_for_board_kiss_omits_unset_tuning(monkeypatch): # Unset keys must not be forwarded, so the wrapper keeps its own defaults. for key in ("kiss_persistence", "kiss_slottime_ms", "tx_delay_ms", "kiss_full_duplex"): assert key not in rc + + +# ─── sync word warning for USB/TCP modems ────────────────────────────── + + +def _capture_modem(monkeypatch, radio_type): + module, cls = { + "modem_tcp": ("openhop_core.hardware.tcp_radio", "TCPLoRaRadio"), + "modem_usb": ("openhop_core.hardware.usb_radio", "USBLoRaRadio"), + }[radio_type] + pytest.importorskip(module) + captured = {} + + class _Dummy(_DummyRadio): + def __init__(self, **kwargs): + captured.update(kwargs) + + monkeypatch.setattr(f"{module}.{cls}", _Dummy) + return captured + + +def _modem_board(radio_type, radio): + section = {"host": "modem.local"} if radio_type == "modem_tcp" else {"port": "/dev/ttyACM0"} + return {"radio_type": radio_type, radio_type: section, "radio": radio} + + +@pytest.mark.parametrize("radio_type", ["modem_tcp", "modem_usb"]) +def test_modem_on_a_foreign_sync_word_is_warned_about_but_not_overridden( + monkeypatch, caplog, radio_type +): + """A modem on another sync word connects, answers pings and transmits but + hears no MeshCore traffic, so say so -- without second-guessing the value.""" + captured = _capture_modem(monkeypatch, radio_type) + radio = {**_modem_radio_cfg(), "sync_word": 0x3444} + + with caplog.at_level("WARNING", logger="Config"): + get_radio_for_board(_modem_board(radio_type, radio)) + + assert captured["sync_word"] == 0x3444 + assert "sync_word is 0x3444, not MeshCore's 0x12" in caplog.text + assert radio_type in caplog.text + + +@pytest.mark.parametrize("radio_type", ["modem_tcp", "modem_usb"]) +@pytest.mark.parametrize("sync_word", [0x12, "0x12", 18, None]) +def test_modem_on_meshcores_sync_word_is_not_warned_about( + monkeypatch, caplog, radio_type, sync_word +): + captured = _capture_modem(monkeypatch, radio_type) + radio = _modem_radio_cfg() + if sync_word is None: + radio.pop("sync_word") # the default + else: + radio["sync_word"] = sync_word + + with caplog.at_level("WARNING", logger="Config"): + get_radio_for_board(_modem_board(radio_type, radio)) + + assert captured["sync_word"] == 0x12 + assert "sync_word" not in caplog.text + + +def test_a_radios_entry_inheriting_a_stray_top_level_sync_word_is_named(monkeypatch, caplog): + """The case that bit a real node: a top-level radio.sync_word the SX1262 + driver ignores, inherited by a modem entry under radios[].""" + from repeater.config import _merge_radio_entry + + captured = _capture_modem(monkeypatch, "modem_tcp") + global_config = {"radio_type": "sx1262", "radio": {**_modem_radio_cfg(), "sync_word": 13380}} + entry = { + "id": "narrow", + "radio_type": "modem_tcp", + "radio": {"frequency": 910525000}, + "modem_tcp": {"host": "192.168.50.154"}, + } + + with caplog.at_level("WARNING", logger="Config"): + get_radio_for_board(_merge_radio_entry(global_config, entry)) + + assert captured["sync_word"] == 0x3444 + assert "Radio 'narrow' (modem_tcp) sync_word is 0x3444" in caplog.text + assert "inherit the top-level radio.sync_word" in caplog.text