""" Unified Configuration Management for Trixy ML Integration. This module provides comprehensive configuration management for all ML components in the Trixy voice assistant system, including wakeword detection, voice recognition, audio pipeline, and system integration. Key Features: - Unified configuration for all ML components - Environment-specific configurations (dev, prod, debug) - Configuration validation and compatibility checking - Dynamic configuration updates and hot-reloading - Configuration file management (JSON, YAML) - Profile-based configuration switching - Security features for model passwords and paths - Performance tuning presets - Integration with application container Usage: from trixy_core.ml import MLConfiguration, ConfigManager # Load configuration config_manager = ConfigManager() ml_config = config_manager.load_ml_config("production") # Create ML components with unified config ml_manager = MLManager(ml_config.ml_manager) audio_pipeline = AudioPipeline(ml_config.audio_pipeline) """ import json import yaml import logging import os from pathlib import Path from typing import Dict, Any, Optional, List, Union, Type from dataclasses import dataclass, field, asdict from enum import Enum import hashlib from datetime import datetime # Import component configurations from .ml_manager import MLConfig as MLManagerConfig, MLSystemMode from .audio_pipeline import AudioPipelineConfig, AudioFormat, BufferType from .wakeword import WakewordConfig, ModelConfig as WakewordModelConfig from .voice_recognition import VoiceRecognitionConfig, ModelConfig as VoiceModelConfig class ConfigFormat(Enum): """Supported configuration file formats.""" JSON = "json" YAML = "yaml" TOML = "toml" class ConfigProfile(Enum): """Configuration profiles for different environments.""" DEVELOPMENT = "development" TESTING = "testing" PRODUCTION = "production" DEBUG = "debug" STANDALONE = "standalone" CLIENT = "client" SERVER = "server" class SecurityLevel(Enum): """Security levels for configuration.""" NONE = "none" BASIC = "basic" ENHANCED = "enhanced" STRICT = "strict" @dataclass class SecurityConfig: """Security configuration for ML system.""" level: SecurityLevel = SecurityLevel.BASIC encrypt_model_paths: bool = False require_model_passwords: bool = True validate_model_checksums: bool = True secure_config_storage: bool = False audit_configuration_changes: bool = True # Model security allowed_model_extensions: List[str] = field(default_factory=lambda: [".pth", ".pt", ".onnx"]) model_path_whitelist: List[str] = field(default_factory=list) require_signed_models: bool = False # Runtime security sandbox_model_loading: bool = False memory_protection: bool = True secure_temp_files: bool = True @dataclass class PerformanceConfig: """Performance configuration for ML system.""" # General performance optimization_level: str = "balanced" # "speed", "balanced", "quality" enable_gpu_acceleration: bool = True enable_model_quantization: bool = False enable_model_pruning: bool = False # Memory management max_memory_usage_mb: int = 1024 enable_memory_pooling: bool = True garbage_collection_threshold: int = 100 # Threading max_worker_threads: int = 4 thread_priority: str = "normal" # "low", "normal", "high" enable_thread_affinity: bool = False # Caching enable_model_caching: bool = True cache_size_mb: int = 256 cache_ttl_seconds: int = 3600 # Monitoring enable_performance_monitoring: bool = True monitoring_interval_seconds: float = 30.0 enable_profiling: bool = False @dataclass class NetworkConfig: """Network configuration for client-server communication.""" # Server connection server_host: Optional[str] = None server_port: Optional[int] = None use_ssl: bool = False ssl_cert_path: Optional[str] = None ssl_key_path: Optional[str] = None # Connection settings connection_timeout_seconds: float = 10.0 read_timeout_seconds: float = 30.0 max_retry_attempts: int = 3 retry_delay_seconds: float = 1.0 keepalive_interval_seconds: float = 30.0 # Audio streaming audio_buffer_size: int = 8192 audio_compression: bool = False audio_encryption: bool = False # Protocol settings protocol_version: str = "1.0" enable_heartbeat: bool = True enable_message_validation: bool = True @dataclass class LoggingConfig: """Logging configuration for ML system.""" # General logging level: str = "INFO" enable_file_logging: bool = True enable_console_logging: bool = True log_file_path: str = "/var/log/trixy/ml.log" # Log formatting log_format: str = "%(asctime)s - %(name)s - %(levelname)s - %(message)s" date_format: str = "%Y-%m-%d %H:%M:%S" # Log rotation max_file_size_mb: int = 100 backup_count: int = 5 # Component logging ml_manager_level: str = "INFO" audio_pipeline_level: str = "INFO" wakeword_level: str = "INFO" voice_recognition_level: str = "INFO" # Debug logging enable_debug_logging: bool = False debug_log_path: str = "/var/log/trixy/ml_debug.log" log_audio_metadata: bool = False log_model_operations: bool = False @dataclass class MLConfiguration: """Unified ML configuration containing all component configurations.""" # Profile information profile: ConfigProfile = ConfigProfile.PRODUCTION version: str = "1.0.0" created_at: str = field(default_factory=lambda: datetime.now().isoformat()) # Component configurations ml_manager: MLManagerConfig = field(default_factory=lambda: MLManagerConfig( wakeword_model_path="./models/wakeword/default/model.pth", wakeword_model_password="", voice_recognition_model_path="./models/voice_recognition/default/model.pth", voice_recognition_model_password="" )) audio_pipeline: AudioPipelineConfig = field(default_factory=AudioPipelineConfig) wakeword: WakewordConfig = field(default_factory=lambda: WakewordConfig( model_path="./models/wakeword/default/model.pth" )) voice_recognition: VoiceRecognitionConfig = field(default_factory=lambda: VoiceRecognitionConfig( model_path="./models/voice_recognition/default/model.pth" )) # System configurations security: SecurityConfig = field(default_factory=SecurityConfig) performance: PerformanceConfig = field(default_factory=PerformanceConfig) network: NetworkConfig = field(default_factory=NetworkConfig) logging: LoggingConfig = field(default_factory=LoggingConfig) # Metadata description: str = "Trixy ML Configuration" tags: List[str] = field(default_factory=list) custom_settings: Dict[str, Any] = field(default_factory=dict) def validate(self) -> bool: """ Validate the configuration. Returns: True if configuration is valid """ try: # Validate individual component configs if not self.ml_manager.validate(): return False # Validate cross-component compatibility if self.ml_manager.audio_sample_rate != self.audio_pipeline.sample_rate: return False if self.ml_manager.audio_chunk_size != self.audio_pipeline.chunk_size: return False # Validate security settings if self.security.require_model_passwords: if not self.ml_manager.wakeword_model_password: return False if not self.ml_manager.voice_recognition_model_password: return False # Validate performance settings if self.performance.max_memory_usage_mb < 256: return False # Minimum memory requirement return True except Exception: return False def get_checksum(self) -> str: """ Calculate configuration checksum for change detection. Returns: MD5 checksum of configuration """ config_dict = asdict(self) # Remove timestamp fields for consistent checksums config_dict.pop('created_at', None) config_str = json.dumps(config_dict, sort_keys=True) return hashlib.md5(config_str.encode()).hexdigest() class ConfigValidationError(Exception): """Configuration validation error.""" pass class ConfigManager: """ Configuration manager for ML system. Handles loading, saving, validating, and managing ML configurations across different environments and profiles. """ def __init__( self, config_dir: str = "./config", logger: Optional[logging.Logger] = None ): """ Initialize configuration manager. Args: config_dir: Directory containing configuration files logger: Logger instance """ self.config_dir = Path(config_dir) self.logger = logger or logging.getLogger(__name__) # Ensure config directory exists self.config_dir.mkdir(parents=True, exist_ok=True) # Configuration cache self._config_cache: Dict[str, MLConfiguration] = {} self._config_checksums: Dict[str, str] = {} # File watchers for hot reloading self._file_watchers: Dict[str, float] = {} self.logger.info(f"Configuration manager initialized: {config_dir}") def create_default_config(self, profile: ConfigProfile = ConfigProfile.PRODUCTION) -> MLConfiguration: """ Create default configuration for a profile. Args: profile: Configuration profile Returns: Default ML configuration """ config = MLConfiguration(profile=profile) if profile == ConfigProfile.DEVELOPMENT: config = self._apply_development_settings(config) elif profile == ConfigProfile.TESTING: config = self._apply_testing_settings(config) elif profile == ConfigProfile.PRODUCTION: config = self._apply_production_settings(config) elif profile == ConfigProfile.DEBUG: config = self._apply_debug_settings(config) elif profile == ConfigProfile.STANDALONE: config = self._apply_standalone_settings(config) elif profile == ConfigProfile.CLIENT: config = self._apply_client_settings(config) elif profile == ConfigProfile.SERVER: config = self._apply_server_settings(config) return config def load_config( self, config_name: str, config_format: ConfigFormat = ConfigFormat.JSON, use_cache: bool = True ) -> MLConfiguration: """ Load configuration from file. Args: config_name: Name of configuration file (without extension) config_format: Configuration file format use_cache: Whether to use cached configuration Returns: Loaded ML configuration """ # Check cache first if use_cache and config_name in self._config_cache: if not self._is_config_file_modified(config_name, config_format): return self._config_cache[config_name] # Load from file config_file = self._get_config_file_path(config_name, config_format) if not config_file.exists(): raise FileNotFoundError(f"Configuration file not found: {config_file}") try: with open(config_file, 'r') as f: if config_format == ConfigFormat.JSON: config_data = json.load(f) elif config_format == ConfigFormat.YAML: config_data = yaml.safe_load(f) else: raise ValueError(f"Unsupported config format: {config_format}") # Convert to MLConfiguration config = self._dict_to_config(config_data) # Validate configuration if not config.validate(): raise ConfigValidationError(f"Invalid configuration: {config_name}") # Cache configuration if use_cache: self._config_cache[config_name] = config self._config_checksums[config_name] = config.get_checksum() self._file_watchers[config_name] = config_file.stat().st_mtime self.logger.info(f"Configuration loaded successfully: {config_name}") return config except Exception as e: self.logger.error(f"Failed to load configuration {config_name}: {e}") raise def save_config( self, config: MLConfiguration, config_name: str, config_format: ConfigFormat = ConfigFormat.JSON, create_backup: bool = True ) -> None: """ Save configuration to file. Args: config: ML configuration to save config_name: Name of configuration file config_format: Configuration file format create_backup: Whether to create backup of existing file """ config_file = self._get_config_file_path(config_name, config_format) # Create backup if requested and file exists if create_backup and config_file.exists(): backup_file = config_file.with_suffix(f".backup.{config_format.value}") config_file.rename(backup_file) self.logger.info(f"Configuration backup created: {backup_file}") try: # Convert configuration to dictionary config_data = asdict(config) # Save to file with open(config_file, 'w') as f: if config_format == ConfigFormat.JSON: json.dump(config_data, f, indent=2, default=str) elif config_format == ConfigFormat.YAML: yaml.dump(config_data, f, default_flow_style=False) else: raise ValueError(f"Unsupported config format: {config_format}") # Update cache self._config_cache[config_name] = config self._config_checksums[config_name] = config.get_checksum() self._file_watchers[config_name] = config_file.stat().st_mtime self.logger.info(f"Configuration saved successfully: {config_name}") except Exception as e: self.logger.error(f"Failed to save configuration {config_name}: {e}") raise def validate_config(self, config: MLConfiguration) -> List[str]: """ Validate configuration and return list of issues. Args: config: Configuration to validate Returns: List of validation issues (empty if valid) """ issues = [] try: # Basic validation if not config.validate(): issues.append("Basic validation failed") # Check model file paths if config.ml_manager.wakeword_model_path and not Path(config.ml_manager.wakeword_model_path).exists(): issues.append(f"Wakeword model file not found: {config.ml_manager.wakeword_model_path}") if config.ml_manager.voice_recognition_model_path and not Path(config.ml_manager.voice_recognition_model_path).exists(): issues.append(f"Voice recognition model file not found: {config.ml_manager.voice_recognition_model_path}") # Check security requirements if config.security.require_model_passwords: if not config.ml_manager.wakeword_model_password: issues.append("Wakeword model password required but not provided") if not config.ml_manager.voice_recognition_model_password: issues.append("Voice recognition model password required but not provided") # Check performance constraints total_threads = ( config.performance.max_worker_threads + config.audio_pipeline.processing_threads ) if total_threads > 16: # Reasonable limit issues.append(f"Too many threads configured: {total_threads}") # Check memory requirements estimated_memory = ( config.performance.max_memory_usage_mb + config.audio_pipeline.max_buffer_size_mb + config.performance.cache_size_mb ) if estimated_memory > 4096: # 4GB limit issues.append(f"Memory usage too high: {estimated_memory}MB") except Exception as e: issues.append(f"Validation error: {e}") return issues def get_config_profiles(self) -> List[str]: """ Get list of available configuration profiles. Returns: List of configuration profile names """ profiles = [] for config_file in self.config_dir.glob("*.json"): profiles.append(config_file.stem) for config_file in self.config_dir.glob("*.yaml"): if config_file.stem not in profiles: profiles.append(config_file.stem) return sorted(profiles) def create_profile_configs(self) -> None: """Create default configuration files for all profiles.""" for profile in ConfigProfile: config = self.create_default_config(profile) self.save_config(config, profile.value, ConfigFormat.JSON) self.logger.info(f"Created default configuration for profile: {profile.value}") def update_config( self, config_name: str, updates: Dict[str, Any], config_format: ConfigFormat = ConfigFormat.JSON ) -> MLConfiguration: """ Update specific configuration values. Args: config_name: Name of configuration to update updates: Dictionary of updates to apply config_format: Configuration file format Returns: Updated configuration """ # Load current configuration config = self.load_config(config_name, config_format) # Apply updates config_dict = asdict(config) self._deep_update(config_dict, updates) # Convert back to configuration object updated_config = self._dict_to_config(config_dict) # Validate updated configuration issues = self.validate_config(updated_config) if issues: raise ConfigValidationError(f"Updated configuration is invalid: {issues}") # Save updated configuration self.save_config(updated_config, config_name, config_format) return updated_config def _apply_development_settings(self, config: MLConfiguration) -> MLConfiguration: """Apply development-specific settings.""" config.logging.level = "DEBUG" config.logging.enable_debug_logging = True config.performance.enable_profiling = True config.security.level = SecurityLevel.BASIC config.tags.append("development") return config def _apply_testing_settings(self, config: MLConfiguration) -> MLConfiguration: """Apply testing-specific settings.""" config.logging.level = "INFO" config.performance.optimization_level = "speed" config.security.level = SecurityLevel.BASIC config.tags.append("testing") return config def _apply_production_settings(self, config: MLConfiguration) -> MLConfiguration: """Apply production-specific settings.""" config.logging.level = "WARNING" config.logging.enable_debug_logging = False config.performance.optimization_level = "balanced" config.security.level = SecurityLevel.ENHANCED config.security.audit_configuration_changes = True config.performance.enable_performance_monitoring = True config.tags.append("production") return config def _apply_debug_settings(self, config: MLConfiguration) -> MLConfiguration: """Apply debug-specific settings.""" config.logging.level = "DEBUG" config.logging.enable_debug_logging = True config.logging.log_audio_metadata = True config.logging.log_model_operations = True config.performance.enable_profiling = True config.audio_pipeline.enable_debug_logging = True config.audio_pipeline.save_debug_audio = True config.tags.append("debug") return config def _apply_standalone_settings(self, config: MLConfiguration) -> MLConfiguration: """Apply standalone-specific settings.""" config.ml_manager.system_mode = MLSystemMode.STANDALONE config.network.server_host = None config.network.server_port = None config.tags.append("standalone") return config def _apply_client_settings(self, config: MLConfiguration) -> MLConfiguration: """Apply client-specific settings.""" config.ml_manager.system_mode = MLSystemMode.CLIENT config.network.server_host = "localhost" config.network.server_port = 2101 config.tags.append("client") return config def _apply_server_settings(self, config: MLConfiguration) -> MLConfiguration: """Apply server-specific settings.""" # Server typically doesn't run ML components directly config.performance.max_memory_usage_mb = 2048 config.network.server_host = "0.0.0.0" config.network.server_port = 2101 config.tags.append("server") return config def _get_config_file_path(self, config_name: str, config_format: ConfigFormat) -> Path: """Get full path to configuration file.""" return self.config_dir / f"{config_name}.{config_format.value}" def _is_config_file_modified(self, config_name: str, config_format: ConfigFormat) -> bool: """Check if configuration file has been modified.""" config_file = self._get_config_file_path(config_name, config_format) if not config_file.exists(): return True if config_name not in self._file_watchers: return True return config_file.stat().st_mtime > self._file_watchers[config_name] def _dict_to_config(self, config_dict: Dict[str, Any]) -> MLConfiguration: """Convert dictionary to MLConfiguration object.""" # This is a simplified conversion - in practice, you'd want more robust handling try: # Handle enum conversions if 'profile' in config_dict and isinstance(config_dict['profile'], str): config_dict['profile'] = ConfigProfile(config_dict['profile']) # Extract component configurations ml_manager_data = config_dict.get('ml_manager', {}) audio_pipeline_data = config_dict.get('audio_pipeline', {}) wakeword_data = config_dict.get('wakeword', {}) voice_recognition_data = config_dict.get('voice_recognition', {}) security_data = config_dict.get('security', {}) performance_data = config_dict.get('performance', {}) network_data = config_dict.get('network', {}) logging_data = config_dict.get('logging', {}) # Create component configurations with basic handling ml_manager_config = MLManagerConfig( wakeword_model_path=ml_manager_data.get('wakeword_model_path', ''), wakeword_model_password=ml_manager_data.get('wakeword_model_password', ''), voice_recognition_model_path=ml_manager_data.get('voice_recognition_model_path', ''), voice_recognition_model_password=ml_manager_data.get('voice_recognition_model_password', ''), **{k: v for k, v in ml_manager_data.items() if k not in [ 'wakeword_model_path', 'wakeword_model_password', 'voice_recognition_model_path', 'voice_recognition_model_password' ]} ) audio_pipeline_config = AudioPipelineConfig(**audio_pipeline_data) wakeword_config = WakewordConfig( model_path=wakeword_data.get('model_path', ''), **{k: v for k, v in wakeword_data.items() if k != 'model_path'} ) voice_recognition_config = VoiceRecognitionConfig( model_path=voice_recognition_data.get('model_path', ''), **{k: v for k, v in voice_recognition_data.items() if k != 'model_path'} ) # Handle security config enums if 'level' in security_data and isinstance(security_data['level'], str): security_data['level'] = SecurityLevel(security_data['level']) security_config = SecurityConfig(**security_data) performance_config = PerformanceConfig(**performance_data) network_config = NetworkConfig(**network_data) logging_config = LoggingConfig(**logging_data) # Create main configuration config = MLConfiguration( profile=config_dict.get('profile', ConfigProfile.PRODUCTION), version=config_dict.get('version', '1.0.0'), created_at=config_dict.get('created_at', datetime.now().isoformat()), ml_manager=ml_manager_config, audio_pipeline=audio_pipeline_config, wakeword=wakeword_config, voice_recognition=voice_recognition_config, security=security_config, performance=performance_config, network=network_config, logging=logging_config, description=config_dict.get('description', 'Trixy ML Configuration'), tags=config_dict.get('tags', []), custom_settings=config_dict.get('custom_settings', {}) ) return config except Exception as e: raise ConfigValidationError(f"Failed to convert dictionary to configuration: {e}") def _deep_update(self, base_dict: Dict[str, Any], update_dict: Dict[str, Any]) -> None: """Deep update dictionary with nested values.""" for key, value in update_dict.items(): if key in base_dict and isinstance(base_dict[key], dict) and isinstance(value, dict): self._deep_update(base_dict[key], value) else: base_dict[key] = value def create_config_manager(config_dir: str = "./config") -> ConfigManager: """ Create configuration manager. Args: config_dir: Configuration directory path Returns: Configuration manager instance """ return ConfigManager(config_dir) def load_ml_config( config_name: str = "production", config_dir: str = "./config" ) -> MLConfiguration: """ Load ML configuration. Args: config_name: Configuration name config_dir: Configuration directory Returns: ML configuration """ config_manager = ConfigManager(config_dir) return config_manager.load_config(config_name) def create_default_configs(config_dir: str = "./config") -> None: """ Create default configuration files for all profiles. Args: config_dir: Configuration directory """ config_manager = ConfigManager(config_dir) config_manager.create_profile_configs()