""" Application Container Integration for Trixy ML System. This module provides comprehensive integration between the ML components and the Trixy application container system. It handles component lifecycle management, dependency injection, and seamless integration with the event system, configuration management, and other Trixy core components. Key Features: - ML component factory for application container - Lifecycle management (initialization, startup, shutdown) - Configuration integration with config manager - Event system integration - Health monitoring and status reporting - Network manager integration for client mode - Performance monitoring integration - Error handling and recovery Usage: from trixy_core.ml.integration import MLContainerIntegration # Create ML integration for application container ml_integration = MLContainerIntegration(application_container) # Initialize ML components ml_integration.initialize() # Start ML system ml_integration.start() """ import logging import threading import time from typing import Dict, List, Optional, Any, Callable, Union from dataclasses import dataclass, field from datetime import datetime from enum import Enum import traceback # Import ML components from .ml_manager import MLManager, MLConfig, MLSystemMode as MLMode, create_ml_manager, create_default_ml_config from .audio_pipeline import AudioPipeline, AudioPipelineConfig, create_audio_pipeline from .workflow import MLWorkflowManager, WorkflowConfig, create_workflow_manager from .events import MLEventManager, create_ml_event_manager from .monitoring import MLPerformanceMonitor, PerformanceConfig, create_performance_monitor from .config import MLConfigManager, MLSystemConfig, create_ml_config # Import application container components try: from ..application import ApplicationContainer, ComponentStatus, ComponentHealth from ..events import EventHandler from ..config import ConfigManager, DeploymentMode from ..network import NetworkManager APPLICATION_CONTAINER_AVAILABLE = True except ImportError: APPLICATION_CONTAINER_AVAILABLE = False ApplicationContainer = object ComponentStatus = object ComponentHealth = object EventHandler = object ConfigManager = object DeploymentMode = object NetworkManager = object # Configure logging logger = logging.getLogger(__name__) class MLIntegrationState(Enum): """ML integration states.""" NOT_INITIALIZED = "not_initialized" INITIALIZING = "initializing" INITIALIZED = "initialized" STARTING = "starting" RUNNING = "running" STOPPING = "stopping" STOPPED = "stopped" ERROR = "error" @dataclass class MLComponentStatus: """Status of individual ML components.""" name: str status: MLIntegrationState health: ComponentStatus last_check: float errors: List[str] = field(default_factory=list) warnings: List[str] = field(default_factory=list) info: Dict[str, Any] = field(default_factory=dict) def is_healthy(self) -> bool: """Check if component is healthy.""" return ( self.status in [MLIntegrationState.RUNNING, MLIntegrationState.INITIALIZED] and len(self.errors) == 0 ) @dataclass class MLIntegrationConfig: """Configuration for ML integration.""" # Component enablement enable_ml_manager: bool = True enable_audio_pipeline: bool = True enable_workflow_manager: bool = True enable_event_integration: bool = True enable_performance_monitoring: bool = True # Integration settings auto_start_components: bool = True health_check_interval: float = 10.0 component_timeout_seconds: float = 30.0 enable_component_recovery: bool = True max_recovery_attempts: int = 3 # Performance settings startup_delay_seconds: float = 1.0 shutdown_timeout_seconds: float = 10.0 # Debug settings enable_integration_logging: bool = True log_component_status: bool = False class MLContainerIntegrationError(Exception): """ML container integration errors.""" pass class MLContainerIntegration: """ ML system integration with Trixy application container. This class provides seamless integration between the ML components and the Trixy application container, handling lifecycle management, dependency injection, and component coordination. """ def __init__( self, application_container: Any, integration_config: Optional[MLIntegrationConfig] = None ): """ Initialize ML container integration. Args: application_container: Trixy application container integration_config: Integration configuration """ self.application_container = application_container self.integration_config = integration_config or MLIntegrationConfig() # Integration state self._state = MLIntegrationState.NOT_INITIALIZED self._state_lock = threading.RLock() # ML components self._ml_manager: Optional[MLManager] = None self._audio_pipeline: Optional[AudioPipeline] = None self._workflow_manager: Optional[MLWorkflowManager] = None self._event_manager: Optional[MLEventManager] = None self._performance_monitor: Optional[MLPerformanceMonitor] = None self._ml_config_manager: Optional[MLConfigManager] = None # Component status tracking self._component_status: Dict[str, MLComponentStatus] = {} self._health_check_thread: Optional[threading.Thread] = None self._stop_health_checks = threading.Event() # Configuration self._ml_config: Optional[MLSystemConfig] = None # Container dependencies self._event_handler: Optional[EventHandler] = None self._config_manager: Optional[ConfigManager] = None self._network_manager: Optional[NetworkManager] = None # Recovery tracking self._recovery_attempts: Dict[str, int] = {} logger.info("ML Container Integration initialized") def initialize(self) -> None: """Initialize the ML integration system.""" with self._state_lock: if self._state != MLIntegrationState.NOT_INITIALIZED: logger.warning("ML integration already initialized") return logger.info("Initializing ML container integration...") self._state = MLIntegrationState.INITIALIZING try: # Get container dependencies self._get_container_dependencies() # Load ML configuration self._load_ml_configuration() # Initialize ML components self._initialize_ml_components() # Start health monitoring if self.integration_config.health_check_interval > 0: self._start_health_monitoring() self._state = MLIntegrationState.INITIALIZED logger.info("ML container integration initialized successfully") except Exception as e: self._state = MLIntegrationState.ERROR logger.error(f"Failed to initialize ML integration: {e}") raise MLContainerIntegrationError(f"Initialization failed: {e}") from e def start(self) -> None: """Start the ML integration system.""" with self._state_lock: if self._state not in [MLIntegrationState.INITIALIZED, MLIntegrationState.STOPPED]: logger.warning(f"Cannot start ML integration from state: {self._state.value}") return logger.info("Starting ML container integration...") self._state = MLIntegrationState.STARTING try: # Start components in dependency order if self.integration_config.auto_start_components: self._start_ml_components() self._state = MLIntegrationState.RUNNING logger.info("ML container integration started successfully") except Exception as e: self._state = MLIntegrationState.ERROR logger.error(f"Failed to start ML integration: {e}") raise MLContainerIntegrationError(f"Start failed: {e}") from e def stop(self) -> None: """Stop the ML integration system.""" with self._state_lock: if self._state in [MLIntegrationState.STOPPED, MLIntegrationState.NOT_INITIALIZED]: return logger.info("Stopping ML container integration...") old_state = self._state self._state = MLIntegrationState.STOPPING try: # Stop health monitoring self._stop_health_monitoring() # Stop ML components self._stop_ml_components() self._state = MLIntegrationState.STOPPED logger.info("ML container integration stopped successfully") except Exception as e: self._state = MLIntegrationState.ERROR logger.error(f"Error stopping ML integration: {e}") def get_state(self) -> MLIntegrationState: """Get current integration state.""" with self._state_lock: return self._state def get_ml_manager(self) -> Optional[MLManager]: """Get ML manager instance.""" return self._ml_manager def get_audio_pipeline(self) -> Optional[AudioPipeline]: """Get audio pipeline instance.""" return self._audio_pipeline def get_workflow_manager(self) -> Optional[MLWorkflowManager]: """Get workflow manager instance.""" return self._workflow_manager def get_event_manager(self) -> Optional[MLEventManager]: """Get ML event manager instance.""" return self._event_manager def get_performance_monitor(self) -> Optional[MLPerformanceMonitor]: """Get performance monitor instance.""" return self._performance_monitor def get_component_health(self) -> Dict[str, MLComponentStatus]: """Get health status of all ML components.""" return self._component_status.copy() def is_healthy(self) -> bool: """Check if all ML components are healthy.""" if self._state != MLIntegrationState.RUNNING: return False return all(status.is_healthy() for status in self._component_status.values()) def restart_component(self, component_name: str) -> bool: """ Restart a specific ML component. Args: component_name: Name of component to restart Returns: True if restart was successful """ logger.info(f"Restarting ML component: {component_name}") try: # Stop component if component_name == "ml_manager" and self._ml_manager: self._ml_manager.stop() elif component_name == "audio_pipeline" and self._audio_pipeline: self._audio_pipeline.stop() elif component_name == "workflow_manager" and self._workflow_manager: self._workflow_manager.stop() elif component_name == "event_manager" and self._event_manager: self._event_manager.stop() elif component_name == "performance_monitor" and self._performance_monitor: self._performance_monitor.stop() # Wait a moment time.sleep(1.0) # Start component if component_name == "ml_manager" and self._ml_manager: self._ml_manager.start() elif component_name == "audio_pipeline" and self._audio_pipeline: self._audio_pipeline.start() elif component_name == "workflow_manager" and self._workflow_manager: self._workflow_manager.start() elif component_name == "event_manager" and self._event_manager: self._event_manager.start() elif component_name == "performance_monitor" and self._performance_monitor: self._performance_monitor.start() logger.info(f"Component {component_name} restarted successfully") return True except Exception as e: logger.error(f"Failed to restart component {component_name}: {e}") return False def _get_container_dependencies(self) -> None: """Get dependencies from application container.""" if not APPLICATION_CONTAINER_AVAILABLE: logger.warning("Application container not available") return try: # Get event handler if hasattr(self.application_container, 'get_event_handler'): self._event_handler = self.application_container.get_event_handler() logger.debug("Event handler obtained from container") # Get config manager if hasattr(self.application_container, 'get_config_manager'): self._config_manager = self.application_container.get_config_manager() logger.debug("Config manager obtained from container") # Get network manager (for client mode) if hasattr(self.application_container, 'get_network_manager'): try: self._network_manager = self.application_container.get_network_manager() logger.debug("Network manager obtained from container") except Exception as e: logger.warning(f"Network manager not available: {e}") except Exception as e: logger.error(f"Error getting container dependencies: {e}") raise def _load_ml_configuration(self) -> None: """Load ML configuration.""" try: # Create ML config manager self._ml_config_manager = MLConfigManager() # Determine deployment mode mode = MLMode.CLIENT # Default if self.application_container and hasattr(self.application_container, 'get_mode'): container_mode = self.application_container.get_mode() if hasattr(container_mode, 'value'): mode_str = container_mode.value if mode_str == "server": mode = MLMode.SERVER elif mode_str == "standalone": mode = MLMode.STANDALONE elif mode_str == "client": mode = MLMode.CLIENT # Load configuration self._ml_config = self._ml_config_manager.load_config(mode) logger.info(f"ML configuration loaded for {mode.value} mode") except Exception as e: logger.error(f"Error loading ML configuration: {e}") # Create default configuration self._ml_config = create_ml_config() def _initialize_ml_components(self) -> None: """Initialize all ML components.""" logger.info("Initializing ML components...") # Initialize ML Event Manager first (others may depend on it) if self.integration_config.enable_event_integration: self._initialize_event_manager() # Initialize Audio Pipeline if self.integration_config.enable_audio_pipeline: self._initialize_audio_pipeline() # Initialize ML Manager if self.integration_config.enable_ml_manager: self._initialize_ml_manager() # Initialize Workflow Manager if self.integration_config.enable_workflow_manager: self._initialize_workflow_manager() # Initialize Performance Monitor if self.integration_config.enable_performance_monitoring: self._initialize_performance_monitor() logger.info("ML components initialized") def _initialize_event_manager(self) -> None: """Initialize ML event manager.""" try: logger.debug("Initializing ML event manager...") self._event_manager = create_ml_event_manager( main_event_handler=self._event_handler, enable_correlation=True, enable_persistence=False, max_event_history=1000 ) self._update_component_status( "event_manager", MLIntegrationState.INITIALIZED, ComponentStatus.INITIALIZED ) logger.debug("ML event manager initialized") except Exception as e: logger.error(f"Error initializing ML event manager: {e}") self._update_component_status( "event_manager", MLIntegrationState.ERROR, ComponentStatus.ERROR, errors=[str(e)] ) raise def _initialize_audio_pipeline(self) -> None: """Initialize audio pipeline.""" try: logger.debug("Initializing audio pipeline...") # Create audio pipeline configuration if self._ml_config and self._ml_config.audio_pipeline_config: audio_config = self._ml_config.audio_pipeline_config else: audio_config = AudioPipelineConfig( sample_rate=self._ml_config.sample_rate if self._ml_config else 16000, channels=self._ml_config.channels if self._ml_config else 1, chunk_size=self._ml_config.chunk_size if self._ml_config else 1024 ) self._audio_pipeline = create_audio_pipeline( config=audio_config, ml_manager=None # Will be set later ) self._update_component_status( "audio_pipeline", MLIntegrationState.INITIALIZED, ComponentStatus.INITIALIZED ) logger.debug("Audio pipeline initialized") except Exception as e: logger.error(f"Error initializing audio pipeline: {e}") self._update_component_status( "audio_pipeline", MLIntegrationState.ERROR, ComponentStatus.ERROR, errors=[str(e)] ) raise def _initialize_ml_manager(self) -> None: """Initialize ML manager.""" try: logger.debug("Initializing ML manager...") # Create ML configuration if self._ml_config: ml_config = MLConfig( mode=self._ml_config.deployment_mode, enable_wakeword=self._ml_config.enable_wakeword, enable_voice_recognition=self._ml_config.enable_voice_recognition, enable_continuous_recognition=self._ml_config.enable_continuous_recognition, wakeword_config=self._ml_config.wakeword_config, voice_recognition_config=self._ml_config.voice_recognition_config, sample_rate=self._ml_config.sample_rate, chunk_size=self._ml_config.chunk_size ) else: ml_config = create_default_ml_config() self._ml_manager = create_ml_manager( config=ml_config, application=self.application_container, event_handler=self._event_handler, network_manager=self._network_manager ) # Link with audio pipeline if self._audio_pipeline: self._audio_pipeline.ml_manager = self._ml_manager self._update_component_status( "ml_manager", MLIntegrationState.INITIALIZED, ComponentStatus.INITIALIZED ) logger.debug("ML manager initialized") except Exception as e: logger.error(f"Error initializing ML manager: {e}") self._update_component_status( "ml_manager", MLIntegrationState.ERROR, ComponentStatus.ERROR, errors=[str(e)] ) raise def _initialize_workflow_manager(self) -> None: """Initialize workflow manager.""" try: logger.debug("Initializing workflow manager...") if not self._ml_manager: logger.warning("ML manager not available for workflow manager") return # Create workflow configuration workflow_config = WorkflowConfig( enable_workflow_logging=self.integration_config.enable_integration_logging ) self._workflow_manager = create_workflow_manager( ml_manager=self._ml_manager, network_manager=self._network_manager, event_handler=self._event_handler, config=workflow_config ) self._update_component_status( "workflow_manager", MLIntegrationState.INITIALIZED, ComponentStatus.INITIALIZED ) logger.debug("Workflow manager initialized") except Exception as e: logger.error(f"Error initializing workflow manager: {e}") self._update_component_status( "workflow_manager", MLIntegrationState.ERROR, ComponentStatus.ERROR, errors=[str(e)] ) raise def _initialize_performance_monitor(self) -> None: """Initialize performance monitor.""" try: logger.debug("Initializing performance monitor...") # Create performance configuration performance_config = PerformanceConfig( enable_performance_logging=self.integration_config.enable_integration_logging, monitoring_interval_seconds=1.0, enable_automatic_optimization=True ) self._performance_monitor = create_performance_monitor( config=performance_config, ml_manager=self._ml_manager, audio_pipeline=self._audio_pipeline ) self._update_component_status( "performance_monitor", MLIntegrationState.INITIALIZED, ComponentStatus.INITIALIZED ) logger.debug("Performance monitor initialized") except Exception as e: logger.error(f"Error initializing performance monitor: {e}") self._update_component_status( "performance_monitor", MLIntegrationState.ERROR, ComponentStatus.ERROR, errors=[str(e)] ) # Performance monitor is not critical - don't raise def _start_ml_components(self) -> None: """Start all ML components.""" logger.info("Starting ML components...") # Start components in dependency order startup_order = [ ("event_manager", self._event_manager), ("audio_pipeline", self._audio_pipeline), ("ml_manager", self._ml_manager), ("workflow_manager", self._workflow_manager), ("performance_monitor", self._performance_monitor) ] for component_name, component in startup_order: if component: try: logger.debug(f"Starting {component_name}...") # Add startup delay if configured if self.integration_config.startup_delay_seconds > 0: time.sleep(self.integration_config.startup_delay_seconds) component.start() self._update_component_status( component_name, MLIntegrationState.RUNNING, ComponentStatus.RUNNING ) logger.debug(f"{component_name} started") except Exception as e: logger.error(f"Error starting {component_name}: {e}") self._update_component_status( component_name, MLIntegrationState.ERROR, ComponentStatus.ERROR, errors=[str(e)] ) # Some components are critical for operation if component_name in ["ml_manager", "workflow_manager"]: raise logger.info("ML components started") def _stop_ml_components(self) -> None: """Stop all ML components.""" logger.info("Stopping ML components...") # Stop components in reverse order shutdown_order = [ ("performance_monitor", self._performance_monitor), ("workflow_manager", self._workflow_manager), ("ml_manager", self._ml_manager), ("audio_pipeline", self._audio_pipeline), ("event_manager", self._event_manager) ] for component_name, component in shutdown_order: if component: try: logger.debug(f"Stopping {component_name}...") component.stop() self._update_component_status( component_name, MLIntegrationState.STOPPED, ComponentStatus.STOPPED ) logger.debug(f"{component_name} stopped") except Exception as e: logger.error(f"Error stopping {component_name}: {e}") self._update_component_status( component_name, MLIntegrationState.ERROR, ComponentStatus.ERROR, errors=[str(e)] ) logger.info("ML components stopped") def _start_health_monitoring(self) -> None: """Start health monitoring thread.""" if self._health_check_thread and self._health_check_thread.is_alive(): return logger.debug("Starting ML health monitoring...") self._stop_health_checks.clear() self._health_check_thread = threading.Thread( target=self._health_monitoring_loop, daemon=True, name="MLHealthMonitor" ) self._health_check_thread.start() logger.debug("ML health monitoring started") def _stop_health_monitoring(self) -> None: """Stop health monitoring thread.""" if not self._health_check_thread or not self._health_check_thread.is_alive(): return logger.debug("Stopping ML health monitoring...") self._stop_health_checks.set() self._health_check_thread.join(timeout=5.0) logger.debug("ML health monitoring stopped") def _health_monitoring_loop(self) -> None: """Health monitoring loop.""" logger.debug("ML health monitoring loop started") while not self._stop_health_checks.is_set(): try: self._check_component_health() # Log component status if enabled if self.integration_config.log_component_status: self._log_component_status() # Wait for next check self._stop_health_checks.wait(self.integration_config.health_check_interval) except Exception as e: logger.error(f"Error in health monitoring loop: {e}") time.sleep(5.0) logger.debug("ML health monitoring loop stopped") def _check_component_health(self) -> None: """Check health of all ML components.""" components = [ ("event_manager", self._event_manager), ("audio_pipeline", self._audio_pipeline), ("ml_manager", self._ml_manager), ("workflow_manager", self._workflow_manager), ("performance_monitor", self._performance_monitor) ] for component_name, component in components: if component: try: # Check if component is responsive is_healthy = self._check_component_responsiveness(component_name, component) if is_healthy: self._update_component_status( component_name, MLIntegrationState.RUNNING, ComponentStatus.RUNNING ) else: self._handle_component_failure(component_name, "Health check failed") except Exception as e: logger.error(f"Error checking health of {component_name}: {e}") self._handle_component_failure(component_name, str(e)) def _check_component_responsiveness(self, component_name: str, component: Any) -> bool: """Check if component is responsive.""" try: # Basic responsiveness checks if hasattr(component, 'is_running'): return component.is_running() elif hasattr(component, 'get_current_state'): state = component.get_current_state() return state is not None elif hasattr(component, '_is_running'): return getattr(component, '_is_running', False) else: # Assume healthy if no specific check available return True except Exception as e: logger.warning(f"Error checking responsiveness of {component_name}: {e}") return False def _handle_component_failure(self, component_name: str, error_message: str) -> None: """Handle component failure.""" logger.warning(f"Component failure detected: {component_name} - {error_message}") self._update_component_status( component_name, MLIntegrationState.ERROR, ComponentStatus.ERROR, errors=[error_message] ) # Attempt recovery if enabled if self.integration_config.enable_component_recovery: self._attempt_component_recovery(component_name, error_message) def _attempt_component_recovery(self, component_name: str, error_message: str) -> None: """Attempt to recover a failed component.""" recovery_count = self._recovery_attempts.get(component_name, 0) if recovery_count >= self.integration_config.max_recovery_attempts: logger.error(f"Max recovery attempts reached for {component_name}") return logger.info(f"Attempting recovery for {component_name} (attempt {recovery_count + 1})") try: success = self.restart_component(component_name) if success: logger.info(f"Recovery successful for {component_name}") self._recovery_attempts[component_name] = 0 # Reset counter else: self._recovery_attempts[component_name] = recovery_count + 1 except Exception as e: logger.error(f"Recovery failed for {component_name}: {e}") self._recovery_attempts[component_name] = recovery_count + 1 def _update_component_status( self, component_name: str, state: MLIntegrationState, health: ComponentStatus, errors: Optional[List[str]] = None, warnings: Optional[List[str]] = None, info: Optional[Dict[str, Any]] = None ) -> None: """Update component status.""" self._component_status[component_name] = MLComponentStatus( name=component_name, status=state, health=health, last_check=time.time(), errors=errors or [], warnings=warnings or [], info=info or {} ) def _log_component_status(self) -> None: """Log current component status.""" if not self.integration_config.log_component_status: return healthy_count = sum(1 for status in self._component_status.values() if status.is_healthy()) total_count = len(self._component_status) logger.info(f"ML Component Health: {healthy_count}/{total_count} healthy") for name, status in self._component_status.items(): if not status.is_healthy(): logger.warning(f"Component {name}: {status.status.value} - {status.errors}") def create_ml_integration( application_container: Any, integration_config: Optional[MLIntegrationConfig] = None ) -> MLContainerIntegration: """ Factory function to create ML container integration. Args: application_container: Trixy application container integration_config: Integration configuration Returns: Configured ML container integration """ return MLContainerIntegration( application_container=application_container, integration_config=integration_config ) # Export main classes and functions __all__ = [ 'MLContainerIntegration', 'MLIntegrationConfig', 'MLIntegrationState', 'MLComponentStatus', 'MLContainerIntegrationError', 'create_ml_integration' ]