reterminal-dm4/backup-from-device/gnss-guard/tm-gnss-guard/config.py

#!/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,
        }