Merge pull request #488 from agessaman/feat/letsmesh-airtime-utilization-edd3fa

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"""

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,

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 {}

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(

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("<I", bytes(data[-4:]))[0]
+
+
+def _status_engine(**overrides) -> 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)

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
 # ---------------------------------------------------------------------------

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