diff --git a/config.yaml.example b/config.yaml.example index a6434c1..ec9412a 100644 --- a/config.yaml.example +++ b/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, diff --git a/repeater/config.py b/repeater/config.py index 3fb265d..25720c6 100644 --- a/repeater/config.py +++ b/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. diff --git a/repeater/data_acquisition/mqtt_handler.py b/repeater/data_acquisition/mqtt_handler.py index ce0eaba..726559e 100644 --- a/repeater/data_acquisition/mqtt_handler.py +++ b/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) diff --git a/repeater/data_acquisition/storage_collector.py b/repeater/data_acquisition/storage_collector.py index f1deabc..c16384b 100644 --- a/repeater/data_acquisition/storage_collector.py +++ b/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: diff --git a/repeater/data_acquisition/storage_utils.py b/repeater/data_acquisition/storage_utils.py index 532a564..c4c7ce7 100644 --- a/repeater/data_acquisition/storage_utils.py +++ b/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} diff --git a/tests/test_observer_radio_attribution.py b/tests/test_observer_radio_attribution.py new file mode 100644 index 0000000..9c62e7d --- /dev/null +++ b/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