""" Configuration Data Models for Trixy Application This module defines type-safe configuration data structures for all deployment modes: - ServerConfig: Configuration for server mode - ClientConfig: Configuration for client/satellite mode - StandaloneConfig: Configuration for standalone mode Each configuration class provides: - Type-safe attribute access - Validation methods - Serialization/deserialization - Environment variable override support - Thread-safe access patterns """ from dataclasses import dataclass, field, asdict from typing import Dict, List, Optional, Union, Any, Set from pathlib import Path import os from enum import Enum class LogLevel(Enum): """Logging levels for the application.""" DEBUG = "debug" INFO = "info" WARNING = "warning" ERROR = "error" CRITICAL = "critical" class DeploymentMode(Enum): """Deployment modes for the Trixy application.""" SERVER = "server" CLIENT = "client" STANDALONE = "standalone" @dataclass class NetworkConfig: """Network configuration for sockets and connections.""" # Command socket configuration command_port: int = 2101 command_bind_address: str = "0.0.0.0" command_max_connections: int = 50 command_timeout: int = 30 command_buffer_size: int = 8192 # Audio stream ports raw_audio_input_port: int = 2102 raw_audio_output_port: int = 2103 raw_music_output_port: int = 2104 # Audio stream configuration audio_bind_address: str = "0.0.0.0" audio_timeout: int = 10 audio_buffer_size: int = 16384 # Connection settings heartbeat_interval: int = 30 max_retry_attempts: int = 5 retry_delay: int = 5 connection_timeout: int = 10 # Security settings enable_encryption: bool = False encryption_key_file: Optional[str] = None enable_compression: bool = True compression_level: int = 6 def validate(self) -> List[str]: """Validate network configuration and return any errors.""" errors = [] # Port validation ports = [ self.command_port, self.raw_audio_input_port, self.raw_audio_output_port, self.raw_music_output_port ] for port in ports: if not (1024 <= port <= 65535): errors.append(f"Port {port} must be between 1024 and 65535") # Check for port conflicts if len(set(ports)) != len(ports): errors.append("Network ports must be unique") # Validate addresses if not self.command_bind_address: errors.append("Command bind address cannot be empty") if not self.audio_bind_address: errors.append("Audio bind address cannot be empty") # Validate timeouts and limits if self.command_timeout <= 0: errors.append("Command timeout must be positive") if self.audio_timeout <= 0: errors.append("Audio timeout must be positive") if self.command_max_connections <= 0: errors.append("Max connections must be positive") return errors @dataclass class AudioConfig: """Audio processing configuration.""" # Audio format settings sample_rate: int = 16000 channels: int = 1 # Mono bit_depth: int = 16 frame_size: int = 1024 # Recording settings max_recording_duration: int = 60 # seconds silence_threshold: float = 0.01 silence_duration: float = 3.0 # seconds pre_recording_buffer: float = 1.0 # seconds # Audio processing enable_noise_reduction: bool = True enable_auto_gain: bool = True gain_level: float = 1.0 # VAD (Voice Activity Detection) settings vad_enabled: bool = True vad_aggressiveness: int = 2 # 0-3, higher = more aggressive vad_frame_duration: int = 30 # milliseconds # Music playback settings music_sample_rate: int = 48000 music_channels: int = 2 # Stereo music_bit_depth: int = 16 def validate(self) -> List[str]: """Validate audio configuration and return any errors.""" errors = [] # Sample rate validation valid_sample_rates = [8000, 16000, 22050, 44100, 48000] if self.sample_rate not in valid_sample_rates: errors.append(f"Sample rate must be one of {valid_sample_rates}") if self.music_sample_rate not in valid_sample_rates: errors.append(f"Music sample rate must be one of {valid_sample_rates}") # Channels validation if self.channels not in [1, 2]: errors.append("Channels must be 1 (mono) or 2 (stereo)") if self.music_channels not in [1, 2]: errors.append("Music channels must be 1 (mono) or 2 (stereo)") # Bit depth validation if self.bit_depth not in [16, 24, 32]: errors.append("Bit depth must be 16, 24, or 32") if self.music_bit_depth not in [16, 24, 32]: errors.append("Music bit depth must be 16, 24, or 32") # Duration and threshold validation if self.max_recording_duration <= 0: errors.append("Max recording duration must be positive") if not (0.0 <= self.silence_threshold <= 1.0): errors.append("Silence threshold must be between 0.0 and 1.0") if self.silence_duration <= 0: errors.append("Silence duration must be positive") if self.pre_recording_buffer < 0: errors.append("Pre-recording buffer must be non-negative") # VAD validation if not (0 <= self.vad_aggressiveness <= 3): errors.append("VAD aggressiveness must be between 0 and 3") if self.vad_frame_duration not in [10, 20, 30]: errors.append("VAD frame duration must be 10, 20, or 30 ms") return errors @dataclass class MLConfig: """Machine Learning model configuration.""" # Model paths wakeword_model_path: str = "models/wakeword/default.pth" voice_recognition_model_path: str = "models/voice_recognition/default.pth" # Model settings wakeword_threshold: float = 0.8 voice_recognition_threshold: float = 0.7 model_cache_size: int = 3 # Processing settings enable_gpu: bool = True gpu_device_id: int = 0 batch_size: int = 1 max_inference_time: float = 5.0 # seconds # Training settings training_data_path: str = "trainer/data" model_output_path: str = "models" training_batch_size: int = 32 training_epochs: int = 100 learning_rate: float = 0.001 def validate(self) -> List[str]: """Validate ML configuration and return any errors.""" errors = [] # Threshold validation if not (0.0 <= self.wakeword_threshold <= 1.0): errors.append("Wakeword threshold must be between 0.0 and 1.0") if not (0.0 <= self.voice_recognition_threshold <= 1.0): errors.append("Voice recognition threshold must be between 0.0 and 1.0") # Hardware validation if self.gpu_device_id < 0: errors.append("GPU device ID must be non-negative") if self.batch_size <= 0: errors.append("Batch size must be positive") if self.max_inference_time <= 0: errors.append("Max inference time must be positive") # Training validation if self.training_batch_size <= 0: errors.append("Training batch size must be positive") if self.training_epochs <= 0: errors.append("Training epochs must be positive") if self.learning_rate <= 0: errors.append("Learning rate must be positive") # Path validation (basic existence check will be done by validator) if not self.wakeword_model_path: errors.append("Wakeword model path cannot be empty") # Voice recognition model path is optional - if not provided, voice recognition will be disabled return errors @dataclass class PluginConfig: """Plugin system configuration.""" # Plugin paths and loading plugin_directories: List[str] = field(default_factory=lambda: ["plugins"]) auto_load_plugins: bool = True plugin_timeout: int = 30 # seconds max_plugin_memory: int = 512 # MB # Plugin security enable_plugin_sandboxing: bool = True allowed_plugin_apis: Set[str] = field(default_factory=lambda: { "event_handler", "satellite_manager", "asset_manager", "config_manager", "scheduler" }) # Plugin management enable_hot_reload: bool = True reload_check_interval: int = 5 # seconds max_reload_attempts: int = 3 def validate(self) -> List[str]: """Validate plugin configuration and return any errors.""" errors = [] if not self.plugin_directories: errors.append("At least one plugin directory must be specified") if self.plugin_timeout <= 0: errors.append("Plugin timeout must be positive") if self.max_plugin_memory <= 0: errors.append("Max plugin memory must be positive") if self.reload_check_interval <= 0: errors.append("Reload check interval must be positive") if self.max_reload_attempts <= 0: errors.append("Max reload attempts must be positive") return errors @dataclass class SatelliteConfig: """Satellite/client specific configuration.""" # Connection settings server_host: str = "localhost" server_port: int = 2101 auto_reconnect: bool = True reconnect_interval: int = 5 # seconds # Satellite identification satellite_id: Optional[str] = None room_name: str = "default" alias_name: str = "trixy-satellite" mac_address: Optional[str] = None # Local processing local_wakeword_processing: bool = True local_voice_recognition: bool = False buffer_audio_locally: bool = True # Audio streaming stream_quality: str = "high" # low, medium, high compression_enabled: bool = True def validate(self) -> List[str]: """Validate satellite configuration and return any errors.""" errors = [] if not self.server_host: errors.append("Server host cannot be empty") if not (1024 <= self.server_port <= 65535): errors.append("Server port must be between 1024 and 65535") if self.reconnect_interval <= 0: errors.append("Reconnect interval must be positive") if not self.room_name: errors.append("Room name cannot be empty") if not self.alias_name: errors.append("Alias name cannot be empty") if self.stream_quality not in ["low", "medium", "high"]: errors.append("Stream quality must be 'low', 'medium', or 'high'") return errors @dataclass class SecurityConfig: """Security and authentication configuration.""" # Encryption settings enable_encryption: bool = False encryption_algorithm: str = "AES-256-GCM" key_rotation_interval: int = 86400 # 24 hours in seconds # Authentication require_authentication: bool = False auth_token_expiry: int = 3600 # 1 hour in seconds max_auth_attempts: int = 3 auth_lockout_duration: int = 300 # 5 minutes in seconds # Access control allowed_clients: List[str] = field(default_factory=list) blocked_clients: List[str] = field(default_factory=list) max_connections_per_client: int = 5 # Security features enable_audit_logging: bool = True audit_log_file: str = "logs/security_audit.log" enable_intrusion_detection: bool = False def validate(self) -> List[str]: """Validate security configuration and return any errors.""" errors = [] if self.key_rotation_interval <= 0: errors.append("Key rotation interval must be positive") if self.auth_token_expiry <= 0: errors.append("Auth token expiry must be positive") if self.max_auth_attempts <= 0: errors.append("Max auth attempts must be positive") if self.auth_lockout_duration <= 0: errors.append("Auth lockout duration must be positive") if self.max_connections_per_client <= 0: errors.append("Max connections per client must be positive") # Check for conflicts between allowed and blocked clients allowed_set = set(self.allowed_clients) blocked_set = set(self.blocked_clients) conflicts = allowed_set & blocked_set if conflicts: errors.append(f"Clients cannot be both allowed and blocked: {conflicts}") return errors @dataclass class LoggingConfig: """Logging configuration.""" # Basic logging settings log_level: LogLevel = LogLevel.INFO log_format: str = "%(asctime)s - %(name)s - %(levelname)s - %(message)s" date_format: str = "%Y-%m-%d %H:%M:%S" # File logging enable_file_logging: bool = True log_file_path: str = "logs/trixy.log" max_log_file_size: int = 10485760 # 10MB log_file_backup_count: int = 5 # Console logging enable_console_logging: bool = True console_log_level: LogLevel = LogLevel.INFO # Specialized logging enable_event_logging: bool = True event_log_file: str = "logs/events.log" enable_performance_logging: bool = False performance_log_file: str = "logs/performance.log" # Debug mode settings debug_mode: bool = False verbose_errors: bool = False log_stack_traces: bool = False def validate(self) -> List[str]: """Validate logging configuration and return any errors.""" errors = [] if self.max_log_file_size <= 0: errors.append("Max log file size must be positive") if self.log_file_backup_count < 0: errors.append("Log file backup count must be non-negative") # Validate log file paths log_files = [ self.log_file_path, self.event_log_file, self.performance_log_file ] for log_file in log_files: if not log_file: errors.append(f"Log file path cannot be empty: {log_file}") return errors @dataclass class ServerConfig: """Complete configuration for server mode.""" # Basic settings mode: DeploymentMode = DeploymentMode.SERVER debug_mode: bool = False data_directory: str = "data" config_file: str = "config/server_config.json" # Component configurations network: NetworkConfig = field(default_factory=NetworkConfig) audio: AudioConfig = field(default_factory=AudioConfig) ml: MLConfig = field(default_factory=MLConfig) plugins: PluginConfig = field(default_factory=PluginConfig) security: SecurityConfig = field(default_factory=SecurityConfig) logging: LoggingConfig = field(default_factory=LoggingConfig) # Server-specific settings registration_mode_timeout: int = 60 # seconds max_satellites: int = 50 satellite_heartbeat_timeout: int = 60 # seconds enable_web_interface: bool = False web_interface_port: int = 8080 # Asset management asset_profile: str = "default" asset_cache_size: int = 100 # Conversation management conversation_timeout: int = 300 # 5 minutes max_concurrent_conversations: int = 10 def validate(self) -> List[str]: """Validate complete server configuration.""" errors = [] # Validate component configurations errors.extend(self.network.validate()) errors.extend(self.audio.validate()) errors.extend(self.ml.validate()) errors.extend(self.plugins.validate()) errors.extend(self.security.validate()) errors.extend(self.logging.validate()) # Validate server-specific settings if self.registration_mode_timeout <= 0: errors.append("Registration mode timeout must be positive") if self.max_satellites <= 0: errors.append("Max satellites must be positive") if self.satellite_heartbeat_timeout <= 0: errors.append("Satellite heartbeat timeout must be positive") if self.web_interface_port and not (1024 <= self.web_interface_port <= 65535): errors.append("Web interface port must be between 1024 and 65535") if self.conversation_timeout <= 0: errors.append("Conversation timeout must be positive") if self.max_concurrent_conversations <= 0: errors.append("Max concurrent conversations must be positive") if self.asset_cache_size <= 0: errors.append("Asset cache size must be positive") # Check for port conflicts with web interface if (self.enable_web_interface and self.web_interface_port in [ self.network.command_port, self.network.raw_audio_input_port, self.network.raw_audio_output_port, self.network.raw_music_output_port ]): errors.append("Web interface port conflicts with network ports") return errors def to_dict(self) -> Dict[str, Any]: """Convert configuration to dictionary for serialization.""" return asdict(self) @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'ServerConfig': """Create configuration from dictionary.""" # Handle nested configurations if 'network' in data and isinstance(data['network'], dict): data['network'] = NetworkConfig(**data['network']) if 'audio' in data and isinstance(data['audio'], dict): data['audio'] = AudioConfig(**data['audio']) if 'ml' in data and isinstance(data['ml'], dict): data['ml'] = MLConfig(**data['ml']) if 'plugins' in data and isinstance(data['plugins'], dict): data['plugins'] = PluginConfig(**data['plugins']) if 'security' in data and isinstance(data['security'], dict): data['security'] = SecurityConfig(**data['security']) if 'logging' in data and isinstance(data['logging'], dict): data['logging'] = LoggingConfig(**data['logging']) # Handle enums if 'mode' in data and isinstance(data['mode'], str): data['mode'] = DeploymentMode(data['mode']) return cls(**data) @dataclass class ClientConfig: """Complete configuration for client/satellite mode.""" # Basic settings mode: DeploymentMode = DeploymentMode.CLIENT debug_mode: bool = False data_directory: str = "data" config_file: str = "config/client_config.json" # Component configurations satellite: SatelliteConfig = field(default_factory=SatelliteConfig) audio: AudioConfig = field(default_factory=AudioConfig) ml: MLConfig = field(default_factory=MLConfig) logging: LoggingConfig = field(default_factory=LoggingConfig) # Client-specific settings enable_local_processing: bool = True local_model_cache_size: int = 50 audio_buffer_size: int = 8192 # Status reporting status_report_interval: int = 30 # seconds health_check_interval: int = 10 # seconds def validate(self) -> List[str]: """Validate complete client configuration.""" errors = [] # Validate component configurations errors.extend(self.satellite.validate()) errors.extend(self.audio.validate()) errors.extend(self.ml.validate()) errors.extend(self.logging.validate()) # Validate client-specific settings if self.local_model_cache_size <= 0: errors.append("Local model cache size must be positive") if self.audio_buffer_size <= 0: errors.append("Audio buffer size must be positive") if self.status_report_interval <= 0: errors.append("Status report interval must be positive") if self.health_check_interval <= 0: errors.append("Health check interval must be positive") return errors def to_dict(self) -> Dict[str, Any]: """Convert configuration to dictionary for serialization.""" return asdict(self) @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'ClientConfig': """Create configuration from dictionary.""" # Handle nested configurations if 'satellite' in data and isinstance(data['satellite'], dict): data['satellite'] = SatelliteConfig(**data['satellite']) if 'audio' in data and isinstance(data['audio'], dict): data['audio'] = AudioConfig(**data['audio']) if 'ml' in data and isinstance(data['ml'], dict): data['ml'] = MLConfig(**data['ml']) if 'logging' in data and isinstance(data['logging'], dict): data['logging'] = LoggingConfig(**data['logging']) # Handle enums if 'mode' in data and isinstance(data['mode'], str): data['mode'] = DeploymentMode(data['mode']) return cls(**data) @dataclass class StandaloneConfig: """Complete configuration for standalone mode.""" # Basic settings mode: DeploymentMode = DeploymentMode.STANDALONE debug_mode: bool = False data_directory: str = "data" config_file: str = "config/standalone_config.json" # Component configurations (full server capabilities) network: NetworkConfig = field(default_factory=NetworkConfig) audio: AudioConfig = field(default_factory=AudioConfig) ml: MLConfig = field(default_factory=MLConfig) plugins: PluginConfig = field(default_factory=PluginConfig) security: SecurityConfig = field(default_factory=SecurityConfig) logging: LoggingConfig = field(default_factory=LoggingConfig) # Standalone-specific settings enable_server_sync: bool = False sync_server_host: Optional[str] = None sync_server_port: Optional[int] = None sync_interval: int = 300 # 5 minutes # Local capabilities (same as server but with defaults for standalone) registration_mode_timeout: int = 60 max_satellites: int = 10 # Lower default for standalone satellite_heartbeat_timeout: int = 60 enable_web_interface: bool = True # More likely to be used in standalone web_interface_port: int = 8080 # Asset management asset_profile: str = "default" asset_cache_size: int = 50 # Lower default for standalone # Conversation management conversation_timeout: int = 300 max_concurrent_conversations: int = 5 # Lower default for standalone def validate(self) -> List[str]: """Validate complete standalone configuration.""" errors = [] # Validate component configurations errors.extend(self.network.validate()) errors.extend(self.audio.validate()) errors.extend(self.ml.validate()) errors.extend(self.plugins.validate()) errors.extend(self.security.validate()) errors.extend(self.logging.validate()) # Validate standalone-specific settings if self.enable_server_sync: if not self.sync_server_host: errors.append("Sync server host must be specified when server sync is enabled") if not self.sync_server_port or not (1024 <= self.sync_server_port <= 65535): errors.append("Sync server port must be between 1024 and 65535 when server sync is enabled") if self.sync_interval <= 0: errors.append("Sync interval must be positive") if self.registration_mode_timeout <= 0: errors.append("Registration mode timeout must be positive") if self.max_satellites <= 0: errors.append("Max satellites must be positive") if self.satellite_heartbeat_timeout <= 0: errors.append("Satellite heartbeat timeout must be positive") if self.web_interface_port and not (1024 <= self.web_interface_port <= 65535): errors.append("Web interface port must be between 1024 and 65535") if self.conversation_timeout <= 0: errors.append("Conversation timeout must be positive") if self.max_concurrent_conversations <= 0: errors.append("Max concurrent conversations must be positive") if self.asset_cache_size <= 0: errors.append("Asset cache size must be positive") # Check for port conflicts network_ports = [ self.network.command_port, self.network.raw_audio_input_port, self.network.raw_audio_output_port, self.network.raw_music_output_port ] if self.enable_web_interface and self.web_interface_port in network_ports: errors.append("Web interface port conflicts with network ports") if self.enable_server_sync and self.sync_server_port in network_ports: errors.append("Sync server port conflicts with network ports") return errors def to_dict(self) -> Dict[str, Any]: """Convert configuration to dictionary for serialization.""" return asdict(self) @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'StandaloneConfig': """Create configuration from dictionary.""" # Handle nested configurations if 'network' in data and isinstance(data['network'], dict): data['network'] = NetworkConfig(**data['network']) if 'audio' in data and isinstance(data['audio'], dict): data['audio'] = AudioConfig(**data['audio']) if 'ml' in data and isinstance(data['ml'], dict): data['ml'] = MLConfig(**data['ml']) if 'plugins' in data and isinstance(data['plugins'], dict): data['plugins'] = PluginConfig(**data['plugins']) if 'security' in data and isinstance(data['security'], dict): data['security'] = SecurityConfig(**data['security']) if 'logging' in data and isinstance(data['logging'], dict): data['logging'] = LoggingConfig(**data['logging']) # Handle enums if 'mode' in data and isinstance(data['mode'], str): data['mode'] = DeploymentMode(data['mode']) return cls(**data) # Type aliases for convenience ConfigType = Union[ServerConfig, ClientConfig, StandaloneConfig] ConfigDict = Dict[str, Any] # Configuration class mapping CONFIG_CLASSES = { DeploymentMode.SERVER: ServerConfig, DeploymentMode.CLIENT: ClientConfig, DeploymentMode.STANDALONE: StandaloneConfig, } def get_config_class(mode: Union[str, DeploymentMode]) -> type: """Get the appropriate configuration class for a deployment mode.""" if isinstance(mode, str): mode = DeploymentMode(mode.lower()) return CONFIG_CLASSES[mode] def create_default_config(mode: Union[str, DeploymentMode]) -> ConfigType: """Create a default configuration for the specified mode.""" config_class = get_config_class(mode) return config_class() def merge_configs(base_config: ConfigType, override_config: Dict[str, Any]) -> ConfigType: """Merge override configuration into base configuration.""" base_dict = base_config.to_dict() # Deep merge the dictionaries def deep_merge(base: Dict, override: Dict) -> Dict: result = base.copy() for key, value in override.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 merged_dict = deep_merge(base_dict, override_config) return base_config.__class__.from_dict(merged_dict)