#!/usr/bin/env python3 """ Flask web server for GNSS Guard dashboard Provides real-time monitoring of GPS sources with auto-refresh """ import json import logging import sqlite3 import time from datetime import datetime, timezone from pathlib import Path from typing import Dict, Any, Optional from flask import Flask, render_template, jsonify, request logger = logging.getLogger("gnss_guard.web") class WebServer: """Web server for GNSS Guard dashboard""" def __init__(self, config, database, buzzer_service=None): """ Initialize web server Args: config: Config instance database: Database instance buzzer_service: Optional BuzzerService instance for alarm control """ self.config = config self.database = database self.buzzer_service = buzzer_service # Create Flask app self.app = Flask( __name__, template_folder=str(Path(__file__).parent / "templates"), static_folder=str(Path(__file__).parent / "static") ) # Setup routes self.app.add_url_rule('/', 'index', self.index) self.app.add_url_rule('/api/status', 'api_status', self.api_status) self.app.add_url_rule('/api/route', 'api_route', self.api_route) self.app.add_url_rule('/api/alarm/acknowledge', 'api_alarm_acknowledge', self.api_alarm_acknowledge, methods=['POST']) self.app.add_url_rule('/api/alarm/status', 'api_alarm_status', self.api_alarm_status) def index(self): """Render main dashboard page""" return render_template('dashboard.html', show_route=self.config.web_show_route) def api_status(self): """ API endpoint returning latest validation data Returns: JSON response with current status of all GPS sources """ try: # Get latest validation record conn = sqlite3.connect(str(self.database.database_path), check_same_thread=False, timeout=5.0) cursor = conn.cursor() cursor.execute( """ SELECT validation_timestamp, validation_timestamp_unix, is_valid, sources_missing, sources_stale, coordinate_differences, source_coordinates, validation_details FROM positions_validation ORDER BY validation_timestamp_unix DESC LIMIT 1 """ ) row = cursor.fetchone() conn.close() if not row: return jsonify({ "error": "No validation data available", "timestamp": datetime.now(timezone.utc).isoformat() }), 404 # Parse row data validation_timestamp = row[0] validation_timestamp_unix = row[1] is_valid = bool(row[2]) sources_missing = json.loads(row[3]) if row[3] else [] sources_stale = json.loads(row[4]) if row[4] else [] coordinate_differences = json.loads(row[5]) if row[5] else {} source_coordinates = json.loads(row[6]) if row[6] else {} validation_details = json.loads(row[7]) if row[7] else {} # Get enabled sources from config enabled_sources = self.config.get_enabled_sources() # Source name mapping for display source_display_names = { "nmea_primary": "Primary GPS", "nmea_secondary": "Secondary GPS", "tm_ais": "TM AIS GPS", "starlink_gps": "Starlink GPS", "starlink_location": "Starlink Location" } # Build sources status sources = {} all_source_names = ["nmea_primary", "nmea_secondary", "tm_ais", "starlink_gps", "starlink_location"] for source_name in all_source_names: display_name = source_display_names.get(source_name, source_name) # Check if source is enabled if source_name not in enabled_sources: sources[source_name] = { "display_name": display_name, "enabled": False, "status": "not_configured", "is_stale": False, "coordinates": None, "last_update": None, "last_update_unix": None } continue # Source is enabled source_data = source_coordinates.get(source_name) # Check if this source is stale is_stale = source_name in sources_stale if not source_data: # Enabled but missing sources[source_name] = { "display_name": display_name, "enabled": True, "status": "missing", "is_stale": is_stale, "coordinates": None, "last_update": None, "last_update_unix": None } else: # Has data - check if stale status = "stale" if is_stale else "ok" sources[source_name] = { "display_name": display_name, "enabled": True, "status": status, "is_stale": is_stale, "coordinates": { "latitude": source_data.get("latitude"), "longitude": source_data.get("longitude") }, "last_update": source_data.get("timestamp"), "last_update_unix": source_data.get("timestamp_unix") } # Get threshold from validation_details threshold_meters = validation_details.get("threshold_meters", 100.0) # Calculate maximum distance for alert banner max_distance_km = None max_distance_m = 0.0 if not is_valid and coordinate_differences: # Find maximum distance in coordinate_differences for source_pair, diff_data in coordinate_differences.items(): if isinstance(diff_data, dict): # Try both field names (distance_meters is the correct one) distance = diff_data.get("distance_meters", diff_data.get("distance_m", 0)) if distance > max_distance_m: max_distance_m = distance # Only show distance alert if max distance exceeds threshold if max_distance_m > threshold_meters: max_distance_km = max_distance_m / 1000.0 # Determine alert state # Show alert if validation fails AND distance exceeds threshold (GPS jamming/spoofing) has_alert = (not is_valid and max_distance_km is not None) or len(sources_missing) > 0 # Find center coordinate for map (priority: nmea_primary > tm_ais > starlink_location) map_center = None for priority_source in ["nmea_primary", "tm_ais", "starlink_location"]: if sources.get(priority_source, {}).get("coordinates"): coords = sources[priority_source]["coordinates"] if coords.get("latitude") and coords.get("longitude"): map_center = coords break # If no priority source, use any available if not map_center: for source_name, source_data in sources.items(): if source_data.get("coordinates"): coords = source_data["coordinates"] if coords.get("latitude") and coords.get("longitude"): map_center = coords break # Build response response = { "timestamp": datetime.now(timezone.utc).isoformat(), "validation_timestamp": validation_timestamp, "validation_timestamp_unix": validation_timestamp_unix, "is_valid": is_valid, "has_alert": has_alert, "max_distance_km": max_distance_km, "threshold_meters": threshold_meters, "sources": sources, "sources_stale": sources_stale, "map_center": map_center, "asset_name": self.config.asset_name } return jsonify(response) except Exception as e: logger.error(f"Error in api_status: {e}") return jsonify({ "error": str(e), "timestamp": datetime.now(timezone.utc).isoformat() }), 500 def api_route(self): """ API endpoint returning 24h route data for map visualization Returns: JSON array of route points with coordinates and validation status """ if not self.config.web_show_route: return jsonify({"error": "Route feature is disabled"}), 403 try: hours = 24 conn = sqlite3.connect(str(self.database.database_path), check_same_thread=False, timeout=5.0) cursor = conn.cursor() # Determine reference time for the 24h window if self.config.demo_unit: # Demo mode: use the PREVIOUS-TO-LAST historical record's timestamp as reference # The last record is the "live" one that gets deleted, so we skip it # Exclude recent "live" records (last 5 minutes) added during demo session five_minutes_ago = time.time() - 300 cursor.execute( """ SELECT validation_timestamp_unix FROM positions_validation WHERE validation_timestamp_unix < ? ORDER BY validation_timestamp_unix DESC LIMIT 1 OFFSET 1 """, (five_minutes_ago,) ) result = cursor.fetchone() reference_time = result[0] if result and result[0] else time.time() else: # Normal mode: use current time as reference reference_time = time.time() # Get validations from the 24h window cutoff_unix = reference_time - (hours * 3600) cursor.execute( """ SELECT validation_timestamp, validation_timestamp_unix, is_valid, sources_missing, sources_stale, source_coordinates, validation_details FROM positions_validation WHERE validation_timestamp_unix >= ? AND validation_timestamp_unix <= ? ORDER BY validation_timestamp_unix DESC """, (cutoff_unix, reference_time) ) rows = cursor.fetchall() conn.close() route_points = [] for row in rows: validation_timestamp = row[0] validation_timestamp_unix = row[1] is_valid = bool(row[2]) sources_missing = json.loads(row[3]) if row[3] else [] sources_stale = json.loads(row[4]) if row[4] else [] source_coordinates = json.loads(row[5]) if row[5] else {} validation_details = json.loads(row[6]) if row[6] else {} # Find best coordinate (priority: nmea_primary > tm_ais > starlink_location) coord = None for priority_source in ["nmea_primary", "tm_ais", "starlink_location", "starlink_gps"]: if priority_source in source_coordinates: src_data = source_coordinates[priority_source] if src_data.get("latitude") and src_data.get("longitude"): coord = { "latitude": src_data["latitude"], "longitude": src_data["longitude"] } break if not coord: continue # Determine status threshold = validation_details.get("threshold_meters", 200) max_distance = validation_details.get("max_difference_meters", 0) if not is_valid and max_distance > threshold: status = "alert" elif sources_missing or sources_stale: status = "degraded" else: status = "valid" route_points.append({ "lat": coord["latitude"], "lng": coord["longitude"], "timestamp": validation_timestamp, "timestamp_unix": validation_timestamp_unix, "status": status, "is_valid": is_valid, "sources_missing": sources_missing, "sources_stale": sources_stale, "max_distance_m": max_distance, "threshold_m": threshold }) return jsonify(route_points) except Exception as e: logger.error(f"Error in api_route: {e}") return jsonify({ "error": str(e), "timestamp": datetime.now(timezone.utc).isoformat() }), 500 def api_alarm_acknowledge(self): """ API endpoint to acknowledge the buzzer alarm. POST /api/alarm/acknowledge Returns: JSON response with acknowledgment status """ try: if not self.buzzer_service: return jsonify({ "success": False, "error": "Buzzer service not available", "alarm_active": False, "alarm_acknowledged": False }), 503 # Acknowledge the alarm was_active = self.buzzer_service.is_alarm_active() acknowledged = self.buzzer_service.acknowledge_alarm() logger.info(f"Alarm acknowledge request: was_active={was_active}, acknowledged={acknowledged}") return jsonify({ "success": True, "acknowledged": acknowledged, "alarm_active": self.buzzer_service.is_alarm_active(), "alarm_acknowledged": self.buzzer_service.is_alarm_acknowledged(), "timestamp": datetime.now(timezone.utc).isoformat() }) except Exception as e: logger.error(f"Error in api_alarm_acknowledge: {e}") return jsonify({ "success": False, "error": str(e), "timestamp": datetime.now(timezone.utc).isoformat() }), 500 def api_alarm_status(self): """ API endpoint to get current alarm status. GET /api/alarm/status Returns: JSON response with alarm status """ try: if not self.buzzer_service: return jsonify({ "available": False, "alarm_active": False, "alarm_acknowledged": False, "buzzer_status": "unavailable" }) return jsonify({ "available": True, "alarm_active": self.buzzer_service.is_alarm_active(), "alarm_acknowledged": self.buzzer_service.is_alarm_acknowledged(), "buzzer_status": self.buzzer_service.get_status(), "timestamp": datetime.now(timezone.utc).isoformat() }) except Exception as e: logger.error(f"Error in api_alarm_status: {e}") return jsonify({ "available": False, "error": str(e), "timestamp": datetime.now(timezone.utc).isoformat() }), 500 def run(self, host='0.0.0.0', port=8080, debug=False): """ Run the Flask web server Args: host: Host to bind to port: Port to bind to debug: Enable debug mode """ logger.info(f"Starting web server on {host}:{port}") self.app.run(host=host, port=port, debug=debug, threaded=True)