openhop_repeater/tests/test_per_radio_analytics.py

"""Per-radio analytics for multi-radio nodes.

Packet stats, route stats, packet rates and neighbour-link history gain a
per-radio breakdown when two or more radio profiles are active. A single-radio
node must get exactly the answer it had, so each of those is checked against
the same database queried with no profiles at all.
"""

from __future__ import annotations

import time

import pytest
from openhop_core.protocol import Packet
from openhop_core.protocol.constants import PH_TYPE_SHIFT, ROUTE_TYPE_FLOOD

from repeater.data_acquisition.sqlite_handler import (
    RADIO_PACKET_RATES_QUERY,
    RadioResolver,
    SQLiteHandler,
    packet_carriers,
)
from repeater.neighbour_links import NeighbourLinkTracker

LOCAL = {
    "radio_id": "local",
    "frequency_hz": 910100000,
    "bandwidth_hz": 500000,
    "spreading_factor": 7,
    "coding_rate": 5,
    "preamble_length": 17,
}
LINK = dict(LOCAL, radio_id="link", frequency_hz=910525000, bandwidth_hz=62500)
BRIDGE = [LOCAL, LINK]
SINGLE = [dict(LOCAL, radio_id="radio0")]


def _store(handler: SQLiteHandler, **overrides) -> None:
    record = {
        "timestamp": time.time() - 60,
        "type": 1,
        "route": 1,
        "length": 40,
        "transmitted": False,
        "is_duplicate": False,
        "rssi": -90,
        "snr": 5.0,
    }
    record.update(overrides)
    handler.store_packet(record)


@pytest.fixture
def handler(tmp_path) -> SQLiteHandler:
    return SQLiteHandler(tmp_path)


@pytest.fixture
def bridge_db(handler) -> SQLiteHandler:
    """One of every attribution case on a two-radio bridge."""
    # Heard on local, not repeated.
    _store(handler, rx_radio_id="local", rssi=-80, snr=8.0)
    # A duplicate heard on link.
    _store(
        handler, rx_radio_id="link", rssi=-100, snr=-2.0, is_duplicate=True, drop_reason="Duplicate"
    )
    # Heard on local, relayed out both radios.
    _store(
        handler,
        route=2,
        rx_radio_id="local",
        rssi=-70,
        snr=10.0,
        transmitted=True,
        tx_radio_id="link",
        tx_radio_ids=["link", "local"],
    )
    # Originated here and sent on both radios; no radio received it.
    _store(
        handler,
        route=2,
        rssi=0,
        snr=0.0,
        transmitted=True,
        tx_radio_id="local",
        tx_radio_ids=["local", "link"],
    )
    # A reception whose radio is not recorded.
    _store(handler, rx_radio_id=None)
    return handler


# ---------------------------------------------------------------------------
# Shared attribution rules
# ---------------------------------------------------------------------------


@pytest.mark.parametrize(
    "row, expected",
    [
        ((False, "local", None, None), (["local"], [])),
        ((False, None, None, None), ([None], [])),
        ((True, "local", "link", '["link", "local"]'), (["local"], ["link", "local"])),
        ((True, None, "local", '["local"]'), ([], ["local"])),
        ((True, "link", "local", None), (["link"], ["local"])),  # before fan-out
        ((True, "link", "local", "not json"), (["link"], ["local"])),
    ],
)
def test_packet_carriers(row, expected):
    assert packet_carriers(*row) == expected


def test_resolver_gives_one_radio_everything_and_two_radios_only_known_ids():
    single = RadioResolver(SINGLE)
    assert not single.multi
    assert single.resolve(None) == "radio0"
    assert single.resolve("anything") == "radio0"

    bridge = RadioResolver(BRIDGE)
    assert bridge.multi and bridge.order == ["local", "link"]
    assert bridge.resolve("link") == "link"
    assert bridge.resolve(None) is None
    assert bridge.resolve("decommissioned") is None


# ---------------------------------------------------------------------------
# /packet_stats
# ---------------------------------------------------------------------------


def test_single_radio_packet_stats_are_unchanged(bridge_db):
    assert bridge_db.get_packet_stats(
        hours=24, radio_profiles=SINGLE
    ) == bridge_db.get_packet_stats(hours=24)
    assert "radios" not in bridge_db.get_packet_stats(hours=24, radio_profiles=SINGLE)


def test_bridge_packet_stats_keep_the_legacy_totals(bridge_db):
    legacy = bridge_db.get_packet_stats(hours=24)
    bridge = bridge_db.get_packet_stats(hours=24, radio_profiles=BRIDGE)

    for key, value in legacy.items():
        assert bridge[key] == value, key
    # One transmitted packet per relay, however many radios carried it.
    assert bridge["transmitted_packets"] == 2


def test_bridge_packet_stats_break_down_per_radio(bridge_db):
    stats = bridge_db.get_packet_stats(hours=24, radio_profiles=BRIDGE)
    local, link = stats["radios"]

    assert local == {
        "radio_id": "local",
        "received": 2,
        "duplicates": 0,
        "dropped": 1,
        "transmissions": 2,
        "avg_rssi": -75.0,
        "avg_snr": 9.0,
    }
    assert link == {
        "radio_id": "link",
        "received": 1,
        "duplicates": 1,
        "dropped": 1,
        "transmissions": 2,
        "avg_rssi": -100.0,
        "avg_snr": -2.0,
    }
    assert stats["unattributed_rx_count"] == 1
    assert stats["unattributed_tx_count"] == 0


def test_packet_stats_cache_does_not_mix_single_and_bridge_answers(bridge_db):
    assert "radios" not in bridge_db.get_packet_stats(hours=24)
    assert "radios" in bridge_db.get_packet_stats(hours=24, radio_profiles=BRIDGE)
    assert "radios" not in bridge_db.get_packet_stats(hours=24)


def test_bridge_packet_stats_on_an_empty_window_match_legacy(handler):
    assert {
        k: v
        for k, v in handler.get_packet_stats(hours=24, radio_profiles=BRIDGE).items()
        if k not in ("radios", "unattributed_rx_count", "unattributed_tx_count")
    } == handler.get_packet_stats(hours=24)


# ---------------------------------------------------------------------------
# /route_stats
# ---------------------------------------------------------------------------


def test_single_radio_route_stats_are_unchanged(bridge_db):
    single = bridge_db.get_route_stats(hours=24, radio_profiles=SINGLE)
    assert single == bridge_db.get_route_stats(hours=24)
    assert single["route_totals"] == {"Flood": 3, "Direct": 2}
    assert "radios" not in single


def test_bridge_route_stats_split_receptions_per_radio(bridge_db):
    stats = bridge_db.get_route_stats(hours=24, radio_profiles=BRIDGE)

    assert stats["route_totals"] == bridge_db.get_route_stats(hours=24)["route_totals"]
    assert stats["radios"] == [
        {"radio_id": "local", "route_totals": {"Flood": 1, "Direct": 1}, "total_packets": 2},
        {"radio_id": "link", "route_totals": {"Flood": 1}, "total_packets": 1},
    ]
    assert stats["originated"] == {"route_totals": {"Direct": 1}, "total_packets": 1}
    assert stats["unattributed_count"] == 1


# ---------------------------------------------------------------------------
# /radio_packet_rates
# ---------------------------------------------------------------------------


def test_packet_rates_follow_the_airtime_attribution(bridge_db):
    now = time.time()
    rates = bridge_db.get_radio_packet_rates(now - 3600, now, 3600, radio_profiles=BRIDGE)
    airtime = bridge_db.get_airtime_buckets(now - 3600, now, 3600, radio_profiles=BRIDGE)

    by_radio = {r["radio_id"]: r for r in rates["radios"]}
    assert (by_radio["local"]["rx_total"], by_radio["local"]["tx_total"]) == (2, 2)
    assert (by_radio["link"]["rx_total"], by_radio["link"]["tx_total"]) == (1, 2)
    for radio in airtime["radios"]:
        assert by_radio[radio["radio_id"]]["rx_total"] == radio["rx_total"]
        assert by_radio[radio["radio_id"]]["tx_total"] == radio["tx_total"]
    assert rates["unattributed_rx_count"] == airtime["unattributed_rx_count"] == 1
    assert rates["unattributed_tx_count"] == airtime["unattributed_tx_count"] == 0

    # Every row sits in one bucket, so each radio's bucket counts sum to its totals.
    for radio in rates["radios"]:
        assert sum(b["rx_count"] for b in radio["buckets"]) == radio["rx_total"]
        assert sum(b["tx_count"] for b in radio["buckets"]) == radio["tx_total"]
        assert all(b["timestamp"] % 3600 == 0 for b in radio["buckets"])


def test_packet_rates_attribute_null_ids_to_a_single_radio(handler):
    _store(handler, rx_radio_id=None)
    _store(handler, transmitted=True, rx_radio_id=None, tx_radio_id=None)
    now = time.time()

    rates = handler.get_radio_packet_rates(now - 3600, now, 3600, radio_profiles=SINGLE)

    assert [(r["radio_id"], r["rx_total"], r["tx_total"]) for r in rates["radios"]] == [
        ("radio0", 1, 1)
    ]
    assert rates["unattributed_rx_count"] == 0


def test_week_of_packet_rates_is_served_from_the_covering_index(handler):
    now = time.time()
    with handler._connect() as conn:
        plan = conn.execute(
            "EXPLAIN QUERY PLAN " + RADIO_PACKET_RATES_QUERY, (3600, 3600, now - 7 * 86400, now)
        ).fetchall()
    assert any("COVERING INDEX idx_packets_airtime" in str(tuple(row)) for row in plan), plan


# ---------------------------------------------------------------------------
# /neighbor_link_history
# ---------------------------------------------------------------------------


@pytest.fixture
def link_history(handler) -> SQLiteHandler:
    base = time.time() - 600
    for i, radio_id in enumerate(["local", "link", "local", None]):
        _store(
            handler,
            timestamp=base + i,
            upstream_hash="AB",
            upstream_hash_size=1,
            rx_radio_id=radio_id,
            rssi=-80 - i,
        )
    return handler


def test_history_rows_name_the_receiving_radio(link_history):
    rows = link_history.get_neighbor_link_history(peer_hash="ab", path_hash_size=1)

    assert [row.get("rx_radio_id") for row in rows] == ["local", "link", "local", None]
    assert "rx_radio_id" not in rows[-1]  # absent, not null, when unrecorded


def test_history_filters_to_one_radio(link_history):
    rows = link_history.get_neighbor_link_history(
        peer_hash="AB", path_hash_size=1, radio_id="local"
    )
    assert [row["rssi"] for row in rows] == [-80, -82]

    buckets = link_history.get_neighbor_link_history(
        peer_hash="AB", path_hash_size=1, bucket_seconds=3600, radio_id="link"
    )
    assert sum(b["n"] for b in buckets) == 1


def test_history_buckets_split_per_radio_only_when_asked(link_history):
    merged = link_history.get_neighbor_link_history(
        peer_hash="AB", path_hash_size=1, bucket_seconds=86400
    )
    assert len(merged) == 1 and merged[0]["n"] == 4
    assert "radio_id" not in merged[0]

    split = link_history.get_neighbor_link_history(
        peer_hash="AB", path_hash_size=1, bucket_seconds=86400, by_radio=True
    )
    assert {b["radio_id"]: b["n"] for b in split} == {"local": 2, "link": 1, None: 1}


# ---------------------------------------------------------------------------
# Neighbour link tracker
# ---------------------------------------------------------------------------


def _flood_from(peer: int) -> Packet:
    pkt = Packet()
    pkt.header = ROUTE_TYPE_FLOOD | (1 << PH_TYPE_SHIFT)
    pkt.payload = bytearray(b"\x01\x02\x03")
    pkt.payload_len = 3
    pkt.path = bytearray([peer])
    pkt.path_len = 1
    return pkt


def _observe(tracker, rx_radio_id, rssi, is_duplicate=False):
    tracker.observe(
        _flood_from(0xAB),
        route_type=ROUTE_TYPE_FLOOD,
        payload_type=1,
        rssi=rssi,
        snr=5.0,
        score=0.5,
        is_duplicate=is_duplicate,
        rx_radio_id=rx_radio_id,
    )


def test_tracker_keeps_merged_and_per_radio_link_stats():
    tracker = NeighbourLinkTracker({})
    _observe(tracker, "link", -100)
    _observe(tracker, "local", -70)
    _observe(tracker, "local", -60, is_duplicate=True)

    (link,) = tracker.snapshot(radio_ids=["local", "link"])

    assert link["sample_count"] == 3 and link["duplicate_sample_count"] == 1
    assert [r["radio_id"] for r in link["radios"]] == ["local", "link"]  # configured order
    local, far = link["radios"]
    assert (local["sample_count"], local["duplicate_sample_count"]) == (2, 1)
    assert (local["last_rssi"], far["last_rssi"]) == (-60.0, -100.0)
    assert far["sample_count"] == 1 and far["ewma_rssi"] == -100.0


def test_tracker_snapshot_is_unchanged_without_two_radios():
    tracker = NeighbourLinkTracker({})
    _observe(tracker, None, -80)
    _observe(tracker, "radio0", -80)

    for radio_ids in (None, [], ["radio0"]):
        (link,) = tracker.snapshot(radio_ids=radio_ids)
        assert "radios" not in link


def test_tracker_lists_a_radio_no_longer_configured_after_the_configured_ones():
    tracker = NeighbourLinkTracker({})
    _observe(tracker, "old", -90)
    _observe(tracker, "link", -80)

    (link,) = tracker.snapshot(radio_ids=["local", "link"])

    assert [r["radio_id"] for r in link["radios"]] == ["link", "old"]