From 11d6391d41ea2c56f23f7693edf9c38e34d5af4a Mon Sep 17 00:00:00 2001 From: agessaman Date: Mon, 21 Sep 2026 15:55:46 -0700 Subject: [PATCH 1/5] feat(mqtt): report RX airtime in the status message An observer derives channel utilisation from the change in transmitted plus received airtime between two status messages, so a node that publishes only tx_air_secs cannot be metered: LetsMesh reports no utilisation for an openHop bridge, and a consumer that defaults the missing key to zero sees TX-only occupancy, which badly understates a repeater that hears far more than it repeats. utilization_percent, already published, is not that figure and cannot stand in for it. It is this node's spend against its own duty-cycle budget over a 60-second window, so at the default 3600 ms/min it reads 100% at 6% of wall clock -- a "how close am I to my TX cap" gauge, not channel occupancy. AirtimeBudgets already tracks received airtime per channel and node_stats() sums it, so this is a reporting gap rather than a missing measurement. Both counters go out as whole seconds, matching the firmware's uint32 fields and the two paths that already report them that way -- the MeshCore wire response in handler_helpers/protocol_request.py and companion stats in main.py. Worth knowing when reading the figure: RX airtime is estimated per packet from the modulation that carried it, over packets that decoded. CRC failures, preamble-only detections and traffic on other channels are not counted, so it is a floor on receiver busy time rather than a measurement of it. Firmware's counter has the same character, so the two remain comparable. --- .../data_acquisition/storage_collector.py | 3 +- tests/test_storage_collector_live_stats.py | 99 +++++++++++++++++++ 2 files changed, 101 insertions(+), 1 deletion(-) create mode 100644 tests/test_storage_collector_live_stats.py diff --git a/repeater/data_acquisition/storage_collector.py b/repeater/data_acquisition/storage_collector.py index b271af8..1241a6e 100644 --- a/repeater/data_acquisition/storage_collector.py +++ b/repeater/data_acquisition/storage_collector.py @@ -217,7 +217,8 @@ class StorageCollector: # Add airtime stats if airtime_stats: - stats["tx_air_secs"] = airtime_stats["total_airtime_ms"] / 1000 + stats["tx_air_secs"] = int(airtime_stats["total_airtime_ms"] / 1000) + stats["rx_air_secs"] = int(airtime_stats.get("total_rx_airtime_ms", 0) / 1000) stats["current_airtime_ms"] = airtime_stats["current_airtime_ms"] stats["utilization_percent"] = airtime_stats["utilization_percent"] diff --git a/tests/test_storage_collector_live_stats.py b/tests/test_storage_collector_live_stats.py new file mode 100644 index 0000000..c3cad4d --- /dev/null +++ b/tests/test_storage_collector_live_stats.py @@ -0,0 +1,99 @@ +"""Coverage for the stats the MQTT status message carries. + +``_get_live_stats`` is what a LetsMesh-style observer reads on every heartbeat, +so the figures it reports have to be the node's own. +""" + +import sys +import threading +import types +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +from repeater.data_acquisition.storage_collector import StorageCollector + +sys.modules.setdefault("psutil", types.ModuleType("psutil")) + +nacl_module = types.ModuleType("nacl") +nacl_signing_module = types.ModuleType("nacl.signing") + + +class _SigningKeyStub: + pass + + +nacl_signing_module.SigningKey = _SigningKeyStub +nacl_module.signing = nacl_signing_module + +sys.modules.setdefault("nacl", nacl_module) +sys.modules.setdefault("nacl.signing", nacl_signing_module) + + +def _make_collector() -> StorageCollector: + with ( + patch("repeater.data_acquisition.storage_collector.SQLiteHandler"), + patch("repeater.data_acquisition.storage_collector.RRDToolHandler"), + patch("repeater.data_acquisition.hardware_stats.HardwareStatsCollector"), + ): + collector = StorageCollector( + config={"storage": {"storage_dir": "/tmp/openhop_repeater_test"}} + ) + + collector._stats_stop_event.set() + if collector._stats_thread is not None: + collector._stats_thread.join(timeout=1) + collector._stats_stop_event = threading.Event() + collector._stats_thread = None + + collector.sqlite_handler = MagicMock() + return collector + + +def _handler(**overrides) -> SimpleNamespace: + handler = SimpleNamespace( + start_time=0.0, + forwarded_count=3, + rx_count=11, + airtime_stats=lambda: { + "current_airtime_ms": 120.0, + "max_airtime_ms": 3600, + "utilization_percent": 3.3, + "total_airtime_ms": 61_500.0, + "total_rx_airtime_ms": 240_900.0, + }, + get_cached_noise_floor=lambda: -118.0, + ) + for key, value in overrides.items(): + setattr(handler, key, value) + return handler + + +def test_reports_both_directions_of_airtime_in_whole_seconds(): + # An observer derives channel utilisation from the change in tx+rx airtime + # between heartbeats, so RX has to be there; firmware counts both in whole + # seconds and the other two stats paths already match that. + collector = _make_collector() + collector.repeater_handler = _handler() + + stats = collector._get_live_stats() + + assert stats["tx_air_secs"] == 61 + assert stats["rx_air_secs"] == 240 + assert isinstance(stats["tx_air_secs"], int) + assert isinstance(stats["rx_air_secs"], int) + + +def test_airtime_survives_a_handler_that_reports_no_rx(): + collector = _make_collector() + collector.repeater_handler = _handler( + airtime_stats=lambda: { + "current_airtime_ms": 0.0, + "utilization_percent": 0.0, + "total_airtime_ms": 2_000.0, + } + ) + + stats = collector._get_live_stats() + + assert stats["tx_air_secs"] == 2 + assert stats["rx_air_secs"] == 0 From a103465baa536c3a074c63a5fed811bb773c71e5 Mon Sep 17 00:00:00 2001 From: agessaman Date: Mon, 21 Sep 2026 15:56:04 -0700 Subject: [PATCH 2/5] fix(mqtt): report the default radio's noise floor, not the newest sample Per-radio sampling made the status message describe the wrong radio on a Fabric node. _get_live_stats read the newest row in the noise_floor table with no radio filter, which was right while only the default radio was ever stored. _record_noise_floor_async now stores every radio's sample, default first, so the newest row is the *other* radio's -- a bridge published its backhaul's noise floor as the node's. That is the opposite of the contract the sampler documents: _sampling_radios puts the default first precisely because /stats, the MQTT status message and companion stats all report that radio, and record_noise_floor honours it by publishing only the default's sample. Reading the table directly went around the rule. Read the engine's cached reading instead, which is default-only by construction, is what /stats reports, and agrees with the collector's own cached value on the WebSocket path -- all three now say the same thing. It also drops a SQLite query per heartbeat, which matters on an SD-card-bound Pi. One behaviour change: the old query looked back 30 minutes, so a node restarted within that window published a pre-restart reading as current. Now the key is omitted until the first sample lands, 30 seconds in. The field was already optional, and a reading from before the restart was not a current one. Single-radio nodes are unaffected: their sole sample is the default's either way. --- .../data_acquisition/storage_collector.py | 18 ++++++---- tests/test_storage_collector_live_stats.py | 36 ++++++++++++++++++- 2 files changed, 46 insertions(+), 8 deletions(-) diff --git a/repeater/data_acquisition/storage_collector.py b/repeater/data_acquisition/storage_collector.py index 1241a6e..95046e8 100644 --- a/repeater/data_acquisition/storage_collector.py +++ b/repeater/data_acquisition/storage_collector.py @@ -198,14 +198,18 @@ class StorageCollector: # a bridge's stored history covers every radio it transmits on. airtime_stats = _node_airtime_stats(self.repeater_handler) or {} - # Get latest noise floor from database + # The default radio's last reading, from memory rather than the newest + # row in the table. Every radio's sample is stored now and the default is + # sampled first, so the newest row on a Fabric node is the *other* + # radio's -- the opposite of the figure a status message reports. The + # engine holds the default radio's reading, which is what /stats reads. noise_floor = None - try: - recent_noise = self.sqlite_handler.get_noise_floor_history(hours=0.5, limit=1) - if recent_noise and len(recent_noise) > 0: - noise_floor = recent_noise[-1].get("noise_floor_dbm") - except Exception as e: - logger.debug(f"Could not fetch noise floor: {e}") + cached_noise_floor = getattr(self.repeater_handler, "get_cached_noise_floor", None) + if callable(cached_noise_floor): + try: + noise_floor = cached_noise_floor() + except Exception as e: + logger.debug(f"Could not read cached noise floor: {e}") stats = { "uptime_secs": uptime_secs, diff --git a/tests/test_storage_collector_live_stats.py b/tests/test_storage_collector_live_stats.py index c3cad4d..60845f9 100644 --- a/tests/test_storage_collector_live_stats.py +++ b/tests/test_storage_collector_live_stats.py @@ -1,7 +1,8 @@ """Coverage for the stats the MQTT status message carries. ``_get_live_stats`` is what a LetsMesh-style observer reads on every heartbeat, -so the figures it reports have to be the node's own. +so the figures it reports have to be the node's own: both directions of airtime, +and the default radio's noise floor rather than whichever radio sampled last. """ import sys @@ -97,3 +98,36 @@ def test_airtime_survives_a_handler_that_reports_no_rx(): assert stats["tx_air_secs"] == 2 assert stats["rx_air_secs"] == 0 + + +def test_noise_floor_is_the_default_radios_not_the_newest_sample(): + # Every radio's sample lands in the table and the default is sampled first, + # so the newest row on a Fabric node belongs to the other radio. Reading it + # would report the backhaul's noise floor as the node's. + collector = _make_collector() + collector.repeater_handler = _handler(get_cached_noise_floor=lambda: -118.0) + collector.sqlite_handler.get_noise_floor_history.return_value = [ + {"timestamp": 2.0, "noise_floor_dbm": -92.0, "radio_id": "link"} + ] + + stats = collector._get_live_stats() + + assert stats["noise_floor"] == -118.0 + collector.sqlite_handler.get_noise_floor_history.assert_not_called() + + +def test_noise_floor_is_omitted_until_a_radio_has_been_sampled(): + collector = _make_collector() + collector.repeater_handler = _handler(get_cached_noise_floor=lambda: None) + + assert "noise_floor" not in collector._get_live_stats() + + +def test_noise_floor_is_omitted_when_the_handler_cannot_report_one(): + collector = _make_collector() + handler = _handler() + del handler.get_cached_noise_floor + + collector.repeater_handler = handler + + assert "noise_floor" not in collector._get_live_stats() From ab35e017ef02b175cb874e18e0b0af878b8b7fde Mon Sep 17 00:00:00 2001 From: agessaman Date: Mon, 21 Sep 2026 16:04:36 -0700 Subject: [PATCH 3/5] fix(mqtt): report real receive errors in the status message stats.errors was a literal 0 on every heartbeat, so an observer could not tell a quiet node from a deaf one. It was hardcoded twice: once where the collector builds the live stats, and again in publish_status, where the defaults were spread *after* live_stats and so overrode anything the provider reported. The second one would have pinned the field to 0 even once the first was fixed. The node does count receive errors -- the radio's CRC failure counter, already sampled on a timer, persisted as per-radio deltas and reported on the MeshCore wire response as n_recv_errors. Report that, and let the defaults be defaults. get_crc_error_count() on the engine is the single reader of the hardware counter, replacing the inline read in get_stats so /stats and the status message cannot drift apart. It reports the default radio's count rather than the node's sum, matching the wire response, /stats, the noise floor beside it in the same payload, and record_crc_errors' rule that only the default radio's delta is published. A bridge's link radio errors are stored against their radio id and queryable from the crc_errors table; they are not in this scalar. queue_len stays 0. The Python repeater has no TX queue to measure, which is what it has always reported -- the ordering fix just stops that being a coincidence. --- repeater/data_acquisition/mqtt_handler.py | 5 +- .../data_acquisition/storage_collector.py | 14 ++- repeater/engine.py | 15 ++- tests/test_storage_collector_live_stats.py | 93 ++++++++++++++++++- 4 files changed, 122 insertions(+), 5 deletions(-) diff --git a/repeater/data_acquisition/mqtt_handler.py b/repeater/data_acquisition/mqtt_handler.py index 9979867..549c70d 100644 --- a/repeater/data_acquisition/mqtt_handler.py +++ b/repeater/data_acquisition/mqtt_handler.py @@ -1005,7 +1005,10 @@ class MeshCoreToMqttPusher: "radio": radio_config or self.radio_config, "client_version": f"openhop_repeater/{self.app_version}", "repeat": repeat_state, - "stats": {**live_stats, "errors": 0, "queue_len": 0, **(extra_stats or {})}, + # Defaults first: a provider that reports real figures must win over + # them. Spread after live_stats, as they were, they pinned errors to + # 0 no matter what the node had counted. + "stats": {"errors": 0, "queue_len": 0, **live_stats, **(extra_stats or {})}, } radios = self._radio_map() diff --git a/repeater/data_acquisition/storage_collector.py b/repeater/data_acquisition/storage_collector.py index 95046e8..33c53b2 100644 --- a/repeater/data_acquisition/storage_collector.py +++ b/repeater/data_acquisition/storage_collector.py @@ -211,11 +211,23 @@ class StorageCollector: except Exception as e: logger.debug(f"Could not read cached noise floor: {e}") + # Receive errors: the radio's CRC failure counter, which is the only + # error the node actually counts. It was published as a literal 0 for as + # long as the field has existed, so an observer could not tell a quiet + # node from a deaf one. + errors = 0 + crc_error_count = getattr(self.repeater_handler, "get_crc_error_count", None) + if callable(crc_error_count): + try: + errors = int(crc_error_count() or 0) + except Exception as e: + logger.debug(f"Could not read CRC error count: {e}") + stats = { "uptime_secs": uptime_secs, "packets_sent": self.repeater_handler.forwarded_count, "packets_received": self.repeater_handler.rx_count, - "errors": 0, + "errors": errors, "queue_len": 0, # N/A for Python repeater } diff --git a/repeater/engine.py b/repeater/engine.py index dcfe0e0..1c3f8d3 100644 --- a/repeater/engine.py +++ b/repeater/engine.py @@ -2454,6 +2454,18 @@ class RepeaterHandler(BaseHandler): """Last sample per radio id. Empty on a single-radio node.""" return dict(self._cached_noise_floor_by_radio) + def get_crc_error_count(self) -> int: + """The default radio's cumulative CRC error count, straight from hardware. + + The default radio's, not the node's sum, because this is the figure the + MeshCore wire response already reports as ``n_recv_errors`` and /stats + as ``crc_error_count``. Per-radio counts are stored and queryable from + the crc_errors table; what a single scalar means here has to match what + the other readers of it mean. + """ + radio = self.dispatcher.radio if self.dispatcher else None + return int(getattr(radio, "crc_error_count", 0) or 0) if radio else 0 + def get_stats(self) -> dict: runtime_config = normalize_modem_config(self.config, warn=False) redact_modem_tokens_in_place(runtime_config) @@ -2478,8 +2490,7 @@ class RepeaterHandler(BaseHandler): noise_floor_dbm = self.get_cached_noise_floor() # Get CRC error count from radio hardware - radio = self.dispatcher.radio if self.dispatcher else None - crc_error_count = getattr(radio, "crc_error_count", 0) if radio else 0 + crc_error_count = self.get_crc_error_count() # Get neighbors from database neighbors = self.storage.get_neighbors() if self.storage else {} diff --git a/tests/test_storage_collector_live_stats.py b/tests/test_storage_collector_live_stats.py index 60845f9..6151f92 100644 --- a/tests/test_storage_collector_live_stats.py +++ b/tests/test_storage_collector_live_stats.py @@ -2,15 +2,19 @@ ``_get_live_stats`` is what a LetsMesh-style observer reads on every heartbeat, so the figures it reports have to be the node's own: both directions of airtime, -and the default radio's noise floor rather than whichever radio sampled last. +the default radio's noise floor rather than whichever radio sampled last, and a +receive-error count rather than a literal zero. The last test covers the other +half of that -- publish_status must let those figures through. """ +import json import sys import threading import types from types import SimpleNamespace from unittest.mock import MagicMock, patch +from repeater.data_acquisition.mqtt_handler import MeshCoreToMqttPusher from repeater.data_acquisition.storage_collector import StorageCollector sys.modules.setdefault("psutil", types.ModuleType("psutil")) @@ -63,6 +67,7 @@ def _handler(**overrides) -> SimpleNamespace: "total_rx_airtime_ms": 240_900.0, }, get_cached_noise_floor=lambda: -118.0, + get_crc_error_count=lambda: 47, ) for key, value in overrides.items(): setattr(handler, key, value) @@ -131,3 +136,89 @@ def test_noise_floor_is_omitted_when_the_handler_cannot_report_one(): collector.repeater_handler = handler assert "noise_floor" not in collector._get_live_stats() + + +def test_errors_reports_the_radios_crc_failures(): + # Published as a literal 0 for as long as the field existed, which made a + # deaf node look like a quiet one. + collector = _make_collector() + collector.repeater_handler = _handler() + + assert collector._get_live_stats()["errors"] == 47 + + +def test_errors_falls_back_to_zero_when_the_handler_cannot_count_them(): + collector = _make_collector() + handler = _handler() + del handler.get_crc_error_count + + collector.repeater_handler = handler + + assert collector._get_live_stats()["errors"] == 0 + + +class _FakeIdentity: + def __init__(self, public_key_hex: str): + self._pk = bytes.fromhex(public_key_hex) + + def get_public_key(self) -> bytes: + return self._pk + + +def _publish_status(stats_provider) -> dict: + config = { + "repeater": {"node_name": "test-node", "mode": "forward"}, + "radio": { + "frequency": 869618000, + "bandwidth": 62500, + "spreading_factor": 8, + "coding_rate": 8, + }, + "mqtt_brokers": { + "iata_code": "LAX", + "status_interval": 0, + "brokers": [ + { + "name": "test-broker", + "enabled": True, + "host": "broker.example", + "port": 1883, + "transport": "tcp", + "format": "letsmesh", + "use_jwt_auth": False, + "tls": {"enabled": False, "insecure": False}, + } + ], + }, + } + pusher = MeshCoreToMqttPusher( + local_identity=_FakeIdentity("AB" * 32), + config=config, + stats_provider=stats_provider, + ) + conn = pusher.connections[0] + captured = [] + conn._running = True + conn.client = MagicMock() + conn.client.publish = lambda topic, payload, retain=False, qos=0: captured.append(payload) + + pusher.publish_status(state="online") + + assert len(captured) == 1 + return json.loads(captured[0]) + + +def test_publish_status_does_not_overwrite_a_real_error_count(): + # The defaults used to be spread *after* live_stats, so a node that counted + # errors still published 0. + status = _publish_status(lambda: {"uptime_secs": 9, "errors": 47, "queue_len": 2}) + + assert status["stats"]["errors"] == 47 + assert status["stats"]["queue_len"] == 2 + + +def test_publish_status_still_defaults_the_fields_a_provider_omits(): + status = _publish_status(lambda: {"uptime_secs": 9}) + + assert status["stats"]["errors"] == 0 + assert status["stats"]["queue_len"] == 0 From 566498ce7ddc5d98876797e02f16eca07644c31a Mon Sep 17 00:00:00 2001 From: agessaman Date: Mon, 21 Sep 2026 16:14:58 -0700 Subject: [PATCH 4/5] feat(mqtt): report each radio's own telemetry in the status radio map The figures in stats are one set for the whole node, so on a two-radio bridge they cannot say which side is carrying the traffic or which side has gone deaf. Airtime is summed across channels and utilisation is the highest of them, which is right for a node-level number and useless for telling the local radio from the backhaul. Noise floor and errors are the default radio's outright, so the link radio is simply unreported. Every radio now carries its own tx_air_secs, rx_air_secs, current_airtime_ms, utilization_percent, noise_floor and errors in its radios[] entry, under the same field names stats uses so there is no second vocabulary to learn. The node already computes all three -- AirtimeBudgets.per_radio_stats, the per-radio noise floor cache, and the radios' CRC counters -- and /stats has reported the first two as airtime_radios and noise_floor_radios since per-radio analytics landed. This is the observer feed catching up. Additive by construction: radios[] is published only where it already was, on a node with more than one band, and a consumer that ignores it reads exactly what it read before. Telemetry is merged into the map rather than gating it, and each source contributes on its own, so a radio the fabric cannot hand back keeps its air settings and still attributes its packets. Two figures worth knowing before comparing them: - stats.errors is the *default radio's* CRC count, matching /stats and the MeshCore wire response, so on a bridge it does not equal the sum of radios[].errors. Making the node scalar a sum would change three readers that have always meant one radio, which is a decision of its own. - tx/rx_air_secs truncate to whole seconds per radio and again for the node, so the per-radio values can sum to one second less than the node's. --- repeater/data_acquisition/mqtt_handler.py | 24 ++- .../data_acquisition/storage_collector.py | 62 ++++++ repeater/engine.py | 23 +++ tests/test_storage_collector_live_stats.py | 192 +++++++++++++++++- 4 files changed, 297 insertions(+), 4 deletions(-) diff --git a/repeater/data_acquisition/mqtt_handler.py b/repeater/data_acquisition/mqtt_handler.py index 549c70d..222a028 100644 --- a/repeater/data_acquisition/mqtt_handler.py +++ b/repeater/data_acquisition/mqtt_handler.py @@ -642,6 +642,7 @@ class MeshCoreToMqttPusher: config: dict, jwt_expiry_minutes: int = 10, stats_provider: Optional[Callable[[], dict]] = None, + radio_stats_provider: Optional[Callable[[], dict]] = None, ): self.config = config # Store local identity and get public key @@ -664,6 +665,7 @@ class MeshCoreToMqttPusher: self.app_version = __version__ self.radio_config = node_info["radio_config"] self.stats_provider = stats_provider + self.radio_stats_provider = radio_stats_provider self._status_task = None self._running = False self._shutdown_requested = False @@ -1052,6 +1054,13 @@ class MeshCoreToMqttPusher: Rebuilt per status publish rather than cached at construction, because air settings can be changed from the web UI while the node runs. + + Each entry also carries that radio's own telemetry where the node can + report it -- the figures in ``stats`` are one set for the whole node, so + on a bridge they cannot say which side is busy or which side has gone + deaf. An observer that ignores ``radios`` reads exactly what it read + before, and telemetry is merged in rather than replacing the map, so a + radio the node has no figures for still attributes its packets. """ from ..config import build_radio_status_entries @@ -1060,7 +1069,20 @@ class MeshCoreToMqttPusher: except Exception as exc: # pragma: no cover - reporting must not break status logger.debug(f"Could not build radio map for status: {exc}") return [] - return entries if len(entries) > 1 else [] + if len(entries) <= 1: + return [] + telemetry = self._radio_telemetry() + return [{**entry, **telemetry.get(entry["id"], {})} for entry in entries] + + def _radio_telemetry(self) -> dict: + """``{radio_id: {...}}`` from the provider, or empty when there is none.""" + if not self.radio_stats_provider: + return {} + try: + return self.radio_stats_provider() or {} + except Exception as exc: # pragma: no cover - reporting must not break status + logger.debug(f"Could not read per-radio telemetry for status: {exc}") + return {} def publish(self, subtopic: str, payload: dict, retain: bool = False, qos: int = 0): """Publish message to all connected brokers""" diff --git a/repeater/data_acquisition/storage_collector.py b/repeater/data_acquisition/storage_collector.py index 33c53b2..9ab4372 100644 --- a/repeater/data_acquisition/storage_collector.py +++ b/repeater/data_acquisition/storage_collector.py @@ -98,6 +98,7 @@ class StorageCollector: local_identity=local_identity, config=config, stats_provider=self._get_live_stats, + radio_stats_provider=self._get_radio_stats, ) self.mqtt_handler.connect() @@ -244,6 +245,67 @@ class StorageCollector: return stats + def _get_radio_stats(self) -> dict: + """``{radio_id: {...}}`` for the status message's radio map. + + Empty on a single-radio node, whose one radio is already what every + figure in ``stats`` describes. On a bridge the node-wide figures cannot + say which side is carrying the traffic or which side has gone deaf, so + each radio reports its own, under the same field names ``stats`` uses. + + Each source contributes independently: a radio with a noise floor but no + airtime budget still reports the noise floor. Airtime is whole seconds + here too, matching the node-level counters. + """ + if not self.repeater_handler: + return {} + + radios: dict = {} + + def entry(radio_id) -> dict: + return radios.setdefault(str(radio_id), {}) + + for source, apply_to in ( + ("airtime_stats_by_radio", self._apply_radio_airtime), + ("get_cached_noise_floor_by_radio", self._apply_radio_noise_floor), + ("get_crc_error_count_by_radio", self._apply_radio_errors), + ): + getter = getattr(self.repeater_handler, source, None) + if not callable(getter): + continue + try: + apply_to(getter(), entry) + except Exception as e: + logger.debug(f"Could not read {source} for the status radio map: {e}") + + return radios + + @staticmethod + def _apply_radio_airtime(per_radio: list, entry) -> None: + for radio in per_radio or []: + radio_id = radio.get("radio_id") + if radio_id is None: + continue + entry(radio_id).update( + { + "tx_air_secs": int(radio.get("total_airtime_ms", 0) / 1000), + "rx_air_secs": int(radio.get("total_rx_airtime_ms", 0) / 1000), + "current_airtime_ms": radio.get("current_airtime_ms", 0), + "utilization_percent": radio.get("utilization_percent", 0), + } + ) + + @staticmethod + def _apply_radio_noise_floor(by_radio: dict, entry) -> None: + for radio_id, noise_floor_dbm in (by_radio or {}).items(): + if noise_floor_dbm is not None: + entry(radio_id)["noise_floor"] = noise_floor_dbm + + @staticmethod + def _apply_radio_errors(by_radio: dict, entry) -> None: + for radio_id, count in (by_radio or {}).items(): + entry(radio_id)["errors"] = int(count) + def record_packet( self, packet_record: dict, diff --git a/repeater/engine.py b/repeater/engine.py index 1c3f8d3..e837cad 100644 --- a/repeater/engine.py +++ b/repeater/engine.py @@ -2144,6 +2144,14 @@ class RepeaterHandler(BaseHandler): """ return self.airtime_budgets.node_stats() + def airtime_stats_by_radio(self) -> list: + """``[{radio_id, ...}]`` per radio, empty on a single-radio node. + + Radios sharing a channel report the same figures because they are the + same budget, which is the statement rather than a duplication. + """ + return self.airtime_budgets.per_radio_stats() + def _egress_can_transmit( self, packet: Packet, tx_radio_ids: Optional[Tuple[str, ...]] ) -> Tuple[bool, float]: @@ -2466,6 +2474,21 @@ class RepeaterHandler(BaseHandler): radio = self.dispatcher.radio if self.dispatcher else None return int(getattr(radio, "crc_error_count", 0) or 0) if radio else 0 + def get_crc_error_count_by_radio(self) -> Dict[str, int]: + """Each radio's cumulative CRC error count. Empty on a single-radio node. + + Read from the hardware counters rather than the sampler's baselines, so + a status message is not pinned to whenever the CRC tick last ran. A + radio the fabric cannot hand back is omitted rather than reported as + zero: absent and error-free are not the same claim. + """ + counts = {} + for radio_id, radio in self._sampling_radios(): + if radio_id is None or radio is None: + continue + counts[radio_id] = int(getattr(radio, "crc_error_count", 0) or 0) + return counts + def get_stats(self) -> dict: runtime_config = normalize_modem_config(self.config, warn=False) redact_modem_tokens_in_place(runtime_config) diff --git a/tests/test_storage_collector_live_stats.py b/tests/test_storage_collector_live_stats.py index 6151f92..dda4c92 100644 --- a/tests/test_storage_collector_live_stats.py +++ b/tests/test_storage_collector_live_stats.py @@ -3,8 +3,9 @@ ``_get_live_stats`` is what a LetsMesh-style observer reads on every heartbeat, so the figures it reports have to be the node's own: both directions of airtime, the default radio's noise floor rather than whichever radio sampled last, and a -receive-error count rather than a literal zero. The last test covers the other -half of that -- publish_status must let those figures through. +receive-error count rather than a literal zero. The last tests cover the other +half of that -- publish_status must let those figures through, and on a bridge +each radio must report its own beside the node's. """ import json @@ -165,7 +166,31 @@ class _FakeIdentity: return self._pk -def _publish_status(stats_provider) -> dict: +LOCAL_RADIO = { + "id": "local", + "radio_type": "sx1262", + "radio": { + "frequency": 869618000, + "bandwidth": 62500, + "spreading_factor": 8, + "coding_rate": 8, + "preamble_length": 32, + }, +} +LINK_RADIO = { + "id": "link", + "radio_type": "sx1262", + "radio": { + "frequency": 864200000, + "bandwidth": 62500, + "spreading_factor": 11, + "coding_rate": 8, + "preamble_length": 32, + }, +} + + +def _publish_status(stats_provider, radios=None, radio_stats_provider=None) -> dict: config = { "repeater": {"node_name": "test-node", "mode": "forward"}, "radio": { @@ -191,10 +216,14 @@ def _publish_status(stats_provider) -> dict: ], }, } + if radios: + config["radios"] = radios + config["fabric"] = {"tx_mode": "bridge", "default_radio": "local"} pusher = MeshCoreToMqttPusher( local_identity=_FakeIdentity("AB" * 32), config=config, stats_provider=stats_provider, + radio_stats_provider=radio_stats_provider, ) conn = pusher.connections[0] captured = [] @@ -222,3 +251,160 @@ def test_publish_status_still_defaults_the_fields_a_provider_omits(): assert status["stats"]["errors"] == 0 assert status["stats"]["queue_len"] == 0 + + +PER_RADIO_AIRTIME = [ + { + "radio_id": "local", + "current_airtime_ms": 388.0, + "max_airtime_ms": 3600, + "utilization_percent": 10.8, + "total_airtime_ms": 2_947_200.0, + "total_rx_airtime_ms": 8_610_400.0, + }, + { + "radio_id": "link", + "current_airtime_ms": 24.0, + "max_airtime_ms": 3600, + "utilization_percent": 0.7, + "total_airtime_ms": 235_400.0, + "total_rx_airtime_ms": 433_700.0, + }, +] + + +def _bridge_handler(**overrides) -> SimpleNamespace: + per_radio = { + "airtime_stats_by_radio": lambda: PER_RADIO_AIRTIME, + "get_cached_noise_floor_by_radio": lambda: {"local": -118.5, "link": -126.0}, + "get_crc_error_count_by_radio": lambda: {"local": 2401, "link": 82}, + } + return _handler(**{**per_radio, **overrides}) + + +def test_radio_stats_carry_each_radios_own_figures(): + collector = _make_collector() + collector.repeater_handler = _bridge_handler() + + assert collector._get_radio_stats() == { + "local": { + "tx_air_secs": 2947, + "rx_air_secs": 8610, + "current_airtime_ms": 388.0, + "utilization_percent": 10.8, + "noise_floor": -118.5, + "errors": 2401, + }, + "link": { + "tx_air_secs": 235, + "rx_air_secs": 433, + "current_airtime_ms": 24.0, + "utilization_percent": 0.7, + "noise_floor": -126.0, + "errors": 82, + }, + } + + +def test_radio_stats_are_empty_on_a_single_radio_node(): + # Every figure in ``stats`` already describes that one radio. + collector = _make_collector() + collector.repeater_handler = _handler() + + assert collector._get_radio_stats() == {} + + +def test_radio_stats_report_the_sources_a_node_does_have(): + # A radio the fabric cannot hand back has no CRC count, which must not cost + # the radios that do have one their airtime and noise floor. + collector = _make_collector() + collector.repeater_handler = _bridge_handler( + get_crc_error_count_by_radio=lambda: {"local": 2401} + ) + + radios = collector._get_radio_stats() + + assert radios["local"]["errors"] == 2401 + assert "errors" not in radios["link"] + assert radios["link"]["tx_air_secs"] == 235 + assert radios["link"]["noise_floor"] == -126.0 + + +def test_one_unreadable_source_does_not_cost_the_others(): + def boom(): + raise RuntimeError("fabric went away") + + collector = _make_collector() + collector.repeater_handler = _bridge_handler(get_cached_noise_floor_by_radio=boom) + + radios = collector._get_radio_stats() + + assert "noise_floor" not in radios["local"] + assert radios["local"]["tx_air_secs"] == 2947 + assert radios["local"]["errors"] == 2401 + + +def test_status_radio_map_carries_telemetry_beside_the_air_settings(): + status = _publish_status( + lambda: {"uptime_secs": 9}, + radios=[LOCAL_RADIO, LINK_RADIO], + radio_stats_provider=lambda: { + "local": {"tx_air_secs": 2947, "noise_floor": -118.5, "errors": 2401}, + "link": {"tx_air_secs": 235, "noise_floor": -126.0, "errors": 82}, + }, + ) + + assert status["radios"] == [ + { + "id": "local", + "radio": "869.618,62.5,8,8", + "tx_air_secs": 2947, + "noise_floor": -118.5, + "errors": 2401, + }, + { + "id": "link", + "radio": "864.2,62.5,11,8", + "tx_air_secs": 235, + "noise_floor": -126.0, + "errors": 82, + }, + ] + + +def test_a_radio_with_no_telemetry_still_attributes_its_packets(): + # The map's first job is saying which band an id is. Telemetry is merged in, + # never a precondition for the entry. + status = _publish_status( + lambda: {"uptime_secs": 9}, + radios=[LOCAL_RADIO, LINK_RADIO], + radio_stats_provider=lambda: {"local": {"errors": 2401}}, + ) + + assert status["radios"] == [ + {"id": "local", "radio": "869.618,62.5,8,8", "errors": 2401}, + {"id": "link", "radio": "864.2,62.5,11,8"}, + ] + + +def test_an_unreadable_provider_leaves_the_map_intact(): + def boom(): + raise RuntimeError("no handler") + + status = _publish_status( + lambda: {"uptime_secs": 9}, radios=[LOCAL_RADIO, LINK_RADIO], radio_stats_provider=boom + ) + + assert status["radios"] == [ + {"id": "local", "radio": "869.618,62.5,8,8"}, + {"id": "link", "radio": "864.2,62.5,11,8"}, + ] + + +def test_a_single_radio_node_publishes_no_radio_map(): + status = _publish_status( + lambda: {"uptime_secs": 9}, + radio_stats_provider=lambda: {"local": {"errors": 2401}}, + ) + + assert "radios" not in status From a439c2427c5d199d083dc1c4c350a97130362184 Mon Sep 17 00:00:00 2001 From: agessaman Date: Mon, 21 Sep 2026 16:31:30 -0700 Subject: [PATCH 5/5] fix(stats): count CRC errors for the node, not the default radio The three readers of the CRC scalar all answered for one receiver. On a bridge that is half the node: a backhaul going deaf moved no number an operator or an observer could see, while the sum of radios[].errors in the same status message already said so. All three now read RepeaterHandler.get_crc_error_count(), which sums the radios -- stats.errors in the MQTT status message, crc_error_count in /stats, and n_recv_errors in the MeshCore wire response, which previously read self.radio directly and so described the default endpoint. stats.errors and the per-radio entries beside it now reconcile. This matches what node_stats() already does with the airtime counters that sit next to it: a node-level counter describes the node, and which radio contributed what stays available per radio, in radios[] and in the crc_errors table. The sum prefers each radio's live hardware counter and falls back to the last count the sampler read from a radio the fabric cannot hand back. These are lifetime counters, so dropping a radio would walk the node's total backwards and hand an observer reading deltas a negative interval. A single-radio node still reports its one unlabelled counter, read live, exactly as before. --- .../data_acquisition/storage_collector.py | 8 +-- repeater/engine.py | 32 +++++++++-- repeater/handler_helpers/protocol_request.py | 10 +++- ...test_handler_helpers_path_protocol_text.py | 51 +++++++++++++++++ tests/test_per_radio_rf_samples.py | 56 +++++++++++++++++++ tests/test_storage_collector_live_stats.py | 18 +++++- 6 files changed, 163 insertions(+), 12 deletions(-) diff --git a/repeater/data_acquisition/storage_collector.py b/repeater/data_acquisition/storage_collector.py index 9ab4372..1c66fbe 100644 --- a/repeater/data_acquisition/storage_collector.py +++ b/repeater/data_acquisition/storage_collector.py @@ -212,10 +212,10 @@ class StorageCollector: except Exception as e: logger.debug(f"Could not read cached noise floor: {e}") - # Receive errors: the radio's CRC failure counter, which is the only - # error the node actually counts. It was published as a literal 0 for as - # long as the field has existed, so an observer could not tell a quiet - # node from a deaf one. + # Receive errors: CRC failures, the only error the node actually counts, + # summed over its radios so this reconciles with radios[].errors below. + # It was published as a literal 0 for as long as the field has existed, + # so an observer could not tell a quiet node from a deaf one. errors = 0 crc_error_count = getattr(self.repeater_handler, "get_crc_error_count", None) if callable(crc_error_count): diff --git a/repeater/engine.py b/repeater/engine.py index e837cad..e8cb57f 100644 --- a/repeater/engine.py +++ b/repeater/engine.py @@ -2463,14 +2463,34 @@ class RepeaterHandler(BaseHandler): return dict(self._cached_noise_floor_by_radio) def get_crc_error_count(self) -> int: - """The default radio's cumulative CRC error count, straight from hardware. + """The node's cumulative CRC error count, summed over its radios. - The default radio's, not the node's sum, because this is the figure the - MeshCore wire response already reports as ``n_recv_errors`` and /stats - as ``crc_error_count``. Per-radio counts are stored and queryable from - the crc_errors table; what a single scalar means here has to match what - the other readers of it mean. + The node's, not the default radio's: on a bridge the default radio is + one of two receivers, and reporting only its errors leaves a deaf + backhaul invisible in every scalar the node publishes. Summing matches + what AirtimeBudgets.node_stats does with the airtime counters beside it. + + Every reader of this figure sees the same number -- the MQTT status + message's ``stats.errors``, /stats' ``crc_error_count`` and the MeshCore + wire response's ``n_recv_errors`` -- so the sum of ``radios[].errors`` + in a status message reconciles with the scalar. Which radio contributed + what stays available per radio, both there and in the crc_errors table. """ + # Live counts win; a radio the fabric cannot hand back right now falls + # back to the last count the sampler read from it. These are lifetime + # counters, so dropping a radio from the sum would walk the node's total + # backwards and hand an observer reading deltas a negative interval -- + # the same reason AirtimeBudgets keeps its retired channels' totals. + counts = { + radio_id: count + for radio_id, count in self._crc_error_baselines.items() + if radio_id is not None + } + counts.update(self.get_crc_error_count_by_radio()) + if counts: + return sum(counts.values()) + + # Single-radio node: one unlabelled counter, read live. radio = self.dispatcher.radio if self.dispatcher else None return int(getattr(radio, "crc_error_count", 0) or 0) if radio else 0 diff --git a/repeater/handler_helpers/protocol_request.py b/repeater/handler_helpers/protocol_request.py index d148f9d..4998192 100644 --- a/repeater/handler_helpers/protocol_request.py +++ b/repeater/handler_helpers/protocol_request.py @@ -239,7 +239,15 @@ class ProtocolRequestHelper: n_recv_direct = getattr(self.engine, "recv_direct_count", 0) if self.engine else 0 n_direct_dups = getattr(self.engine, "direct_dup_count", 0) if self.engine else 0 n_flood_dups = getattr(self.engine, "flood_dup_count", 0) if self.engine else 0 - n_recv_errors = int(getattr(self.radio, "crc_error_count", 0) or 0) if self.radio else 0 + # Via the engine, so this reports the whole node: on a bridge, reading + # self.radio answers for the default endpoint only and a deaf backhaul + # never reaches the client asking for status. + n_recv_errors = 0 + node_crc_errors = getattr(self.engine, "get_crc_error_count", None) if self.engine else None + if callable(node_crc_errors): + n_recv_errors = int(node_crc_errors() or 0) + elif self.radio: + n_recv_errors = int(getattr(self.radio, "crc_error_count", 0) or 0) # Pack 56-byte RepeaterStats (layout matches firmware) stats = struct.pack( diff --git a/tests/test_handler_helpers_path_protocol_text.py b/tests/test_handler_helpers_path_protocol_text.py index f050e50..d7ac4f2 100644 --- a/tests/test_handler_helpers_path_protocol_text.py +++ b/tests/test_handler_helpers_path_protocol_text.py @@ -361,6 +361,57 @@ def test_protocol_request_handle_get_status_builds_56_byte_payload(): assert len(data) == 56 +def _status_recv_errors(engine, radio) -> int: + """n_recv_errors, the last uint32 of the 56-byte RepeaterStats.""" + helper = ProtocolRequestHelper( + identity_manager=MagicMock(), + packet_injector=AsyncMock(), + radio=radio, + engine=engine, + ) + data = helper._handle_get_status(client=None, timestamp=0, req_data=b"") + return struct.unpack(" SimpleNamespace: + engine = SimpleNamespace( + start_time=time.time() - 120, + rx_count=7, + forwarded_count=5, + sent_flood_count=2, + sent_direct_count=3, + recv_flood_count=4, + recv_direct_count=1, + direct_dup_count=6, + flood_dup_count=8, + airtime_mgr=SimpleNamespace(total_airtime_ms=9300, total_rx_airtime_ms=4200), + ) + for key, value in overrides.items(): + setattr(engine, key, value) + return engine + + +def _status_radio(crc_error_count=11) -> SimpleNamespace: + return SimpleNamespace( + get_noise_floor=lambda: -110, + get_last_rssi=lambda: -70, + get_last_snr=lambda: 2.5, + crc_error_count=crc_error_count, + ) + + +def test_status_recv_errors_counts_the_whole_node(): + # Reading self.radio answers for the default endpoint only, so a bridge's + # deaf backhaul never reached a client asking for status. + engine = _status_engine(get_crc_error_count=lambda: 13) + + assert _status_recv_errors(engine, _status_radio(crc_error_count=4)) == 13 + + +def test_status_recv_errors_falls_back_to_the_radio_counter(): + assert _status_recv_errors(_status_engine(), _status_radio(crc_error_count=11)) == 11 + + def test_protocol_request_access_list_admin_and_reserved_rules(): admin = SimpleNamespace(is_admin=lambda: True) not_admin = SimpleNamespace(is_admin=lambda: False) diff --git a/tests/test_per_radio_rf_samples.py b/tests/test_per_radio_rf_samples.py index 58207cd..d3f04a3 100644 --- a/tests/test_per_radio_rf_samples.py +++ b/tests/test_per_radio_rf_samples.py @@ -384,6 +384,62 @@ async def test_crc_errors_publish_only_the_default_radio(): assert published == {"local"} +def test_node_crc_count_sums_every_radio(): + # The scalar every reader shares -- stats.errors, /stats crc_error_count and + # the wire response's n_recv_errors. Reading the default radio alone would + # leave a deaf backhaul invisible in all three. + handler, _, _ = _bridge_handler() + + assert handler.get_crc_error_count() == 13 + assert handler.get_crc_error_count_by_radio() == {"local": 4, "link": 9} + + +def test_node_crc_count_reconciles_with_the_per_radio_figures(): + handler, _, _ = _bridge_handler() + + by_radio = handler.get_crc_error_count_by_radio() + + assert handler.get_crc_error_count() == sum(by_radio.values()) + + +def test_a_radio_the_fabric_cannot_hand_back_keeps_its_last_count(): + # Lifetime counters: a radio going missing does not un-drop the packets it + # dropped, and an observer reading these as deltas must never see a negative + # interval. The sampler's last reading stands in until the radio is back. + handler, local, link = _bridge_handler() + handler._crc_error_baselines = {"local": 4, "link": 9} + + def no_link(radio_id): + if radio_id == "link": + raise KeyError(radio_id) + return local + + handler.dispatcher.radio.fabric.get_radio = no_link + + assert handler.get_crc_error_count_by_radio() == {"local": 4} + assert handler.get_crc_error_count() == 13 + + +def test_a_fabric_down_to_one_radio_still_counts_the_one_that_left(): + handler, local, _ = _bridge_handler() + handler._crc_error_baselines = {"local": 4, "link": 9} + + handler.dispatcher.radio.fabric = _Fabric([("local", local)], "local") + + # One radio needs no label, so it reports unattributed again ... + assert handler.get_crc_error_count_by_radio() == {} + # ... but the node total does not drop the departed radio's 9. + assert handler.get_crc_error_count() == 13 + + +def test_single_radio_node_still_reports_its_one_counter(): + handler = _make_handler() + handler.dispatcher.radio.crc_error_count = 5 + + assert handler.get_crc_error_count_by_radio() == {} + assert handler.get_crc_error_count() == 5 + + # --------------------------------------------------------------------------- # API boundary: the endpoints call the storage facade, not the SQLite handler # --------------------------------------------------------------------------- diff --git a/tests/test_storage_collector_live_stats.py b/tests/test_storage_collector_live_stats.py index dda4c92..ee1bf3c 100644 --- a/tests/test_storage_collector_live_stats.py +++ b/tests/test_storage_collector_live_stats.py @@ -139,7 +139,7 @@ def test_noise_floor_is_omitted_when_the_handler_cannot_report_one(): assert "noise_floor" not in collector._get_live_stats() -def test_errors_reports_the_radios_crc_failures(): +def test_errors_reports_the_nodes_crc_failures(): # Published as a literal 0 for as long as the field existed, which made a # deaf node look like a quiet one. collector = _make_collector() @@ -408,3 +408,19 @@ def test_a_single_radio_node_publishes_no_radio_map(): ) assert "radios" not in status + + +def test_status_errors_reconcile_with_the_per_radio_counts(): + # The node scalar is the sum over its radios, so a consumer can add up + # radios[].errors and land on stats.errors rather than on a mismatch. + collector = _make_collector() + collector.repeater_handler = _bridge_handler(get_crc_error_count=lambda: 2483) + + status = _publish_status( + collector._get_live_stats, + radios=[LOCAL_RADIO, LINK_RADIO], + radio_stats_provider=collector._get_radio_stats, + ) + + assert status["stats"]["errors"] == 2483 + assert sum(radio["errors"] for radio in status["radios"]) == 2483