""" Application Container and Factory System for Trixy This module provides the central ApplicationContainer class that implements: - Dependency injection pattern with lazy loading - Component lifecycle management (initialization, startup, shutdown) - Thread-safe singleton pattern for different deployment modes - Integration with event system and configuration system - Component health monitoring and status reporting - Graceful shutdown with proper cleanup The ApplicationContainer serves as the central hub for all Trixy components, providing easy access to: - Event handler system - Configuration manager - Plugin system - Satellite manager - And all other core components Usage: from trixy_core.application import ApplicationContainer # Create application container application = ApplicationContainer( mode="server", config_path="config/server_config.json", debug_mode=False ) # Start the application application.start() # Access components event_handler = application.get_event_handler() plugin_system = application.get_plugin_system() satellite_manager = application.get_satellite_manager() config_manager = application.get_config_manager() # Graceful shutdown application.shutdown() """ import os import sys import threading import time import traceback import weakref from typing import Dict, Any, Optional, Set, List, Type, Callable, Union from pathlib import Path from dataclasses import dataclass from enum import Enum import json from pprint import pformat # Import configuration system try: from .config import ( ConfigManager, DeploymentMode, ServerConfig, ClientConfig, StandaloneConfig, ConfigurationError ) from .config import pprint as config_pprint except ImportError as e: print(f"Failed to import configuration system: {e}") sys.exit(1) # Import event system try: from .events import ( EventHandler, TrixyEvent, SystemStartupEventData, SystemShutdownEventData, EventDataFactory ) from .events import pprint as events_pprint except ImportError as e: print(f"Failed to import event system: {e}") sys.exit(1) class ComponentStatus(Enum): """Status of individual components in the application container.""" NOT_INITIALIZED = "not_initialized" INITIALIZING = "initializing" INITIALIZED = "initialized" STARTING = "starting" RUNNING = "running" STOPPING = "stopping" STOPPED = "stopped" ERROR = "error" class ApplicationStatus(Enum): """Overall status of the application container.""" NOT_STARTED = "not_started" STARTING = "starting" RUNNING = "running" SHUTTING_DOWN = "shutting_down" STOPPED = "stopped" ERROR = "error" @dataclass class ComponentHealth: """Health information for a component.""" name: str status: ComponentStatus last_check: float errors: List[str] warnings: List[str] info: Dict[str, Any] def is_healthy(self) -> bool: """Check if component is healthy.""" return self.status in [ComponentStatus.INITIALIZED, ComponentStatus.RUNNING] and len(self.errors) == 0 class ApplicationContainerError(Exception): """Base exception for application container errors.""" pass class ComponentInitializationError(ApplicationContainerError): """Raised when a component fails to initialize.""" pass class ComponentLifecycleError(ApplicationContainerError): """Raised when a component lifecycle operation fails.""" pass def pprint(message: str) -> None: """ Application container logging function that adapts based on mode. Uses the same pattern as specified in CLAUDE.md. """ # For now use simple printing - will integrate with proper logging print(f"[APPLICATION] {message}") class ApplicationContainer: """ Central application container providing dependency injection and lifecycle management. This is a thread-safe singleton that manages all core components of the Trixy application. It supports three deployment modes: server, client, and standalone. Features: - Lazy component initialization - Dependency injection - Component lifecycle management - Health monitoring - Graceful shutdown - Thread-safe access - Event-driven architecture """ _instances: Dict[str, 'ApplicationContainer'] = {} _lock = threading.RLock() def __init__( self, mode: Union[str, DeploymentMode], config_path: Optional[str] = None, debug_mode: bool = False, **kwargs ): """ Initialize the application container. Args: mode: Deployment mode (server, client, standalone) config_path: Path to configuration file debug_mode: Enable debug mode **kwargs: Additional configuration overrides """ # Convert string mode to enum if isinstance(mode, str): mode = DeploymentMode(mode.lower()) self._mode = mode self._config_path = config_path self._debug_mode = debug_mode self._kwargs = kwargs # Application state self._status = ApplicationStatus.NOT_STARTED self._startup_time: Optional[float] = None self._shutdown_time: Optional[float] = None # Component storage self._components: Dict[str, Any] = {} self._component_factories: Dict[str, Callable] = {} self._component_health: Dict[str, ComponentHealth] = {} self._component_dependencies: Dict[str, Set[str]] = {} self._component_lock = threading.RLock() # Core components that will be initialized self._config_manager: Optional[ConfigManager] = None self._event_handler: Optional[EventHandler] = None # Initialize logging self._init_logging() pprint(f"ApplicationContainer created for {mode.value} mode") # Register component factories self._register_component_factories() # Store in singleton registry with self._lock: instance_key = f"{mode.value}_{id(self)}" self._instances[instance_key] = self self._instance_key = instance_key def _init_logging(self) -> None: """Initialize logging for the application container.""" # This will be enhanced when integrated with the main logging system pass def _register_component_factories(self) -> None: """Register factory methods for all components.""" self._component_factories = { 'config_manager': self._create_config_manager, 'event_handler': self._create_event_handler, 'plugin_system': self._create_plugin_system, 'satellite_manager': self._create_satellite_manager, 'network_manager': self._create_network_manager, 'asset_manager': self._create_asset_manager, 'scheduler': self._create_scheduler, 'conversation_manager': self._create_conversation_manager, 'arbitration_manager': self._create_arbitration_manager, 'ml_system': self._create_ml_system, } # Define component dependencies self._component_dependencies = { 'config_manager': set(), # No dependencies 'event_handler': {'config_manager'}, 'plugin_system': {'config_manager', 'event_handler'}, 'satellite_manager': {'config_manager', 'event_handler'}, 'network_manager': {'config_manager', 'event_handler'}, 'asset_manager': {'config_manager'}, 'scheduler': {'config_manager', 'event_handler'}, 'conversation_manager': {'config_manager', 'event_handler'}, 'arbitration_manager': {'config_manager', 'event_handler'}, 'ml_system': {'config_manager', 'event_handler'}, } pprint(f"Registered {len(self._component_factories)} component factories") def start(self) -> None: """ Start the application container and all components. This performs: 1. Configuration loading and validation 2. Core component initialization 3. Component startup in dependency order 4. Event system startup 5. Application ready signal Raises: ApplicationContainerError: If startup fails """ if self._status != ApplicationStatus.NOT_STARTED: raise ApplicationContainerError(f"Application already started (status: {self._status.value})") pprint("Starting application container...") self._status = ApplicationStatus.STARTING self._startup_time = time.time() try: # Initialize configuration first (it has no dependencies) config_manager = self.get_config_manager() pprint("Configuration manager initialized") # Initialize event handler early (depends on config) event_handler = self.get_event_handler() pprint("Event handler initialized") # Trigger startup event startup_data = EventDataFactory.create_event_data( "system_startup", mode=self._mode.value, config_file=self._config_path or "default", debug_mode=self._debug_mode, startup_time_ms=self._startup_time * 1000 ) event_handler.trigger_event("system_startup", startup_data) # Initialize other components based on mode self._initialize_mode_specific_components() self._status = ApplicationStatus.RUNNING pprint(f"Application container started successfully in {time.time() - self._startup_time:.2f} seconds") except Exception as e: self._status = ApplicationStatus.ERROR pprint(f"Failed to start application container: {e}") if self._debug_mode: pprint(traceback.format_exc()) raise ApplicationContainerError(f"Application startup failed: {e}") from e def _initialize_mode_specific_components(self) -> None: """Initialize components specific to the current deployment mode.""" if self._mode == DeploymentMode.SERVER: pprint("Initializing server mode components...") self.get_plugin_system() self.get_satellite_manager() network_manager = self.get_network_manager() self.get_scheduler() self.get_conversation_manager() self.get_arbitration_manager() # ML system is not typically needed on server # Start network server for SERVER mode self._start_network_server(network_manager) elif self._mode == DeploymentMode.CLIENT: pprint("Initializing client mode components...") network_manager = self.get_network_manager() ml_system = self.get_ml_system() # Client needs ML for wakeword/voice recognition # Client mode has minimal components # Connect to server for CLIENT mode self._connect_to_network_server(network_manager) # Start ML system for wakeword detection and voice recognition self._start_ml_system(ml_system) elif self._mode == DeploymentMode.STANDALONE: pprint("Initializing standalone mode components...") self.get_plugin_system() network_manager = self.get_network_manager() self.get_scheduler() self.get_conversation_manager() self.get_ml_system() # Standalone also needs ML processing # Standalone has most components but no satellite manager # Start network server for STANDALONE mode (can accept connections) self._start_network_server(network_manager) # Asset manager is available in all modes self.get_asset_manager() def _start_network_server(self, network_manager) -> None: """ Start the network server with configuration from config manager. Args: network_manager: NetworkManager instance to start """ try: pprint("Starting network server...") # Get configuration config_manager = self.get_config_manager() config = config_manager.get_config() # Extract network configuration network_config = getattr(config, 'network', None) if not network_config: raise ApplicationContainerError("No network configuration found") # Determine ports from configuration command_port = getattr(network_config, 'command_port', 2101) audio_ports = { 'input': getattr(network_config, 'raw_audio_input_port', 2102), 'output': getattr(network_config, 'raw_audio_output_port', 2103), 'music': getattr(network_config, 'raw_music_output_port', 2104) } # Start the network server network_manager.start_server( command_port=command_port, audio_ports=audio_ports ) pprint(f"Network server started successfully on command port {command_port}") pprint(f"Audio streams: Input={audio_ports['input']}, Output={audio_ports['output']}, Music={audio_ports['music']}") # Update component health self._update_component_health('network_manager', ComponentStatus.RUNNING) except Exception as e: pprint(f"Failed to start network server: {e}") if self._debug_mode: pprint(traceback.format_exc()) self._update_component_health('network_manager', ComponentStatus.ERROR, errors=[str(e)]) raise ApplicationContainerError(f"Network server startup failed: {e}") from e def _start_ml_system(self, ml_system) -> None: """ Start the ML system for wakeword detection and voice recognition. Args: ml_system: MLManager instance to start """ try: pprint("Starting ML system...") # Start ML system asynchronously import asyncio loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) loop.run_until_complete(ml_system.start()) pprint("ML system started successfully") self._update_component_health('ml_system', ComponentStatus.RUNNING) except Exception as e: pprint(f"Failed to start ML system: {e}") if self._debug_mode: pprint(traceback.format_exc()) self._update_component_health('ml_system', ComponentStatus.ERROR, errors=[str(e)]) raise ApplicationContainerError(f"ML system startup failed: {e}") from e def _connect_to_network_server(self, network_manager) -> None: """ Connect to a network server with configuration from config manager. Args: network_manager: NetworkManager instance to use for connection """ try: pprint("Connecting to network server...") # Get configuration config_manager = self.get_config_manager() config = config_manager.get_config() # Extract satellite configuration for client mode satellite_config = getattr(config, 'satellite', None) if not satellite_config: raise ApplicationContainerError("No satellite configuration found for client mode") # Get server connection details server_host = getattr(satellite_config, 'server_host', 'localhost') server_port = getattr(satellite_config, 'server_port', 2101) # Attempt to connect to server success = network_manager.connect_to_server(server_host, server_port) if success: pprint(f"Connected to server at {server_host}:{server_port}") self._update_component_health('network_manager', ComponentStatus.RUNNING) else: pprint(f"Failed to connect to server at {server_host}:{server_port}") self._update_component_health('network_manager', ComponentStatus.ERROR, errors=["Connection failed"]) # Check if auto-reconnect is enabled auto_reconnect = getattr(satellite_config, 'auto_reconnect', True) if auto_reconnect: pprint("Auto-reconnect is enabled - connection will be retried automatically") else: raise ApplicationContainerError(f"Failed to connect to server at {server_host}:{server_port}") except Exception as e: pprint(f"Failed to connect to network server: {e}") if self._debug_mode: pprint(traceback.format_exc()) self._update_component_health('network_manager', ComponentStatus.ERROR, errors=[str(e)]) raise ApplicationContainerError(f"Network server connection failed: {e}") from e def shutdown(self) -> None: """ Gracefully shutdown the application container. This performs: 1. Component shutdown in reverse dependency order 2. Resource cleanup 3. Event system shutdown 4. Final status update """ if self._status in [ApplicationStatus.SHUTTING_DOWN, ApplicationStatus.STOPPED]: pprint("Application container already shutting down or stopped") return pprint("Shutting down application container...") self._status = ApplicationStatus.SHUTTING_DOWN self._shutdown_time = time.time() try: # Trigger shutdown event if event handler is available if self._event_handler: shutdown_data = EventDataFactory.create_event_data( "system_shutdown", reason="normal_shutdown", graceful_shutdown=True, uptime_seconds=self._shutdown_time - (self._startup_time or 0) ) self._event_handler.trigger_event("system_shutdown", shutdown_data) # Shutdown components in reverse dependency order shutdown_order = self._get_shutdown_order() for component_name in shutdown_order: self._shutdown_component(component_name) self._status = ApplicationStatus.STOPPED pprint("Application container shutdown completed") except Exception as e: self._status = ApplicationStatus.ERROR pprint(f"Error during shutdown: {e}") if self._debug_mode: pprint(traceback.format_exc()) finally: # Clean up from singleton registry with self._lock: if hasattr(self, '_instance_key') and self._instance_key in self._instances: del self._instances[self._instance_key] def _get_shutdown_order(self) -> List[str]: """Get the order in which components should be shut down (reverse of dependency order).""" # For now, use a simple reverse order # This could be enhanced with proper topological sorting order = [ 'arbitration_manager', 'conversation_manager', 'scheduler', 'ml_system', # Shutdown ML system before core components 'asset_manager', 'network_manager', 'satellite_manager', 'plugin_system', 'event_handler', 'config_manager' ] return [name for name in order if name in self._components] def _shutdown_component(self, component_name: str) -> None: """Shutdown a specific component safely.""" if component_name not in self._components: return component = self._components[component_name] try: pprint(f"Shutting down {component_name}...") # Call shutdown method if available if hasattr(component, 'shutdown'): component.shutdown() elif hasattr(component, 'stop'): component.stop() elif hasattr(component, 'close'): component.close() # Update health status if component_name in self._component_health: self._component_health[component_name].status = ComponentStatus.STOPPED pprint(f"{component_name} shutdown completed") except Exception as e: pprint(f"Error shutting down {component_name}: {e}") if self._debug_mode: pprint(traceback.format_exc()) # Component Factory Methods def _create_config_manager(self) -> ConfigManager: """Create and initialize the configuration manager.""" pprint("Creating configuration manager...") config_manager = ConfigManager( mode=self._mode, config_file=self._config_path, debug_mode=self._debug_mode ) # Load configuration config = config_manager.load_config() pprint(f"Configuration loaded successfully for {self._mode.value} mode") self._update_component_health('config_manager', ComponentStatus.INITIALIZED) return config_manager def _create_event_handler(self) -> EventHandler: """Create and initialize the event handler.""" pprint("Creating event handler...") # Get configuration for event handler settings config_manager = self.get_config_manager() config = config_manager.get_config() # Create event handler with configuration event_handler = EventHandler( max_history_size=getattr(config.logging, 'max_event_history', 1000) ) # Enable debug mode if needed if self._debug_mode: event_handler.enable_debug_mode(True) pprint("Event handler created successfully") self._update_component_health('event_handler', ComponentStatus.INITIALIZED) return event_handler def _create_plugin_system(self): """Create and initialize the plugin system.""" pprint("Creating plugin system...") try: # Import the real plugin manager from .plugins import PluginManager # Get configuration for plugin system config_manager = self.get_config_manager() config = config_manager.get_config() # Create plugin manager with configuration plugin_manager = PluginManager( application=self, plugins_directory=getattr(config, 'plugins_directory', None), auto_load=getattr(config, 'auto_load_plugins', True), enable_hot_reload=self._debug_mode, # Enable hot-reload in debug mode health_check_interval=getattr(config, 'plugin_health_check_interval', 60.0), max_error_count=getattr(config, 'plugin_max_error_count', 5) ) pprint("Plugin manager created successfully") self._update_component_health('plugin_system', ComponentStatus.INITIALIZED) return plugin_manager except ImportError as e: pprint(f"Failed to import PluginManager: {e}") # Fall back to placeholder plugin_system = PluginSystemPlaceholder(self) self._update_component_health('plugin_system', ComponentStatus.INITIALIZED) return plugin_system except Exception as e: pprint(f"Failed to create plugin manager: {e}") # Fall back to placeholder on any error plugin_system = PluginSystemPlaceholder(self) self._update_component_health('plugin_system', ComponentStatus.ERROR, errors=[str(e)]) return plugin_system def _create_satellite_manager(self): """Create and initialize the satellite manager.""" pprint("Creating satellite manager...") # Satellite manager will be implemented in future # For now, return a placeholder satellite_manager = SatelliteManagerPlaceholder(self) self._update_component_health('satellite_manager', ComponentStatus.INITIALIZED) return satellite_manager def _create_network_manager(self): """Create and initialize the network manager.""" pprint("Creating network manager...") # Import the actual network manager try: from .network import NetworkManager network_manager = NetworkManager(self) pprint("Network manager created successfully") except ImportError as e: pprint(f"Failed to import NetworkManager: {e}") # Fall back to placeholder for now network_manager = NetworkManagerPlaceholder(self) self._update_component_health('network_manager', ComponentStatus.INITIALIZED) return network_manager def _create_asset_manager(self): """Create and initialize the asset manager.""" pprint("Creating asset manager...") try: # Import the real asset manager from .assets import AssetManager, AssetManagerConfig, ProfileMode # Get configuration for asset manager config_manager = self.get_config_manager() config = config_manager.get_config() # Determine assets directory and profile from configuration assets_directory = getattr(config, 'assets_directory', './assets') asset_profile = getattr(config, 'asset_profile', None) # Create asset manager configuration asset_config = AssetManagerConfig( assets_directory=assets_directory, current_profile=asset_profile, cache_enabled=getattr(config, 'asset_cache_enabled', True), cache_max_size=getattr(config, 'asset_cache_max_size', 100), cache_max_memory=getattr(config, 'asset_cache_max_memory', 100 * 1024 * 1024), validation_enabled=getattr(config, 'asset_validation_enabled', True), hot_reload_enabled=self._debug_mode, # Enable hot-reload in debug mode hot_reload_debounce=getattr(config, 'asset_hot_reload_debounce', 0.5), preload_patterns=getattr(config, 'asset_preload_patterns', [ "audio/success.wav", "audio/error.wav", "audio/notification.wav", "audio/wakeword_detected.wav" ]), profile_mode=ProfileMode.FALLBACK, thread_pool_size=getattr(config, 'asset_thread_pool_size', 4), enable_integrity_checks=getattr(config, 'asset_integrity_checks', True), auto_create_directories=True, compress_cached_assets=getattr(config, 'asset_cache_compression', False), asset_timeout=getattr(config, 'asset_timeout', 30.0) ) # Create asset manager with configuration asset_manager = AssetManager( application_container=self, config=asset_config ) pprint("Asset manager created successfully") self._update_component_health('asset_manager', ComponentStatus.INITIALIZED) return asset_manager except ImportError as e: pprint(f"Failed to import AssetManager: {e}") # Fall back to placeholder asset_manager = AssetManagerPlaceholder(self) self._update_component_health('asset_manager', ComponentStatus.INITIALIZED) return asset_manager except Exception as e: pprint(f"Failed to create asset manager: {e}") # Fall back to placeholder on any error asset_manager = AssetManagerPlaceholder(self) self._update_component_health('asset_manager', ComponentStatus.ERROR, errors=[str(e)]) return asset_manager def _create_scheduler(self): """Create and initialize the scheduler.""" pprint("Creating scheduler...") # Scheduler will be implemented in future # For now, return a placeholder scheduler = SchedulerPlaceholder(self) self._update_component_health('scheduler', ComponentStatus.INITIALIZED) return scheduler def _create_conversation_manager(self): """Create and initialize the conversation manager.""" pprint("Creating conversation manager...") try: # Import the real conversation manager from .conversation import ConversationManager # Get configuration for conversation manager config_manager = self.get_config_manager() config = config_manager.get_config() # Create conversation manager with configuration conversation_manager = ConversationManager( application=self, max_concurrent_conversations=getattr(config, 'max_concurrent_conversations', 100), default_conversation_timeout=getattr(config, 'default_conversation_timeout', 300.0), plugin_question_timeout=getattr(config, 'plugin_question_timeout', 30.0), cleanup_interval_seconds=getattr(config, 'conversation_cleanup_interval', 60.0), max_conversation_history=getattr(config, 'max_conversation_history', 1000), enable_context_persistence=getattr(config, 'enable_conversation_persistence', False), context_storage_path=getattr(config, 'conversation_storage_path', None), enable_analytics=getattr(config, 'enable_conversation_analytics', True) ) pprint("Conversation manager created successfully") self._update_component_health('conversation_manager', ComponentStatus.INITIALIZED) return conversation_manager except ImportError as e: pprint(f"Failed to import ConversationManager: {e}") # Fall back to placeholder conversation_manager = ConversationManagerPlaceholder(self) self._update_component_health('conversation_manager', ComponentStatus.INITIALIZED) return conversation_manager except Exception as e: pprint(f"Failed to create conversation manager: {e}") # Fall back to placeholder on any error conversation_manager = ConversationManagerPlaceholder(self) self._update_component_health('conversation_manager', ComponentStatus.ERROR, errors=[str(e)]) return conversation_manager def _create_arbitration_manager(self): """Create and initialize the arbitration manager.""" pprint("Creating arbitration manager...") # Arbitration manager will be implemented in future # For now, return a placeholder arbitration_manager = ArbitrationManagerPlaceholder(self) self._update_component_health('arbitration_manager', ComponentStatus.INITIALIZED) return arbitration_manager def _create_ml_system(self): """Create and initialize the ML system.""" pprint("Creating ML system...") try: # Import the ML system from .ml import create_client_ml_system, create_standalone_ml_system, MLMode # Get configuration for ML system config_manager = self.get_config_manager() config = config_manager.get_config() # Get event handler for ML system integration event_handler = self.get_event_handler() # Determine which ML system to create based on deployment mode if self._mode == DeploymentMode.CLIENT: # Create client ML system ml_system = create_client_ml_system( application_container=self, enable_wakeword=getattr(config, 'enable_wakeword', True), enable_voice_recognition=getattr(config, 'enable_voice_recognition', True), enable_performance_monitoring=getattr(config, 'enable_ml_monitoring', True) ) elif self._mode == DeploymentMode.STANDALONE: # Create standalone ML system ml_system = create_standalone_ml_system( application_container=self, enable_wakeword=getattr(config, 'enable_wakeword', True), enable_voice_recognition=getattr(config, 'enable_voice_recognition', True), enable_continuous_recognition=getattr(config, 'enable_continuous_recognition', True), enable_performance_monitoring=getattr(config, 'enable_ml_monitoring', True) ) else: # For server mode, ML system might not be needed # Return a placeholder for now pprint("ML system not needed for server mode, using placeholder") ml_system = MLSystemPlaceholder(self) pprint("ML system created successfully") self._update_component_health('ml_system', ComponentStatus.INITIALIZED) return ml_system except ImportError as e: pprint(f"Failed to import ML system: {e}") # Fall back to placeholder ml_system = MLSystemPlaceholder(self) self._update_component_health('ml_system', ComponentStatus.INITIALIZED) return ml_system except Exception as e: pprint(f"Failed to create ML system: {e}") # Fall back to placeholder on any error ml_system = MLSystemPlaceholder(self) self._update_component_health('ml_system', ComponentStatus.ERROR, errors=[str(e)]) return ml_system # Component Access Methods (Public API as specified in CLAUDE.md) def get_config_manager(self) -> ConfigManager: """Get the configuration manager instance.""" return self._get_component('config_manager') def get_event_handler(self) -> EventHandler: """Get the event handler instance.""" return self._get_component('event_handler') def get_plugin_system(self): """Get the plugin system instance.""" return self._get_component('plugin_system') def get_satellite_manager(self): """Get the satellite manager instance.""" return self._get_component('satellite_manager') def get_network_manager(self): """Get the network manager instance.""" return self._get_component('network_manager') def get_asset_manager(self): """Get the asset manager instance.""" return self._get_component('asset_manager') def get_scheduler(self): """Get the scheduler instance.""" return self._get_component('scheduler') def get_conversation_manager(self): """Get the conversation manager instance.""" return self._get_component('conversation_manager') def get_arbitration_manager(self): """Get the arbitration manager instance.""" return self._get_component('arbitration_manager') def get_ml_system(self): """Get the ML system instance.""" return self._get_component('ml_system') def _get_component(self, component_name: str): """ Get a component instance, creating it if necessary (lazy loading). Args: component_name: Name of the component Returns: Component instance Raises: ComponentInitializationError: If component creation fails """ with self._component_lock: # Return existing component if available if component_name in self._components: return self._components[component_name] # Check if factory exists if component_name not in self._component_factories: raise ComponentInitializationError(f"No factory registered for component: {component_name}") # Initialize dependencies first self._ensure_dependencies(component_name) # Create the component try: self._update_component_health(component_name, ComponentStatus.INITIALIZING) factory = self._component_factories[component_name] component = factory() self._components[component_name] = component self._update_component_health(component_name, ComponentStatus.RUNNING) pprint(f"Component '{component_name}' initialized successfully") return component except Exception as e: self._update_component_health(component_name, ComponentStatus.ERROR, errors=[str(e)]) raise ComponentInitializationError(f"Failed to initialize {component_name}: {e}") from e def _ensure_dependencies(self, component_name: str) -> None: """Ensure all dependencies for a component are initialized.""" if component_name not in self._component_dependencies: return dependencies = self._component_dependencies[component_name] for dep_name in dependencies: if dep_name not in self._components: pprint(f"Initializing dependency '{dep_name}' for '{component_name}'") self._get_component(dep_name) def _update_component_health( self, component_name: str, status: ComponentStatus, errors: Optional[List[str]] = None, warnings: Optional[List[str]] = None, info: Optional[Dict[str, Any]] = None ) -> None: """Update the health status of a component.""" self._component_health[component_name] = ComponentHealth( name=component_name, status=status, last_check=time.time(), errors=errors or [], warnings=warnings or [], info=info or {} ) # Health and Status Methods def get_status(self) -> ApplicationStatus: """Get the current application status.""" return self._status def get_health_report(self) -> Dict[str, Any]: """ Get a comprehensive health report for the application and all components. Returns: Dictionary containing health information """ return { 'application': { 'status': self._status.value, 'mode': self._mode.value, 'uptime': time.time() - (self._startup_time or time.time()), 'startup_time': self._startup_time, 'debug_mode': self._debug_mode, }, 'components': { name: { 'status': health.status.value, 'is_healthy': health.is_healthy(), 'last_check': health.last_check, 'errors': health.errors, 'warnings': health.warnings, 'info': health.info } for name, health in self._component_health.items() }, 'summary': { 'total_components': len(self._component_health), 'healthy_components': sum(1 for h in self._component_health.values() if h.is_healthy()), 'error_components': sum(1 for h in self._component_health.values() if h.status == ComponentStatus.ERROR), } } def is_healthy(self) -> bool: """Check if the application and all components are healthy.""" if self._status != ApplicationStatus.RUNNING: return False return all(health.is_healthy() for health in self._component_health.values()) # Utility Methods def get_version(self) -> str: """Get the application version.""" return "1.0.0" def get_mode(self) -> DeploymentMode: """Get the deployment mode.""" return self._mode def is_debug_mode(self) -> bool: """Check if debug mode is enabled.""" return self._debug_mode def get_uptime(self) -> float: """Get application uptime in seconds.""" if self._startup_time is None: return 0.0 return time.time() - self._startup_time @classmethod def get_instance(cls, mode: Union[str, DeploymentMode] = None) -> Optional['ApplicationContainer']: """ Get an existing application container instance. Args: mode: Deployment mode to find (optional) Returns: ApplicationContainer instance or None if not found """ with cls._lock: if mode is None: # Return any available instance return next(iter(cls._instances.values()), None) if isinstance(mode, str): mode = DeploymentMode(mode.lower()) # Find instance with matching mode for instance in cls._instances.values(): if instance._mode == mode: return instance return None @classmethod def cleanup_instances(cls) -> None: """Clean up all application container instances.""" with cls._lock: for instance in list(cls._instances.values()): try: instance.shutdown() except Exception as e: pprint(f"Error shutting down instance: {e}") cls._instances.clear() # Placeholder classes for components that will be implemented later # These provide the basic interface expected by the application container class ComponentPlaceholder: """Base placeholder for components that haven't been implemented yet.""" def __init__(self, application: ApplicationContainer): self.application = application self.application_ref = weakref.ref(application) def shutdown(self): """Shutdown method for proper cleanup.""" pass class PluginSystemPlaceholder(ComponentPlaceholder): """Placeholder for the plugin system.""" pass class SatelliteManagerPlaceholder(ComponentPlaceholder): """Placeholder for the satellite manager.""" pass class NetworkManagerPlaceholder(ComponentPlaceholder): """Placeholder for the network manager.""" pass class AssetManagerPlaceholder(ComponentPlaceholder): """Placeholder for the asset manager.""" pass class SchedulerPlaceholder(ComponentPlaceholder): """Placeholder for the scheduler.""" pass class ConversationManagerPlaceholder(ComponentPlaceholder): """Placeholder for the conversation manager.""" pass class ArbitrationManagerPlaceholder(ComponentPlaceholder): """Placeholder for the arbitration manager.""" pass class MLSystemPlaceholder(ComponentPlaceholder): """Placeholder for the ML system.""" pass # Utility Functions def create_application_container( mode: Union[str, DeploymentMode], config_path: Optional[str] = None, debug_mode: bool = False, **kwargs ) -> ApplicationContainer: """ Convenience function to create an ApplicationContainer. Args: mode: Deployment mode config_path: Path to configuration file debug_mode: Enable debug mode **kwargs: Additional arguments Returns: ApplicationContainer instance """ return ApplicationContainer( mode=mode, config_path=config_path, debug_mode=debug_mode, **kwargs ) def get_application_container(mode: Union[str, DeploymentMode] = None) -> Optional[ApplicationContainer]: """ Get an existing application container instance. Args: mode: Deployment mode (optional) Returns: ApplicationContainer instance or None """ return ApplicationContainer.get_instance(mode) def cleanup_application_containers() -> None: """Clean up all application container instances.""" ApplicationContainer.cleanup_instances() # Module exports __all__ = [ 'ApplicationContainer', 'ApplicationContainerError', 'ComponentInitializationError', 'ComponentLifecycleError', 'ApplicationStatus', 'ComponentStatus', 'ComponentHealth', 'create_application_container', 'get_application_container', 'cleanup_application_containers', 'pprint' ]