feat(mqtt): attribute published packets to the radio that carried them

A two-radio bridge is one MeshCore node with one pubkey, and the path
byte it appends on repeat is the same whichever radio carried the
packet. Nothing on the air says which band a hop used, and an observer
receiving on one side can only infer the other by already knowing its
neighbour's frequency. The node itself does know, so say it.

Two halves, both needed:

- Packets gain rx_radio_id (the radio that heard it) and tx_radio_ids
  (every radio that retransmitted it - a bridge repeating on both sides
  lists both). tx_radio_ids is always a list, so a consumer never has to
  branch on the JSON type, and is absent rather than empty when the
  packet was not repeated.
- Status gains radios: [{id, radio}], mapping each id to its
  MHz,kHz,SF,CR settings. Without it a radio id is just an
  operator-chosen label. Ids are unique only within a node, so consumers
  key on (origin_id, id). Rebuilt per publish, since air settings can
  change from the web UI while the node runs.

Single-radio nodes publish neither and their payload stays byte-for-byte
what it was: the ids would say nothing the status message doesn't
already. A node whose radio settings cannot all be read publishes no map
at all rather than a partial one that would place a packet on the wrong
band.

On a Fabric node the top-level status radio field now reports the
default radio rather than the top-level radio: block, which is the base
entries inherit from and can describe a band no radio is on.

config.yaml.example

@@ -758,6 +758,18 @@ mqtt_brokers:
   #   enabled: true|false # Enable TLS. If the endpoint's certificate is self-signed, the Root CA should be added to the OS's certificate store. 
   #   insecure: true|false # Validate TLS certificates
 
+  # Multi-radio nodes (see `radios:` above) additionally publish:
+  #   - `rx_radio_id` and `tx_radio_ids` on each packet: the radio that heard
+  #     it, and every radio that retransmitted it (a bridge repeating on both
+  #     sides lists both). `tx_radio_ids` is always a list; it is absent when
+  #     the packet was not repeated.
+  #   - `radios` in the status message: `[{id, radio}]` mapping each radio id
+  #     to its `MHz,kHz,SF,CR` settings, so a consumer can resolve those ids to
+  #     frequencies. Ids are unique only within a node - key on
+  #     (origin_id, id). The top-level `radio` field then reports the default
+  #     radio.
+  # Single-radio nodes publish neither, exactly as before.
+
   # Block specific packet types from being published to the MQTT endpoint
   # If not specified or empty list, all types are published
   # Available types: REQ, RESPONSE, TXT_MSG, ACK, ADVERT, GRP_TXT,

repeater/config.py

@@ -162,6 +162,11 @@ def resolve_storage_dir(
     return storage_dir
 
 
+def format_radio_config_str(frequency_hz, bandwidth_hz, spreading_factor, coding_rate) -> str:
+    """Air settings as the MC2MQTT status ``radio`` string: MHz,kHz,SF,CR."""
+    return f"{frequency_hz / 1_000_000},{bandwidth_hz / 1_000},{spreading_factor},{coding_rate}"
+
+
 def get_node_info(config: Dict[str, Any]) -> Dict[str, Any]:
     """
     Extract node name, radio configuration, and MQTT settings from config.
@@ -174,14 +179,12 @@ def get_node_info(config: Dict[str, Any]) -> Dict[str, Any]:
     """
     node_name = config.get("repeater", {}).get("node_name", "PyMC-Repeater")
     radio_config = config.get("radio", {})
-    radio_freq = radio_config.get("frequency", 0.0)
-    radio_bw = radio_config.get("bandwidth", 0.0)
-    radio_sf = radio_config.get("spreading_factor", 7)
-    radio_cr = radio_config.get("coding_rate", 5)
-    # Format frequency in MHz and bandwidth in kHz
-    radio_freq_mhz = radio_freq / 1_000_000
-    radio_bw_khz = radio_bw / 1_000
-    radio_config_str = f"{radio_freq_mhz},{radio_bw_khz},{radio_sf},{radio_cr}"
+    radio_config_str = format_radio_config_str(
+        radio_config.get("frequency", 0.0),
+        radio_config.get("bandwidth", 0.0),
+        radio_config.get("spreading_factor", 7),
+        radio_config.get("coding_rate", 5),
+    )
 
     # Handle getting the config from mqtt brokers, falling back to letsmesh if it doesn't exist
     mqtt_config = config.get("mqtt_brokers", config.get("letsmesh", {}))
@@ -1216,6 +1219,38 @@ def build_metering_profiles(config: dict) -> list:
     return profiles
 
 
+def build_radio_status_entries(config: dict) -> list:
+    """Map every radio id to its air settings, for the MQTT status message.
+
+    Observers join this against the ``rx_radio_id`` / ``tx_radio_ids`` fields
+    published on each packet. The ids are operator-chosen and unique only
+    within a node, so a consumer keys on ``(origin_id, id)``. The ``radio``
+    value repeats the ``MHz,kHz,SF,CR`` shape of the top-level status field so
+    no second parser is needed.
+
+    Returns an empty list when any radio's settings are unreadable: a partial
+    map would silently attribute a packet to the wrong band, which is worse
+    than an observer knowing the map is unavailable.
+    """
+    profiles = build_radio_profiles(config)
+    entries = []
+    for profile in profiles:
+        fields = (
+            profile.get("frequency_hz"),
+            profile.get("bandwidth_hz"),
+            profile.get("spreading_factor"),
+            profile.get("coding_rate"),
+        )
+        if any(value is None for value in fields):
+            logger.warning(
+                "Radio %s has unreadable air settings; omitting the status radio map",
+                profile.get("radio_id"),
+            )
+            return []
+        entries.append({"id": profile["radio_id"], "radio": format_radio_config_str(*fields)})
+    return entries
+
+
 def build_radio_stack(config: dict):
     """Build single- or multi-radio stack for the repeater.
 

repeater/data_acquisition/mqtt_handler.py

@@ -988,11 +988,57 @@ class MeshCoreToMqttPusher:
             "stats": {**live_stats, "errors": 0, "queue_len": 0, **(extra_stats or {})},
         }
 
+        radios = self._radio_map()
+        if radios:
+            status["radios"] = radios
+            # On a Fabric node the top-level ``radio:`` block is the base that
+            # entries inherit from, so it can describe a band no radio is
+            # actually on. Report the default radio instead, which is the one
+            # an observer that ignores ``radios`` should assume.
+            status["radio"] = self._default_radio_str(radios) or status["radio"]
+
         if location:
             status["location"] = location
 
         return self.publish("status", status, retain=True, qos=1)
 
+    def _default_radio_str(self, entries: list) -> Optional[str]:
+        """Air settings of the radio Fabric transmits on by default.
+
+        Mirrors build_radio_stack's rule: ``fabric.default_radio`` when set,
+        otherwise the first configured radio.
+        """
+        fabric = self.config.get("fabric")
+        fabric = fabric if isinstance(fabric, dict) else {}
+        default_id = fabric.get("default_radio") or fabric.get("default_radio_id")
+        if default_id:
+            for entry in entries:
+                if entry["id"] == str(default_id):
+                    return entry["radio"]
+        return entries[0]["radio"] if entries else None
+
+    def _radio_map(self) -> list:
+        """Per-radio air settings for multi-radio nodes, else an empty list.
+
+        This is what makes the ``rx_radio_id`` / ``tx_radio_ids`` on published
+        packets mean something: on its own a radio id is an operator-chosen
+        label, and only this map says which frequency it was. A node runs one
+        radio in the overwhelming majority of deployments, where the existing
+        ``radio`` field already says everything, so the map is published only
+        when there is genuinely more than one band to tell apart.
+
+        Rebuilt per status publish rather than cached at construction, because
+        air settings can be changed from the web UI while the node runs.
+        """
+        from ..config import build_radio_status_entries
+
+        try:
+            entries = build_radio_status_entries(self.config)
+        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 []
+
     def publish(self, subtopic: str, payload: dict, retain: bool = False, qos: int = 0):
         """Publish message to all connected brokers"""
         message = json.dumps(payload)

repeater/data_acquisition/storage_collector.py

@@ -63,6 +63,11 @@ class StorageCollector:
             max_workers=1, thread_name_prefix="storage-writer"
         )
 
+        # Radio count is fixed at boot (build_radio_stack opens the hardware),
+        # so this is resolved once rather than per packet. Air settings can
+        # still change at runtime; those are re-read when status is published.
+        self._multi_radio = len(build_radio_profiles(config)) > 1
+
         self.storage_dir = resolve_storage_dir(config)
         self.storage_dir.mkdir(parents=True, exist_ok=True)
 
@@ -368,6 +373,10 @@ class StorageCollector:
         ``packet_record['airtime_ms']``, populated upstream by
         RepeaterHandler._build_packet_record using the Semtech reference
         time-on-air formula. No recomputation is needed here.
+
+        On a multi-radio node the payload also carries the ingress radio and
+        every successful egress, which an observer resolves to frequencies
+        through the ``radios`` map in this node's status message.
         """
         if not self.mqtt_handler:
             return
@@ -380,7 +389,10 @@ class StorageCollector:
 
             node_name = self.config.get("repeater", {}).get("node_name", "Unknown")
             packet = PacketRecord.from_packet_record(
-                packet_record, origin=node_name, origin_id=self.mqtt_handler.public_key
+                packet_record,
+                origin=node_name,
+                origin_id=self.mqtt_handler.public_key,
+                include_radio_ids=self._multi_radio,
             )
 
             if packet:

repeater/data_acquisition/storage_utils.py

@@ -2,7 +2,7 @@
 
 from dataclasses import asdict, dataclass
 from datetime import datetime, timezone
-from typing import Optional
+from typing import List, Optional
 
 
 @dataclass
@@ -30,10 +30,20 @@ class PacketRecord:
     score: str
     duration: str
     hash: str
+    # Multi-radio only: which interface carried this packet. A single-radio
+    # node omits both, keeping its payload byte-identical to pre-Fabric
+    # builds. The ids are the operator's ``radios[].id`` values, resolved
+    # against the ``radios`` map in the status message.
+    rx_radio_id: Optional[str] = None
+    tx_radio_ids: Optional[List[str]] = None
 
     @classmethod
     def from_packet_record(
-        cls, packet_record: dict, origin: str, origin_id: str
+        cls,
+        packet_record: dict,
+        origin: str,
+        origin_id: str,
+        include_radio_ids: bool = False,
     ) -> Optional["PacketRecord"]:
         """
         Create PacketRecord from internal packet_record format.
@@ -44,10 +54,15 @@ class PacketRecord:
         Records produced by older code paths that pre-date that field fall
         back to 0 to preserve legacy behavior.
 
+        ``include_radio_ids`` is set by callers running more than one radio.
+        On a single-radio node the ids carry no information an observer cannot
+        already read from the status message, so they are left off the wire.
+
         Args:
             packet_record: Internal packet record dictionary
             origin: Node name
             origin_id: Public key of the node
+            include_radio_ids: Emit rx_radio_id / tx_radio_ids (multi-radio nodes)
 
         Returns:
             PacketRecord instance or None if raw_packet is missing
@@ -65,6 +80,12 @@ class PacketRecord:
 
         airtime_ms = float(packet_record.get("airtime_ms", 0.0) or 0.0)
 
+        rx_radio_id = None
+        tx_radio_ids = None
+        if include_radio_ids:
+            rx_radio_id = packet_record.get("rx_radio_id") or None
+            tx_radio_ids = cls._tx_radio_ids(packet_record)
+
         return cls(
             origin=origin,
             origin_id=origin_id,
@@ -83,8 +104,30 @@ class PacketRecord:
             score=str(int(packet_record.get("score", 0) * 1000)),
             duration=str(int(round(airtime_ms))),
             hash=packet_record.get("packet_hash", ""),
+            rx_radio_id=rx_radio_id,
+            tx_radio_ids=tx_radio_ids,
         )
 
+    @staticmethod
+    def _tx_radio_ids(packet_record: dict) -> Optional[List[str]]:
+        """Every radio that successfully transmitted this packet, or None.
+
+        Always a list, even for the single egress of a non-bridge TX, so a
+        consumer never has to branch on the JSON type. A packet this node
+        heard but did not repeat has no egress at all: the key is then absent
+        rather than empty, which is the same statement the record makes.
+        """
+        ids = packet_record.get("tx_radio_ids")
+        if isinstance(ids, (list, tuple)):
+            resolved = [str(value) for value in ids if value]
+            return resolved or None
+        primary = packet_record.get("tx_radio_id")
+        return [str(primary)] if primary else None
+
     def to_dict(self) -> dict:
-        """Convert to dictionary for JSON serialization"""
-        return asdict(self)
+        """Convert to dictionary for JSON serialization.
+
+        Unset optional fields are dropped so a single-radio node publishes the
+        exact payload it did before multi-radio attribution existed.
+        """
+        return {key: value for key, value in asdict(self).items() if value is not None}

tests/test_observer_radio_attribution.py

@@ -0,0 +1,345 @@
+"""Radio attribution on the MQTT observer feed.
+
+A two-radio bridge is one MeshCore node with one pubkey, and the path byte it
+appends is the same whichever radio carried the packet, so nothing on the air
+says which band a hop used. The node itself does know, and these tests lock the
+two halves that let an observer read it: the ingress/egress radio ids on each
+published packet, and the id -> frequency map in the status message that gives
+those ids meaning.
+
+Single-radio nodes must publish exactly what they published before any of this
+existed.
+"""
+
+import json
+from unittest.mock import MagicMock
+
+from repeater.config import build_radio_status_entries, get_node_info
+from repeater.data_acquisition.mqtt_handler import MeshCoreToMqttPusher
+from repeater.data_acquisition.storage_utils import PacketRecord
+
+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,
+    },
+}
+
+
+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 _packet_record(**overrides) -> dict:
+    record = {
+        "timestamp": 1700000000.0,
+        "type": 4,
+        "route": 1,
+        "rssi": -90,
+        "snr": 7.5,
+        "score": 0.5,
+        "payload_length": 32,
+        "packet_hash": "DEADBEEF" + "00" * 4,
+        "raw_packet": bytes(range(40)).hex(),
+        "airtime_ms": 123.0,
+    }
+    record.update(overrides)
+    return record
+
+
+def _serialize(record: dict, include_radio_ids: bool) -> dict:
+    packet = PacketRecord.from_packet_record(
+        record,
+        origin="test-node",
+        origin_id="AB" * 32,
+        include_radio_ids=include_radio_ids,
+    )
+    assert packet is not None
+    return packet.to_dict()
+
+
+# --------------------------------------------------------------------
+# Packet payload
+# --------------------------------------------------------------------
+def test_single_radio_payload_carries_no_radio_ids():
+    """The ids say nothing a single-radio status message doesn't already say."""
+    payload = _serialize(
+        _packet_record(rx_radio_id="radio0", tx_radio_id="radio0", tx_radio_ids=["radio0"]),
+        include_radio_ids=False,
+    )
+
+    assert "rx_radio_id" not in payload
+    assert "tx_radio_ids" not in payload
+
+
+def test_single_radio_payload_is_unchanged_from_pre_attribution_builds():
+    """Observers parsing the old schema must see byte-identical JSON."""
+    expected_keys = {
+        "origin",
+        "origin_id",
+        "timestamp",
+        "type",
+        "direction",
+        "time",
+        "date",
+        "len",
+        "packet_type",
+        "route",
+        "payload_len",
+        "raw",
+        "SNR",
+        "RSSI",
+        "score",
+        "duration",
+        "hash",
+    }
+
+    assert set(_serialize(_packet_record(), include_radio_ids=False)) == expected_keys
+
+
+def test_bridge_egress_on_both_sides_is_one_packet_with_both_radios():
+    """Repeating on both sides duplicates the TX, not the upload: one payload
+    naming both egress radios, not two payloads."""
+    payload = _serialize(
+        _packet_record(
+            rx_radio_id="local",
+            tx_radio_id="link",
+            tx_radio_ids=["link", "local"],
+        ),
+        include_radio_ids=True,
+    )
+
+    assert payload["rx_radio_id"] == "local"
+    assert payload["tx_radio_ids"] == ["link", "local"]
+
+
+def test_tx_radio_ids_is_a_list_even_for_a_single_egress():
+    """A consumer must never have to branch on the JSON type of this field."""
+    payload = _serialize(
+        _packet_record(rx_radio_id="local", tx_radio_id="link"),
+        include_radio_ids=True,
+    )
+
+    assert payload["tx_radio_ids"] == ["link"]
+
+
+def test_packet_heard_but_not_repeated_has_no_egress_key():
+    """Absent, not empty: the node received it and transmitted nothing."""
+    payload = _serialize(
+        _packet_record(rx_radio_id="link", tx_radio_id=None, tx_radio_ids=None),
+        include_radio_ids=True,
+    )
+
+    assert payload["rx_radio_id"] == "link"
+    assert "tx_radio_ids" not in payload
+
+
+def test_pre_fabric_rows_with_null_ids_omit_the_fields():
+    """Rows written before radio attribution existed carry NULL ids."""
+    payload = _serialize(_packet_record(), include_radio_ids=True)
+
+    assert "rx_radio_id" not in payload
+    assert "tx_radio_ids" not in payload
+
+
+# --------------------------------------------------------------------
+# Status radio map
+# --------------------------------------------------------------------
+def test_radio_status_entries_map_each_id_to_its_air_settings():
+    entries = build_radio_status_entries(
+        {"radios": [LOCAL_RADIO, LINK_RADIO], "fabric": {"default_radio": "local"}}
+    )
+
+    assert entries == [
+        {"id": "local", "radio": "869.618,62.5,8,8"},
+        {"id": "link", "radio": "864.2,62.5,11,8"},
+    ]
+
+
+def test_radio_status_entry_matches_the_single_radio_status_string():
+    """The per-radio value reuses the top-level ``radio`` format, so consumers
+    need no second parser."""
+    config = {
+        "radio_type": "sx1262",
+        "radio": LOCAL_RADIO["radio"],
+        "repeater": {"node_name": "test-node"},
+    }
+
+    entries = build_radio_status_entries(config)
+
+    assert len(entries) == 1
+    assert entries[0]["radio"] == get_node_info(config)["radio_config"]
+
+
+def test_unreadable_radio_settings_publish_no_map_at_all():
+    """A partial map would attribute a packet to the wrong band."""
+    broken = {"id": "link", "radio_type": "sx1262", "radio": {"frequency": "not-a-number"}}
+
+    assert build_radio_status_entries({"radios": [LOCAL_RADIO, broken]}) == []
+
+
+# --------------------------------------------------------------------
+# Status message on the wire
+# --------------------------------------------------------------------
+def _make_config(radios=None) -> dict:
+    config = {
+        "repeater": {"node_name": "test-node"},
+        "radio_type": "sx1262",
+        "radio": LOCAL_RADIO["radio"],
+        "duty_cycle": {"max_airtime_per_minute": 3600},
+        "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": "link"}
+    return config
+
+
+def _publish_status(config: dict) -> dict:
+    pusher = MeshCoreToMqttPusher(local_identity=_FakeIdentity("AB" * 32), config=config)
+    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_status_publishes_the_radio_map_for_a_two_radio_node():
+    status = _publish_status(_make_config(radios=[LOCAL_RADIO, LINK_RADIO]))
+
+    assert status["radios"] == [
+        {"id": "local", "radio": "869.618,62.5,8,8"},
+        {"id": "link", "radio": "864.2,62.5,11,8"},
+    ]
+
+
+def test_status_radio_field_reports_the_default_radio_on_a_fabric_node():
+    """The top-level ``radio:`` block is the base entries inherit from, so it
+    can name a band no radio is on. Observers ignoring ``radios`` get the
+    radio Fabric transmits on by default."""
+    status = _publish_status(_make_config(radios=[LOCAL_RADIO, LINK_RADIO]))
+
+    assert status["radio"] == "864.2,62.5,11,8"
+
+
+def test_single_radio_status_has_no_radio_map():
+    status = _publish_status(_make_config())
+
+    assert "radios" not in status
+    assert status["radio"] == "869.618,62.5,8,8"
+
+
+def test_status_ids_resolve_every_radio_id_a_packet_can_name():
+    """The two halves must join: every id the packet payload can carry has an
+    entry in the status map."""
+    config = _make_config(radios=[LOCAL_RADIO, LINK_RADIO])
+    status = _publish_status(config)
+    payload = _serialize(
+        _packet_record(rx_radio_id="local", tx_radio_id="link", tx_radio_ids=["link", "local"]),
+        include_radio_ids=True,
+    )
+
+    known = {entry["id"] for entry in status["radios"]}
+    assert payload["rx_radio_id"] in known
+    assert set(payload["tx_radio_ids"]) <= known
+
+
+# --------------------------------------------------------------------
+# Collector wiring
+# --------------------------------------------------------------------
+def _make_collector(tmp_path, radios=None):
+    """A StorageCollector with its storage backends stubbed out."""
+    from unittest.mock import patch
+
+    from repeater.data_acquisition.storage_collector import StorageCollector
+
+    config = {"storage": {"storage_dir": str(tmp_path)}, "repeater": {"node_name": "test-node"}}
+    config["radio_type"] = "sx1262"
+    config["radio"] = LOCAL_RADIO["radio"]
+    if radios:
+        config["radios"] = radios
+
+    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=config)
+
+    collector._stats_stop_event.set()
+    if collector._stats_thread is not None:
+        collector._stats_thread.join(timeout=1)
+    collector.mqtt_handler = MagicMock()
+    collector.mqtt_handler.public_key = "AB" * 32
+    return collector
+
+
+def _published_payload(collector, record) -> dict:
+    collector._publish_packet_to_mqtt(record)
+    collector.mqtt_handler.publish_packet.assert_called_once()
+    return collector.mqtt_handler.publish_packet.call_args[0][0]
+
+
+def test_collector_publishes_radio_ids_on_a_two_radio_node(tmp_path):
+    collector = _make_collector(tmp_path, radios=[LOCAL_RADIO, LINK_RADIO])
+    assert collector._multi_radio is True
+
+    payload = _published_payload(
+        collector,
+        _packet_record(rx_radio_id="local", tx_radio_id="link", tx_radio_ids=["link", "local"]),
+    )
+
+    assert payload["rx_radio_id"] == "local"
+    assert payload["tx_radio_ids"] == ["link", "local"]
+
+
+def test_collector_omits_radio_ids_on_a_single_radio_node(tmp_path):
+    collector = _make_collector(tmp_path)
+    assert collector._multi_radio is False
+
+    payload = _published_payload(
+        collector, _packet_record(rx_radio_id="radio0", tx_radio_id="radio0")
+    )
+
+    assert "rx_radio_id" not in payload
+    assert "tx_radio_ids" not in payload