#!/usr/bin/env python3 """ Configuration management for GNSS Guard Loads configuration from .env or .env.prod files """ import os from pathlib import Path from typing import Dict, Any from dotenv import load_dotenv class Config: """Configuration manager for GNSS Guard""" @staticmethod def _get_int_env(key: str, default: int) -> int: """Get integer environment variable, handling empty strings""" value = os.getenv(key, "") if not value or value.strip() == "": return default try: return int(value) except ValueError: return default def __init__(self): # Determine environment file to load # Priority: 1) ENV=prod -> .env.prod, 2) .env.prod exists -> .env.prod, 3) .env base_path = Path(__file__).parent if os.getenv("ENV") == "prod": env_file = ".env.prod" elif (base_path / ".env.prod").exists(): env_file = ".env.prod" else: env_file = ".env" # Load environment variables env_path = base_path / env_file if env_path.exists(): load_dotenv(env_path) else: # Try loading from current directory as fallback load_dotenv() # Asset configuration self.asset_name = os.getenv("ASSET_NAME", "unknown") # Timing configuration self.iteration_period_seconds = self._get_int_env("ITERATION_PERIOD_SECONDS", 10) self.stale_threshold_seconds = self._get_int_env("STALE_THRESHOLD_SECONDS", 60) self.validation_threshold_meters = float(os.getenv("VALIDATION_THRESHOLD_METERS", "200")) self.startup_warmup_seconds = self._get_int_env("STARTUP_WARMUP_SECONDS", 5) # Data retention configuration self.positions_raw_retention_days = self._get_int_env("POSITIONS_RAW_RETENTION_DAYS", 14) self.positions_validation_retention_days = self._get_int_env("POSITIONS_VALIDATION_RETENTION_DAYS", 31) self.log_retention_days = self._get_int_env("LOG_RETENTION_DAYS", 14) # TM AIS GPS configuration self.tm_ais_url = os.getenv("TM_AIS_URL", "https://localhost:8443/location") # Trim whitespace from token (common issue with .env files) self.tm_ais_token = os.getenv("TM_AIS_TOKEN", "").strip() self.tm_ais_max_retries = self._get_int_env("TM_AIS_MAX_RETRIES", 3) # Starlink configuration self.starlink_ip = os.getenv("STARLINK_IP", "10.130.60.70") self.starlink_port = self._get_int_env("STARLINK_PORT", 9200) self.starlink_max_retries = self._get_int_env("STARLINK_MAX_RETRIES", 3) # NMEA Primary GPS configuration self.nmea_primary_ip = os.getenv("NMEA_PRIMARY_IP", "") self.nmea_primary_port = self._get_int_env("NMEA_PRIMARY_PORT", 0) # NMEA Secondary GPS configuration self.nmea_secondary_ip = os.getenv("NMEA_SECONDARY_IP", "") self.nmea_secondary_port = self._get_int_env("NMEA_SECONDARY_PORT", 0) # Database configuration self.database_path = Path(os.getenv("DATABASE_PATH", "data/gnss_guard.db")) # Logs configuration self.logs_base_path = Path(os.getenv("LOGS_BASE_PATH", "logs")) # Web server configuration self.web_enabled = os.getenv("WEB_ENABLED", "true").lower() in ("true", "1", "yes") self.web_host = os.getenv("WEB_HOST", "0.0.0.0") self.web_port = self._get_int_env("WEB_PORT", 8080) self.web_show_route = os.getenv("WEB_SHOW_ROUTE", "false").lower() in ("true", "1", "yes") # Demo mode - when enabled, route shows last 24h of available data instead of current time self.demo_unit = os.getenv("DEMO_UNIT", "false").lower() in ("true", "1", "yes") # Source enablement flags self.tm_ais_enabled = os.getenv("TM_AIS_ENABLED", "true").lower() in ("true", "1", "yes") self.starlink_enabled = os.getenv("STARLINK_ENABLED", "true").lower() in ("true", "1", "yes") self.nmea_primary_enabled = os.getenv("NMEA_PRIMARY_ENABLED", "false").lower() in ("true", "1", "yes") self.nmea_secondary_enabled = os.getenv("NMEA_SECONDARY_ENABLED", "false").lower() in ("true", "1", "yes") # NMEA verbose logging (log all NMEA sentences, not just GGA) self.nmea_verbose_logging = os.getenv("NMEA_VERBOSE_LOGGING", "false").lower() in ("true", "1", "yes") # Server sync configuration self.server_enabled = os.getenv("SERVER_ENABLED", "false").lower() in ("true", "1", "yes") self.server_url = os.getenv("SERVER_URL", "").strip() self.server_token = os.getenv("SERVER_TOKEN", "").strip() self.server_sync_batch_size = self._get_int_env("SERVER_SYNC_BATCH_SIZE", 100) self.server_sync_max_queue = self._get_int_env("SERVER_SYNC_MAX_QUEUE", 1000) def get_enabled_sources(self) -> list: """Get list of enabled source names""" sources = [] if self.tm_ais_enabled: sources.append("tm_ais") if self.starlink_enabled: sources.extend(["starlink_location", "starlink_gps"]) if self.nmea_primary_enabled: sources.append("nmea_primary") if self.nmea_secondary_enabled: sources.append("nmea_secondary") return sources def to_dict(self) -> Dict[str, Any]: """Convert configuration to dictionary""" return { "asset_name": self.asset_name, "iteration_period_seconds": self.iteration_period_seconds, "stale_threshold_seconds": self.stale_threshold_seconds, "validation_threshold_meters": self.validation_threshold_meters, "startup_warmup_seconds": self.startup_warmup_seconds, "positions_raw_retention_days": self.positions_raw_retention_days, "positions_validation_retention_days": self.positions_validation_retention_days, "log_retention_days": self.log_retention_days, "tm_ais_url": self.tm_ais_url, "tm_ais_enabled": self.tm_ais_enabled, "tm_ais_max_retries": self.tm_ais_max_retries, "starlink_ip": self.starlink_ip, "starlink_port": self.starlink_port, "starlink_enabled": self.starlink_enabled, "starlink_max_retries": self.starlink_max_retries, "nmea_primary_enabled": self.nmea_primary_enabled, "nmea_secondary_enabled": self.nmea_secondary_enabled, "database_path": str(self.database_path), "logs_base_path": str(self.logs_base_path), "web_enabled": self.web_enabled, "web_host": self.web_host, "web_port": self.web_port, "web_show_route": self.web_show_route, }