feat(config): warn when a USB/TCP modem is not on MeshCore's sync word

A modem on another sync word connects, answers pings and transmits, but hears
no MeshCore traffic, so the failure is silent. Since per-radio sections
overlay the top-level ones, a radios[] modem entry inherits the top-level
radio.sync_word, which the SX1262 driver ignores; a stray value there (e.g.
13380 / 0x3444) left a real node's TCP modem deaf.

Log a warning naming the radio, the value and the inheritance, and keep
passing the configured value through so a deliberate private network is not
overridden. Only the USB and TCP modem drivers push the sync word to the
radio, so only they are checked.
This commit is contained in:
agessaman authored and Rightup committed 2026-10-02 10:57:24 +01:00
1 parent ed89cb86a4
commit 688689fbbd
2 files changed
+123 -2

No files matched your search

+41 -2
View File
@@ -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)),
+82
View File
@@ -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