Spaces:
Sleeping
Sleeping
| """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}") | |