""" Error Handling and Fallback Mechanisms for Trixy ML Integration. This module provides comprehensive error handling, recovery mechanisms, and fallback strategies for the ML components in the Trixy voice assistant. It ensures robust operation and graceful degradation under various failure scenarios. Key Features: - Comprehensive error classification and handling - Automatic recovery mechanisms - Fallback strategies for component failures - Circuit breaker patterns for fault isolation - Health monitoring and diagnostics - Error reporting and logging - Performance degradation handling - Resource exhaustion management Usage: from trixy_core.ml import MLErrorHandler, CircuitBreaker, FallbackManager # Create error handler error_handler = MLErrorHandler() # Register with ML components ml_manager = MLManager(config, error_handler=error_handler) # Configure fallback strategies fallback_manager = FallbackManager(error_handler) fallback_manager.configure_fallbacks() """ import asyncio import logging import time import threading import traceback from typing import Dict, Any, Optional, List, Callable, Union, Type from dataclasses import dataclass, field from enum import Enum import psutil import gc from collections import defaultdict, deque from datetime import datetime, timedelta # ML component imports from .ml_manager import MLManager, MLManagerError from .audio_pipeline import AudioPipeline, AudioPipelineError from .wakeword import WakewordDetectorError from .voice_recognition import VoiceRecognitionDetectorError class ErrorSeverity(Enum): """Error severity levels.""" LOW = 1 MEDIUM = 2 HIGH = 3 CRITICAL = 4 class ErrorCategory(Enum): """Error categories for classification.""" SYSTEM = "system" NETWORK = "network" MODEL = "model" AUDIO = "audio" MEMORY = "memory" PERFORMANCE = "performance" CONFIGURATION = "configuration" USER_INPUT = "user_input" class RecoveryStrategy(Enum): """Recovery strategies.""" RETRY = "retry" FALLBACK = "fallback" RESTART = "restart" DEGRADE = "degrade" IGNORE = "ignore" ESCALATE = "escalate" class CircuitState(Enum): """Circuit breaker states.""" CLOSED = "closed" OPEN = "open" HALF_OPEN = "half_open" @dataclass class ErrorInfo: """Error information data structure.""" error_id: str timestamp: float severity: ErrorSeverity category: ErrorCategory component: str error_type: str message: str traceback: Optional[str] = None context: Dict[str, Any] = field(default_factory=dict) recovery_strategy: Optional[RecoveryStrategy] = None retry_count: int = 0 resolved: bool = False resolution_time: Optional[float] = None def duration(self) -> Optional[float]: """Get error duration if resolved.""" if self.resolved and self.resolution_time: return self.resolution_time - self.timestamp return None @dataclass class HealthStatus: """Component health status.""" component: str is_healthy: bool last_check: float error_count: int = 0 performance_score: float = 1.0 memory_usage_mb: float = 0.0 cpu_usage_percent: float = 0.0 uptime_seconds: float = 0.0 issues: List[str] = field(default_factory=list) class CircuitBreaker: """ Circuit breaker for fault isolation and automatic recovery. Implements the circuit breaker pattern to prevent cascading failures and provide automatic recovery mechanisms. """ def __init__( self, name: str, failure_threshold: int = 5, recovery_timeout: float = 60.0, success_threshold: int = 3 ): """ Initialize circuit breaker. Args: name: Circuit breaker name failure_threshold: Number of failures to trigger open state recovery_timeout: Time to wait before trying half-open state success_threshold: Number of successes to close circuit """ self.name = name self.failure_threshold = failure_threshold self.recovery_timeout = recovery_timeout self.success_threshold = success_threshold self._state = CircuitState.CLOSED self._failure_count = 0 self._success_count = 0 self._last_failure_time = 0.0 self._lock = threading.RLock() def call(self, func: Callable, *args, **kwargs) -> Any: """ Execute function through circuit breaker. Args: func: Function to execute *args: Function arguments **kwargs: Function keyword arguments Returns: Function result Raises: CircuitBreakerError: If circuit is open """ with self._lock: if self._state == CircuitState.OPEN: if time.time() - self._last_failure_time >= self.recovery_timeout: self._state = CircuitState.HALF_OPEN self._success_count = 0 else: raise CircuitBreakerError(f"Circuit breaker {self.name} is open") try: result = func(*args, **kwargs) self._on_success() return result except Exception as e: self._on_failure() raise async def async_call(self, func: Callable, *args, **kwargs) -> Any: """ Execute async function through circuit breaker. Args: func: Async function to execute *args: Function arguments **kwargs: Function keyword arguments Returns: Function result """ with self._lock: if self._state == CircuitState.OPEN: if time.time() - self._last_failure_time >= self.recovery_timeout: self._state = CircuitState.HALF_OPEN self._success_count = 0 else: raise CircuitBreakerError(f"Circuit breaker {self.name} is open") try: result = await func(*args, **kwargs) self._on_success() return result except Exception as e: self._on_failure() raise def _on_success(self) -> None: """Handle successful operation.""" with self._lock: self._failure_count = 0 if self._state == CircuitState.HALF_OPEN: self._success_count += 1 if self._success_count >= self.success_threshold: self._state = CircuitState.CLOSED def _on_failure(self) -> None: """Handle failed operation.""" with self._lock: self._failure_count += 1 self._last_failure_time = time.time() if self._failure_count >= self.failure_threshold: self._state = CircuitState.OPEN def get_state(self) -> CircuitState: """Get current circuit state.""" return self._state def reset(self) -> None: """Reset circuit breaker to closed state.""" with self._lock: self._state = CircuitState.CLOSED self._failure_count = 0 self._success_count = 0 class FallbackStrategy: """Base fallback strategy.""" def __init__(self, name: str): self.name = name self.enabled = True async def execute(self, error_info: ErrorInfo, context: Dict[str, Any]) -> bool: """ Execute fallback strategy. Args: error_info: Error information context: Execution context Returns: True if fallback was successful """ raise NotImplementedError class RetryFallback(FallbackStrategy): """Retry fallback strategy with exponential backoff.""" def __init__(self, max_retries: int = 3, base_delay: float = 1.0, max_delay: float = 60.0): super().__init__("retry") self.max_retries = max_retries self.base_delay = base_delay self.max_delay = max_delay async def execute(self, error_info: ErrorInfo, context: Dict[str, Any]) -> bool: """Execute retry fallback.""" if error_info.retry_count >= self.max_retries: return False # Calculate delay with exponential backoff delay = min(self.base_delay * (2 ** error_info.retry_count), self.max_delay) await asyncio.sleep(delay) try: # Retry the original operation original_func = context.get('original_function') if original_func: await original_func() return True except Exception: pass return False class DegradedModeFallback(FallbackStrategy): """Degraded mode fallback strategy.""" def __init__(self): super().__init__("degraded_mode") async def execute(self, error_info: ErrorInfo, context: Dict[str, Any]) -> bool: """Execute degraded mode fallback.""" component = context.get('component') if error_info.component == 'wakeword': # Fallback to energy-based detection return await self._enable_energy_based_detection(context) elif error_info.component == 'voice_recognition': # Fallback to generic speaker identification return await self._enable_generic_speaker_id(context) elif error_info.component == 'audio_pipeline': # Reduce audio quality settings return await self._reduce_audio_quality(context) return False async def _enable_energy_based_detection(self, context: Dict[str, Any]) -> bool: """Enable energy-based wakeword detection.""" try: # This would switch to a simpler energy-based detection logging.info("Switching to energy-based wakeword detection") return True except Exception: return False async def _enable_generic_speaker_id(self, context: Dict[str, Any]) -> bool: """Enable generic speaker identification.""" try: # This would use a simpler speaker identification approach logging.info("Switching to generic speaker identification") return True except Exception: return False async def _reduce_audio_quality(self, context: Dict[str, Any]) -> bool: """Reduce audio quality settings.""" try: # This would reduce audio processing quality to save resources logging.info("Reducing audio quality for degraded mode") return True except Exception: return False class RestartFallback(FallbackStrategy): """Component restart fallback strategy.""" def __init__(self): super().__init__("restart") async def execute(self, error_info: ErrorInfo, context: Dict[str, Any]) -> bool: """Execute restart fallback.""" component = context.get('component') try: if hasattr(component, 'restart'): await component.restart() return True elif hasattr(component, 'stop') and hasattr(component, 'start'): await component.stop() await asyncio.sleep(1.0) # Brief pause await component.start() return True except Exception as e: logging.error(f"Failed to restart component {error_info.component}: {e}") return False class CircuitBreakerError(Exception): """Circuit breaker exception.""" pass class MLErrorHandler: """ Comprehensive error handler for ML components. Provides centralized error handling, classification, recovery, and fallback mechanisms for the ML system. """ def __init__(self, logger: Optional[logging.Logger] = None): """ Initialize ML error handler. Args: logger: Logger instance """ self.logger = logger or logging.getLogger(__name__) # Error tracking self._errors: Dict[str, ErrorInfo] = {} self._error_history = deque(maxlen=1000) self._error_counts = defaultdict(int) # Circuit breakers self._circuit_breakers: Dict[str, CircuitBreaker] = {} # Fallback strategies self._fallback_strategies: Dict[RecoveryStrategy, FallbackStrategy] = { RecoveryStrategy.RETRY: RetryFallback(), RecoveryStrategy.FALLBACK: DegradedModeFallback(), RecoveryStrategy.RESTART: RestartFallback() } # Health monitoring self._health_status: Dict[str, HealthStatus] = {} self._health_check_interval = 30.0 self._health_check_task: Optional[asyncio.Task] = None # Component references self._components: Dict[str, Any] = {} # Configuration self._max_critical_errors = 5 self._error_escalation_threshold = 10 self.logger.info("ML Error Handler initialized") def register_component(self, name: str, component: Any) -> None: """ Register component for error handling. Args: name: Component name component: Component instance """ self._components[name] = component # Create circuit breaker for component self._circuit_breakers[name] = CircuitBreaker(name) # Initialize health status self._health_status[name] = HealthStatus( component=name, is_healthy=True, last_check=time.time() ) self.logger.info(f"Registered component for error handling: {name}") async def handle_error( self, error: Exception, component: str, context: Optional[Dict[str, Any]] = None, severity: Optional[ErrorSeverity] = None ) -> bool: """ Handle an error with appropriate recovery strategy. Args: error: Exception that occurred component: Component that generated the error context: Additional context information severity: Error severity (auto-detected if not provided) Returns: True if error was handled successfully """ try: # Create error info error_info = self._create_error_info(error, component, context, severity) # Store error self._errors[error_info.error_id] = error_info self._error_history.append(error_info) self._error_counts[error_info.category] += 1 # Update component health await self._update_component_health(component, error_info) # Log error self._log_error(error_info) # Determine recovery strategy recovery_strategy = self._determine_recovery_strategy(error_info) error_info.recovery_strategy = recovery_strategy # Execute recovery strategy success = await self._execute_recovery_strategy(error_info, context or {}) if success: error_info.resolved = True error_info.resolution_time = time.time() self.logger.info(f"Error {error_info.error_id} resolved using {recovery_strategy.value}") else: # Escalate if recovery failed await self._escalate_error(error_info) return success except Exception as e: self.logger.error(f"Error in error handler: {e}") return False async def start_health_monitoring(self) -> None: """Start health monitoring task.""" if self._health_check_task: return self._health_check_task = asyncio.create_task(self._health_monitoring_loop()) self.logger.info("Health monitoring started") async def stop_health_monitoring(self) -> None: """Stop health monitoring task.""" if self._health_check_task: self._health_check_task.cancel() try: await self._health_check_task except asyncio.CancelledError: pass self._health_check_task = None self.logger.info("Health monitoring stopped") def get_circuit_breaker(self, component: str) -> Optional[CircuitBreaker]: """Get circuit breaker for component.""" return self._circuit_breakers.get(component) def get_error_statistics(self) -> Dict[str, Any]: """Get error statistics.""" total_errors = len(self._error_history) resolved_errors = sum(1 for e in self._error_history if e.resolved) # Calculate error rates by category category_rates = {} for category, count in self._error_counts.items(): category_rates[category.value] = count / max(total_errors, 1) # Calculate average resolution time resolution_times = [e.duration() for e in self._error_history if e.resolved and e.duration()] avg_resolution_time = sum(resolution_times) / len(resolution_times) if resolution_times else 0 return { "total_errors": total_errors, "resolved_errors": resolved_errors, "resolution_rate": resolved_errors / max(total_errors, 1), "category_rates": category_rates, "average_resolution_time": avg_resolution_time, "active_circuit_breakers": { name: cb.get_state().value for name, cb in self._circuit_breakers.items() } } def get_health_status(self, component: Optional[str] = None) -> Union[HealthStatus, Dict[str, HealthStatus]]: """Get health status for component(s).""" if component: return self._health_status.get(component) return self._health_status.copy() def _create_error_info( self, error: Exception, component: str, context: Optional[Dict[str, Any]], severity: Optional[ErrorSeverity] ) -> ErrorInfo: """Create error information object.""" import uuid # Auto-detect severity if not provided if severity is None: severity = self._classify_error_severity(error) # Classify error category category = self._classify_error_category(error, component) return ErrorInfo( error_id=str(uuid.uuid4()), timestamp=time.time(), severity=severity, category=category, component=component, error_type=type(error).__name__, message=str(error), traceback=traceback.format_exc(), context=context or {} ) def _classify_error_severity(self, error: Exception) -> ErrorSeverity: """Classify error severity.""" if isinstance(error, (MemoryError, SystemError)): return ErrorSeverity.CRITICAL elif isinstance(error, (MLManagerError, AudioPipelineError)): return ErrorSeverity.HIGH elif isinstance(error, (WakewordDetectorError, VoiceRecognitionDetectorError)): return ErrorSeverity.MEDIUM else: return ErrorSeverity.LOW def _classify_error_category(self, error: Exception, component: str) -> ErrorCategory: """Classify error category.""" if isinstance(error, (MemoryError, OSError)): return ErrorCategory.SYSTEM elif isinstance(error, (ConnectionError, TimeoutError)): return ErrorCategory.NETWORK elif 'model' in str(error).lower() or 'tensor' in str(error).lower(): return ErrorCategory.MODEL elif 'audio' in component.lower(): return ErrorCategory.AUDIO elif isinstance(error, MemoryError): return ErrorCategory.MEMORY else: return ErrorCategory.SYSTEM def _determine_recovery_strategy(self, error_info: ErrorInfo) -> RecoveryStrategy: """Determine appropriate recovery strategy.""" if error_info.severity == ErrorSeverity.CRITICAL: return RecoveryStrategy.RESTART elif error_info.category == ErrorCategory.NETWORK: return RecoveryStrategy.RETRY elif error_info.category in [ErrorCategory.MODEL, ErrorCategory.MEMORY]: return RecoveryStrategy.FALLBACK elif error_info.retry_count < 2: return RecoveryStrategy.RETRY else: return RecoveryStrategy.FALLBACK async def _execute_recovery_strategy( self, error_info: ErrorInfo, context: Dict[str, Any] ) -> bool: """Execute recovery strategy.""" strategy = error_info.recovery_strategy if not strategy or strategy not in self._fallback_strategies: return False fallback = self._fallback_strategies[strategy] if not fallback.enabled: return False try: # Add component reference to context context['component'] = self._components.get(error_info.component) # Execute fallback strategy success = await fallback.execute(error_info, context) if success: error_info.retry_count += 1 return success except Exception as e: self.logger.error(f"Error executing recovery strategy {strategy.value}: {e}") return False async def _escalate_error(self, error_info: ErrorInfo) -> None: """Escalate unresolved error.""" self.logger.critical(f"Escalating unresolved error: {error_info.error_id}") # Check if we need to shut down the system critical_errors = sum( 1 for e in self._error_history if e.severity == ErrorSeverity.CRITICAL and not e.resolved ) if critical_errors >= self._max_critical_errors: self.logger.critical("Maximum critical errors reached - system shutdown may be required") async def _update_component_health(self, component: str, error_info: ErrorInfo) -> None: """Update component health status.""" if component in self._health_status: health = self._health_status[component] health.error_count += 1 health.last_check = time.time() # Determine if component is healthy if error_info.severity in [ErrorSeverity.HIGH, ErrorSeverity.CRITICAL]: health.is_healthy = False health.issues.append(f"{error_info.error_type}: {error_info.message}") # Keep only recent issues if len(health.issues) > 10: health.issues = health.issues[-10:] def _log_error(self, error_info: ErrorInfo) -> None: """Log error with appropriate level.""" log_msg = f"Error {error_info.error_id} in {error_info.component}: {error_info.message}" if error_info.severity == ErrorSeverity.CRITICAL: self.logger.critical(log_msg) elif error_info.severity == ErrorSeverity.HIGH: self.logger.error(log_msg) elif error_info.severity == ErrorSeverity.MEDIUM: self.logger.warning(log_msg) else: self.logger.info(log_msg) # Log traceback for high/critical errors if error_info.severity in [ErrorSeverity.HIGH, ErrorSeverity.CRITICAL] and error_info.traceback: self.logger.debug(f"Traceback for {error_info.error_id}:\n{error_info.traceback}") async def _health_monitoring_loop(self) -> None: """Health monitoring loop.""" while True: try: await self._perform_health_checks() await asyncio.sleep(self._health_check_interval) except asyncio.CancelledError: break except Exception as e: self.logger.error(f"Error in health monitoring: {e}") await asyncio.sleep(5.0) async def _perform_health_checks(self) -> None: """Perform health checks on all components.""" for component_name, component in self._components.items(): try: health = self._health_status[component_name] # Update system metrics process = psutil.Process() health.memory_usage_mb = process.memory_info().rss / 1024 / 1024 health.cpu_usage_percent = process.cpu_percent() # Check component-specific health if hasattr(component, 'is_running') and hasattr(component, 'get_statistics'): is_running = component.is_running() if is_running: stats = component.get_statistics() health.performance_score = self._calculate_performance_score(stats) else: health.is_healthy = False health.issues.append("Component not running") health.last_check = time.time() except Exception as e: self.logger.warning(f"Health check failed for {component_name}: {e}") def _calculate_performance_score(self, stats: Dict[str, Any]) -> float: """Calculate performance score from component statistics.""" # This is a simplified performance scoring algorithm score = 1.0 # Factor in error rates if 'processing_errors' in stats and 'chunks_processed' in stats: processed = stats['chunks_processed'] errors = stats['processing_errors'] if processed > 0: error_rate = errors / processed score *= (1.0 - min(error_rate, 0.5)) # Cap error impact at 50% # Factor in processing time if 'average_processing_time' in stats: avg_time = stats['average_processing_time'] if avg_time > 0.1: # Consider 100ms as baseline score *= max(0.1, 1.0 - (avg_time - 0.1)) return max(0.0, min(1.0, score)) class FallbackManager: """ Manager for fallback strategies and degraded operation modes. Coordinates fallback mechanisms across ML components to maintain system functionality under various failure conditions. """ def __init__(self, error_handler: MLErrorHandler): """ Initialize fallback manager. Args: error_handler: ML error handler instance """ self.error_handler = error_handler self.logger = logging.getLogger(__name__) # Fallback configurations self._fallback_configs: Dict[str, Dict[str, Any]] = {} self._active_fallbacks: Dict[str, str] = {} self.logger.info("Fallback Manager initialized") def configure_fallbacks(self) -> None: """Configure fallback strategies for all components.""" # Configure wakeword fallbacks self._fallback_configs['wakeword'] = { 'energy_detection': { 'enabled': True, 'threshold': 0.05, 'confidence_penalty': 0.3 }, 'silence_detection': { 'enabled': True, 'timeout': 5.0 } } # Configure voice recognition fallbacks self._fallback_configs['voice_recognition'] = { 'generic_speaker': { 'enabled': True, 'default_confidence': 0.5 }, 'skip_identification': { 'enabled': True, 'anonymous_mode': True } } # Configure audio pipeline fallbacks self._fallback_configs['audio_pipeline'] = { 'reduced_quality': { 'enabled': True, 'sample_rate': 8000, 'channels': 1 }, 'basic_processing': { 'enabled': True, 'disable_enhancement': True } } self.logger.info("Fallback configurations set up") async def activate_fallback(self, component: str, fallback_type: str) -> bool: """ Activate fallback for component. Args: component: Component name fallback_type: Type of fallback to activate Returns: True if fallback was activated successfully """ try: config = self._fallback_configs.get(component, {}).get(fallback_type) if not config or not config.get('enabled'): return False # Activate fallback self._active_fallbacks[component] = fallback_type self.logger.info(f"Activated fallback {fallback_type} for {component}") return True except Exception as e: self.logger.error(f"Failed to activate fallback {fallback_type} for {component}: {e}") return False async def deactivate_fallback(self, component: str) -> bool: """ Deactivate fallback for component. Args: component: Component name Returns: True if fallback was deactivated successfully """ if component in self._active_fallbacks: fallback_type = self._active_fallbacks.pop(component) self.logger.info(f"Deactivated fallback {fallback_type} for {component}") return True return False def get_active_fallbacks(self) -> Dict[str, str]: """Get currently active fallbacks.""" return self._active_fallbacks.copy() def create_ml_error_handler(logger: Optional[logging.Logger] = None) -> MLErrorHandler: """ Create ML error handler. Args: logger: Logger instance Returns: ML error handler instance """ return MLErrorHandler(logger) def create_fallback_manager(error_handler: MLErrorHandler) -> FallbackManager: """ Create fallback manager. Args: error_handler: ML error handler Returns: Fallback manager instance """ return FallbackManager(error_handler)