From b2eb45b199862d668246c79f4d54e047bcdcaf20 Mon Sep 17 00:00:00 2001 From: Lloyd Date: Fri, 10 Jul 2026 13:36:49 +0100 Subject: [PATCH] feat: Add LBT diagnostics endpoint with correlation analysis - Implemented `lbt_diagnostics` API endpoint to return aggregated Listen Before Talk (LBT) diagnostics aligned with RF metrics. - Introduced methods for calculating Pearson correlation coefficients and auto-bucket sizing for diagnostics. - Enhanced data aggregation logic in `StorageCollector` for LBT diagnostics. - Updated OpenAPI specification to include new endpoint and response schemas. - Added comprehensive unit tests for LBT diagnostics, including validation of correlation calculations and data integrity. --- repeater/data_acquisition/sqlite_handler.py | 519 ++++++++++++++++++ .../data_acquisition/storage_collector.py | 14 + repeater/web/api_endpoints.py | 290 ++++++++++ repeater/web/openapi.yaml | 435 +++++++++++++++ tests/test_api_endpoints_core_coverage.py | 172 ++++++ tests/test_sqlite_handler_easy.py | 163 ++++++ 6 files changed, 1593 insertions(+) diff --git a/repeater/data_acquisition/sqlite_handler.py b/repeater/data_acquisition/sqlite_handler.py index 4c6a4b8..73becd8 100644 --- a/repeater/data_acquisition/sqlite_handler.py +++ b/repeater/data_acquisition/sqlite_handler.py @@ -1,6 +1,7 @@ import base64 import json import logging +import math import secrets import sqlite3 import threading @@ -944,6 +945,524 @@ class SQLiteHandler: logger.error(f"Failed to get policy event counts: {e}") return [] + def get_lbt_diagnostics( + self, + start_timestamp: float, + end_timestamp: float, + bucket_seconds: int = 300, + severe_attempt_threshold: int = 4, + ) -> dict: + """Return aggregated LBT diagnostics for TX-path packets. + + LBT metadata in packets is persisted as "extra attempts/backoffs" where: + - lbt_attempts == 0 means first CAD/LBT check was clear + - total attempts/checks ~= lbt_attempts + 1 + + This method avoids returning raw packet rows and instead returns + bucketed aggregates + summary metrics for efficient dashboard refreshes. + """ + + def _weighted_percentile(attempt_counts: dict, q: float) -> Optional[float]: + total = sum(int(v) for v in attempt_counts.values()) + if total <= 0: + return None + + q = max(0.0, min(1.0, float(q))) + # Use nearest-rank percentile so p95 on sparse samples doesn't + # systematically under-report tail attempts. + rank = max(1, int(math.ceil(total * q))) + running = 0 + for attempt in sorted(int(k) for k in attempt_counts.keys()): + running += int(attempt_counts.get(attempt, 0)) + if running >= rank: + return float(attempt) + return float(max(int(k) for k in attempt_counts.keys())) + + def _packet_type_name(pkt_type: int) -> str: + try: + from openhop_core.protocol.utils import PAYLOAD_TYPES as _PT + + labels = { + "REQ": "Request", + "RESPONSE": "Response", + "TXT_MSG": "Plain Text Message", + "ACK": "Acknowledgment", + "ADVERT": "Node Advertisement", + "GRP_TXT": "Group Text Message", + "GRP_DATA": "Group Datagram", + "ANON_REQ": "Anonymous Request", + "PATH": "Returned Path", + "TRACE": "Trace", + "MULTIPART": "Multi-part Packet", + "CONTROL": "Control", + "RAW_CUSTOM": "Custom Packet", + } + code = _PT.get(pkt_type) + if not code: + return ( + f"Reserved Type {pkt_type}" if 0 <= pkt_type <= 15 else f"Type {pkt_type}" + ) + return f"{labels.get(code, code.replace('_', ' ').title())} ({code})" + except Exception: + return f"Reserved Type {pkt_type}" if 0 <= pkt_type <= 15 else f"Type {pkt_type}" + + try: + bucket_seconds = max(60, min(int(bucket_seconds), 3600)) + severe_attempt_threshold = max(2, int(severe_attempt_threshold)) + + if end_timestamp < start_timestamp: + start_timestamp, end_timestamp = end_timestamp, start_timestamp + + tx_filter = "(transmitted = 1 OR lbt_attempts > 0 OR drop_reason LIKE 'TX failed%')" + + with self._connect() as conn: + conn.row_factory = sqlite3.Row + + aggregate_rows = conn.execute( + f""" + WITH tx_packets AS ( + SELECT + CAST(timestamp / ? AS INTEGER) * ? AS bucket_ts, + CASE + WHEN lbt_attempts IS NULL OR lbt_attempts < 0 THEN 1 + ELSE lbt_attempts + 1 + END AS attempts_total, + CASE WHEN transmitted = 1 THEN 1 ELSE 0 END AS tx_success, + CASE + WHEN transmitted = 0 AND drop_reason LIKE 'TX failed%' THEN 1 + ELSE 0 + END AS failed_tx, + CASE WHEN COALESCE(lbt_channel_busy, 0) = 1 THEN 1 ELSE 0 END AS busy + FROM packets INDEXED BY idx_packets_timestamp + WHERE timestamp >= ? + AND timestamp <= ? + AND {tx_filter} + ) + SELECT + bucket_ts, + COUNT(*) AS transmissions, + SUM(attempts_total) AS total_attempts, + SUM(CASE WHEN attempts_total = 1 THEN 1 ELSE 0 END) AS attempts_1, + SUM(CASE WHEN attempts_total = 2 THEN 1 ELSE 0 END) AS attempts_2, + SUM(CASE WHEN attempts_total = 3 THEN 1 ELSE 0 END) AS attempts_3, + SUM(CASE WHEN attempts_total >= 4 THEN 1 ELSE 0 END) AS attempts_4_plus, + SUM(CASE WHEN attempts_total > 1 THEN 1 ELSE 0 END) AS retry_packets, + SUM(CASE WHEN tx_success = 1 AND attempts_total = 1 THEN 1 ELSE 0 END) AS first_attempt_success, + SUM(failed_tx) AS failed_transmissions, + SUM(busy) AS busy_channel_events, + SUM(CASE WHEN attempts_total >= ? THEN 1 ELSE 0 END) AS severe_contention_count, + MAX(attempts_total) AS max_attempts + FROM tx_packets + GROUP BY bucket_ts + ORDER BY bucket_ts ASC + """, + ( + bucket_seconds, + bucket_seconds, + float(start_timestamp), + float(end_timestamp), + severe_attempt_threshold, + ), + ).fetchall() + + dist_rows = conn.execute( + f""" + WITH tx_packets AS ( + SELECT + CAST(timestamp / ? AS INTEGER) * ? AS bucket_ts, + CASE + WHEN lbt_attempts IS NULL OR lbt_attempts < 0 THEN 1 + ELSE lbt_attempts + 1 + END AS attempts_total + FROM packets INDEXED BY idx_packets_timestamp + WHERE timestamp >= ? + AND timestamp <= ? + AND {tx_filter} + ) + SELECT bucket_ts, attempts_total, COUNT(*) AS cnt + FROM tx_packets + GROUP BY bucket_ts, attempts_total + ORDER BY bucket_ts ASC, attempts_total ASC + """, + ( + bucket_seconds, + bucket_seconds, + float(start_timestamp), + float(end_timestamp), + ), + ).fetchall() + + type_rows = conn.execute( + f""" + WITH tx_packets AS ( + SELECT + CAST(timestamp / ? AS INTEGER) * ? AS bucket_ts, + type AS packet_type, + CASE + WHEN lbt_attempts IS NULL OR lbt_attempts < 0 THEN 1 + ELSE lbt_attempts + 1 + END AS attempts_total, + CASE WHEN transmitted = 1 THEN 1 ELSE 0 END AS tx_success, + CASE + WHEN transmitted = 0 AND drop_reason LIKE 'TX failed%' THEN 1 + ELSE 0 + END AS failed_tx + FROM packets INDEXED BY idx_packets_timestamp + WHERE timestamp >= ? + AND timestamp <= ? + AND {tx_filter} + ) + SELECT + bucket_ts, + packet_type, + COUNT(*) AS transmissions, + SUM(attempts_total) AS total_attempts, + SUM(CASE WHEN attempts_total = 1 THEN 1 ELSE 0 END) AS attempts_1, + SUM(CASE WHEN attempts_total = 2 THEN 1 ELSE 0 END) AS attempts_2, + SUM(CASE WHEN attempts_total = 3 THEN 1 ELSE 0 END) AS attempts_3, + SUM(CASE WHEN attempts_total >= 4 THEN 1 ELSE 0 END) AS attempts_4_plus, + SUM(CASE WHEN attempts_total > 1 THEN 1 ELSE 0 END) AS retry_packets, + SUM(CASE WHEN tx_success = 1 AND attempts_total = 1 THEN 1 ELSE 0 END) AS first_attempt_success, + SUM(failed_tx) AS failed_transmissions, + SUM(CASE WHEN attempts_total >= ? THEN 1 ELSE 0 END) AS severe_contention_count, + MAX(attempts_total) AS max_attempts + FROM tx_packets + GROUP BY bucket_ts, packet_type + ORDER BY bucket_ts ASC, packet_type ASC + """, + ( + bucket_seconds, + bucket_seconds, + float(start_timestamp), + float(end_timestamp), + severe_attempt_threshold, + ), + ).fetchall() + + dist_by_bucket: dict = {} + overall_dist: dict = {} + for row in dist_rows: + bucket_ts = int(row["bucket_ts"]) + attempt = int(row["attempts_total"]) + count = int(row["cnt"]) + bucket_dist = dist_by_bucket.setdefault(bucket_ts, {}) + bucket_dist[attempt] = bucket_dist.get(attempt, 0) + count + overall_dist[attempt] = overall_dist.get(attempt, 0) + count + + bucket_map: dict = {} + start_bucket = int(float(start_timestamp) // bucket_seconds) * bucket_seconds + end_bucket = int(float(end_timestamp) // bucket_seconds) * bucket_seconds + for bucket_ts in range(start_bucket, end_bucket + 1, bucket_seconds): + bucket_map[bucket_ts] = { + "timestamp": bucket_ts, + "transmissions": 0, + "total_attempts": 0, + "attempts_1": 0, + "attempts_2": 0, + "attempts_3": 0, + "attempts_4_plus": 0, + "retry_packets": 0, + "first_attempt_success": 0, + "failed_transmissions": 0, + "busy_channel_events": 0, + "severe_contention_count": 0, + "max_attempts": 0, + } + + for row in aggregate_rows: + bucket_ts = int(row["bucket_ts"]) + if bucket_ts not in bucket_map: + bucket_map[bucket_ts] = { + "timestamp": bucket_ts, + "transmissions": 0, + "total_attempts": 0, + "attempts_1": 0, + "attempts_2": 0, + "attempts_3": 0, + "attempts_4_plus": 0, + "retry_packets": 0, + "first_attempt_success": 0, + "failed_transmissions": 0, + "busy_channel_events": 0, + "severe_contention_count": 0, + "max_attempts": 0, + } + bucket_map[bucket_ts].update( + { + "transmissions": int(row["transmissions"] or 0), + "total_attempts": int(row["total_attempts"] or 0), + "attempts_1": int(row["attempts_1"] or 0), + "attempts_2": int(row["attempts_2"] or 0), + "attempts_3": int(row["attempts_3"] or 0), + "attempts_4_plus": int(row["attempts_4_plus"] or 0), + "retry_packets": int(row["retry_packets"] or 0), + "first_attempt_success": int(row["first_attempt_success"] or 0), + "failed_transmissions": int(row["failed_transmissions"] or 0), + "busy_channel_events": int(row["busy_channel_events"] or 0), + "severe_contention_count": int(row["severe_contention_count"] or 0), + "max_attempts": int(row["max_attempts"] or 0), + } + ) + + buckets = [] + for bucket_ts in sorted(bucket_map.keys()): + bucket = bucket_map[bucket_ts] + transmissions = int(bucket["transmissions"]) + total_attempts = int(bucket["total_attempts"]) + attempts_3_plus = int(bucket["attempts_3"] + bucket["attempts_4_plus"]) + + median_attempts = _weighted_percentile(dist_by_bucket.get(bucket_ts, {}), 0.5) + p95_attempts = _weighted_percentile(dist_by_bucket.get(bucket_ts, {}), 0.95) + + retry_rate_pct = None + first_attempt_success_rate_pct = None + avg_attempts = None + attempts_3_plus_pct = None + attempts_4_plus_pct = None + severe_contention_pct = None + + if transmissions > 0: + retry_rate_pct = (bucket["retry_packets"] * 100.0) / transmissions + first_attempt_success_rate_pct = ( + bucket["first_attempt_success"] * 100.0 + ) / transmissions + avg_attempts = total_attempts / transmissions + attempts_3_plus_pct = (attempts_3_plus * 100.0) / transmissions + attempts_4_plus_pct = (bucket["attempts_4_plus"] * 100.0) / transmissions + severe_contention_pct = ( + bucket["severe_contention_count"] * 100.0 + ) / transmissions + + buckets.append( + { + "timestamp": bucket_ts, + "transmissions": transmissions, + "total_attempts": total_attempts, + "first_attempt_success": int(bucket["first_attempt_success"]), + "retry_packets": int(bucket["retry_packets"]), + "retry_rate_pct": retry_rate_pct, + "first_attempt_success_rate_pct": first_attempt_success_rate_pct, + "avg_attempts": avg_attempts, + "median_attempts": median_attempts, + "p95_attempts": p95_attempts, + "max_attempts": int(bucket["max_attempts"]), + "attempts_1": int(bucket["attempts_1"]), + "attempts_2": int(bucket["attempts_2"]), + "attempts_3": int(bucket["attempts_3"]), + "attempts_4_plus": int(bucket["attempts_4_plus"]), + "attempts_3_plus": int(attempts_3_plus), + "attempts_3_plus_pct": attempts_3_plus_pct, + "attempts_4_plus_pct": attempts_4_plus_pct, + "failed_transmissions": int(bucket["failed_transmissions"]), + "busy_channel_events": int(bucket["busy_channel_events"]), + "severe_contention_count": int(bucket["severe_contention_count"]), + "severe_contention_pct": severe_contention_pct, + } + ) + + total_transmissions = int(sum(b["transmissions"] for b in buckets)) + total_attempts = int(sum(b["total_attempts"] for b in buckets)) + first_attempt_success = int(sum(b["first_attempt_success"] for b in buckets)) + retry_packets = int(sum(b["retry_packets"] for b in buckets)) + attempts_1 = int(sum(b["attempts_1"] for b in buckets)) + attempts_2 = int(sum(b["attempts_2"] for b in buckets)) + attempts_3 = int(sum(b["attempts_3"] for b in buckets)) + attempts_4_plus = int(sum(b["attempts_4_plus"] for b in buckets)) + attempts_3_plus = int(attempts_3 + attempts_4_plus) + failed_transmissions = int(sum(b["failed_transmissions"] for b in buckets)) + busy_channel_events = int(sum(b["busy_channel_events"] for b in buckets)) + severe_contention_count = int(sum(b["severe_contention_count"] for b in buckets)) + max_attempts = int(max([b["max_attempts"] for b in buckets], default=0)) + + retry_rate_pct = None + first_attempt_success_rate_pct = None + avg_attempts = None + attempts_3_plus_pct = None + attempts_4_plus_pct = None + severe_contention_pct = None + + if total_transmissions > 0: + retry_rate_pct = (retry_packets * 100.0) / total_transmissions + first_attempt_success_rate_pct = ( + first_attempt_success * 100.0 + ) / total_transmissions + avg_attempts = total_attempts / total_transmissions + attempts_3_plus_pct = (attempts_3_plus * 100.0) / total_transmissions + attempts_4_plus_pct = (attempts_4_plus * 100.0) / total_transmissions + severe_contention_pct = (severe_contention_count * 100.0) / total_transmissions + + worst_bucket = None + scored_buckets = [ + b + for b in buckets + if int(b.get("transmissions", 0)) > 0 and b.get("retry_rate_pct") is not None + ] + if scored_buckets: + worst = max( + scored_buckets, key=lambda item: float(item.get("retry_rate_pct") or 0.0) + ) + worst_bucket = { + "timestamp": int(worst["timestamp"]), + "retry_rate_pct": float(worst.get("retry_rate_pct") or 0.0), + "attempts_3_plus_pct": float(worst.get("attempts_3_plus_pct") or 0.0), + "max_attempts": int(worst.get("max_attempts") or 0), + "transmissions": int(worst.get("transmissions") or 0), + } + + summary = { + "total_transmissions": total_transmissions, + "total_attempts": total_attempts, + "first_attempt_success": first_attempt_success, + "retry_packets": retry_packets, + "retry_rate_pct": retry_rate_pct, + "first_attempt_success_rate_pct": first_attempt_success_rate_pct, + "avg_attempts": avg_attempts, + "median_attempts": _weighted_percentile(overall_dist, 0.5), + "p95_attempts": _weighted_percentile(overall_dist, 0.95), + "max_attempts": max_attempts, + "attempts_1": attempts_1, + "attempts_2": attempts_2, + "attempts_3": attempts_3, + "attempts_4_plus": attempts_4_plus, + "attempts_3_plus": attempts_3_plus, + "attempts_3_plus_pct": attempts_3_plus_pct, + "attempts_4_plus_pct": attempts_4_plus_pct, + "failed_transmissions": failed_transmissions, + "busy_channel_events": busy_channel_events, + "severe_contention_count": severe_contention_count, + "severe_contention_pct": severe_contention_pct, + "severe_attempt_threshold": severe_attempt_threshold, + "has_lbt_data": total_transmissions > 0, + "worst_bucket": worst_bucket, + } + + packet_type_totals: dict = {} + packet_type_buckets = [] + for row in type_rows: + bucket_ts = int(row["bucket_ts"]) + packet_type = int(row["packet_type"] if row["packet_type"] is not None else -1) + transmissions = int(row["transmissions"] or 0) + total_attempts_for_type = int(row["total_attempts"] or 0) + attempts_3_plus = int((row["attempts_3"] or 0) + (row["attempts_4_plus"] or 0)) + + retry_rate_pct_for_type = None + first_attempt_success_rate_pct_for_type = None + avg_attempts_for_type = None + attempts_3_plus_pct_for_type = None + if transmissions > 0: + retry_rate_pct_for_type = ( + int(row["retry_packets"] or 0) * 100.0 + ) / transmissions + first_attempt_success_rate_pct_for_type = ( + int(row["first_attempt_success"] or 0) * 100.0 + ) / transmissions + avg_attempts_for_type = total_attempts_for_type / transmissions + attempts_3_plus_pct_for_type = (attempts_3_plus * 100.0) / transmissions + + packet_type_buckets.append( + { + "timestamp": bucket_ts, + "packet_type": packet_type, + "packet_type_label": _packet_type_name(packet_type), + "transmissions": transmissions, + "total_attempts": total_attempts_for_type, + "first_attempt_success": int(row["first_attempt_success"] or 0), + "retry_packets": int(row["retry_packets"] or 0), + "retry_rate_pct": retry_rate_pct_for_type, + "first_attempt_success_rate_pct": first_attempt_success_rate_pct_for_type, + "avg_attempts": avg_attempts_for_type, + "attempts_1": int(row["attempts_1"] or 0), + "attempts_2": int(row["attempts_2"] or 0), + "attempts_3": int(row["attempts_3"] or 0), + "attempts_4_plus": int(row["attempts_4_plus"] or 0), + "attempts_3_plus": attempts_3_plus, + "attempts_3_plus_pct": attempts_3_plus_pct_for_type, + "max_attempts": int(row["max_attempts"] or 0), + "failed_transmissions": int(row["failed_transmissions"] or 0), + "severe_contention_count": int(row["severe_contention_count"] or 0), + } + ) + + total_entry = packet_type_totals.setdefault( + packet_type, + { + "packet_type": packet_type, + "packet_type_label": _packet_type_name(packet_type), + "transmissions": 0, + "retry_packets": 0, + }, + ) + total_entry["transmissions"] += transmissions + total_entry["retry_packets"] += int(row["retry_packets"] or 0) + + packet_types = [] + for pkt_type in sorted( + packet_type_totals.keys(), + key=lambda key: packet_type_totals[key]["transmissions"], + reverse=True, + ): + entry = packet_type_totals[pkt_type] + transmissions = int(entry["transmissions"]) + retry_rate_pct_for_type = None + if transmissions > 0: + retry_rate_pct_for_type = (int(entry["retry_packets"]) * 100.0) / transmissions + packet_types.append( + { + "packet_type": int(entry["packet_type"]), + "packet_type_label": str(entry["packet_type_label"]), + "transmissions": transmissions, + "retry_packets": int(entry["retry_packets"]), + "retry_rate_pct": retry_rate_pct_for_type, + } + ) + + return { + "start_time": int(start_timestamp), + "end_time": int(end_timestamp), + "bucket_seconds": bucket_seconds, + "summary": summary, + "buckets": buckets, + "packet_types": packet_types, + "packet_type_buckets": packet_type_buckets, + } + + except Exception as e: + logger.error(f"Failed to get LBT diagnostics: {e}") + return { + "start_time": int(start_timestamp), + "end_time": int(end_timestamp), + "bucket_seconds": max(60, min(int(bucket_seconds), 3600)), + "summary": { + "total_transmissions": 0, + "total_attempts": 0, + "first_attempt_success": 0, + "retry_packets": 0, + "retry_rate_pct": None, + "first_attempt_success_rate_pct": None, + "avg_attempts": None, + "median_attempts": None, + "p95_attempts": None, + "max_attempts": 0, + "attempts_1": 0, + "attempts_2": 0, + "attempts_3": 0, + "attempts_4_plus": 0, + "attempts_3_plus": 0, + "attempts_3_plus_pct": None, + "attempts_4_plus_pct": None, + "failed_transmissions": 0, + "busy_channel_events": 0, + "severe_contention_count": 0, + "severe_contention_pct": None, + "severe_attempt_threshold": max(2, int(severe_attempt_threshold)), + "has_lbt_data": False, + "worst_bucket": None, + }, + "buckets": [], + "packet_types": [], + "packet_type_buckets": [], + } + def get_packet_stats(self, hours: int = 24) -> dict: try: now = time.time() diff --git a/repeater/data_acquisition/storage_collector.py b/repeater/data_acquisition/storage_collector.py index 89fff18..e9a2a1d 100644 --- a/repeater/data_acquisition/storage_collector.py +++ b/repeater/data_acquisition/storage_collector.py @@ -380,6 +380,20 @@ class StorageCollector: bucket_seconds=bucket_seconds, ) + def get_lbt_diagnostics( + self, + start_timestamp: float, + end_timestamp: float, + bucket_seconds: int = 300, + severe_attempt_threshold: int = 4, + ) -> dict: + return self.sqlite_handler.get_lbt_diagnostics( + start_timestamp=start_timestamp, + end_timestamp=end_timestamp, + bucket_seconds=bucket_seconds, + severe_attempt_threshold=severe_attempt_threshold, + ) + def get_packet_stats(self, hours: int = 24) -> dict: return self.sqlite_handler.get_packet_stats(hours) diff --git a/repeater/web/api_endpoints.py b/repeater/web/api_endpoints.py index dfd2196..5b77445 100644 --- a/repeater/web/api_endpoints.py +++ b/repeater/web/api_endpoints.py @@ -447,6 +447,131 @@ class APIEndpoints: values = [v if v is not None else 0 for v in data_points] return [[timestamps_ms[i], values[i]] for i in range(min(len(values), len(timestamps_ms)))] + @staticmethod + def _pearson_correlation(left: list[float], right: list[float]) -> Optional[float]: + if len(left) != len(right) or len(left) < 5: + return None + + mean_left = sum(left) / len(left) + mean_right = sum(right) / len(right) + + numerator = 0.0 + left_variance = 0.0 + right_variance = 0.0 + for i in range(len(left)): + dx = left[i] - mean_left + dy = right[i] - mean_right + numerator += dx * dy + left_variance += dx * dx + right_variance += dy * dy + + denominator = (left_variance * right_variance) ** 0.5 + if denominator <= 0: + return None + return numerator / denominator + + @staticmethod + def _auto_bucket_seconds(range_seconds: int) -> int: + if range_seconds <= 0: + return 60 + target = max(60, int(range_seconds / 120)) + rounded = ((target + 59) // 60) * 60 + return max(60, min(rounded, 3600)) + + def _build_rrd_bucket_metrics(self, rrd_data: Optional[dict], bucket_seconds: int) -> dict: + if not rrd_data: + return {} + + timestamps = rrd_data.get("timestamps") or [] + metrics = rrd_data.get("metrics") or {} + if not isinstance(timestamps, list) or not isinstance(metrics, dict): + return {} + + def _counter_delta(values: list) -> list[float]: + output = [] + previous = None + for item in values: + if item is None: + output.append(0.0) + elif previous is None: + output.append(0.0) + previous = item + else: + output.append(float(max(0, item - previous))) + previous = item + return output + + rx_values = _counter_delta(metrics.get("rx_count", [])) + tx_values = _counter_delta(metrics.get("tx_count", [])) + drop_values = _counter_delta(metrics.get("drop_count", [])) + rssi_values = metrics.get("avg_rssi", []) or [] + snr_values = metrics.get("avg_snr", []) or [] + + bucket_map: dict = {} + + max_len = len(timestamps) + for i in range(max_len): + ts = int(timestamps[i]) + bucket_ts = int(ts / bucket_seconds) * bucket_seconds + bucket = bucket_map.setdefault( + bucket_ts, + { + "rx_count": 0.0, + "tx_count": 0.0, + "drop_count": 0.0, + "avg_rssi_sum": 0.0, + "avg_rssi_samples": 0, + "avg_snr_sum": 0.0, + "avg_snr_samples": 0, + }, + ) + + if i < len(rx_values): + bucket["rx_count"] += float(rx_values[i] or 0.0) + if i < len(tx_values): + bucket["tx_count"] += float(tx_values[i] or 0.0) + if i < len(drop_values): + bucket["drop_count"] += float(drop_values[i] or 0.0) + + if i < len(rssi_values) and rssi_values[i] is not None: + bucket["avg_rssi_sum"] += float(rssi_values[i]) + bucket["avg_rssi_samples"] += 1 + + if i < len(snr_values) and snr_values[i] is not None: + bucket["avg_snr_sum"] += float(snr_values[i]) + bucket["avg_snr_samples"] += 1 + + finalized: dict = {} + for bucket_ts, raw in bucket_map.items(): + rx_count = float(raw["rx_count"]) + tx_count = float(raw["tx_count"]) + drop_count = float(raw["drop_count"]) + tx_drop_total = tx_count + drop_count + + avg_rssi = None + if raw["avg_rssi_samples"] > 0: + avg_rssi = raw["avg_rssi_sum"] / raw["avg_rssi_samples"] + + avg_snr = None + if raw["avg_snr_samples"] > 0: + avg_snr = raw["avg_snr_sum"] / raw["avg_snr_samples"] + + packet_loss_rate_pct = None + if tx_drop_total > 0: + packet_loss_rate_pct = (drop_count * 100.0) / tx_drop_total + + finalized[bucket_ts] = { + "rx_count": int(round(rx_count)), + "tx_count": int(round(tx_count)), + "drop_count": int(round(drop_count)), + "traffic_volume": int(round(rx_count + tx_count)), + "packet_loss_rate_pct": packet_loss_rate_pct, + "avg_rssi": avg_rssi, + "avg_snr": avg_snr, + } + + return finalized + def _setup_status_from_config(self, config: dict) -> tuple[bool, dict]: """Return whether first-run setup should still be available.""" node_name = config.get("repeater", {}).get("node_name", "") @@ -3268,6 +3393,171 @@ class APIEndpoints: logger.error(f"Error getting metrics graph data: {e}") return self._error(e) + @cherrypy.expose + @cherrypy.tools.json_out() + def lbt_diagnostics( + self, + hours=24, + start_timestamp=None, + end_timestamp=None, + bucket_seconds=None, + severe_attempt_threshold=4, + ): + """Return aggregated LBT diagnostics aligned to RF-health buckets.""" + try: + max_hours = 168 + hours_int = max(1, min(int(hours), max_hours)) + + now = time.time() + if start_timestamp is not None or end_timestamp is not None: + if start_timestamp is None and end_timestamp is not None: + end_ts = float(end_timestamp) + start_ts = end_ts - (hours_int * 3600) + elif end_timestamp is None and start_timestamp is not None: + start_ts = float(start_timestamp) + end_ts = now + else: + start_ts = float(start_timestamp) + end_ts = float(end_timestamp) + else: + start_ts, end_ts = self._get_time_range(hours_int) + start_ts = float(start_ts) + end_ts = float(end_ts) + + if end_ts < start_ts: + start_ts, end_ts = end_ts, start_ts + + range_seconds = int(end_ts - start_ts) + if range_seconds <= 0: + return self._error("Invalid time range") + + if range_seconds > max_hours * 3600: + return self._error(f"Time range too large. Max range is {max_hours} hours") + + if bucket_seconds is None: + bucket_s = self._auto_bucket_seconds(range_seconds) + else: + bucket_s = max(60, min(int(bucket_seconds), 3600)) + + severe_threshold = max(2, min(int(severe_attempt_threshold), 16)) + + storage = self._get_storage() + lbt = storage.get_lbt_diagnostics( + start_timestamp=start_ts, + end_timestamp=end_ts, + bucket_seconds=bucket_s, + severe_attempt_threshold=severe_threshold, + ) + + rrd_data = storage.get_rrd_data( + start_time=int(start_ts), + end_time=int(end_ts), + resolution="average", + ) + rf_by_bucket = self._build_rrd_bucket_metrics(rrd_data, bucket_s) + + merged_buckets = [] + for bucket in lbt.get("buckets", []): + bucket_ts = int(bucket.get("timestamp", 0)) + rf = rf_by_bucket.get(bucket_ts, {}) + merged_buckets.append( + { + **bucket, + "rf": { + "avg_rssi": rf.get("avg_rssi"), + "avg_snr": rf.get("avg_snr"), + "packet_loss_rate_pct": rf.get("packet_loss_rate_pct"), + "traffic_volume": rf.get("traffic_volume", 0), + "rx_count": rf.get("rx_count", 0), + "tx_count": rf.get("tx_count", 0), + "drop_count": rf.get("drop_count", 0), + }, + } + ) + + merged_packet_type_buckets = [] + for bucket in lbt.get("packet_type_buckets", []): + bucket_ts = int(bucket.get("timestamp", 0)) + rf = rf_by_bucket.get(bucket_ts, {}) + merged_packet_type_buckets.append( + { + **bucket, + "rf": { + "avg_rssi": rf.get("avg_rssi"), + "avg_snr": rf.get("avg_snr"), + "packet_loss_rate_pct": rf.get("packet_loss_rate_pct"), + "traffic_volume": rf.get("traffic_volume", 0), + "rx_count": rf.get("rx_count", 0), + "tx_count": rf.get("tx_count", 0), + "drop_count": rf.get("drop_count", 0), + }, + } + ) + + def _build_correlation(metric_getter): + left = [] + right = [] + for item in merged_buckets: + retry_rate = item.get("retry_rate_pct") + metric_value = metric_getter(item) + if retry_rate is None or metric_value is None: + continue + left.append(float(retry_rate)) + right.append(float(metric_value)) + + coeff = self._pearson_correlation(left, right) + if coeff is None: + return { + "coefficient": None, + "sample_count": len(left), + "note": "Insufficient or non-varying samples", + } + return { + "coefficient": coeff, + "sample_count": len(left), + "note": None, + } + + correlations = { + "retry_rate_vs_avg_snr": _build_correlation( + lambda item: item.get("rf", {}).get("avg_snr") + ), + "retry_rate_vs_avg_rssi": _build_correlation( + lambda item: item.get("rf", {}).get("avg_rssi") + ), + "retry_rate_vs_packet_loss_rate": _build_correlation( + lambda item: item.get("rf", {}).get("packet_loss_rate_pct") + ), + "retry_rate_vs_traffic_volume": _build_correlation( + lambda item: item.get("rf", {}).get("traffic_volume") + ), + } + + diagnostics = { + "start_time": int(start_ts), + "end_time": int(end_ts), + "bucket_seconds": int(bucket_s), + "severe_attempt_threshold": severe_threshold, + "summary": lbt.get("summary", {}), + "buckets": merged_buckets, + "packet_types": lbt.get("packet_types", []), + "packet_type_buckets": merged_packet_type_buckets, + "correlations": correlations, + "limitations": [ + "LBT attempts are derived from stored per-packet retry counts (lbt_attempts + 1).", + "Per-attempt RSSI/SNR and channel frequency are not recorded for each LBT attempt.", + "Airtime utilisation is not available in the current RRD metric set for direct alignment.", + ], + } + + return self._success(diagnostics) + + except ValueError as e: + return self._error(f"Invalid parameter format: {e}") + except Exception as e: + logger.error(f"Error getting LBT diagnostics: {e}") + return self._error(e) + @cherrypy.expose @cherrypy.tools.json_out() @cherrypy.tools.json_in() diff --git a/repeater/web/openapi.yaml b/repeater/web/openapi.yaml index b874674..4943717 100644 --- a/repeater/web/openapi.yaml +++ b/repeater/web/openapi.yaml @@ -1132,6 +1132,66 @@ paths: items: type: number + /lbt_diagnostics: + get: + tags: [Charts] + summary: Get LBT diagnostics aligned with RF metrics + description: | + Returns aggregated Listen Before Talk (LBT) diagnostics for transmission-path + packets, aligned to RF health buckets for correlation analysis. + + Notes: + - LBT total attempts are derived as `lbt_attempts + 1` from stored packet metadata. + - Buckets are bounded and aggregated server-side for efficient dashboard refresh. + parameters: + - name: hours + in: query + schema: + type: integer + minimum: 1 + maximum: 168 + default: 24 + description: Fallback range in hours when explicit timestamps are not provided. + - name: start_timestamp + in: query + schema: + type: number + format: float + description: Inclusive start timestamp (Unix epoch seconds). + - name: end_timestamp + in: query + schema: + type: number + format: float + description: Inclusive end timestamp (Unix epoch seconds). + - name: bucket_seconds + in: query + schema: + type: integer + minimum: 60 + maximum: 3600 + description: Bucket width in seconds. If omitted, server auto-selects based on range. + - name: severe_attempt_threshold + in: query + schema: + type: integer + minimum: 2 + maximum: 16 + default: 4 + description: Attempt threshold used to classify severe contention events. + responses: + '200': + description: LBT diagnostics response + content: + application/json: + schema: + type: object + properties: + success: + type: boolean + data: + $ref: '#/components/schemas/LbtDiagnosticsResponse' + /noise_floor_history: get: tags: [Noise Floor] @@ -4158,6 +4218,381 @@ components: type: string description: Error message + LbtDiagnosticsResponse: + type: object + required: [start_time, end_time, bucket_seconds, severe_attempt_threshold, summary, buckets, packet_types, packet_type_buckets, correlations, limitations] + properties: + start_time: + type: integer + description: Selected range start timestamp (Unix epoch seconds) + end_time: + type: integer + description: Selected range end timestamp (Unix epoch seconds) + bucket_seconds: + type: integer + minimum: 60 + maximum: 3600 + description: Aggregation interval in seconds + severe_attempt_threshold: + type: integer + minimum: 2 + maximum: 16 + description: Attempt count threshold used for severe contention classification + summary: + $ref: '#/components/schemas/LbtSummary' + buckets: + type: array + items: + $ref: '#/components/schemas/LbtBucket' + packet_types: + type: array + description: Packet types present in the selected range, sorted by transmission volume. + items: + $ref: '#/components/schemas/LbtPacketTypeSummary' + packet_type_buckets: + type: array + description: Sparse per-packet-type bucket diagnostics (only combinations with transmissions). + items: + $ref: '#/components/schemas/LbtPacketTypeBucket' + correlations: + $ref: '#/components/schemas/LbtCorrelationSet' + limitations: + type: array + items: + type: string + + LbtPacketTypeSummary: + type: object + required: [packet_type, packet_type_label, transmissions, retry_packets] + properties: + packet_type: + type: integer + description: Raw packet type code from packet records. + packet_type_label: + type: string + description: Human-readable packet type label. + transmissions: + type: integer + minimum: 0 + retry_packets: + type: integer + minimum: 0 + retry_rate_pct: + type: number + format: float + nullable: true + + LbtPacketTypeBucket: + type: object + required: [timestamp, packet_type, packet_type_label, transmissions, total_attempts, first_attempt_success, retry_packets, attempts_1, attempts_2, attempts_3, attempts_4_plus, attempts_3_plus, max_attempts, failed_transmissions, severe_contention_count, rf] + properties: + timestamp: + type: integer + description: Bucket timestamp (Unix epoch seconds) + packet_type: + type: integer + description: Raw packet type code from packet records. + packet_type_label: + type: string + description: Human-readable packet type label. + transmissions: + type: integer + minimum: 0 + total_attempts: + type: integer + minimum: 0 + first_attempt_success: + type: integer + minimum: 0 + retry_packets: + type: integer + minimum: 0 + retry_rate_pct: + type: number + format: float + nullable: true + first_attempt_success_rate_pct: + type: number + format: float + nullable: true + avg_attempts: + type: number + format: float + nullable: true + attempts_1: + type: integer + minimum: 0 + attempts_2: + type: integer + minimum: 0 + attempts_3: + type: integer + minimum: 0 + attempts_4_plus: + type: integer + minimum: 0 + attempts_3_plus: + type: integer + minimum: 0 + attempts_3_plus_pct: + type: number + format: float + nullable: true + max_attempts: + type: integer + minimum: 0 + failed_transmissions: + type: integer + minimum: 0 + severe_contention_count: + type: integer + minimum: 0 + rf: + $ref: '#/components/schemas/LbtRfBucket' + + LbtSummary: + type: object + required: [total_transmissions, total_attempts, first_attempt_success, retry_packets, max_attempts, attempts_1, attempts_2, attempts_3, attempts_4_plus, attempts_3_plus, failed_transmissions, busy_channel_events, severe_contention_count, severe_attempt_threshold, has_lbt_data] + properties: + total_transmissions: + type: integer + minimum: 0 + total_attempts: + type: integer + minimum: 0 + first_attempt_success: + type: integer + minimum: 0 + retry_packets: + type: integer + minimum: 0 + retry_rate_pct: + type: number + format: float + nullable: true + first_attempt_success_rate_pct: + type: number + format: float + nullable: true + avg_attempts: + type: number + format: float + nullable: true + median_attempts: + type: number + format: float + nullable: true + p95_attempts: + type: number + format: float + nullable: true + max_attempts: + type: integer + minimum: 0 + attempts_1: + type: integer + minimum: 0 + attempts_2: + type: integer + minimum: 0 + attempts_3: + type: integer + minimum: 0 + attempts_4_plus: + type: integer + minimum: 0 + attempts_3_plus: + type: integer + minimum: 0 + attempts_3_plus_pct: + type: number + format: float + nullable: true + attempts_4_plus_pct: + type: number + format: float + nullable: true + failed_transmissions: + type: integer + minimum: 0 + busy_channel_events: + type: integer + minimum: 0 + severe_contention_count: + type: integer + minimum: 0 + severe_contention_pct: + type: number + format: float + nullable: true + severe_attempt_threshold: + type: integer + minimum: 2 + maximum: 16 + has_lbt_data: + type: boolean + worst_bucket: + $ref: '#/components/schemas/LbtWorstBucket' + nullable: true + + LbtWorstBucket: + type: object + required: [timestamp, retry_rate_pct, attempts_3_plus_pct, max_attempts, transmissions] + properties: + timestamp: + type: integer + retry_rate_pct: + type: number + format: float + attempts_3_plus_pct: + type: number + format: float + max_attempts: + type: integer + minimum: 0 + transmissions: + type: integer + minimum: 0 + + LbtBucket: + type: object + required: [timestamp, transmissions, total_attempts, first_attempt_success, retry_packets, max_attempts, attempts_1, attempts_2, attempts_3, attempts_4_plus, attempts_3_plus, failed_transmissions, busy_channel_events, severe_contention_count, rf] + properties: + timestamp: + type: integer + description: Bucket timestamp (Unix epoch seconds) + transmissions: + type: integer + minimum: 0 + total_attempts: + type: integer + minimum: 0 + first_attempt_success: + type: integer + minimum: 0 + retry_packets: + type: integer + minimum: 0 + retry_rate_pct: + type: number + format: float + nullable: true + first_attempt_success_rate_pct: + type: number + format: float + nullable: true + avg_attempts: + type: number + format: float + nullable: true + median_attempts: + type: number + format: float + nullable: true + p95_attempts: + type: number + format: float + nullable: true + max_attempts: + type: integer + minimum: 0 + attempts_1: + type: integer + minimum: 0 + attempts_2: + type: integer + minimum: 0 + attempts_3: + type: integer + minimum: 0 + attempts_4_plus: + type: integer + minimum: 0 + attempts_3_plus: + type: integer + minimum: 0 + attempts_3_plus_pct: + type: number + format: float + nullable: true + attempts_4_plus_pct: + type: number + format: float + nullable: true + failed_transmissions: + type: integer + minimum: 0 + busy_channel_events: + type: integer + minimum: 0 + severe_contention_count: + type: integer + minimum: 0 + severe_contention_pct: + type: number + format: float + nullable: true + rf: + $ref: '#/components/schemas/LbtRfBucket' + + LbtRfBucket: + type: object + required: [traffic_volume, rx_count, tx_count, drop_count] + properties: + avg_rssi: + type: number + format: float + nullable: true + avg_snr: + type: number + format: float + nullable: true + packet_loss_rate_pct: + type: number + format: float + nullable: true + traffic_volume: + type: integer + minimum: 0 + rx_count: + type: integer + minimum: 0 + tx_count: + type: integer + minimum: 0 + drop_count: + type: integer + minimum: 0 + + LbtCorrelationSet: + type: object + required: [retry_rate_vs_avg_snr, retry_rate_vs_avg_rssi, retry_rate_vs_packet_loss_rate, retry_rate_vs_traffic_volume] + properties: + retry_rate_vs_avg_snr: + $ref: '#/components/schemas/LbtCorrelationValue' + retry_rate_vs_avg_rssi: + $ref: '#/components/schemas/LbtCorrelationValue' + retry_rate_vs_packet_loss_rate: + $ref: '#/components/schemas/LbtCorrelationValue' + retry_rate_vs_traffic_volume: + $ref: '#/components/schemas/LbtCorrelationValue' + + LbtCorrelationValue: + type: object + required: [coefficient, sample_count] + properties: + coefficient: + type: number + format: float + nullable: true + description: Pearson coefficient in range [-1, 1] when enough samples exist + sample_count: + type: integer + minimum: 0 + note: + type: string + nullable: true + RoomMessage: type: object required: [id, author_pubkey, post_timestamp, message_text, txt_type] diff --git a/tests/test_api_endpoints_core_coverage.py b/tests/test_api_endpoints_core_coverage.py index ca82074..b2a3b6f 100644 --- a/tests/test_api_endpoints_core_coverage.py +++ b/tests/test_api_endpoints_core_coverage.py @@ -1743,6 +1743,178 @@ def test_metrics_graph_data_includes_policy_events(cherrypy_ctx): assert series_by_type["policy_events"]["data"] == [[100000, 1], [160000, 3], [220000, 2]] +def test_lbt_diagnostics_aligns_with_rrd_and_returns_correlations(cherrypy_ctx): + del cherrypy_ctx + api = _make_api() + + storage = SimpleNamespace( + get_lbt_diagnostics=MagicMock( + return_value={ + "start_time": 60, + "end_time": 240, + "bucket_seconds": 60, + "summary": { + "total_transmissions": 10, + "total_attempts": 14, + "first_attempt_success": 7, + "retry_packets": 3, + "retry_rate_pct": 30.0, + "first_attempt_success_rate_pct": 70.0, + "avg_attempts": 1.4, + "median_attempts": 1.0, + "p95_attempts": 3.0, + "max_attempts": 4, + "attempts_1": 7, + "attempts_2": 2, + "attempts_3": 1, + "attempts_4_plus": 0, + "attempts_3_plus": 1, + "attempts_3_plus_pct": 10.0, + "attempts_4_plus_pct": 0.0, + "failed_transmissions": 0, + "busy_channel_events": 3, + "severe_contention_count": 0, + "severe_contention_pct": 0.0, + "severe_attempt_threshold": 4, + "has_lbt_data": True, + "worst_bucket": { + "timestamp": 120, + "retry_rate_pct": 50.0, + "attempts_3_plus_pct": 20.0, + "max_attempts": 3, + "transmissions": 5, + }, + }, + "buckets": [ + { + "timestamp": 60, + "transmissions": 3, + "total_attempts": 3, + "first_attempt_success": 3, + "retry_packets": 0, + "retry_rate_pct": 0.0, + "first_attempt_success_rate_pct": 100.0, + "avg_attempts": 1.0, + "median_attempts": 1.0, + "p95_attempts": 1.0, + "max_attempts": 1, + "attempts_1": 3, + "attempts_2": 0, + "attempts_3": 0, + "attempts_4_plus": 0, + "attempts_3_plus": 0, + "attempts_3_plus_pct": 0.0, + "attempts_4_plus_pct": 0.0, + "failed_transmissions": 0, + "busy_channel_events": 0, + "severe_contention_count": 0, + "severe_contention_pct": 0.0, + }, + { + "timestamp": 120, + "transmissions": 5, + "total_attempts": 8, + "first_attempt_success": 2, + "retry_packets": 3, + "retry_rate_pct": 60.0, + "first_attempt_success_rate_pct": 40.0, + "avg_attempts": 1.6, + "median_attempts": 2.0, + "p95_attempts": 3.0, + "max_attempts": 3, + "attempts_1": 2, + "attempts_2": 2, + "attempts_3": 1, + "attempts_4_plus": 0, + "attempts_3_plus": 1, + "attempts_3_plus_pct": 20.0, + "attempts_4_plus_pct": 0.0, + "failed_transmissions": 0, + "busy_channel_events": 3, + "severe_contention_count": 0, + "severe_contention_pct": 0.0, + }, + ], + "packet_types": [ + { + "packet_type": 1, + "packet_type_label": "Response (RESPONSE)", + "transmissions": 8, + "retry_packets": 3, + "retry_rate_pct": 37.5, + } + ], + "packet_type_buckets": [ + { + "timestamp": 120, + "packet_type": 1, + "packet_type_label": "Response (RESPONSE)", + "transmissions": 5, + "total_attempts": 8, + "first_attempt_success": 2, + "retry_packets": 3, + "retry_rate_pct": 60.0, + "first_attempt_success_rate_pct": 40.0, + "avg_attempts": 1.6, + "attempts_1": 2, + "attempts_2": 2, + "attempts_3": 1, + "attempts_4_plus": 0, + "attempts_3_plus": 1, + "attempts_3_plus_pct": 20.0, + "max_attempts": 3, + "failed_transmissions": 0, + "severe_contention_count": 0, + } + ], + } + ), + get_rrd_data=MagicMock( + return_value={ + "timestamps": [100, 160, 220], + "metrics": { + "rx_count": [100, 110, 125], + "tx_count": [50, 55, 70], + "drop_count": [10, 11, 14], + "avg_rssi": [-80.0, -90.0, -95.0], + "avg_snr": [8.0, 2.0, -1.0], + }, + } + ), + ) + _attach_storage(api, storage) + + out = api.lbt_diagnostics(hours="24", bucket_seconds="60", severe_attempt_threshold="4") + assert out["success"] is True + assert out["data"]["bucket_seconds"] == 60 + assert out["data"]["severe_attempt_threshold"] == 4 + + buckets = out["data"]["buckets"] + assert len(buckets) == 2 + assert "rf" in buckets[0] + assert buckets[0]["rf"]["traffic_volume"] >= 0 + assert buckets[0]["rf"]["avg_snr"] is not None + + correlations = out["data"]["correlations"] + assert "retry_rate_vs_avg_snr" in correlations + assert "retry_rate_vs_traffic_volume" in correlations + assert "sample_count" in correlations["retry_rate_vs_avg_snr"] + + assert out["data"]["packet_types"][0]["packet_type"] == 1 + assert out["data"]["packet_type_buckets"][0]["packet_type_label"] == "Response (RESPONSE)" + assert "rf" in out["data"]["packet_type_buckets"][0] + + +def test_lbt_diagnostics_rejects_unbounded_large_time_range(cherrypy_ctx): + del cherrypy_ctx + api = _make_api() + _attach_storage(api, SimpleNamespace()) + + out = api.lbt_diagnostics(start_timestamp="0", end_timestamp=str(200 * 3600)) + assert out["success"] is False + assert "Max range" in out["error"] + + def test_advert_contact_and_rate_limit_stats_endpoints(cherrypy_ctx): del cherrypy_ctx api = _make_api() diff --git a/tests/test_sqlite_handler_easy.py b/tests/test_sqlite_handler_easy.py index bb0d3f9..cc31c16 100644 --- a/tests/test_sqlite_handler_easy.py +++ b/tests/test_sqlite_handler_easy.py @@ -361,6 +361,10 @@ def test_sync_transport_keys_validation_and_tree_apply(tmp_path, monkeypatch): ] ) + +def test_sync_transport_keys_parent_and_tree_apply(tmp_path, monkeypatch): + h = _make_handler(tmp_path) + with pytest.raises(ValueError, match="Parent node 'missing'"): h.sync_transport_keys( [ @@ -405,3 +409,162 @@ def test_sync_transport_keys_validation_and_tree_apply(tmp_path, monkeypatch): assert rows[1][2] == "deny" assert rows[1][3] == "GEN-KEY" assert rows[1][4] == rows[0][0] + + +def test_get_lbt_diagnostics_aggregates_retry_distribution_and_summary(tmp_path): + h = _make_handler(tmp_path) + + packets = [ + { + "timestamp": 10.0, + "type": 1, + "route": 1, + "length": 8, + "transmitted": True, + "packet_hash": "lbt-1", + "lbt_attempts": 0, + "lbt_channel_busy": False, + }, + { + "timestamp": 20.0, + "type": 1, + "route": 1, + "length": 8, + "transmitted": True, + "packet_hash": "lbt-2", + "lbt_attempts": 1, + "lbt_channel_busy": True, + }, + { + "timestamp": 30.0, + "type": 1, + "route": 1, + "length": 8, + "transmitted": True, + "packet_hash": "lbt-3", + "lbt_attempts": 2, + "lbt_channel_busy": True, + }, + { + "timestamp": 40.0, + "type": 1, + "route": 1, + "length": 8, + "transmitted": False, + "drop_reason": "TX failed", + "packet_hash": "lbt-4", + "lbt_attempts": 4, + "lbt_channel_busy": True, + }, + { + # Excluded from TX-path diagnostics by filter. + "timestamp": 50.0, + "type": 1, + "route": 1, + "length": 8, + "transmitted": False, + "drop_reason": "Duplicate", + "packet_hash": "lbt-excluded", + "lbt_attempts": 0, + "lbt_channel_busy": False, + }, + { + "timestamp": 70.0, + "type": 1, + "route": 1, + "length": 8, + "transmitted": True, + "packet_hash": "lbt-5", + "lbt_attempts": 0, + "lbt_channel_busy": False, + }, + ] + + for record in packets: + h.store_packet(record) + + out = h.get_lbt_diagnostics( + start_timestamp=0, + end_timestamp=180, + bucket_seconds=60, + severe_attempt_threshold=4, + ) + + summary = out["summary"] + assert summary["total_transmissions"] == 5 + assert summary["total_attempts"] == 12 + assert summary["first_attempt_success"] == 2 + assert summary["retry_packets"] == 3 + assert summary["retry_rate_pct"] == pytest.approx(60.0) + assert summary["avg_attempts"] == pytest.approx(2.4) + assert summary["max_attempts"] == 5 + assert summary["median_attempts"] == pytest.approx(2.0) + assert summary["p95_attempts"] == pytest.approx(5.0) + assert summary["attempts_1"] == 2 + assert summary["attempts_2"] == 1 + assert summary["attempts_3"] == 1 + assert summary["attempts_4_plus"] == 1 + assert summary["attempts_3_plus"] == 2 + assert summary["failed_transmissions"] == 1 + assert summary["busy_channel_events"] == 3 + assert summary["severe_contention_count"] == 1 + assert summary["has_lbt_data"] is True + assert summary["worst_bucket"] is not None + assert summary["worst_bucket"]["timestamp"] == 0 + + buckets = {int(b["timestamp"]): b for b in out["buckets"]} + first = buckets[0] + assert first["transmissions"] == 4 + assert first["retry_packets"] == 3 + assert first["retry_rate_pct"] == pytest.approx(75.0) + assert first["first_attempt_success_rate_pct"] == pytest.approx(25.0) + assert first["attempts_4_plus"] == 1 + assert first["severe_contention_count"] == 1 + assert first["failed_transmissions"] == 1 + + second = buckets[60] + assert second["transmissions"] == 1 + assert second["retry_packets"] == 0 + assert second["retry_rate_pct"] == pytest.approx(0.0) + assert second["avg_attempts"] == pytest.approx(1.0) + + packet_types = out["packet_types"] + assert len(packet_types) == 1 + assert packet_types[0]["packet_type"] == 1 + assert packet_types[0]["transmissions"] == 5 + assert packet_types[0]["retry_packets"] == 3 + + packet_type_buckets = out["packet_type_buckets"] + assert len(packet_type_buckets) == 2 + first_type_bucket = packet_type_buckets[0] + assert first_type_bucket["packet_type"] == 1 + assert first_type_bucket["timestamp"] == 0 + assert first_type_bucket["retry_rate_pct"] == pytest.approx(75.0) + assert first_type_bucket["attempts_3_plus_pct"] == pytest.approx(50.0) + + +def test_get_lbt_diagnostics_empty_range_preserves_no_data_distinction(tmp_path): + h = _make_handler(tmp_path) + + out = h.get_lbt_diagnostics( + start_timestamp=0, + end_timestamp=180, + bucket_seconds=60, + severe_attempt_threshold=4, + ) + + summary = out["summary"] + assert summary["total_transmissions"] == 0 + assert summary["retry_rate_pct"] is None + assert summary["first_attempt_success_rate_pct"] is None + assert summary["avg_attempts"] is None + assert summary["has_lbt_data"] is False + + assert len(out["buckets"]) >= 3 + for bucket in out["buckets"]: + assert bucket["transmissions"] == 0 + assert bucket["retry_rate_pct"] is None + assert bucket["first_attempt_success_rate_pct"] is None + assert bucket["avg_attempts"] is None + assert out["packet_types"] == [] + assert out["packet_type_buckets"] == []