Tuning params: real float units per firmware; drop no-op TX delays
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 <noreply@anthropic.com>
This commit is contained in:
1 parent
42a22fd228
commit
e4a4e7c03b
3 files changed
+75
-108
No files matched your search
+29
-58
@@ -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
|
||||
|
||||
@@ -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(
|
||||
|
||||
+32
-17
@@ -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 -------------------------------------------------------------------
|
||||
|
||||
Reference in new issue
Block a user