"""Configuration management for TSU-WAVE""" import yaml import json import os from pathlib import Path from typing import Dict, Any, Optional import logging logger = logging.getLogger(__name__) def get_default_config() -> Dict[str, Any]: """Get default configuration""" return { 'system': { 'environment': 'production', 'log_level': 'INFO', 'data_directory': '/var/lib/tsu-wave' }, 'database': { 'host': 'localhost', 'port': 5432, 'name': 'tsuwave', 'user': 'tsuwave_user', 'password': '${DB_PASSWORD}', 'pool_size': 10 }, 'redis': { 'host': 'localhost', 'port': 6379, 'db': 0, 'password': '${REDIS_PASSWORD}' }, 'api': { 'host': '0.0.0.0', 'port': 8000, 'workers': 4, 'jwt_secret': '${JWT_SECRET}', 'token_expiry': 86400, 'rate_limit': 100 }, 'dashboard': { 'host': '0.0.0.0', 'port': 8080, 'refresh_interval': 5 }, 'parameters': { 'wcc': {'alert': 1.35, 'critical': 1.58}, 'kpr': {'alert': 1.6, 'critical': 2.0}, 'hfsi': {'alert': 0.6, 'critical': 0.4}, 'becf': {'alert': 4.0, 'critical': 6.0}, 'sdb': {'alert': 2.5, 'critical': 1.0}, 'sbsp': {'alert': 0.7, 'critical': 1.2}, 'smvi': {'alert': 0.4, 'critical': 0.6} }, 'chi_weights': { 'wcc': 0.12, 'kpr': 0.19, 'hfsi': 0.24, 'becf': 0.21, 'sdb': 0.08, 'sbsp': 0.11, 'smvi': 0.05 } } def load_config(config_path: Optional[str] = None) -> Dict[str, Any]: """Load configuration from file""" if config_path is None: config_path = os.environ.get('TSUWAVE_CONFIG', 'config/config.yml') config = get_default_config() # Try to load from file config_file = Path(config_path) if config_file.exists(): try: with open(config_file, 'r') as f: if config_file.suffix == '.yml' or config_file.suffix == '.yaml': user_config = yaml.safe_load(f) elif config_file.suffix == '.json': user_config = json.load(f) else: logger.warning(f"Unknown config file type: {config_file.suffix}") return config # Deep merge config = _deep_merge(config, user_config) logger.info(f"Loaded configuration from {config_path}") except Exception as e: logger.error(f"Error loading config: {e}") else: logger.warning(f"Config file not found: {config_path}, using defaults") # Override with environment variables config = _apply_env_overrides(config) return config def save_config(config: Dict[str, Any], config_path: str): """Save configuration to file""" config_file = Path(config_path) config_file.parent.mkdir(parents=True, exist_ok=True) try: with open(config_file, 'w') as f: if config_file.suffix == '.yml' or config_file.suffix == '.yaml': yaml.dump(config, f, default_flow_style=False) elif config_file.suffix == '.json': json.dump(config, f, indent=2) else: logger.warning(f"Unknown config file type: {config_file.suffix}") return logger.info(f"Saved configuration to {config_path}") except Exception as e: logger.error(f"Error saving config: {e}") def _deep_merge(base: Dict, update: Dict) -> Dict: """Deep merge two dictionaries""" result = base.copy() for key, value in update.items(): if key in result and isinstance(result[key], dict) and isinstance(value, dict): result[key] = _deep_merge(result[key], value) else: result[key] = value return result def _apply_env_overrides(config: Dict[str, Any]) -> Dict[str, Any]: """Apply environment variable overrides""" # Database overrides if os.environ.get('TSUWAVE_DB_HOST'): config['database']['host'] = os.environ['TSUWAVE_DB_HOST'] if os.environ.get('TSUWAVE_DB_PORT'): config['database']['port'] = int(os.environ['TSUWAVE_DB_PORT']) if os.environ.get('TSUWAVE_DB_NAME'): config['database']['name'] = os.environ['TSUWAVE_DB_NAME'] if os.environ.get('TSUWAVE_DB_USER'): config['database']['user'] = os.environ['TSUWAVE_DB_USER'] if os.environ.get('TSUWAVE_DB_PASSWORD'): config['database']['password'] = os.environ['TSUWAVE_DB_PASSWORD'] # Redis overrides if os.environ.get('TSUWAVE_REDIS_HOST'): config['redis']['host'] = os.environ['TSUWAVE_REDIS_HOST'] if os.environ.get('TSUWAVE_REDIS_PORT'): config['redis']['port'] = int(os.environ['TSUWAVE_REDIS_PORT']) if os.environ.get('TSUWAVE_REDIS_PASSWORD'): config['redis']['password'] = os.environ['TSUWAVE_REDIS_PASSWORD'] # API overrides if os.environ.get('TSUWAVE_API_HOST'): config['api']['host'] = os.environ['TSUWAVE_API_HOST'] if os.environ.get('TSUWAVE_API_PORT'): config['api']['port'] = int(os.environ['TSUWAVE_API_PORT']) if os.environ.get('TSUWAVE_JWT_SECRET'): config['api']['jwt_secret'] = os.environ['TSUWAVE_JWT_SECRET'] # System overrides if os.environ.get('TSUWAVE_ENV'): config['system']['environment'] = os.environ['TSUWAVE_ENV'] if os.environ.get('TSUWAVE_LOG_LEVEL'): config['system']['log_level'] = os.environ['TSUWAVE_LOG_LEVEL'] return config def get_config_value(config: Dict[str, Any], key_path: str, default: Any = None) -> Any: """Get nested config value using dot notation""" keys = key_path.split('.') value = config for key in keys: if isinstance(value, dict) and key in value: value = value[key] else: return default return value def set_config_value(config: Dict[str, Any], key_path: str, value: Any): """Set nested config value using dot notation""" keys = key_path.split('.') target = config for key in keys[:-1]: if key not in target: target[key] = {} target = target[key] target[keys[-1]] = value def validate_config(config: Dict[str, Any]) -> bool: """Validate configuration""" required_keys = [ 'database.host', 'database.port', 'database.name', 'api.port', 'system.environment' ] for key in required_keys: if get_config_value(config, key) is None: logger.error(f"Missing required config: {key}") return False # Validate port numbers api_port = get_config_value(config, 'api.port') if not isinstance(api_port, int) or api_port < 1 or api_port > 65535: logger.error(f"Invalid API port: {api_port}") return False db_port = get_config_value(config, 'database.port') if not isinstance(db_port, int) or db_port < 1 or db_port > 65535: logger.error(f"Invalid database port: {db_port}") return False return True def create_example_config(output_path: str = 'config/config.example.yml'): """Create example configuration file""" config = get_default_config() # Replace sensitive values with placeholders config['database']['password'] = '${DB_PASSWORD}' config['redis']['password'] = '${REDIS_PASSWORD}' config['api']['jwt_secret'] = '${JWT_SECRET}' save_config(config, output_path) logger.info(f"Created example config at {output_path}")