feat(airtime): attribute airtime to the radio that carried it

A two-radio Fabric bridge runs each side at its own SF, so one combined
utilization line computed from a single caller-supplied profile describes
neither radio. Report airtime per radio instead.

build_radio_stack now records the effective air settings of every radio it
builds, in configured order, as radio_stack_meta["radio_profiles"]. The
same builder is exposed as build_radio_profiles() so /stats and the airtime
endpoint can read current settings from the live config rather than a boot
snapshot. Air settings are read the way get_radio_for_board reads them, and
a field that cannot be read is reported as null rather than defaulted, so
nothing is charted from settings that were never on the air.

get_airtime_buckets takes those profiles and charges each stored packet to
the radios that actually carried it: its ingress rx_radio_id and every
successful egress in tx_radio_ids, falling back to the scalar tx_radio_id
for pre-fan-out rows. A relayed packet therefore appears as RX on the radio
that heard it and TX on each radio that sent it, each at that radio's own
time-on-air. Duplicate receptions are included; they occupied the receiver.
With one profile every packet belongs to that radio, so single-radio and
pre-Fabric databases are 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 response is additive: legacy buckets/rx_total/tx_total remain as the
combined series, and radios[] carries one series per radio alongside
unattributed_rx_count/unattributed_tx_count. Airtime comes from the shared
core estimator instead of a second inline Semtech formula, memoised per
(radio, length). The sf/bw_hz/cr/preamble query parameters are kept only as
a fallback; server profiles are authoritative and are never applied across
both radios of a bridge.
This commit is contained in:
agessaman committed 2026-09-11 21:22:06 -07:00
1 parent f510aedf78
commit 76c1a69ca4
6 files changed
+805 -36

No files matched your search

+137 -1
View File
@@ -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
+174 -33
View File
@@ -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:
@@ -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]:
+35 -1
View File
@@ -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:
+100
View File
@@ -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:
+350
View File
@@ -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"}) == []