diff --git a/repeater/data_acquisition/mqtt_handler.py b/repeater/data_acquisition/mqtt_handler.py index 9979867..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 @@ -1005,7 +1007,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() @@ -1049,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 @@ -1057,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 b271af8..1c66fbe 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() @@ -198,26 +199,43 @@ 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}") + + # 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): + 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 } # 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"] @@ -227,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 dcfe0e0..e8cb57f 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]: @@ -2454,6 +2462,53 @@ 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 node's cumulative CRC error count, summed over its radios. + + 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 + + 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) @@ -2478,8 +2533,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/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 new file mode 100644 index 0000000..ee1bf3c --- /dev/null +++ b/tests/test_storage_collector_live_stats.py @@ -0,0 +1,426 @@ +"""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: 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 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 +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")) + +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, + get_crc_error_count=lambda: 47, + ) + 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 + + +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() + + +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() + 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 + + +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": { + "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}, + } + ], + }, + } + 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 = [] + 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 + + +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 + + +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