diff --git a/repeater/engine.py b/repeater/engine.py index a456cf6..e57ddb0 100644 --- a/repeater/engine.py +++ b/repeater/engine.py @@ -20,6 +20,8 @@ from repeater.storage import StorageCollector logger = logging.getLogger("RepeaterHandler") +NOISE_FLOOR_INTERVAL = 30.0 # seconds + class RepeaterHandler(BaseHandler): @@ -80,6 +82,12 @@ class RepeaterHandler(BaseHandler): logger.error(f"Failed to initialize StorageCollector: {e}") self.storage = None + # Initialize background timer tracking + self.last_noise_measurement = time.time() + self.noise_floor_interval = NOISE_FLOOR_INTERVAL # 30 seconds + self._background_task = None + self._start_background_tasks() + async def __call__(self, packet: Packet, metadata: Optional[dict] = None) -> None: if metadata is None: @@ -88,9 +96,6 @@ class RepeaterHandler(BaseHandler): # Track incoming packet self.rx_count += 1 - # Check if it's time to send a periodic advertisement - await self._check_and_send_periodic_advert() - # Check if we're in monitor mode (receive only, no forwarding) mode = self.config.get("repeater", {}).get("mode", "forward") monitor_mode = mode == "monitor" @@ -565,32 +570,6 @@ class RepeaterHandler(BaseHandler): asyncio.create_task(delayed_send()) - async def _check_and_send_periodic_advert(self): - - if self.send_advert_interval_hours <= 0 or not self.send_advert_func: - return - - current_time = time.time() - interval_seconds = self.send_advert_interval_hours * 3600 # Convert hours to seconds - time_since_last_advert = current_time - self.last_advert_time - - # Check if interval has elapsed - if time_since_last_advert >= interval_seconds: - logger.info( - f"Periodic advert interval elapsed ({time_since_last_advert:.0f}s >= " - f"{interval_seconds:.0f}s). Sending advert..." - ) - try: - # Call the send_advert function - success = await self.send_advert_func() - if success: - self.last_advert_time = current_time - logger.info("Periodic advert sent successfully") - else: - logger.warning("Failed to send periodic advert") - except Exception as e: - logger.error(f"Error sending periodic advert: {e}", exc_info=True) - def get_noise_floor(self) -> Optional[float]: try: radio = self.dispatcher.radio if self.dispatcher else None @@ -672,7 +651,73 @@ class RepeaterHandler(BaseHandler): stats.update(self.airtime_mgr.get_stats()) return stats + def _start_background_tasks(self): + if self._background_task is None: + self._background_task = asyncio.create_task(self._background_timer_loop()) + logger.info("Background timer started for noise floor and adverts") + + async def _background_timer_loop(self): + try: + while True: + current_time = time.time() + + # Check noise floor recording (every 30 seconds) + if current_time - self.last_noise_measurement >= self.noise_floor_interval: + await self._record_noise_floor_async() + self.last_noise_measurement = current_time + + # Check advert sending (every N hours) + if self.send_advert_interval_hours > 0 and self.send_advert_func: + interval_seconds = self.send_advert_interval_hours * 3600 + if current_time - self.last_advert_time >= interval_seconds: + await self._send_periodic_advert_async() + self.last_advert_time = current_time + + # Sleep for 5 seconds before next check + await asyncio.sleep(5.0) + + except asyncio.CancelledError: + logger.info("Background timer loop cancelled") + raise + except Exception as e: + logger.error(f"Error in background timer loop: {e}") + # Restart the timer after a delay + await asyncio.sleep(30) + self._background_task = asyncio.create_task(self._background_timer_loop()) + + async def _record_noise_floor_async(self): + if not self.storage: + return + + try: + noise_floor = self.get_noise_floor() + if noise_floor is not None: + self.storage.record_noise_floor(noise_floor) + logger.debug(f"Recorded noise floor: {noise_floor} dBm") + else: + logger.debug("Unable to read noise floor from radio") + except Exception as e: + logger.error(f"Error recording noise floor: {e}") + + async def _send_periodic_advert_async(self): + logger.info(f"Periodic advert timer triggered (interval: {self.send_advert_interval_hours}h)") + try: + if self.send_advert_func: + success = await self.send_advert_func() + if success: + logger.info("Periodic advert sent successfully") + else: + logger.warning("Failed to send periodic advert") + else: + logger.debug("No send_advert_func configured") + except Exception as e: + logger.error(f"Error sending periodic advert: {e}") + def cleanup(self): + if self._background_task and not self._background_task.done(): + self._background_task.cancel() + logger.info("Background timer task cancelled") + if self.storage: try: self.storage.close() diff --git a/repeater/storage.py b/repeater/storage.py index 785f5eb..f4fe7ef 100644 --- a/repeater/storage.py +++ b/repeater/storage.py @@ -90,6 +90,15 @@ class StorageCollector: ) """) + # Noise floor measurements table + conn.execute(""" + CREATE TABLE IF NOT EXISTS noise_floor ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + timestamp REAL NOT NULL, + noise_floor_dbm REAL NOT NULL + ) + """) + # Create indexes for performance conn.execute("CREATE INDEX IF NOT EXISTS idx_packets_timestamp ON packets(timestamp)") conn.execute("CREATE INDEX IF NOT EXISTS idx_packets_type ON packets(type)") @@ -97,6 +106,7 @@ class StorageCollector: conn.execute("CREATE INDEX IF NOT EXISTS idx_packets_transmitted ON packets(transmitted)") conn.execute("CREATE INDEX IF NOT EXISTS idx_adverts_timestamp ON adverts(timestamp)") conn.execute("CREATE INDEX IF NOT EXISTS idx_adverts_pubkey ON adverts(pubkey)") + conn.execute("CREATE INDEX IF NOT EXISTS idx_noise_timestamp ON noise_floor(timestamp)") conn.commit() logger.info(f"SQLite database initialized: {self.sqlite_path}") @@ -122,6 +132,7 @@ class StorageCollector: "--step", "60", # 1-minute steps "--start", str(int(time.time() - 60)), + # Data sources - Basic metrics "DS:rx_count:COUNTER:120:0:U", # Received packets "DS:tx_count:COUNTER:120:0:U", # Transmitted packets @@ -131,6 +142,7 @@ class StorageCollector: "DS:avg_length:GAUGE:120:0:256", # Average packet length "DS:avg_score:GAUGE:120:0:1", # Average packet score "DS:neighbor_count:GAUGE:120:0:U", # Number of neighbors + "DS:noise_floor:GAUGE:120:-150:-50", # Noise floor in dBm # Packet type counters (based on pyMC payload types) "DS:type_0:COUNTER:120:0:U", # Request (PAYLOAD_TYPE_REQ) @@ -200,11 +212,20 @@ class StorageCollector: self._store_advert_sqlite(advert_record) self._publish_mqtt(advert_record, "advert") + def record_noise_floor(self, noise_floor_dbm: float): + """Record noise floor measurement every 30 seconds""" + noise_record = { + "timestamp": time.time(), + "noise_floor_dbm": noise_floor_dbm + } + self._store_noise_floor_sqlite(noise_record) + self._update_rrd_noise_metrics(noise_record) + self._publish_mqtt(noise_record, "noise_floor") + def _store_packet_sqlite(self, record: dict): try: with sqlite3.connect(self.sqlite_path) as conn: - # Ensure fields that are non-sqlite-bindable are serialized orig_path = record.get("original_path") fwd_path = record.get("forwarded_path") try: @@ -308,6 +329,20 @@ class StorageCollector: except Exception as e: logger.error(f"Failed to store advert in SQLite: {e}") + def _store_noise_floor_sqlite(self, record: dict): + + try: + with sqlite3.connect(self.sqlite_path) as conn: + conn.execute(""" + INSERT INTO noise_floor (timestamp, noise_floor_dbm) + VALUES (?, ?) + """, ( + record.get("timestamp", time.time()), + record.get("noise_floor_dbm") + )) + except Exception as e: + logger.error(f"Failed to store noise floor in SQLite: {e}") + def _update_rrd_packet_metrics(self, record: dict): if not RRDTOOL_AVAILABLE or not self.rrd_path.exists(): return @@ -352,8 +387,35 @@ class StorageCollector: except Exception as e: logger.error(f"Failed to update RRD packet metrics: {e}") + def _update_rrd_noise_metrics(self, record: dict): + + if not RRDTOOL_AVAILABLE or not self.rrd_path.exists(): + return + + try: + timestamp = int(record.get("timestamp", time.time())) + noise_floor = record.get("noise_floor_dbm", "U") + + # Skip if trying to update with old data + try: + info = rrdtool.info(str(self.rrd_path)) + last_update = int(info.get("last_update", timestamp - 60)) + if timestamp <= last_update: + return + except Exception: + pass + + # Update RRD with noise floor only, set other metrics to undefined + # Format: timestamp:rx:tx:drop:rssi:snr:length:score:neighbors:noise_floor:type_0:type_1:... + values = f"{timestamp}:0:0:0:U:U:U:U:U:{noise_floor}" + ":0" * 17 # 17 packet type counters + + rrdtool.update(str(self.rrd_path), values) + + except Exception as e: + logger.error(f"Failed to update RRD noise metrics: {e}") + def _publish_mqtt(self, record: dict, record_type: str): - """Publish record to MQTT broker.""" + if not self.mqtt_client: return @@ -426,7 +488,7 @@ class StorageCollector: return {} def get_recent_packets(self, limit: int = 100) -> list: - """Get recent packets with all fields for debugging/analysis.""" + try: with sqlite3.connect(self.sqlite_path) as conn: conn.row_factory = sqlite3.Row @@ -454,7 +516,7 @@ class StorageCollector: start_timestamp: Optional[float] = None, end_timestamp: Optional[float] = None, limit: int = 1000) -> list: - """Get packets filtered by type, route, and timestamp range.""" + try: with sqlite3.connect(self.sqlite_path) as conn: conn.row_factory = sqlite3.Row @@ -506,7 +568,7 @@ class StorageCollector: return [] def get_packet_by_hash(self, packet_hash: str) -> Optional[dict]: - """Get a specific packet by its hash.""" + try: with sqlite3.connect(self.sqlite_path) as conn: conn.row_factory = sqlite3.Row @@ -529,7 +591,7 @@ class StorageCollector: def get_rrd_data(self, start_time: Optional[int] = None, end_time: Optional[int] = None, resolution: str = "average") -> Optional[dict]: - """Get RRD time series data including packet type statistics.""" + if not RRDTOOL_AVAILABLE or not self.rrd_path.exists(): return None @@ -597,7 +659,7 @@ class StorageCollector: return None def get_packet_type_stats(self, hours: int = 24) -> dict: - """Get packet type statistics for the specified time period.""" + try: # Get RRD data for packet types end_time = int(time.time()) @@ -652,7 +714,7 @@ class StorageCollector: return {"error": str(e)} def get_neighbors(self) -> dict: - """Get all neighbors from the database formatted like the in-memory neighbors dict.""" + try: with sqlite3.connect(self.sqlite_path) as conn: conn.row_factory = sqlite3.Row @@ -706,16 +768,146 @@ class StorageCollector: result = conn.execute("DELETE FROM adverts WHERE timestamp < ?", (cutoff,)) adverts_deleted = result.rowcount + # Clean old noise floor measurements + result = conn.execute("DELETE FROM noise_floor WHERE timestamp < ?", (cutoff,)) + noise_deleted = result.rowcount + conn.commit() - if packets_deleted > 0 or adverts_deleted > 0: - logger.info(f"Cleaned up {packets_deleted} old packets and {adverts_deleted} old adverts") + if packets_deleted > 0 or adverts_deleted > 0 or noise_deleted > 0: + logger.info(f"Cleaned up {packets_deleted} old packets, {adverts_deleted} old adverts, {noise_deleted} old noise measurements") except Exception as e: logger.error(f"Failed to cleanup old data: {e}") + def get_noise_floor_history(self, hours: int = 24) -> list: + + try: + cutoff = time.time() - (hours * 3600) + + with sqlite3.connect(self.sqlite_path) as conn: + conn.row_factory = sqlite3.Row + + measurements = conn.execute(""" + SELECT timestamp, noise_floor_dbm + FROM noise_floor + WHERE timestamp > ? + ORDER BY timestamp ASC + """, (cutoff,)).fetchall() + + return [{"timestamp": row["timestamp"], "noise_floor_dbm": row["noise_floor_dbm"]} + for row in measurements] + + except Exception as e: + logger.error(f"Failed to get noise floor history: {e}") + return [] + + def get_noise_floor_stats(self, hours: int = 24) -> dict: + + try: + cutoff = time.time() - (hours * 3600) + + with sqlite3.connect(self.sqlite_path) as conn: + conn.row_factory = sqlite3.Row + + stats = conn.execute(""" + SELECT + COUNT(*) as measurement_count, + AVG(noise_floor_dbm) as avg_noise_floor, + MIN(noise_floor_dbm) as min_noise_floor, + MAX(noise_floor_dbm) as max_noise_floor + FROM noise_floor + WHERE timestamp > ? + """, (cutoff,)).fetchone() + + return { + "measurement_count": stats["measurement_count"], + "avg_noise_floor": round(stats["avg_noise_floor"] or 0, 1), + "min_noise_floor": round(stats["min_noise_floor"] or 0, 1), + "max_noise_floor": round(stats["max_noise_floor"] or 0, 1), + "hours": hours + } + + except Exception as e: + logger.error(f"Failed to get noise floor stats: {e}") + return {} + + def get_noise_floor_rrd(self, hours: int = 24, resolution: str = "average") -> dict: + + if not RRDTOOL_AVAILABLE or not self.rrd_path.exists(): + return {"error": "RRD not available"} + + try: + end_time = int(time.time()) + start_time = end_time - (hours * 3600) + + # Fetch data from RRD + fetch_result = rrdtool.fetch( + str(self.rrd_path), + resolution.upper(), + "--start", str(start_time), + "--end", str(end_time) + ) + + if not fetch_result: + return {"error": "No data available"} + + (start, end, step), data_sources, data_points = fetch_result + + # Find noise_floor data source index + try: + noise_floor_index = data_sources.index('noise_floor') + except ValueError: + return {"error": "Noise floor data not found in RRD"} + + # Extract timestamps and noise floor values + timestamps = [] + noise_values = [] + + current_time = start + for point in data_points: + timestamps.append(current_time * 1000) # Convert to milliseconds for JavaScript + noise_floor_value = point[noise_floor_index] + noise_values.append(noise_floor_value if noise_floor_value is not None else None) + current_time += step + + # Filter out None values and create chart data + chart_data = [] + valid_values = [] + + for i, (ts, value) in enumerate(zip(timestamps, noise_values)): + if value is not None: + chart_data.append([ts, value]) + valid_values.append(value) + + # Calculate statistics + stats = {} + if valid_values: + stats = { + "min": round(min(valid_values), 1), + "max": round(max(valid_values), 1), + "avg": round(sum(valid_values) / len(valid_values), 1), + "count": len(valid_values) + } + + return { + "start_time": start, + "end_time": end, + "step": step, + "hours": hours, + "resolution": resolution, + "data": chart_data, # Array of [timestamp_ms, value] pairs for charting + "timestamps": timestamps, + "values": noise_values, + "stats": stats + } + + except Exception as e: + logger.error(f"Failed to get noise floor RRD data: {e}") + return {"error": str(e)} + def close(self): - """Clean shutdown of storage systems.""" + if self.mqtt_client: self.mqtt_client.loop_stop() self.mqtt_client.disconnect() diff --git a/repeater/web/api_endpoints.py b/repeater/web/api_endpoints.py index 101d0de..2839216 100644 --- a/repeater/web/api_endpoints.py +++ b/repeater/web/api_endpoints.py @@ -451,6 +451,55 @@ class APIEndpoints: logger.error(f"Failed to save config to {config_path}: {e}") raise + @cherrypy.expose + @cherrypy.tools.json_out() + def noise_floor_history(self, hours: int = 24): + try: + storage = self._get_storage() + hours = int(hours) + history = storage.get_noise_floor_history(hours=hours) + + return self._success({ + "history": history, + "hours": hours, + "count": len(history) + }) + except Exception as e: + logger.error(f"Error fetching noise floor history: {e}") + return self._error(e) + + @cherrypy.expose + @cherrypy.tools.json_out() + def noise_floor_stats(self, hours: int = 24): + try: + storage = self._get_storage() + hours = int(hours) + stats = storage.get_noise_floor_stats(hours=hours) + + return self._success({ + "stats": stats, + "hours": hours + }) + except Exception as e: + logger.error(f"Error fetching noise floor stats: {e}") + return self._error(e) + + @cherrypy.expose + @cherrypy.tools.json_out() + def noise_floor_chart_data(self, hours: int = 24): + try: + storage = self._get_storage() + hours = int(hours) + chart_data = storage.get_noise_floor_rrd(hours=hours) + + return self._success({ + "chart_data": chart_data, + "hours": hours + }) + except Exception as e: + logger.error(f"Error fetching noise floor chart data: {e}") + return self._error(e) + @cherrypy.expose def cad_calibration_stream(self): cherrypy.response.headers['Content-Type'] = 'text/event-stream' diff --git a/repeater/web/stats-data-collection.md b/repeater/web/stats-data-collection.md index d1a646e..b520fd8 100644 --- a/repeater/web/stats-data-collection.md +++ b/repeater/web/stats-data-collection.md @@ -16,6 +16,427 @@ This document provides examples for using the pyMC_Repeater API endpoints to cre - `/api/packet_type_stats` - Get packet type distribution - `/api/rrd_data` - Get raw RRD time series data +### Noise Floor Monitoring +- `/api/noise_floor_history` - Get noise floor history (SQLite data) +- `/api/noise_floor_stats` - Get noise floor statistics +- `/api/noise_floor_chart_data` - Get noise floor RRD chart data + +## Noise Floor API Examples + +### Fetch Noise Floor History +```javascript +// Get last 24 hours of noise floor data from SQLite +async function fetchNoiseFloorHistory() { + const response = await fetch('/api/noise_floor_history?hours=24'); + const result = await response.json(); + + if (result.success) { + const history = result.data.history; + console.log(`Found ${history.length} noise floor measurements`); + + // Each record: { timestamp: 1234567890.123, noise_floor_dbm: -95.5 } + history.forEach(record => { + console.log(`${new Date(record.timestamp * 1000).toISOString()}: ${record.noise_floor_dbm} dBm`); + }); + } else { + console.error('Error:', result.error); + } +} +``` + +### Fetch Noise Floor Statistics +```javascript +// Get statistical summary of noise floor data +async function fetchNoiseFloorStats() { + const response = await fetch('/api/noise_floor_stats?hours=24'); + const result = await response.json(); + + if (result.success) { + const stats = result.data.stats; + console.log('Noise Floor Statistics:'); + console.log(`Count: ${stats.count}`); + console.log(`Average: ${stats.average?.toFixed(1)} dBm`); + console.log(`Min: ${stats.min} dBm`); + console.log(`Max: ${stats.max} dBm`); + console.log(`Std Dev: ${stats.std_dev?.toFixed(2)}`); + } +} +``` + +### Fetch Chart-Ready Noise Floor Data +```javascript +// Get RRD-based noise floor data optimized for charts +async function fetchNoiseFloorChartData() { + const response = await fetch('/api/noise_floor_chart_data?hours=24'); + const result = await response.json(); + + if (result.success) { + const chartData = result.data.chart_data; + + // Data points: [[timestamp_ms, noise_floor_dbm], ...] + chartData.data_points.forEach(point => { + const [timestamp_ms, value] = point; + console.log(`${new Date(timestamp_ms).toISOString()}: ${value} dBm`); + }); + + console.log('Statistics:', chartData.statistics); + } +} +``` + +## Noise Floor Chart Examples + +### 1. Noise Floor Time Series (Chart.js) + +```html + + + + + + + + + + + + +``` + +### 2. Noise Floor Distribution Histogram + +```html + + + + + + + + + + + +``` + +### 3. Real-time Noise Floor Monitor + +```html + + + + + + + + +
+

Real-time Noise Floor Monitor

+ +
+
+
-- dBm
+
Current
+
+
+
-- dBm
+
1h Average
+
+
+
-- dBm
+
1h Min
+
+
+
-- dBm
+
1h Max
+
+
+ + +
+ + + + +``` + ## Chart.js Examples ### 1. Packet Type Distribution (Pie Chart)