""" Configuration Validation Module for Trixy Application This module provides comprehensive validation for configuration data across all deployment modes. It includes: - Schema validation - Cross-field validation - System resource validation - Path and file validation - Network port validation - Security validation Validation results include detailed error messages and warnings to help administrators correct configuration issues. """ import os import socket import subprocess import platform from pathlib import Path from typing import Dict, List, Optional, Union, Any, Tuple import re import ipaddress from dataclasses import dataclass from .config_models import ( ServerConfig, ClientConfig, StandaloneConfig, ConfigType, NetworkConfig, AudioConfig, MLConfig, PluginConfig, SatelliteConfig, SecurityConfig, LoggingConfig, DeploymentMode, LogLevel ) @dataclass class ValidationResult: """Result of configuration validation.""" is_valid: bool errors: List[str] warnings: List[str] info: List[str] def add_error(self, message: str) -> None: """Add an error message.""" self.errors.append(message) self.is_valid = False def add_warning(self, message: str) -> None: """Add a warning message.""" self.warnings.append(message) def add_info(self, message: str) -> None: """Add an info message.""" self.info.append(message) def merge(self, other: 'ValidationResult') -> None: """Merge another validation result.""" self.errors.extend(other.errors) self.warnings.extend(other.warnings) self.info.extend(other.info) if not other.is_valid: self.is_valid = False def has_issues(self) -> bool: """Check if there are any errors or warnings.""" return bool(self.errors or self.warnings) def get_summary(self) -> str: """Get a summary of validation results.""" parts = [] if self.errors: parts.append(f"{len(self.errors)} errors") if self.warnings: parts.append(f"{len(self.warnings)} warnings") if self.info: parts.append(f"{len(self.info)} info messages") if not parts: return "Configuration valid" return f"Configuration validation: {', '.join(parts)}" class ConfigValidator: """Comprehensive configuration validator.""" def __init__(self, strict_mode: bool = False): """ Initialize the validator. Args: strict_mode: If True, warnings are treated as errors """ self.strict_mode = strict_mode self._system_info = self._get_system_info() def _get_system_info(self) -> Dict[str, Any]: """Get system information for validation.""" info = { "platform": platform.system().lower(), "available_ports": self._get_available_ports(), "has_gpu": self._check_gpu_availability(), "audio_devices": self._get_audio_devices(), } return info def _get_available_ports(self) -> List[int]: """Get list of available ports.""" available = [] for port in range(1024, 65536): sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) try: sock.bind(('localhost', port)) available.append(port) sock.close() except OSError: pass if len(available) >= 100: # Limit check for performance break return available def _check_gpu_availability(self) -> bool: """Check if GPU is available for ML processing.""" try: # Check for NVIDIA GPU result = subprocess.run(['nvidia-smi'], capture_output=True, text=True) if result.returncode == 0: return True except FileNotFoundError: pass try: # Check for AMD GPU (ROCm) result = subprocess.run(['rocm-smi'], capture_output=True, text=True) if result.returncode == 0: return True except FileNotFoundError: pass return False def _get_audio_devices(self) -> Dict[str, List[str]]: """Get available audio devices.""" devices = {"input": [], "output": []} # This is a placeholder - actual implementation would use system-specific # audio enumeration (ALSA, PulseAudio, WASAPI, CoreAudio) try: if self._system_info["platform"] == "linux": # Check ALSA devices devices["input"] = ["default", "hw:0,0"] devices["output"] = ["default", "hw:0,0"] elif self._system_info["platform"] == "windows": # Check WASAPI devices devices["input"] = ["default"] devices["output"] = ["default"] elif self._system_info["platform"] == "darwin": # Check CoreAudio devices devices["input"] = ["default"] devices["output"] = ["default"] except Exception: pass return devices def validate_config(self, config: ConfigType) -> ValidationResult: """Validate a complete configuration.""" result = ValidationResult(True, [], [], []) # Validate based on config type if isinstance(config, ServerConfig): result.merge(self.validate_server_config(config)) elif isinstance(config, ClientConfig): result.merge(self.validate_client_config(config)) elif isinstance(config, StandaloneConfig): result.merge(self.validate_standalone_config(config)) else: result.add_error(f"Unknown configuration type: {type(config)}") return result def validate_server_config(self, config: ServerConfig) -> ValidationResult: """Validate server configuration.""" result = ValidationResult(True, [], [], []) # Validate base configuration result.merge(self._validate_base_config(config)) # Validate component configurations result.merge(self.validate_network_config(config.network)) result.merge(self.validate_audio_config(config.audio)) result.merge(self.validate_ml_config(config.ml)) result.merge(self.validate_plugin_config(config.plugins)) result.merge(self.validate_security_config(config.security)) result.merge(self.validate_logging_config(config.logging)) # Server-specific validations if config.max_satellites <= 0: result.add_error("Max satellites must be positive") elif config.max_satellites > 1000: result.add_warning("Very high max satellites count may impact performance") if config.satellite_heartbeat_timeout <= 0: result.add_error("Satellite heartbeat timeout must be positive") elif config.satellite_heartbeat_timeout < 10: result.add_warning("Very low heartbeat timeout may cause unnecessary disconnections") if config.conversation_timeout <= 0: result.add_error("Conversation timeout must be positive") if config.max_concurrent_conversations <= 0: result.add_error("Max concurrent conversations must be positive") elif config.max_concurrent_conversations > config.max_satellites: result.add_warning("Max concurrent conversations exceeds max satellites") # Web interface validation if config.enable_web_interface: if not (1024 <= config.web_interface_port <= 65535): result.add_error("Web interface port must be between 1024 and 65535") else: # Check for port conflicts network_ports = [ config.network.command_port, config.network.raw_audio_input_port, config.network.raw_audio_output_port, config.network.raw_music_output_port ] if config.web_interface_port in network_ports: result.add_error("Web interface port conflicts with network ports") return result def validate_client_config(self, config: ClientConfig) -> ValidationResult: """Validate client configuration.""" result = ValidationResult(True, [], [], []) # Validate base configuration result.merge(self._validate_base_config(config)) # Validate component configurations result.merge(self.validate_satellite_config(config.satellite)) result.merge(self.validate_audio_config(config.audio)) result.merge(self.validate_ml_config(config.ml)) result.merge(self.validate_logging_config(config.logging)) # Client-specific validations if config.local_model_cache_size <= 0: result.add_error("Local model cache size must be positive") if config.audio_buffer_size <= 0: result.add_error("Audio buffer size must be positive") elif not (config.audio_buffer_size & (config.audio_buffer_size - 1)) == 0: result.add_warning("Audio buffer size should be a power of 2 for optimal performance") if config.status_report_interval <= 0: result.add_error("Status report interval must be positive") elif config.status_report_interval < 5: result.add_warning("Very frequent status reports may impact network performance") if config.health_check_interval <= 0: result.add_error("Health check interval must be positive") elif config.health_check_interval > config.status_report_interval: result.add_warning("Health check interval should be less than or equal to status report interval") return result def validate_standalone_config(self, config: StandaloneConfig) -> ValidationResult: """Validate standalone configuration.""" result = ValidationResult(True, [], [], []) # Validate base configuration result.merge(self._validate_base_config(config)) # Validate component configurations (same as server) result.merge(self.validate_network_config(config.network)) result.merge(self.validate_audio_config(config.audio)) result.merge(self.validate_ml_config(config.ml)) result.merge(self.validate_plugin_config(config.plugins)) result.merge(self.validate_security_config(config.security)) result.merge(self.validate_logging_config(config.logging)) # Standalone-specific validations if config.enable_server_sync: if not config.sync_server_host: result.add_error("Sync server host required when server sync is enabled") else: result.merge(self._validate_host(config.sync_server_host, "sync_server_host")) if not config.sync_server_port: result.add_error("Sync server port required when server sync is enabled") elif not (1024 <= config.sync_server_port <= 65535): result.add_error("Sync server port must be between 1024 and 65535") if config.sync_interval <= 0: result.add_error("Sync interval must be positive") elif config.sync_interval < 60: result.add_warning("Very frequent sync intervals may impact performance") # Same server validations apply result.merge(self.validate_server_config(config)) return result def _validate_base_config(self, config: ConfigType) -> ValidationResult: """Validate base configuration fields.""" result = ValidationResult(True, [], [], []) # Validate data directory data_dir = Path(config.data_directory) if not data_dir.is_absolute(): result.add_info(f"Using relative data directory: {config.data_directory}") if not data_dir.exists(): result.add_warning(f"Data directory does not exist: {config.data_directory}") elif not data_dir.is_dir(): result.add_error(f"Data directory is not a directory: {config.data_directory}") # Validate config file path if config.config_file: config_file = Path(config.config_file) if not config_file.parent.exists(): result.add_warning(f"Config file parent directory does not exist: {config_file.parent}") return result def validate_network_config(self, config: NetworkConfig) -> ValidationResult: """Validate network configuration.""" result = ValidationResult(True, [], [], []) # Port validation ports = [ ("command_port", config.command_port), ("raw_audio_input_port", config.raw_audio_input_port), ("raw_audio_output_port", config.raw_audio_output_port), ("raw_music_output_port", config.raw_music_output_port), ] for port_name, port_value in ports: if not (1024 <= port_value <= 65535): result.add_error(f"{port_name} must be between 1024 and 65535, got {port_value}") # Check for port uniqueness port_values = [p[1] for p in ports] if len(set(port_values)) != len(port_values): result.add_error("All network ports must be unique") # Validate bind addresses for addr_name, addr_value in [ ("command_bind_address", config.command_bind_address), ("audio_bind_address", config.audio_bind_address), ]: result.merge(self._validate_bind_address(addr_value, addr_name)) # Timeout and buffer validations if config.command_timeout <= 0: result.add_error("Command timeout must be positive") elif config.command_timeout > 300: result.add_warning("Very high command timeout may cause poor user experience") if config.audio_timeout <= 0: result.add_error("Audio timeout must be positive") if config.command_max_connections <= 0: result.add_error("Max connections must be positive") elif config.command_max_connections > 1000: result.add_warning("Very high max connections may impact system resources") # Buffer size validation for buffer_name, buffer_size in [ ("command_buffer_size", config.command_buffer_size), ("audio_buffer_size", config.audio_buffer_size), ]: if buffer_size <= 0: result.add_error(f"{buffer_name} must be positive") elif not (buffer_size & (buffer_size - 1)) == 0: result.add_warning(f"{buffer_name} should be a power of 2 for optimal performance") # Connection retry validation if config.max_retry_attempts < 0: result.add_error("Max retry attempts must be non-negative") elif config.max_retry_attempts > 100: result.add_warning("Very high retry attempts may cause long delays") if config.retry_delay <= 0: result.add_error("Retry delay must be positive") if config.connection_timeout <= 0: result.add_error("Connection timeout must be positive") if config.heartbeat_interval <= 0: result.add_error("Heartbeat interval must be positive") elif config.heartbeat_interval > 300: result.add_warning("Very long heartbeat intervals may delay disconnect detection") # Compression validation if config.enable_compression: if not (1 <= config.compression_level <= 9): result.add_error("Compression level must be between 1 and 9") # Encryption validation if config.enable_encryption and config.encryption_key_file: key_file = Path(config.encryption_key_file) if not key_file.exists(): result.add_error(f"Encryption key file not found: {config.encryption_key_file}") elif not key_file.is_file(): result.add_error(f"Encryption key file is not a file: {config.encryption_key_file}") return result def _validate_bind_address(self, address: str, field_name: str) -> ValidationResult: """Validate a bind address.""" result = ValidationResult(True, [], [], []) if not address: result.add_error(f"{field_name} cannot be empty") return result # Special addresses if address in ["0.0.0.0", "::", "localhost", "127.0.0.1", "::1"]: return result # Try to parse as IP address try: ip = ipaddress.ip_address(address) if ip.is_private: result.add_info(f"{field_name} uses private IP address: {address}") elif ip.is_loopback: result.add_info(f"{field_name} uses loopback address: {address}") return result except ValueError: pass # Try as hostname try: socket.gethostbyname(address) result.add_info(f"{field_name} uses hostname: {address}") except socket.gaierror: result.add_warning(f"{field_name} hostname cannot be resolved: {address}") return result def _validate_host(self, host: str, field_name: str) -> ValidationResult: """Validate a host address.""" result = ValidationResult(True, [], [], []) if not host: result.add_error(f"{field_name} cannot be empty") return result # Try to resolve hostname try: socket.gethostbyname(host) except socket.gaierror: result.add_warning(f"{field_name} hostname cannot be resolved: {host}") return result def validate_audio_config(self, config: AudioConfig) -> ValidationResult: """Validate audio configuration.""" result = ValidationResult(True, [], [], []) # Sample rate validation valid_sample_rates = [8000, 16000, 22050, 44100, 48000] if config.sample_rate not in valid_sample_rates: result.add_error(f"Sample rate must be one of {valid_sample_rates}, got {config.sample_rate}") if config.music_sample_rate not in valid_sample_rates: result.add_error(f"Music sample rate must be one of {valid_sample_rates}, got {config.music_sample_rate}") # Channels validation if config.channels not in [1, 2]: result.add_error(f"Channels must be 1 or 2, got {config.channels}") if config.music_channels not in [1, 2]: result.add_error(f"Music channels must be 1 or 2, got {config.music_channels}") # Bit depth validation valid_bit_depths = [16, 24, 32] if config.bit_depth not in valid_bit_depths: result.add_error(f"Bit depth must be one of {valid_bit_depths}, got {config.bit_depth}") if config.music_bit_depth not in valid_bit_depths: result.add_error(f"Music bit depth must be one of {valid_bit_depths}, got {config.music_bit_depth}") # Frame size validation if config.frame_size <= 0: result.add_error("Frame size must be positive") elif not (config.frame_size & (config.frame_size - 1)) == 0: result.add_warning("Frame size should be a power of 2 for optimal performance") # Recording duration validation if config.max_recording_duration <= 0: result.add_error("Max recording duration must be positive") elif config.max_recording_duration > 600: # 10 minutes result.add_warning("Very long max recording duration may use excessive resources") # Threshold validation if not (0.0 <= config.silence_threshold <= 1.0): result.add_error(f"Silence threshold must be between 0.0 and 1.0, got {config.silence_threshold}") if config.silence_duration <= 0: result.add_error("Silence duration must be positive") elif config.silence_duration > 10: result.add_warning("Very long silence duration may cause poor user experience") if config.pre_recording_buffer < 0: result.add_error("Pre-recording buffer must be non-negative") elif config.pre_recording_buffer > 5: result.add_warning("Very long pre-recording buffer may use excessive memory") # Gain validation if config.gain_level < 0: result.add_error("Gain level must be non-negative") elif config.gain_level > 10: result.add_warning("Very high gain level may cause audio distortion") # VAD validation if config.vad_enabled: if not (0 <= config.vad_aggressiveness <= 3): result.add_error(f"VAD aggressiveness must be between 0 and 3, got {config.vad_aggressiveness}") if config.vad_frame_duration not in [10, 20, 30]: result.add_error(f"VAD frame duration must be 10, 20, or 30 ms, got {config.vad_frame_duration}") return result def validate_ml_config(self, config: MLConfig) -> ValidationResult: """Validate ML configuration.""" result = ValidationResult(True, [], [], []) # Model paths validation # Wakeword model is required if not config.wakeword_model_path: result.add_error("wakeword_model_path cannot be empty") else: path = Path(config.wakeword_model_path) if not path.exists(): result.add_warning(f"Model file not found: {config.wakeword_model_path}") elif not path.is_file(): result.add_error(f"Model path is not a file: {config.wakeword_model_path}") elif path.suffix not in ['.pth', '.pt', '.onnx']: result.add_warning(f"Unexpected model file extension: {path.suffix}") # Voice recognition model is optional - only validate if provided if config.voice_recognition_model_path and config.voice_recognition_model_path.strip(): path = Path(config.voice_recognition_model_path) if not path.exists(): result.add_warning(f"Model file not found: {config.voice_recognition_model_path}") elif not path.is_file(): result.add_error(f"Model path is not a file: {config.voice_recognition_model_path}") elif path.suffix not in ['.pth', '.pt', '.onnx']: result.add_warning(f"Unexpected model file extension: {path.suffix}") # Threshold validation if not (0.0 <= config.wakeword_threshold <= 1.0): result.add_error(f"Wakeword threshold must be between 0.0 and 1.0, got {config.wakeword_threshold}") if not (0.0 <= config.voice_recognition_threshold <= 1.0): result.add_error(f"Voice recognition threshold must be between 0.0 and 1.0, got {config.voice_recognition_threshold}") # Cache and performance validation if config.model_cache_size <= 0: result.add_error("Model cache size must be positive") elif config.model_cache_size > 100: result.add_warning("Very large model cache may use excessive memory") if config.batch_size <= 0: result.add_error("Batch size must be positive") elif config.batch_size > 64: result.add_warning("Very large batch size may use excessive memory") if config.max_inference_time <= 0: result.add_error("Max inference time must be positive") elif config.max_inference_time > 30: result.add_warning("Very long max inference time may cause poor user experience") # GPU validation if config.enable_gpu: if not self._system_info.get("has_gpu", False): result.add_warning("GPU acceleration requested but no GPU detected") if config.gpu_device_id < 0: result.add_error("GPU device ID must be non-negative") # Training configuration validation for path_name, path_value in [ ("training_data_path", config.training_data_path), ("model_output_path", config.model_output_path), ]: if not path_value: result.add_error(f"{path_name} cannot be empty") continue path = Path(path_value) if path.exists() and not path.is_dir(): result.add_error(f"{path_name} exists but is not a directory: {path_value}") if config.training_batch_size <= 0: result.add_error("Training batch size must be positive") elif config.training_batch_size > 512: result.add_warning("Very large training batch size may use excessive memory") if config.training_epochs <= 0: result.add_error("Training epochs must be positive") elif config.training_epochs > 10000: result.add_warning("Very high training epochs count may take excessive time") if config.learning_rate <= 0: result.add_error("Learning rate must be positive") elif config.learning_rate > 1.0: result.add_warning("Very high learning rate may prevent model convergence") elif config.learning_rate < 1e-6: result.add_warning("Very low learning rate may slow training significantly") return result def validate_plugin_config(self, config: PluginConfig) -> ValidationResult: """Validate plugin configuration.""" result = ValidationResult(True, [], [], []) # Plugin directories validation if not config.plugin_directories: result.add_error("At least one plugin directory must be specified") else: for plugin_dir in config.plugin_directories: path = Path(plugin_dir) if not path.exists(): result.add_warning(f"Plugin directory does not exist: {plugin_dir}") elif not path.is_dir(): result.add_error(f"Plugin directory is not a directory: {plugin_dir}") # Timeout and resource validation if config.plugin_timeout <= 0: result.add_error("Plugin timeout must be positive") elif config.plugin_timeout > 300: result.add_warning("Very high plugin timeout may cause poor user experience") if config.max_plugin_memory <= 0: result.add_error("Max plugin memory must be positive") elif config.max_plugin_memory > 8192: # 8GB result.add_warning("Very high plugin memory limit may use excessive resources") # Hot reload validation if config.enable_hot_reload: if config.reload_check_interval <= 0: result.add_error("Reload check interval must be positive") elif config.reload_check_interval < 1: result.add_warning("Very frequent reload checks may impact performance") if config.max_reload_attempts <= 0: result.add_error("Max reload attempts must be positive") elif config.max_reload_attempts > 10: result.add_warning("Very high max reload attempts may cause long delays on failures") # API validation if not config.allowed_plugin_apis: result.add_warning("No plugin APIs are allowed - plugins may not function properly") return result def validate_satellite_config(self, config: SatelliteConfig) -> ValidationResult: """Validate satellite configuration.""" result = ValidationResult(True, [], [], []) # Server connection validation if not config.server_host: result.add_error("Server host cannot be empty") else: result.merge(self._validate_host(config.server_host, "server_host")) if not (1024 <= config.server_port <= 65535): result.add_error(f"Server port must be between 1024 and 65535, got {config.server_port}") if config.reconnect_interval <= 0: result.add_error("Reconnect interval must be positive") elif config.reconnect_interval < 1: result.add_warning("Very frequent reconnect attempts may impact server performance") # Identification validation if not config.room_name: result.add_error("Room name cannot be empty") elif not re.match(r'^[a-zA-Z0-9_-]+$', config.room_name): result.add_warning("Room name should contain only alphanumeric characters, underscores, and hyphens") if not config.alias_name: result.add_error("Alias name cannot be empty") elif not re.match(r'^[a-zA-Z0-9_-]+$', config.alias_name): result.add_warning("Alias name should contain only alphanumeric characters, underscores, and hyphens") # MAC address validation (if provided) if config.mac_address: mac_pattern = r'^([0-9A-Fa-f]{2}[:-]){5}([0-9A-Fa-f]{2})$' if not re.match(mac_pattern, config.mac_address): result.add_error(f"Invalid MAC address format: {config.mac_address}") # Stream quality validation if config.stream_quality not in ["low", "medium", "high"]: result.add_error(f"Stream quality must be 'low', 'medium', or 'high', got {config.stream_quality}") return result def validate_security_config(self, config: SecurityConfig) -> ValidationResult: """Validate security configuration.""" result = ValidationResult(True, [], [], []) # Encryption validation if config.enable_encryption: if config.encryption_algorithm not in ["AES-256-GCM", "AES-256-CBC", "ChaCha20-Poly1305"]: result.add_warning(f"Unrecognized encryption algorithm: {config.encryption_algorithm}") if config.key_rotation_interval <= 0: result.add_error("Key rotation interval must be positive") elif config.key_rotation_interval < 3600: # 1 hour result.add_warning("Very frequent key rotation may impact performance") # Authentication validation if config.require_authentication: if config.auth_token_expiry <= 0: result.add_error("Auth token expiry must be positive") elif config.auth_token_expiry < 300: # 5 minutes result.add_warning("Very short auth token expiry may cause frequent re-authentication") if config.max_auth_attempts <= 0: result.add_error("Max auth attempts must be positive") elif config.max_auth_attempts > 10: result.add_warning("Very high max auth attempts may aid brute force attacks") if config.auth_lockout_duration <= 0: result.add_error("Auth lockout duration must be positive") # Access control validation if config.max_connections_per_client <= 0: result.add_error("Max connections per client must be positive") elif config.max_connections_per_client > 100: result.add_warning("Very high max connections per client may enable resource exhaustion") # Client list validation allowed_set = set(config.allowed_clients) blocked_set = set(config.blocked_clients) conflicts = allowed_set & blocked_set if conflicts: result.add_error(f"Clients cannot be both allowed and blocked: {', '.join(conflicts)}") # Audit logging validation if config.enable_audit_logging: if not config.audit_log_file: result.add_error("Audit log file path cannot be empty when audit logging is enabled") else: log_path = Path(config.audit_log_file) if not log_path.parent.exists(): result.add_warning(f"Audit log directory does not exist: {log_path.parent}") return result def validate_logging_config(self, config: LoggingConfig) -> ValidationResult: """Validate logging configuration.""" result = ValidationResult(True, [], [], []) # Log level validation is handled by enum # File logging validation if config.enable_file_logging: if not config.log_file_path: result.add_error("Log file path cannot be empty when file logging is enabled") else: log_path = Path(config.log_file_path) if not log_path.parent.exists(): result.add_warning(f"Log file directory does not exist: {log_path.parent}") if config.max_log_file_size <= 0: result.add_error("Max log file size must be positive") elif config.max_log_file_size < 1024: # 1KB result.add_warning("Very small max log file size may cause excessive rotation") elif config.max_log_file_size > 1073741824: # 1GB result.add_warning("Very large max log file size may use excessive disk space") if config.log_file_backup_count < 0: result.add_error("Log file backup count must be non-negative") elif config.log_file_backup_count > 100: result.add_warning("Very high backup count may use excessive disk space") # Specialized logging validation if config.enable_event_logging: if not config.event_log_file: result.add_error("Event log file path cannot be empty when event logging is enabled") else: event_log_path = Path(config.event_log_file) if not event_log_path.parent.exists(): result.add_warning(f"Event log directory does not exist: {event_log_path.parent}") if config.enable_performance_logging: if not config.performance_log_file: result.add_error("Performance log file path cannot be empty when performance logging is enabled") else: perf_log_path = Path(config.performance_log_file) if not perf_log_path.parent.exists(): result.add_warning(f"Performance log directory does not exist: {perf_log_path.parent}") return result def validate_file_permissions(self, config: ConfigType) -> ValidationResult: """Validate file and directory permissions.""" result = ValidationResult(True, [], [], []) # Check data directory permissions data_path = Path(config.data_directory) if data_path.exists(): if not os.access(data_path, os.R_OK | os.W_OK): result.add_error(f"Insufficient permissions for data directory: {config.data_directory}") # Check log directory permissions (if applicable) if hasattr(config, 'logging'): log_path = Path(config.logging.log_file_path).parent if log_path.exists(): if not os.access(log_path, os.R_OK | os.W_OK): result.add_error(f"Insufficient permissions for log directory: {log_path}") return result def validate_system_resources(self, config: ConfigType) -> ValidationResult: """Validate system resource requirements.""" result = ValidationResult(True, [], [], []) # Check available disk space try: data_path = Path(config.data_directory) if data_path.exists(): stat = os.statvfs(data_path) available_space = stat.f_bavail * stat.f_frsize # Available space in bytes if available_space < 1073741824: # Less than 1GB result.add_warning(f"Low disk space available: {available_space / 1073741824:.1f} GB") elif available_space < 104857600: # Less than 100MB result.add_error(f"Critically low disk space: {available_space / 1048576:.1f} MB") except (OSError, AttributeError): # os.statvfs not available on Windows pass # Check memory usage estimates if isinstance(config, (ServerConfig, StandaloneConfig)): estimated_memory = self._estimate_memory_usage(config) result.add_info(f"Estimated memory usage: {estimated_memory:.1f} MB") if estimated_memory > 8192: # 8GB result.add_warning("Configuration may require significant memory resources") return result def _estimate_memory_usage(self, config: Union[ServerConfig, StandaloneConfig]) -> float: """Estimate memory usage for server/standalone configurations.""" # Base memory usage base_memory = 256 # MB # Satellite connections satellite_memory = config.max_satellites * 5 # ~5MB per satellite # Audio processing audio_memory = config.audio.frame_size * config.audio.channels * 4 * 10 # ~10 frames in memory audio_memory = audio_memory / (1024 * 1024) # Convert to MB # Model caching if hasattr(config, 'ml'): model_memory = config.ml.model_cache_size * 50 # ~50MB per cached model else: model_memory = 0 # Plugin memory if hasattr(config, 'plugins'): plugin_memory = len(config.plugins.plugin_directories) * config.plugins.max_plugin_memory else: plugin_memory = 0 return base_memory + satellite_memory + audio_memory + model_memory + plugin_memory def validate_config_dict(config_dict: Dict[str, Any], mode: DeploymentMode) -> ValidationResult: """Validate a configuration dictionary before creating config object.""" result = ValidationResult(True, [], [], []) # Import here to avoid circular imports from .config_models import CONFIG_CLASSES try: config_class = CONFIG_CLASSES[mode] # Try to create the config object to catch any data validation errors config = config_class.from_dict(config_dict) # Validate the created config validator = ConfigValidator() result.merge(validator.validate_config(config)) except Exception as e: result.add_error(f"Failed to create {mode.value} configuration: {str(e)}") return result def quick_validate(config_file: str) -> ValidationResult: """Quickly validate a configuration file.""" result = ValidationResult(True, [], [], []) config_path = Path(config_file) if not config_path.exists(): result.add_error(f"Configuration file not found: {config_file}") return result if not config_path.is_file(): result.add_error(f"Configuration path is not a file: {config_file}") return result try: import json with open(config_path, 'r') as f: config_data = json.load(f) # Determine mode from filename or config data filename = config_path.name.lower() if 'server' in filename: mode = DeploymentMode.SERVER elif 'client' in filename: mode = DeploymentMode.CLIENT elif 'standalone' in filename: mode = DeploymentMode.STANDALONE elif 'mode' in config_data: mode = DeploymentMode(config_data['mode']) else: result.add_error("Cannot determine deployment mode from filename or config data") return result result.merge(validate_config_dict(config_data, mode)) except json.JSONDecodeError as e: result.add_error(f"Invalid JSON in configuration file: {str(e)}") except Exception as e: result.add_error(f"Error validating configuration file: {str(e)}") return result