From e4a4e7c03b053590145eb3db1b0f3add4649eab9 Mon Sep 17 00:00:00 2001 From: Jean-Pierre Martineau Date: Fri, 10 Jul 2026 01:14:05 -0400 Subject: [PATCH] Tuning params: real float units per firmware; drop no-op TX delays MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Checked against companion firmware (MyMesh.cpp CMD_SET_TUNING_PARAMS): rx_delay_base and airtime_factor are floats carried x1000 on the wire (constrained to 0-20 s and 0-9 respectively), so the settings now speak real units — previously the raw wire integers were shown and edited, making "2" mean 0.002. The TX delay factors added in the previous commit are removed: the companion handler reads exactly the two fields and ignores trailing bytes, so they were silent no-ops. They exist only as remote-CLI settings on repeater/room-server/sensor firmware. Airtime factor now leads the Tuning section, ahead of RX delay. Co-Authored-By: Claude Fable 5 --- meshterm/core/connection.py | 87 ++++++++++++---------------------- meshterm/core/device_config.py | 47 ++++++------------ tests/test_device_config.py | 49 ++++++++++++------- 3 files changed, 75 insertions(+), 108 deletions(-) diff --git a/meshterm/core/connection.py b/meshterm/core/connection.py index 24a1689..f3f5231 100644 --- a/meshterm/core/connection.py +++ b/meshterm/core/connection.py @@ -430,13 +430,13 @@ class Device(ABC): @abstractmethod async def get_tuning(self) -> dict: - """Return radio tuning parameters. + """Return radio tuning parameters, in their real units. Returns: - A dict with ``rx_delay`` and ``airtime_factor`` (both ints). The TX delay - factors are write-only in the companion protocol (the firmware's tuning - response carries only these two fields), so they are absent here on real - hardware; the simulator includes them for a fuller editing experience. + A dict with ``rx_delay`` (float seconds) and ``airtime_factor`` (float). + The firmware stores both as floats and moves them over the wire scaled + ×1000; implementations undo that scaling so callers only ever see the + real values. """ @abstractmethod @@ -550,21 +550,15 @@ class Device(ABC): """ @abstractmethod - async def set_tuning( - self, - rx_delay: int, - airtime_factor: int, - tx_delay_factor: int = 0, - direct_tx_delay_factor: int = 0, - ) -> None: - """Set radio tuning parameters. + async def set_tuning(self, rx_delay: float, airtime_factor: float) -> None: + """Set radio tuning parameters, in their real units. - The firmware takes all four fields in one command, so a caller changing one must - resend the others. The two TX delay factors (single bytes: the random extra delay - before relayed transmissions, and the delay before direct/zero-hop ones) cannot be - read back from real hardware — see :meth:`get_tuning` — so an unknown current - value defaults to ``0``, which is also what the underlying library always sent - before these were exposed. + The firmware takes both fields in one command, so a caller changing one must + resend the other. Values are the real ones (``rx_delay`` in seconds, 0–20; + ``airtime_factor`` a duty-cycle factor, 0–9); implementations apply the + protocol's ×1000 wire scaling. (The repeater-side TX delay factors are *not* + part of this command — companion firmware reads exactly these two fields and + ignores anything after them; those knobs are remote-CLI settings on repeaters.) """ @abstractmethod @@ -1574,9 +1568,11 @@ class MeshCoreDevice(Device): async def get_tuning(self) -> dict: # noqa: D102 - inherited docstring event = self._ok(await self._require().commands.get_tuning()) payload = getattr(event, "payload", {}) or {} + # The firmware stores floats and reports them ×1000 (rx_delay_base * 1000, + # airtime_factor * 1000); undo that so callers see the real values. return { - "rx_delay": int(payload.get("rx_delay", 0)), - "airtime_factor": int(payload.get("airtime_factor", 0)), + "rx_delay": int(payload.get("rx_delay", 0)) / 1000.0, + "airtime_factor": int(payload.get("airtime_factor", 0)) / 1000.0, } async def get_autoadd_config(self) -> Optional[int]: # noqa: D102 - inherited docstring @@ -1655,27 +1651,13 @@ class MeshCoreDevice(Device): async def set_radio(self, freq: float, bw: float, sf: int, cr: int) -> None: # noqa: D102 self._ok(await self._require().commands.set_radio(freq, bw, sf, cr)) - async def set_tuning( # noqa: D102 - inherited docstring - self, - rx_delay: int, - airtime_factor: int, - tx_delay_factor: int = 0, - direct_tx_delay_factor: int = 0, - ) -> None: - from meshcore import EventType - - # The library's set_tuning hardcodes the two TX delay bytes to zero, so the full - # CMD_SET_TUNING_PARAMS (0x15) frame is built here instead: rx_delay(4 LE) + - # airtime_factor(4 LE) + tx_delay_factor(1) + direct_tx_delay_factor(1). - data = ( - b"\x15" - + int(rx_delay).to_bytes(4, "little") - + int(airtime_factor).to_bytes(4, "little") - + int(tx_delay_factor).to_bytes(1, "little") - + int(direct_tx_delay_factor).to_bytes(1, "little") - ) + async def set_tuning(self, rx_delay: float, airtime_factor: float) -> None: # noqa: D102 + # CMD_SET_TUNING_PARAMS carries both floats ×1000; the firmware divides them + # back out (prefs.rx_delay_base = rx / 1000, prefs.airtime_factor = af / 1000). self._ok( - await self._require().commands.send(data, [EventType.OK, EventType.ERROR]) + await self._require().commands.set_tuning( + round(float(rx_delay) * 1000), round(float(airtime_factor) * 1000) + ) ) async def set_autoadd_config(self, flags: int) -> None: # noqa: D102 @@ -1811,12 +1793,7 @@ class MockDevice(Device): "radio_cr": 5, "simulated": True, } - self._tuning: dict = { - "rx_delay": 0, - "airtime_factor": 0, - "tx_delay_factor": 0, - "direct_tx_delay_factor": 0, - } + self._tuning: dict = {"rx_delay": 0.0, "airtime_factor": 0.0} self._autoadd_config = 0 self._flood_scope = "" # Simulated clock skew (seconds behind the host), so the sync-clock flow has a @@ -1982,18 +1959,12 @@ class MockDevice(Device): async def set_radio(self, freq: float, bw: float, sf: int, cr: int) -> None: # noqa: D102 self._info.update(radio_freq=freq, radio_bw=bw, radio_sf=sf, radio_cr=cr) - async def set_tuning( # noqa: D102 - inherited docstring - self, - rx_delay: int, - airtime_factor: int, - tx_delay_factor: int = 0, - direct_tx_delay_factor: int = 0, - ) -> None: + async def set_tuning(self, rx_delay: float, airtime_factor: float) -> None: # noqa: D102 + # Round-trip through the wire's ×1000 integer scaling so the simulator loses + # precision exactly where real firmware would. self._tuning = { - "rx_delay": rx_delay, - "airtime_factor": airtime_factor, - "tx_delay_factor": tx_delay_factor, - "direct_tx_delay_factor": direct_tx_delay_factor, + "rx_delay": round(float(rx_delay) * 1000) / 1000.0, + "airtime_factor": round(float(airtime_factor) * 1000) / 1000.0, } async def set_autoadd_config(self, flags: int) -> None: # noqa: D102 diff --git a/meshterm/core/device_config.py b/meshterm/core/device_config.py index 70db602..511e1b4 100644 --- a/meshterm/core/device_config.py +++ b/meshterm/core/device_config.py @@ -248,23 +248,12 @@ def _coords_apply(field_name: str) -> Callable[[Device, Any, dict], Awaitable[No def _tuning_apply(field_name: str) -> Callable[[Device, Any, dict], Awaitable[None]]: - """Build an apply that updates one tuning field, preserving the others. - - The TX delay factors cannot be read back from real hardware (see - :meth:`~meshterm.core.connection.Device.get_tuning`), so an unknown current value - falls back to ``0`` — exactly what every tuning write sent for them before they were - exposed as settings. - """ + """Build an apply that updates one tuning field, preserving the other.""" async def apply(device: Device, value: Any, snapshot: dict) -> None: - fields = { - key: snapshot.get(key) - for key in ( - "rx_delay", "airtime_factor", "tx_delay_factor", "direct_tx_delay_factor" - ) - } - fields[field_name] = value - await device.set_tuning(*(int(v or 0) for v in fields.values())) + rx = value if field_name == "rx_delay" else snapshot.get("rx_delay", 0.0) + af = value if field_name == "airtime_factor" else snapshot.get("airtime_factor", 0.0) + await device.set_tuning(float(rx or 0.0), float(af or 0.0)) return apply @@ -404,28 +393,20 @@ DEVICE_SETTINGS: list[SettingSpec] = [ "Radio", "int", minimum=1, maximum=30, max_key="max_tx_power", getter=_get("tx_power"), apply=lambda d, v, s: d.set_tx_power(v), ), - # Tuning + # Tuning. Both are firmware floats moved over the wire ×1000; the ranges are the + # firmware's own constrain() bounds. (The repeater-side TX delay factors are *not* + # here: companion firmware ignores them — they are remote-CLI settings on repeaters.) SettingSpec( - "rx_delay", "RX delay", "Receive-delay tuning", "Tuning", "int", - minimum=0, - getter=_get("rx_delay"), apply=_tuning_apply("rx_delay"), - ), - SettingSpec( - "airtime_factor", "Airtime factor", "Airtime budgeting factor", "Tuning", "int", - minimum=0, + "airtime_factor", "Airtime factor", + "Airtime budget: duty cycle = 100 / (factor + 1) %, so 0 = unlimited", "Tuning", + "float", minimum=0.0, maximum=9.0, getter=_get("airtime_factor"), apply=_tuning_apply("airtime_factor"), ), SettingSpec( - "tx_delay_factor", "TX delay", - "Random delay factor before relayed transmissions (write-only on hardware)", - "Tuning", "int", minimum=0, maximum=255, - getter=_get("tx_delay_factor"), apply=_tuning_apply("tx_delay_factor"), - ), - SettingSpec( - "direct_tx_delay_factor", "Direct TX delay", - "Delay factor before direct, zero-hop transmissions (write-only on hardware)", - "Tuning", "int", minimum=0, maximum=255, - getter=_get("direct_tx_delay_factor"), apply=_tuning_apply("direct_tx_delay_factor"), + "rx_delay", "RX delay", + "Base delay before handling received packets, seconds", "Tuning", + "float", minimum=0.0, maximum=20.0, + getter=_get("rx_delay"), apply=_tuning_apply("rx_delay"), ), # Behavior SettingSpec( diff --git a/tests/test_device_config.py b/tests/test_device_config.py index bfc9646..75ef03c 100644 --- a/tests/test_device_config.py +++ b/tests/test_device_config.py @@ -385,24 +385,39 @@ async def test_autoadd_config_round_trips() -> None: assert (await build_snapshot(device))["autoadd_config"] == 5 -async def test_apply_tx_delay_preserves_other_tuning_fields() -> None: - """Editing one tuning field resends all four without clobbering the others.""" +def test_tuning_fields_are_floats_with_firmware_ranges() -> None: + """RX delay and airtime factor take real (unscaled) float values, firmware-bounded. + + The firmware stores both as floats (rx_delay_base 0–20 s, airtime_factor 0–9) and + only the *wire* carries them ×1000 — the settings must speak the real units. + """ + assert parse_value(get_spec("rx_delay"), "0.5") == 0.5 + assert parse_value(get_spec("airtime_factor"), "2.5") == 2.5 + with pytest.raises(DeviceConfigError, match="<= 20"): + parse_value(get_spec("rx_delay"), "500") # a raw wire value must be rejected + with pytest.raises(DeviceConfigError, match="<= 9"): + parse_value(get_spec("airtime_factor"), "1000") + + +def test_tx_delay_factors_are_not_companion_settings() -> None: + """The repeater-only TX delay factors must not appear as (no-op) device settings. + + Companion firmware's CMD_SET_TUNING_PARAMS reads exactly rx_delay + airtime_factor + and ignores trailing bytes, so offering these knobs here would silently do nothing. + """ + with pytest.raises(DeviceConfigError, match="unknown setting"): + get_spec("tx_delay_factor") + with pytest.raises(DeviceConfigError, match="unknown setting"): + get_spec("direct_tx_delay_factor") + + +async def test_apply_tuning_field_preserves_the_other() -> None: + """Editing one tuning field resends the pair without clobbering its sibling.""" device = await _connected_mock() - await device.set_tuning(7, 2, 0, 0) - spec = get_spec("tx_delay_factor") - await spec.apply(device, 5, await build_snapshot(device)) - tuning = await device.get_tuning() - assert tuning == { - "rx_delay": 7, - "airtime_factor": 2, - "tx_delay_factor": 5, - "direct_tx_delay_factor": 0, - } - # And the other direction: a direct-TX-delay edit keeps the fresh tx_delay_factor. - spec = get_spec("direct_tx_delay_factor") - await spec.apply(device, 3, await build_snapshot(device)) - assert (await device.get_tuning())["tx_delay_factor"] == 5 - assert (await device.get_tuning())["direct_tx_delay_factor"] == 3 + await device.set_tuning(0.5, 2.0) + spec = get_spec("airtime_factor") + await spec.apply(device, 3.5, await build_snapshot(device)) + assert await device.get_tuning() == {"rx_delay": 0.5, "airtime_factor": 3.5} # -- clock sync -------------------------------------------------------------------