""" Trixy Satellite Management System This module provides comprehensive satellite (client) management for the Trixy application. It includes satellite registration, connection management, status monitoring, and advanced access patterns for managing multiple satellites simultaneously. The satellite management system implements the architecture specified in CLAUDE.md: - SatelliteManager: Central management of all registered satellites - Satellite: Individual satellite representation and control - Registration system: MAC-based registration file management - Query system: Advanced satellite filtering and selection - Event integration: Full integration with the event system - Network integration: Socket management and communication Key Features: - Thread-safe operations for multi-satellite support - Advanced access patterns (index, query-based, bulk operations) - Connection state management (registered, not connected, connected) - Satellite registration and blacklist management - Integration with existing event and network systems - Comprehensive logging and error handling Usage Example: from trixy_core.satellites import SatelliteManager, Satellite # Create satellite manager (usually via application container) satellite_manager = SatelliteManager(application) # Access satellites satellite = satellite_manager[0] # Direct index access satellites = satellite_manager["status=connected,room=kitchen"] # Query-based # Bulk operations satellite_manager.disconnect_all("room=living_room") satellite_manager.say_all("Hello everyone!", "status=connected") # Individual satellite operations satellite.say("Hello from the kitchen!") status = satellite.get_status() satellite.disconnect() """ from .satellite_manager import ( SatelliteManager, SatelliteManagerError, SatelliteNotFoundError, SatelliteRegistrationError, SatelliteConnectionError, ConnectionState, SatelliteStats, ) from .satellite import ( Satellite, SatelliteError, SatelliteStatus, SatelliteCapability, AudioPortInfo, SatelliteInfo, ) from .registration import ( SatelliteRegistration, RegistrationManager, RegistrationError, RegistrationStatus, BlacklistEntry, BlacklistManager, ) from .query_parser import ( QueryParser, QueryError, QueryCondition, QueryOperator, parse_query, validate_query, ) # Version information __version__ = "1.0.0" __author__ = "Trixy Development Team" # Public API __all__ = [ # Core classes "SatelliteManager", "Satellite", "SatelliteRegistration", "RegistrationManager", "BlacklistManager", "QueryParser", # Exception classes "SatelliteManagerError", "SatelliteNotFoundError", "SatelliteRegistrationError", "SatelliteConnectionError", "SatelliteError", "RegistrationError", "QueryError", # Enums and data classes "ConnectionState", "SatelliteStatus", "SatelliteCapability", "RegistrationStatus", "QueryOperator", "SatelliteStats", "AudioPortInfo", "SatelliteInfo", "BlacklistEntry", "QueryCondition", # Utility functions "parse_query", "validate_query", ] def pprint(message: str) -> None: """ Satellite system logging function. This uses the same pattern as specified in CLAUDE.md. """ print(f"[SATELLITES] {message}") def create_satellite_manager(application, **kwargs) -> SatelliteManager: """ Convenience function to create a SatelliteManager instance. Args: application: Application container instance **kwargs: Additional arguments for SatelliteManager Returns: SatelliteManager: Configured satellite manager """ return SatelliteManager(application, **kwargs) def get_supported_query_operators(): """ Get a list of all supported query operators. Returns: List[QueryOperator]: All supported query operators """ return list(QueryOperator) def get_supported_satellite_capabilities(): """ Get a list of all supported satellite capabilities. Returns: List[SatelliteCapability]: All supported capabilities """ return list(SatelliteCapability) def validate_mac_address(mac_address: str) -> bool: """ Validate a MAC address format. Args: mac_address: MAC address to validate Returns: bool: True if valid, False otherwise """ if not isinstance(mac_address, str): return False parts = mac_address.split(':') if len(parts) != 6: return False for part in parts: if len(part) != 2: return False try: int(part, 16) except ValueError: return False return True def normalize_mac_address(mac_address: str) -> str: """ Normalize a MAC address to lowercase with colons. Args: mac_address: MAC address to normalize Returns: str: Normalized MAC address Raises: ValueError: If MAC address is invalid """ if not validate_mac_address(mac_address): raise ValueError(f"Invalid MAC address format: {mac_address}") return mac_address.lower() def get_satellite_documentation(): """ Get documentation for satellite management system. Returns: Dict[str, str]: Documentation mapping """ return { "satellite_manager": "Central management system for all registered satellites", "satellite": "Individual satellite representation with control methods", "registration": "MAC-based registration file management system", "query_parser": "Advanced query system for satellite filtering and selection", "connection_states": "registered, not_connected, connected, error", "access_patterns": "Direct index, query-based, bulk operations", "integration": "Full integration with event and network systems", } # Initialize logging for the module def _init_satellite_logging(): """Initialize logging for the satellite management system.""" # This will be enhanced when integrated with proper logging system pass # Initialize the module _init_satellite_logging() # Module-level configuration SATELLITE_CONFIG_DEFAULTS = { "max_satellites": 50, "registration_timeout": 60.0, "connection_timeout": 30.0, "heartbeat_interval": 30.0, "max_reconnect_attempts": 3, "registration_file_dir": "config/satellites", "blacklist_file": "config/satellites/blacklist.json", "enable_auto_registration": False, "default_audio_ports": { "raw_input": 2102, "raw_output": 2103, "music_output": 2104, } } def get_satellite_config_defaults(): """ Get default configuration values for satellite management. Returns: Dict[str, Any]: Default configuration values """ return SATELLITE_CONFIG_DEFAULTS.copy() # Convenience function for quick satellite manager setup def setup_satellite_system( application, registration_dir: str = None, max_satellites: int = 50, enable_auto_registration: bool = False ) -> SatelliteManager: """ Set up a complete satellite management system with reasonable defaults. Args: application: Application container instance registration_dir: Directory for registration files max_satellites: Maximum number of satellites enable_auto_registration: Enable automatic registration Returns: SatelliteManager: Configured and ready-to-use satellite manager """ satellite_manager = SatelliteManager( application, registration_dir=registration_dir, max_satellites=max_satellites, enable_auto_registration=enable_auto_registration ) pprint(f"Satellite system initialized (max: {max_satellites}, auto_reg: {enable_auto_registration})") return satellite_manager # Export additional utility functions __all__.extend([ "create_satellite_manager", "get_supported_query_operators", "get_supported_satellite_capabilities", "validate_mac_address", "normalize_mac_address", "get_satellite_documentation", "get_satellite_config_defaults", "setup_satellite_system", "pprint", ])