diff --git a/repeater/config.py b/repeater/config.py index b01d006..6e4fc50 100644 --- a/repeater/config.py +++ b/repeater/config.py @@ -7,7 +7,7 @@ from typing import Any, Dict, Optional, overload import yaml from repeater.exceptions import ConfigurationError -from repeater.modem_config import normalize_modem_config +from repeater.modem_config import LEGACY_MODEM_RADIO_TYPES, normalize_modem_config from repeater.policy_engine import default_policy_engine_config logger = logging.getLogger("Config") @@ -895,6 +895,128 @@ def _validate_fabric_fanout(fabric_cfg, tx_mode: str, radio_count: int) -> tuple return repeat_on_ingress, local_tx_mode +# Air-setting defaults per radio_type, mirroring ``get_radio_for_board``. The +# SX1262 paths require every field in the config, so their entries only matter +# for a malformed config that never reaches hardware. +_RADIO_AIR_DEFAULTS = { + "sx1262": {"preamble_length": 16}, + "sx1262_ch341": {"preamble_length": 16}, + "kiss": {"preamble_length": 32}, + "modem_tcp": {"preamble_length": 16}, + "modem_usb": {"preamble_length": 16}, +} +_RADIO_AIR_COMMON_DEFAULTS = { + "frequency": 869618000, + "bandwidth": 62500, + "spreading_factor": 8, + "coding_rate": 8, +} + + +def _radio_air_profile(board_config: dict, radio_id: str) -> Optional[dict]: + """Effective LoRa air settings for one built radio, or None when disabled. + + Mirrors the field reads in ``get_radio_for_board`` so the reported profile + is the one the hardware was actually configured with. Reporting-only: a + field that cannot be read as an integer is returned as ``None`` rather than + guessed, so the UI can say the profile is unavailable instead of drawing a + chart from a default that was never on the air. + """ + radio_type_raw = board_config.get("radio_type") + radio_type = "none" if radio_type_raw is None else str(radio_type_raw).lower().strip() + if radio_type in ("", "none", "null", "disabled", "off", "no_radio"): + return None + if radio_type == "kiss-modem": + radio_type = "kiss" + radio_type = LEGACY_MODEM_RADIO_TYPES.get(radio_type, radio_type) + + if radio_type not in _RADIO_AIR_DEFAULTS: + # get_radio_for_board rejects this type, so no hardware was built from + # it. Reporting a default-backed profile would describe a radio that is + # not on the air. + logger.warning( + "No airtime profile for radio %s: unsupported radio_type=%r", radio_id, radio_type_raw + ) + return None + + radio_cfg = board_config.get("radio") + if not isinstance(radio_cfg, dict): + radio_cfg = {} + defaults = dict(_RADIO_AIR_COMMON_DEFAULTS) + defaults.update(_RADIO_AIR_DEFAULTS[radio_type]) + + def _field(name: str) -> Optional[int]: + value = radio_cfg.get(name, defaults.get(name)) + try: + return int(value) + except (TypeError, ValueError): + logger.warning( + "radio %s: unreadable %s=%r; reporting profile field as unknown", + radio_id, + name, + value, + ) + return None + + return { + "radio_id": str(radio_id), + "radio_type": radio_type, + "frequency_hz": _field("frequency"), + "bandwidth_hz": _field("bandwidth"), + "spreading_factor": _field("spreading_factor"), + "coding_rate": _field("coding_rate"), + "preamble_length": _field("preamble_length"), + } + + +def build_radio_profiles(config: dict) -> list: + """Report the air settings of every configured radio, in configured order. + + Uses the same id and inheritance rules as ``build_radio_stack`` so the + profiles line up one-for-one with the radios it builds, without touching + hardware. Callers that need current settings (the airtime endpoint) can + rebuild from the live config dict rather than reading a boot snapshot. + """ + if not isinstance(config, dict): + return [] + + try: + normalized = normalize_modem_config(config, warn=False) + except Exception: # pragma: no cover - normalisation is defensive here + normalized = config + + radios_cfg = normalized.get("radios") + fabric_cfg = normalized.get("fabric") if isinstance(normalized.get("fabric"), dict) else {} + default_radio = fabric_cfg.get("default_radio") or fabric_cfg.get("default_radio_id") + + profiles = [] + if isinstance(radios_cfg, list) and len(radios_cfg) > 0: + for entry in radios_cfg: + try: + merged = _merge_radio_entry(normalized, entry) + except ValueError as exc: + logger.warning("Cannot profile radios[] entry: %s", exc) + continue + profile = _radio_air_profile(merged, merged["_radio_id"]) + if profile is not None: + profiles.append(profile) + if len(profiles) != len(radios_cfg): + # A partial list would read as a smaller Fabric, and a one-element + # list would claim every packet belongs to that radio. Report + # nothing instead and let the caller fall back to what was built. + logger.warning( + "Profiled %d of %d configured radios; reporting no profiles", + len(profiles), + len(radios_cfg), + ) + return [] + return profiles + + rid = str(default_radio) if (default_radio and fabric_cfg.get("use_fabric")) else "radio0" + profile = _radio_air_profile(normalized, rid) + return [profile] if profile is not None else [] + + def build_radio_stack(config: dict): """Build single- or multi-radio stack for the repeater. @@ -923,6 +1045,10 @@ def build_radio_stack(config: dict): "repeat_on_ingress": repeat_on_ingress, "local_tx_mode": local_tx_mode, "fabric": False, + # Air settings per built radio, in configured order. Reporting consumers + # (airtime attribution, /stats) read these instead of assuming the + # top-level ``radio`` section applies to every radio in a Fabric. + "radio_profiles": [], } if isinstance(radios_cfg, list) and len(radios_cfg) > 0: @@ -935,11 +1061,15 @@ def build_radio_stack(config: dict): ) from exc pairs = [] + profiles = [] for entry in radios_cfg: merged = _merge_radio_entry(config, entry) rid = merged.pop("_radio_id") physical = get_radio_for_board(merged) pairs.append((physical, rid)) + profile = _radio_air_profile(merged, rid) + if profile is not None: + profiles.append(profile) default_id = str(default_radio) if default_radio else pairs[0][1] radio = FabricRadio(radios=pairs, default_radio_id=default_id) @@ -951,6 +1081,7 @@ def build_radio_stack(config: dict): "radio_ids": [rid for _, rid in pairs], "default_radio": default_id, "fabric": True, + "radio_profiles": profiles, } ) return radio, meta @@ -958,6 +1089,8 @@ def build_radio_stack(config: dict): physical = get_radio_for_board(config) meta["radio_ids"] = ["radio0"] meta["default_radio"] = "radio0" + single_profile = _radio_air_profile(normalize_modem_config(config, warn=False), "radio0") + meta["radio_profiles"] = [single_profile] if single_profile is not None else [] if use_fabric: try: @@ -969,12 +1102,15 @@ def build_radio_stack(config: dict): rid = str(default_radio or "radio0") radio = FabricRadio(radio=physical, radio_id=rid, default_radio_id=rid) _apply_fabric_tx_mode(radio.fabric, tx_mode) + if single_profile is not None: + single_profile = dict(single_profile, radio_id=rid) meta.update( { "mode": "single_fabric", "radio_ids": [rid], "default_radio": rid, "fabric": True, + "radio_profiles": [single_profile] if single_profile is not None else [], } ) return radio, meta diff --git a/repeater/data_acquisition/sqlite_handler.py b/repeater/data_acquisition/sqlite_handler.py index 3cf7656..282b562 100644 --- a/repeater/data_acquisition/sqlite_handler.py +++ b/repeater/data_acquisition/sqlite_handler.py @@ -9,6 +9,8 @@ import time from pathlib import Path from typing import Any, Dict, List, Optional +from openhop_core.protocol.packet_utils import calculate_lora_airtime_ms + logger = logging.getLogger("SQLiteHandler") @@ -2245,67 +2247,206 @@ class SQLiteHandler: bw_hz: int = 62500, cr: int = 5, preamble: int = 17, - ) -> list: + radio_profiles: Optional[list] = None, + ) -> dict: """Return pre-aggregated airtime buckets for chart rendering. - Applies the Semtech LoRa airtime formula server-side and groups results - into time buckets, drastically reducing response size vs raw packet rows. + Applies the shared core LoRa time-on-air estimator server-side and groups + results into time buckets, drastically reducing response size vs raw + packet rows. + + ``radio_profiles`` is the ordered list of active radio air settings (see + ``build_radio_profiles``). Each stored packet is attributed to the radios + that actually carried it: its ingress ``rx_radio_id`` and every successful + egress in ``tx_radio_ids``, so a relayed packet is charged to both sides of + a Fabric bridge at that side's own SF. With a single profile every packet + belongs to that radio, which keeps pre-Fabric databases (whose radio id + columns are NULL) attributed exactly as before. With two or more, a packet + whose radio cannot be identified is counted as unattributed rather than + guessed onto the default radio. + + The legacy top-level ``buckets``/``rx_total``/``tx_total`` fields remain, + carrying the sum across radios for UI builds that predate ``radios``. """ - import math + profiles = [p for p in (radio_profiles or []) if isinstance(p, dict)] + if not profiles: + profiles = [ + { + "radio_id": "radio0", + "frequency_hz": None, + "bandwidth_hz": int(bw_hz), + "spreading_factor": int(sf), + "coding_rate": int(cr), + "preamble_length": int(preamble), + } + ] + single_radio = len(profiles) == 1 - bw_khz = bw_hz / 1000 - t_sym = (2**sf) / bw_khz # ms per symbol - t_preamble = (preamble + 4.25) * t_sym - de = 1 if sf >= 11 and bw_hz <= 125000 else 0 + def _new_bucket(bucket_ts: int) -> dict: + return { + "timestamp": bucket_ts, + "rx_ms": 0.0, + "tx_ms": 0.0, + "rx_count": 0, + "tx_count": 0, + } - def _airtime_ms(length_bytes: int) -> float: - length_bytes = max(length_bytes or 32, 1) - numerator = max(8 * length_bytes - 4 * sf + 28 + 16, 0) # CRC=1, H=0 - denominator = 4 * (sf - 2 * de) - n_payload = 8 + math.ceil(numerator / denominator) * cr - return t_preamble + n_payload * t_sym + # One accumulator per radio, plus the combined legacy series. Airtime is + # memoised per (radio, length) because a 24 h window is thousands of rows + # over a handful of distinct packet lengths. + series: dict = {} + order: list = [] + for profile in profiles: + radio_id = str(profile.get("radio_id") or "radio0") + if radio_id in series: + continue + order.append(radio_id) + fields = ( + profile.get("spreading_factor"), + profile.get("bandwidth_hz"), + profile.get("coding_rate"), + profile.get("preamble_length"), + ) + series[radio_id] = { + "profile": profile, + "params": fields if all(f is not None for f in fields) else None, + "buckets": {}, + "rx_total": 0, + "tx_total": 0, + "cache": {}, + } + + def _airtime_ms(radio_id: str, length_bytes) -> float: + entry = series[radio_id] + if entry["params"] is None: + return 0.0 + length = max(int(length_bytes or 32), 1) + cached = entry["cache"].get(length) + if cached is None: + sf_v, bw_v, cr_v, preamble_v = entry["params"] + cached = calculate_lora_airtime_ms(length, sf_v, bw_v, cr_v, preamble_v) + entry["cache"][length] = cached + return cached try: with self._connect() as conn: conn.row_factory = sqlite3.Row rows = conn.execute( - "SELECT timestamp, length, transmitted FROM packets " + "SELECT timestamp, length, transmitted, rx_radio_id, tx_radio_id, " + "tx_radio_ids FROM packets " "WHERE timestamp >= ? AND timestamp <= ? ORDER BY timestamp ASC", (start_timestamp, end_timestamp), ).fetchall() - buckets: dict = {} + totals: dict = {} rx_total = 0 tx_total = 0 + unattributed_rx = 0 + unattributed_tx = 0 + + def _resolve(radio_id) -> Optional[str]: + """Map a stored radio id onto an active series, or None.""" + if single_radio: + return order[0] + if radio_id is None: + return None + key = str(radio_id) + return key if key in series else None + + def _record(kind: str, radio_id, length, bucket_ts: int, total_bucket: dict) -> None: + nonlocal rx_total, tx_total, unattributed_rx, unattributed_tx + resolved = _resolve(radio_id) + ms = _airtime_ms(resolved, length) if resolved else 0.0 + if resolved: + entry = series[resolved] + bucket = entry["buckets"].get(bucket_ts) + if bucket is None: + bucket = entry["buckets"][bucket_ts] = _new_bucket(bucket_ts) + bucket[kind + "_ms"] += ms + bucket[kind + "_count"] += 1 + entry[kind + "_total"] += 1 + elif kind == "rx": + unattributed_rx += 1 + else: + unattributed_tx += 1 + total_bucket[kind + "_ms"] += ms + total_bucket[kind + "_count"] += 1 + if kind == "rx": + rx_total += 1 + else: + tx_total += 1 + for row in rows: bucket_ts = int(row["timestamp"] / bucket_seconds) * bucket_seconds - ms = _airtime_ms(row["length"]) - if bucket_ts not in buckets: - buckets[bucket_ts] = { - "timestamp": bucket_ts, - "rx_ms": 0.0, - "tx_ms": 0.0, - "rx_count": 0, - "tx_count": 0, - } + total_bucket = totals.get(bucket_ts) + if total_bucket is None: + total_bucket = totals[bucket_ts] = _new_bucket(bucket_ts) + length = row["length"] + + tx_radio_ids = row["tx_radio_ids"] + if isinstance(tx_radio_ids, str): + try: + decoded = json.loads(tx_radio_ids) + except ValueError: + decoded = None + tx_radio_ids = decoded if isinstance(decoded, list) else None + if row["transmitted"]: - buckets[bucket_ts]["tx_ms"] += ms - buckets[bucket_ts]["tx_count"] += 1 - tx_total += 1 + # A relayed packet is one row: it occupied the ingress radio + # on the way in and every successful egress on the way out. + # Locally originated packets have no ingress radio. + if row["rx_radio_id"] is not None: + _record("rx", row["rx_radio_id"], length, bucket_ts, total_bucket) + if tx_radio_ids: + for tx_radio_id in tx_radio_ids: + _record("tx", tx_radio_id, length, bucket_ts, total_bucket) + else: + _record("tx", row["tx_radio_id"], length, bucket_ts, total_bucket) else: - buckets[bucket_ts]["rx_ms"] += ms - buckets[bucket_ts]["rx_count"] += 1 - rx_total += 1 + # Receptions the repeater did not forward, duplicates included: + # they consumed airtime on the radio that heard them. + _record("rx", row["rx_radio_id"], length, bucket_ts, total_bucket) + + radios = [] + for radio_id in order: + entry = series[radio_id] + profile = entry["profile"] + radios.append( + { + "radio_id": radio_id, + "profile": { + "frequency_hz": profile.get("frequency_hz"), + "bandwidth_hz": profile.get("bandwidth_hz"), + "spreading_factor": profile.get("spreading_factor"), + "coding_rate": profile.get("coding_rate"), + "preamble_length": profile.get("preamble_length"), + }, + "buckets": sorted(entry["buckets"].values(), key=lambda b: b["timestamp"]), + "rx_total": entry["rx_total"], + "tx_total": entry["tx_total"], + } + ) return { - "buckets": sorted(buckets.values(), key=lambda x: x["timestamp"]), + "buckets": sorted(totals.values(), key=lambda b: b["timestamp"]), "bucket_seconds": bucket_seconds, "rx_total": rx_total, "tx_total": tx_total, + "radios": radios, + "unattributed_rx_count": unattributed_rx, + "unattributed_tx_count": unattributed_tx, } except Exception as e: logger.error(f"Failed to get airtime buckets: {e}") - return {"buckets": [], "bucket_seconds": bucket_seconds, "rx_total": 0, "tx_total": 0} + return { + "buckets": [], + "bucket_seconds": bucket_seconds, + "rx_total": 0, + "tx_total": 0, + "radios": [], + "unattributed_rx_count": 0, + "unattributed_tx_count": 0, + } def get_packet_by_hash(self, packet_hash: str) -> Optional[dict]: try: diff --git a/repeater/data_acquisition/storage_collector.py b/repeater/data_acquisition/storage_collector.py index d8401bf..bc87cef 100644 --- a/repeater/data_acquisition/storage_collector.py +++ b/repeater/data_acquisition/storage_collector.py @@ -484,9 +484,17 @@ class StorageCollector: bw_hz: int = 62500, cr: int = 5, preamble: int = 17, + radio_profiles: Optional[list] = None, ) -> dict: return self.sqlite_handler.get_airtime_buckets( - start_timestamp, end_timestamp, bucket_seconds, sf, bw_hz, cr, preamble + start_timestamp, + end_timestamp, + bucket_seconds, + sf, + bw_hz, + cr, + preamble, + radio_profiles=radio_profiles, ) def get_packet_by_hash(self, packet_hash: str) -> Optional[dict]: diff --git a/repeater/web/api_endpoints.py b/repeater/web/api_endpoints.py index 0a9b971..17d8f0a 100644 --- a/repeater/web/api_endpoints.py +++ b/repeater/web/api_endpoints.py @@ -1851,6 +1851,9 @@ class APIEndpoints: if daemon is not None: meta = getattr(daemon, "radio_stack_meta", None) or {} stats["radio_stack"] = meta + # Authoritative per-radio air settings, so clients never have to infer + # which profile applies to which radio of a Fabric. + stats["radio_profiles"] = self._active_radio_profiles() stats["site_name"] = self.config.get("web", {}).get("site_name", "") stats["version"] = __version__ try: @@ -4080,6 +4083,28 @@ class APIEndpoints: logger.error(f"Error getting airtime data: {e}") return self._error(e) + def _active_radio_profiles(self) -> list: + """Air settings of the radios currently on the air, in configured order. + + Rebuilt from the live config rather than read from the boot-time + ``radio_stack_meta`` snapshot, so a live radio reconfiguration is + reflected without a restart. Falls back to the daemon's snapshot if the + config cannot be read. + """ + try: + from repeater.config import build_radio_profiles + + profiles = build_radio_profiles(self.config) + if profiles: + return profiles + except Exception as e: + logger.warning(f"Could not derive radio profiles from config: {e}") + + daemon = getattr(self, "daemon_instance", None) + meta = getattr(daemon, "radio_stack_meta", None) or {} if daemon is not None else {} + snapshot = meta.get("radio_profiles") + return list(snapshot) if isinstance(snapshot, list) else [] + @cherrypy.expose @cherrypy.tools.json_out() def airtime_chart_data( @@ -4095,7 +4120,15 @@ class APIEndpoints: """Server-side aggregated airtime utilization for chart rendering. Returns pre-bucketed rx_ms/tx_ms per time bucket instead of raw packet rows, - reducing response size from potentially hundreds of KB to a few KB. + reducing response size from potentially hundreds of KB to a few KB. The + response carries one series per active radio under ``radios`` alongside the + legacy combined fields. + + The server's own radio configuration decides which profile each radio's + airtime is computed with; a two-radio Fabric must never have one + caller-supplied profile applied to both sides. The ``sf``/``bw_hz``/``cr``/ + ``preamble`` query parameters are kept only as a fallback for callers + talking to a configuration this server cannot read profiles from. """ try: now = __import__("time").time() @@ -4110,6 +4143,7 @@ class APIEndpoints: bw_hz=int(bw_hz), cr=int(cr), preamble=int(preamble), + radio_profiles=self._active_radio_profiles(), ) return self._success(result) except Exception as e: diff --git a/repeater/web/openapi.yaml b/repeater/web/openapi.yaml index d5ac71a..fb02a4e 100644 --- a/repeater/web/openapi.yaml +++ b/repeater/web/openapi.yaml @@ -3768,21 +3768,28 @@ paths: default: 60 - name: sf in: query + description: >- + Fallback spreading factor. Ignored whenever the server can read its own + radio profiles, which are authoritative and are never applied across + two Fabric radios. schema: type: integer default: 9 - name: bw_hz in: query + description: Fallback bandwidth in Hz. See `sf`. schema: type: integer default: 62500 - name: cr in: query + description: Fallback coding-rate denominator (5-8). See `sf`. schema: type: integer default: 5 - name: preamble in: query + description: Fallback preamble length in symbols. See `sf`. schema: type: integer default: 17 @@ -3793,6 +3800,58 @@ paths: application/json: schema: type: object + properties: + success: + type: boolean + data: + type: object + properties: + bucket_seconds: + type: integer + description: Width of each bucket in seconds. + buckets: + type: array + description: >- + Combined series across every radio, retained for UI builds + that predate `radios`. + items: + $ref: '#/components/schemas/AirtimeBucket' + rx_total: + type: integer + description: Reception events across every radio in the window. + tx_total: + type: integer + description: Transmission events across every radio in the window. + radios: + type: array + description: >- + One series per active radio, in configured order. A relayed + packet appears as RX on its ingress radio and TX on every + radio that successfully transmitted it. + items: + type: object + properties: + radio_id: + type: string + profile: + $ref: '#/components/schemas/AirtimeRadioProfile' + buckets: + type: array + items: + $ref: '#/components/schemas/AirtimeBucket' + rx_total: + type: integer + tx_total: + type: integer + unattributed_rx_count: + type: integer + description: >- + Receptions on a multi-radio node whose radio could not be + identified (history predating radio attribution, or a radio + no longer configured). Never guessed onto another radio. + unattributed_tx_count: + type: integer + description: Transmissions with no identifiable radio. See above. /adverts_count_by_contact_type: get: @@ -5155,6 +5214,47 @@ components: enum: [unknown] schemas: + AirtimeBucket: + type: object + properties: + timestamp: + type: integer + description: Unix start of the bucket, floored to bucket_seconds. + rx_ms: + type: number + description: Receive time-on-air in this bucket, in milliseconds. + tx_ms: + type: number + description: Transmit time-on-air in this bucket, in milliseconds. + rx_count: + type: integer + tx_count: + type: integer + + AirtimeRadioProfile: + type: object + description: >- + Air settings this radio's time-on-air was computed with. A null field means + the setting could not be read; airtime is reported as zero rather than + estimated from a default that was never on the air. + properties: + frequency_hz: + type: integer + nullable: true + bandwidth_hz: + type: integer + nullable: true + spreading_factor: + type: integer + nullable: true + coding_rate: + type: integer + nullable: true + description: Coding-rate denominator (5 = 4/5 ... 8 = 4/8). + preamble_length: + type: integer + nullable: true + PluginStatus: type: object properties: diff --git a/tests/test_airtime_per_radio.py b/tests/test_airtime_per_radio.py new file mode 100644 index 0000000..6feb11b --- /dev/null +++ b/tests/test_airtime_per_radio.py @@ -0,0 +1,350 @@ +"""Per-radio airtime attribution for Fabric bridges. + +Covers the reporting contract of ``get_airtime_buckets``: which radio each +stored packet is charged to, that each radio's own LoRa profile is used, and +that single-radio installations keep the response they had before. +""" + +from __future__ import annotations + +import time + +import pytest +from openhop_core.protocol.packet_utils import calculate_lora_airtime_ms + +from repeater.config import build_radio_profiles +from repeater.data_acquisition.sqlite_handler import SQLiteHandler + +LOCAL = { + "radio_id": "local", + "frequency_hz": 869618000, + "bandwidth_hz": 62500, + "spreading_factor": 8, + "coding_rate": 8, + "preamble_length": 32, +} +LINK = { + "radio_id": "link", + "frequency_hz": 864200000, + "bandwidth_hz": 62500, + "spreading_factor": 11, + "coding_rate": 8, + "preamble_length": 32, +} +BRIDGE = [LOCAL, LINK] + +BASE_TS = 1_700_000_000 # fixed so bucket boundaries are deterministic + + +def _store(handler: SQLiteHandler, **overrides) -> None: + record = { + "timestamp": BASE_TS, + "type": 1, + "route": 0, + "length": 64, + "transmitted": False, + "is_duplicate": False, + } + record.update(overrides) + handler.store_packet(record) + + +def _buckets(handler: SQLiteHandler, profiles=None, bucket_seconds: int = 60) -> dict: + return handler.get_airtime_buckets( + start_timestamp=BASE_TS - 3600, + end_timestamp=BASE_TS + 3600, + bucket_seconds=bucket_seconds, + radio_profiles=profiles, + ) + + +def _radio(result: dict, radio_id: str) -> dict: + return next(r for r in result["radios"] if r["radio_id"] == radio_id) + + +def _expected_ms(profile: dict, length: int) -> float: + return calculate_lora_airtime_ms( + length, + profile["spreading_factor"], + profile["bandwidth_hz"], + profile["coding_rate"], + profile["preamble_length"], + ) + + +@pytest.fixture +def handler(tmp_path) -> SQLiteHandler: + return SQLiteHandler(tmp_path) + + +def test_single_radio_response_stays_backward_compatible(handler): + _store(handler, length=40) + _store(handler, length=40, transmitted=True) + + result = _buckets(handler) + + # Legacy fields unchanged in shape and meaning. + assert result["bucket_seconds"] == 60 + assert result["rx_total"] == 1 + assert result["tx_total"] == 1 + assert len(result["buckets"]) == 1 + bucket = result["buckets"][0] + assert bucket["timestamp"] == (BASE_TS // 60) * 60 + assert bucket["rx_count"] == 1 and bucket["tx_count"] == 1 + assert bucket["rx_ms"] > 0 and bucket["tx_ms"] > 0 + + # ...and the additive fields describe the same single radio. + assert len(result["radios"]) == 1 + only = result["radios"][0] + assert only["rx_total"] == 1 and only["tx_total"] == 1 + assert only["buckets"] == result["buckets"] + assert result["unattributed_rx_count"] == 0 + assert result["unattributed_tx_count"] == 0 + + +def test_null_radio_ids_attribute_to_the_only_radio(handler): + """Pre-Fabric rows carry no radio ids; one radio leaves nothing to infer.""" + _store(handler, rx_radio_id=None, tx_radio_id=None, tx_radio_ids=None) + + result = _buckets(handler, [LOCAL]) + + assert _radio(result, "local")["rx_total"] == 1 + assert result["unattributed_rx_count"] == 0 + + +def test_forwarded_packet_charges_ingress_rx_and_every_successful_egress(handler): + _store( + handler, + length=50, + transmitted=True, + rx_radio_id="local", + tx_radio_id="link", + tx_radio_ids=["link", "local"], + ) + + result = _buckets(handler, BRIDGE) + local = _radio(result, "local") + link = _radio(result, "link") + + assert local["rx_total"] == 1 and local["tx_total"] == 1 + assert link["rx_total"] == 0 and link["tx_total"] == 1 + assert local["buckets"][0]["rx_ms"] == pytest.approx(_expected_ms(LOCAL, 50)) + assert local["buckets"][0]["tx_ms"] == pytest.approx(_expected_ms(LOCAL, 50)) + assert link["buckets"][0]["tx_ms"] == pytest.approx(_expected_ms(LINK, 50)) + # Combined legacy series is the sum of both radios. + assert result["rx_total"] == 1 + assert result["tx_total"] == 2 + + +def test_duplicate_reception_charges_only_the_radio_that_heard_it(handler): + _store(handler, rx_radio_id="link", is_duplicate=True, drop_reason="duplicate") + + result = _buckets(handler, BRIDGE) + + assert _radio(result, "link")["rx_total"] == 1 + assert _radio(result, "local")["rx_total"] == 0 + assert _radio(result, "local")["buckets"] == [] + + +def test_local_origin_packet_contributes_tx_only(handler): + _store(handler, transmitted=True, rx_radio_id=None, tx_radio_ids=["local"]) + + result = _buckets(handler, BRIDGE) + + assert _radio(result, "local")["tx_total"] == 1 + assert _radio(result, "local")["rx_total"] == 0 + assert result["rx_total"] == 0 + + +def test_partial_fanout_counts_only_successful_egresses(handler): + """``tx_radio_ids`` holds the radios that actually sent; a refused radio is absent.""" + _store( + handler, + transmitted=True, + rx_radio_id="link", + tx_radio_id="local", + tx_radio_ids=["local"], + ) + + result = _buckets(handler, BRIDGE) + + assert _radio(result, "local")["tx_total"] == 1 + assert _radio(result, "link")["tx_total"] == 0 + assert _radio(result, "link")["rx_total"] == 1 + + +def test_scalar_tx_radio_id_supports_pre_migration_rows(handler): + _store(handler, transmitted=True, rx_radio_id="local", tx_radio_id="link", tx_radio_ids=None) + + result = _buckets(handler, BRIDGE) + + assert _radio(result, "link")["tx_total"] == 1 + assert _radio(result, "local")["rx_total"] == 1 + assert result["unattributed_tx_count"] == 0 + + +def test_unknown_radio_attribution_is_reported_not_guessed(handler): + _store(handler, rx_radio_id=None) # Fabric history with no ingress id + _store(handler, transmitted=True, rx_radio_id=None, tx_radio_id=None, tx_radio_ids=None) + _store(handler, rx_radio_id="decommissioned") # radio no longer configured + + result = _buckets(handler, BRIDGE) + + assert result["unattributed_rx_count"] == 2 + assert result["unattributed_tx_count"] == 1 + for radio in result["radios"]: + assert radio["rx_total"] == 0 and radio["tx_total"] == 0 + # Counted in the legacy totals, but with no airtime invented for them. + assert result["rx_total"] == 2 and result["tx_total"] == 1 + assert result["buckets"][0]["rx_ms"] == 0.0 + + +def test_equal_lengths_at_sf8_and_sf11_differ_in_airtime(handler): + _store(handler, length=64, rx_radio_id="local") + _store(handler, length=64, rx_radio_id="link") + + result = _buckets(handler, BRIDGE) + local_ms = _radio(result, "local")["buckets"][0]["rx_ms"] + link_ms = _radio(result, "link")["buckets"][0]["rx_ms"] + + assert local_ms == pytest.approx(_expected_ms(LOCAL, 64)) + assert link_ms == pytest.approx(_expected_ms(LINK, 64)) + assert link_ms > local_ms * 4 # SF11 is far slower than SF8 at the same bandwidth + + +def test_per_entry_bandwidth_override_affects_only_that_radio(handler): + wide_link = dict(LINK, bandwidth_hz=250000) + _store(handler, length=64, rx_radio_id="local") + _store(handler, length=64, rx_radio_id="link") + + narrow = _buckets(handler, BRIDGE) + wide = _buckets(handler, [LOCAL, wide_link]) + + assert _radio(wide, "local")["buckets"][0]["rx_ms"] == pytest.approx( + _radio(narrow, "local")["buckets"][0]["rx_ms"] + ) + assert _radio(wide, "link")["buckets"][0]["rx_ms"] == pytest.approx(_expected_ms(wide_link, 64)) + assert ( + _radio(wide, "link")["buckets"][0]["rx_ms"] < _radio(narrow, "link")["buckets"][0]["rx_ms"] + ) + + +def test_unreadable_profile_reports_counts_without_inventing_airtime(handler): + broken = dict(LINK, spreading_factor=None) + _store(handler, rx_radio_id="link") + + result = _buckets(handler, [LOCAL, broken]) + link = _radio(result, "link") + + assert link["rx_total"] == 1 + assert link["buckets"][0]["rx_ms"] == 0.0 + assert link["profile"]["spreading_factor"] is None + + +def test_buckets_are_keyed_by_the_same_boundaries_for_every_radio(handler): + """Both panels must be directly comparable, so both must share bucket edges.""" + _store(handler, timestamp=BASE_TS + 5, rx_radio_id="local") + _store(handler, timestamp=BASE_TS + 15, rx_radio_id="link") + _store(handler, timestamp=BASE_TS + 125, rx_radio_id="link") + + result = _buckets(handler, BRIDGE) + local_edges = [b["timestamp"] for b in _radio(result, "local")["buckets"]] + link_edges = [b["timestamp"] for b in _radio(result, "link")["buckets"]] + + assert local_edges == [(BASE_TS // 60) * 60] + assert link_edges[0] == local_edges[0] + assert all(edge % 60 == 0 for edge in local_edges + link_edges) + # Edges come from the same grid, so the UI can index both series by timestamp. + assert set(link_edges) <= {b["timestamp"] for b in result["buckets"]} + + +def test_realistic_day_of_traffic_uses_the_timestamp_index(handler): + """A 24 h window stays a bounded index range-scan, not a table scan.""" + day_start = BASE_TS - 86400 + for i in range(5000): + _store( + handler, + timestamp=day_start + i * 17, + length=32 + (i % 40), + rx_radio_id="local" if i % 2 else "link", + ) + # Traffic outside the window that must not be read. + for i in range(2000): + _store(handler, timestamp=day_start - 10_000 - i, rx_radio_id="local") + + with handler._connect() as conn: + plan = conn.execute( + "EXPLAIN QUERY PLAN SELECT timestamp, length, transmitted, rx_radio_id, " + "tx_radio_id, tx_radio_ids FROM packets " + "WHERE timestamp >= ? AND timestamp <= ? ORDER BY timestamp ASC", + (day_start, BASE_TS), + ).fetchall() + assert any("idx_packets_timestamp" in str(tuple(row)) for row in plan), plan + + started = time.perf_counter() + result = handler.get_airtime_buckets( + start_timestamp=day_start, + end_timestamp=BASE_TS, + bucket_seconds=60, + radio_profiles=BRIDGE, + ) + elapsed = time.perf_counter() - started + + assert result["rx_total"] == 5000 + assert _radio(result, "local")["rx_total"] + _radio(result, "link")["rx_total"] == 5000 + assert elapsed < 5.0, f"24 h aggregation took {elapsed:.2f}s" + + +def test_config_profiles_feed_the_aggregator_in_configured_order(): + config = { + "radios": [ + { + "id": "local", + "radio_type": "sx1262", + "radio": { + "frequency": 869618000, + "bandwidth": 62500, + "spreading_factor": 8, + "coding_rate": 8, + "preamble_length": 32, + }, + }, + { + "id": "link", + "radio_type": "sx1262", + "radio": { + "frequency": 864200000, + "bandwidth": 62500, + "spreading_factor": 11, + "coding_rate": 8, + "preamble_length": 32, + }, + }, + ], + "fabric": {"tx_mode": "bridge", "default_radio": "local"}, + } + + profiles = build_radio_profiles(config) + + assert [p["radio_id"] for p in profiles] == ["local", "link"] + assert profiles[0]["spreading_factor"] == 8 + assert profiles[1]["spreading_factor"] == 11 + assert profiles[1]["frequency_hz"] == 864200000 + + +def test_partial_radio_config_reports_no_profiles_rather_than_a_smaller_fabric(): + """A one-profile answer would claim every packet belongs to that radio.""" + config = { + "radios": [ + {"id": "local", "radio_type": "sx1262", "radio": {"spreading_factor": 8}}, + {"radio_type": "sx1262", "radio": {"spreading_factor": 11}}, # no id + ], + } + + assert build_radio_profiles(config) == [] + + +def test_unsupported_radio_type_has_no_profile(): + """get_radio_for_board would reject it, so no radio was built from it.""" + assert build_radio_profiles({"radio_type": "sx1280", "radio": {"spreading_factor": 8}}) == [] + assert build_radio_profiles({"radio_type": "none"}) == []