From 8008f0963449d4f6d02c72768e2eda754fcfcdd8 Mon Sep 17 00:00:00 2001 From: Thibault VINCENT Date: Sat, 22 Aug 2026 01:28:50 +0200 Subject: [PATCH] feat(ws): carry the sidebar vitals in the stats broadcast, decimate the 24h aggregate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 5s WebSocket broadcast used to carry uptime_seconds alone — while the dashboard sidebar starved: its duty-cycle gauge, noise floor and mode never travel over WS, and the heartbeat marks stats fresh, which suppresses the UI's HTTP fallback poll. The vitals now ride every beat, all cheap in-memory scalars (~200 bytes): utilization_percent, noise_floor_dbm (kept on record, no per-beat SQL), the repeater mode (so a change made over RF or CLI reaches the web badge), and the advert tier counters the sidebar's OK/Drop line shows, injected as a callback from the daemon like the other collector hooks. In exchange the heavy get_packet_stats(24h) SQL aggregate is decimated to one beat in six (30s): nobody reads a 24h cumulative at 5s resolution, and the UI already treats both payload fields as optional, so older dashboards keep working. Net wire traffic goes down. Signed-off-by: Thibault VINCENT --- .../data_acquisition/storage_collector.py | 61 +++++++++++++++---- repeater/main.py | 4 ++ ...est_storage_collector_ws_stats_throttle.py | 43 +++++++++++++ 3 files changed, 96 insertions(+), 12 deletions(-) diff --git a/repeater/data_acquisition/storage_collector.py b/repeater/data_acquisition/storage_collector.py index 0a609e0..ac2dd79 100644 --- a/repeater/data_acquisition/storage_collector.py +++ b/repeater/data_acquisition/storage_collector.py @@ -3,7 +3,7 @@ import concurrent.futures import logging import threading import time -from typing import Optional +from typing import Any, Dict, Optional from repeater.config import resolve_storage_dir @@ -88,6 +88,11 @@ class StorageCollector: # Initialize WebSocket handler for real-time updates self.websocket_available = False self.websocket_has_connected_clients = lambda: False + # Wired by the daemon once the advert helper exists; returns the + # rate-limit stats dict the sidebar's advert tier reads. + self.advert_stats_getter = None + self._last_noise_floor_dbm: Optional[float] = None + self._stats_broadcast_seq = 0 self._ws_stats_broadcast_interval_sec: float = 5.0 self._stats_stop_event = threading.Event() self._stats_thread: Optional[threading.Thread] = None @@ -256,18 +261,49 @@ class StorageCollector: self._publish_packet_to_mqtt(packet_record) + # The heavy 24h SQL aggregate rides one beat in six (30s at the 5s + # default): nobody reads a 24h cumulative at 5s resolution, and the + # sidebar vitals it used to travel with are cheap in-memory scalars. + PACKET_STATS_EVERY_N_BEATS = 6 + def _broadcast_stats_once(self) -> None: - """Compute the 24h aggregate and broadcast it to WebSocket clients.""" - packet_stats_24h = self.sqlite_handler.get_packet_stats(hours=24) - uptime_seconds = ( - time.time() - self.repeater_handler.start_time if self.repeater_handler else 0 - ) - self.websocket_broadcast_stats( - { - "packet_stats": packet_stats_24h, - "system_stats": {"uptime_seconds": uptime_seconds}, - } - ) + """Broadcast the sidebar vitals; the 24h aggregate is decimated. + + The beat carries scalars only — a few hundred bytes — so the + dashboard sidebar can read every vital live on any page. Series + (noise-floor history, charts) stay on their own slow HTTP paths. + """ + system_stats: Dict[str, Any] = { + "uptime_seconds": ( + time.time() - self.repeater_handler.start_time if self.repeater_handler else 0 + ), + "mode": self.config.get("repeater", {}).get("mode", "forward"), + } + airtime_mgr = getattr(self.repeater_handler, "airtime_mgr", None) + if airtime_mgr is not None: + airtime_stats = airtime_mgr.get_stats() + if airtime_stats: + system_stats["utilization_percent"] = airtime_stats["utilization_percent"] + if self._last_noise_floor_dbm is not None: + system_stats["noise_floor_dbm"] = self._last_noise_floor_dbm + if self.advert_stats_getter is not None: + try: + tier = self.advert_stats_getter() + system_stats["advert_tier"] = { + "current_tier": tier.get("adaptive", {}).get("current_tier"), + "adverts_allowed": tier.get("stats", {}).get("adverts_allowed", 0), + "adverts_dropped": tier.get("stats", {}).get("adverts_dropped", 0), + "active_penalties": len(tier.get("active_penalties") or {}), + } + except Exception as e: + logger.debug(f"Advert stats unavailable for broadcast: {e}") + + payload: Dict[str, Any] = {"system_stats": system_stats} + if self._stats_broadcast_seq % self.PACKET_STATS_EVERY_N_BEATS == 0: + payload["packet_stats"] = self.sqlite_handler.get_packet_stats(hours=24) + self._stats_broadcast_seq += 1 + + self.websocket_broadcast_stats(payload) def _stats_broadcast_loop(self) -> None: """Broadcast aggregate stats every interval while clients are connected. @@ -337,6 +373,7 @@ class StorageCollector: def record_noise_floor(self, noise_floor_dbm: float): """Record noise floor to storage and defer network publishing to background tasks.""" + self._last_noise_floor_dbm = noise_floor_dbm noise_record = {"timestamp": time.time(), "noise_floor_dbm": noise_floor_dbm} self.sqlite_handler.store_noise_floor(noise_record) self._schedule_background( diff --git a/repeater/main.py b/repeater/main.py index 3375935..90f1278 100644 --- a/repeater/main.py +++ b/repeater/main.py @@ -532,6 +532,10 @@ class RepeaterDaemon: log_fn=logger.info, ) logger.info("Advert processing helper initialized") + if self.repeater_handler and self.repeater_handler.storage: + self.repeater_handler.storage.advert_stats_getter = ( + self.advert_helper.get_rate_limit_stats + ) # Set up discovery handler if enabled allow_discovery = self.config.get("repeater", {}).get("allow_discovery", True) diff --git a/tests/test_storage_collector_ws_stats_throttle.py b/tests/test_storage_collector_ws_stats_throttle.py index e3162e8..73d93e5 100644 --- a/tests/test_storage_collector_ws_stats_throttle.py +++ b/tests/test_storage_collector_ws_stats_throttle.py @@ -82,6 +82,49 @@ def test_broadcast_stats_once_queries_and_broadcasts(): payload = collector.websocket_broadcast_stats.call_args.args[0] assert payload["packet_stats"] == {"total_packets": 1} assert "uptime_seconds" in payload["system_stats"] + assert payload["system_stats"]["mode"] == "forward" + + +def test_broadcast_decimates_the_packet_stats_aggregate(): + collector = _make_collector() + + for _ in range(collector.PACKET_STATS_EVERY_N_BEATS + 1): + collector._broadcast_stats_once() + + # The heavy 24h aggregate rides the first beat and the (N+1)th, not the + # beats in between; the vitals travel on every beat. + assert collector.sqlite_handler.get_packet_stats.call_count == 2 + payloads = [c.args[0] for c in collector.websocket_broadcast_stats.call_args_list] + assert all("system_stats" in p for p in payloads) + assert [("packet_stats" in p) for p in payloads] == [True] + [False] * ( + collector.PACKET_STATS_EVERY_N_BEATS - 1 + ) + [True] + + +def test_broadcast_carries_the_sidebar_vitals(): + collector = _make_collector() + collector.repeater_handler = SimpleNamespace( + start_time=0.0, + airtime_mgr=SimpleNamespace(get_stats=lambda: {"utilization_percent": 20.0}), + ) + collector._last_noise_floor_dbm = -107.1 + collector.advert_stats_getter = lambda: { + "adaptive": {"current_tier": "NORMAL"}, + "stats": {"adverts_allowed": 15, "adverts_dropped": 0}, + "active_penalties": {}, + } + + collector._broadcast_stats_once() + + vitals = collector.websocket_broadcast_stats.call_args.args[0]["system_stats"] + assert vitals["utilization_percent"] == 20.0 + assert vitals["noise_floor_dbm"] == -107.1 + assert vitals["advert_tier"] == { + "current_tier": "NORMAL", + "adverts_allowed": 15, + "adverts_dropped": 0, + "active_penalties": 0, + } def test_stats_loop_broadcasts_when_clients_connected():