""" Satellite Manager Implementation This module implements the SatelliteManager class that serves as the central management system for all registered satellites in the Trixy application. It provides advanced access patterns, bulk operations, and comprehensive satellite lifecycle management as specified in CLAUDE.md. Key Features: - Direct index access: satellite_manager[0] - Query-based access: satellite_manager["status=connected,room=kitchen"] - Bulk operations: disconnect_all(), reconnect_all(), say_all() - Thread-safe operations for multi-satellite support - Registration file management and blacklist support - Integration with network system for socket management - Integration with event system for satellite events - Connection state management (registered, not connected, connected) - Comprehensive logging and error handling """ import json import threading import time import weakref from typing import List, Dict, Any, Optional, Union, Iterator, Tuple, Set, Callable from dataclasses import dataclass, field from datetime import datetime, timezone from enum import Enum from pathlib import Path import uuid import os from .satellite import ( Satellite, SatelliteStatus, SatelliteInfo, SatelliteCapability, AudioPortInfo, create_satellite, create_satellite_info, SatelliteError, SatelliteConnectionError ) from .query_parser import ( QueryParser, QueryCondition, QueryError, parse_query, filter_satellites ) def pprint(message: str) -> None: """Satellite manager logging function.""" print(f"[SATELLITE_MANAGER] {message}") class ConnectionState(Enum): """Connection states for satellites.""" REGISTERED = "registered" NOT_CONNECTED = "not_connected" CONNECTED = "connected" ERROR = "error" BLACKLISTED = "blacklisted" @dataclass class SatelliteStats: """Statistics for satellite management.""" total_satellites: int = 0 connected_satellites: int = 0 registered_satellites: int = 0 blacklisted_satellites: int = 0 error_satellites: int = 0 total_connections: int = 0 total_messages: int = 0 total_errors: int = 0 uptime_seconds: float = 0.0 last_updated: Optional[datetime] = None def to_dict(self) -> Dict[str, Any]: """Convert to dictionary representation.""" return { "total_satellites": self.total_satellites, "connected_satellites": self.connected_satellites, "registered_satellites": self.registered_satellites, "blacklisted_satellites": self.blacklisted_satellites, "error_satellites": self.error_satellites, "total_connections": self.total_connections, "total_messages": self.total_messages, "total_errors": self.total_errors, "uptime_seconds": self.uptime_seconds, "last_updated": self.last_updated.isoformat() if self.last_updated else None, } class SatelliteManagerError(Exception): """Base exception for satellite manager errors.""" pass class SatelliteNotFoundError(SatelliteManagerError): """Raised when a requested satellite is not found.""" pass class SatelliteRegistrationError(SatelliteManagerError): """Raised when satellite registration fails.""" pass class SatelliteConnectionError(SatelliteManagerError): """Raised when satellite connection operations fail.""" pass class SatelliteManager: """ Central management system for all registered satellites. This class implements the interface specified in CLAUDE.md, providing: - Advanced access patterns (index, query-based, bulk operations) - Thread-safe operations for multi-satellite environments - Registration and blacklist management - Integration with network and event systems - Comprehensive statistics and monitoring Usage Examples: # Direct access satellite = satellite_manager[0] # Query-based access kitchen_satellites = satellite_manager["room=kitchen"] connected_satellites = satellite_manager["status=connected"] # Bulk operations satellite_manager.disconnect_all("room=living_room") satellite_manager.say_all("Hello everyone!", "status=connected") """ def __init__( self, application, registration_dir: Optional[str] = None, blacklist_file: Optional[str] = None, max_satellites: int = 50, enable_auto_registration: bool = False, registration_timeout: float = 60.0, connection_timeout: float = 30.0, heartbeat_interval: float = 30.0 ): """ Initialize the satellite manager. Args: application: Application container reference registration_dir: Directory for registration files blacklist_file: Path to blacklist file max_satellites: Maximum number of satellites enable_auto_registration: Enable automatic registration registration_timeout: Timeout for registration process connection_timeout: Timeout for connections heartbeat_interval: Heartbeat check interval """ self._application = application self._application_ref = weakref.ref(application) if application else None # Configuration self._max_satellites = max_satellites self._enable_auto_registration = enable_auto_registration self._registration_timeout = registration_timeout self._connection_timeout = connection_timeout self._heartbeat_interval = heartbeat_interval # File paths self._registration_dir = Path(registration_dir or "config/satellites") self._blacklist_file = Path(blacklist_file or "config/satellites/blacklist.json") # Ensure directories exist self._registration_dir.mkdir(parents=True, exist_ok=True) self._blacklist_file.parent.mkdir(parents=True, exist_ok=True) # Storage self._satellites: Dict[str, Satellite] = {} # MAC address -> Satellite self._satellites_by_id: Dict[str, Satellite] = {} # satellite_id -> Satellite self._blacklisted_macs: Set[str] = set() # Thread safety self._lock = threading.RLock() # Query parser self._query_parser = QueryParser(case_sensitive=False) # Statistics self._stats = SatelliteStats() self._start_time = datetime.now(timezone.utc) # Registration mode self._registration_mode = False self._registration_timer = None # Event integration self._event_handler = None self._network_manager = None # Load existing data self._load_blacklist() self._load_registered_satellites() pprint(f"SatelliteManager initialized (max: {max_satellites}, auto_reg: {enable_auto_registration})") self._update_stats() # Advanced Access Patterns Implementation def __getitem__(self, key: Union[int, str]) -> Union[Satellite, List[Satellite]]: """ Advanced access pattern implementation. Supports: - Direct index access: satellite_manager[0] - Query-based access: satellite_manager["status=connected,room=kitchen"] Args: key: Index (int) or query string (str) Returns: Satellite or List[Satellite] depending on access pattern Raises: SatelliteNotFoundError: If satellite not found QueryError: If query is invalid """ with self._lock: if isinstance(key, int): # Direct index access satellites = list(self._satellites.values()) if key < 0 or key >= len(satellites): raise SatelliteNotFoundError(f"Satellite index {key} out of range (0-{len(satellites)-1})") return satellites[key] elif isinstance(key, str): # Query-based access try: satellites = list(self._satellites.values()) filtered = self._query_parser.filter_satellites(satellites, key) return filtered except QueryError as e: raise SatelliteManagerError(f"Invalid query '{key}': {e}") else: raise SatelliteManagerError(f"Unsupported key type: {type(key)}") def __len__(self) -> int: """Get the number of registered satellites.""" return len(self._satellites) def __iter__(self) -> Iterator[Satellite]: """Iterate over all satellites.""" with self._lock: return iter(list(self._satellites.values())) def __contains__(self, item: Union[str, Satellite]) -> bool: """Check if satellite exists by MAC address or satellite object.""" with self._lock: if isinstance(item, str): return item in self._satellites elif isinstance(item, Satellite): return item.mac_address in self._satellites return False # Core Management Methods def add_satellite(self, satellite_info: SatelliteInfo) -> Satellite: """ Add a new satellite to the manager. Args: satellite_info: Complete satellite information Returns: Satellite: The created satellite instance Raises: SatelliteRegistrationError: If registration fails """ with self._lock: if len(self._satellites) >= self._max_satellites: raise SatelliteRegistrationError( f"Maximum number of satellites ({self._max_satellites}) reached" ) mac_address = satellite_info.mac_address.lower() # Check blacklist if mac_address in self._blacklisted_macs: raise SatelliteRegistrationError(f"MAC address {mac_address} is blacklisted") # Check if already exists if mac_address in self._satellites: pprint(f"Satellite {mac_address} already registered, updating info") existing = self._satellites[mac_address] # Update satellite info existing._info = satellite_info return existing # Create new satellite satellite = create_satellite( satellite_info, self._application, self._network_manager ) # Add to storage self._satellites[mac_address] = satellite self._satellites_by_id[satellite_info.satellite_id] = satellite # Save registration self._save_satellite_registration(satellite_info) # Update statistics self._update_stats() # Trigger event self._trigger_satellite_registered_event(satellite) pprint(f"Satellite added: {satellite_info.alias_name} ({mac_address})") return satellite def remove_satellite(self, mac_address: str, reason: str = "manual_removal") -> bool: """ Remove a satellite from the manager. Args: mac_address: MAC address of satellite to remove reason: Reason for removal Returns: bool: True if satellite was removed """ with self._lock: mac_address = mac_address.lower() if mac_address not in self._satellites: return False satellite = self._satellites[mac_address] # Disconnect if connected if satellite.is_connected: satellite.disconnect(reason=f"removal: {reason}") # Remove from storage del self._satellites[mac_address] if satellite.satellite_id in self._satellites_by_id: del self._satellites_by_id[satellite.satellite_id] # Remove registration file self._remove_satellite_registration(mac_address) # Update statistics self._update_stats() pprint(f"Satellite removed: {satellite.alias_name} ({mac_address})") return True def get_satellite_by_mac(self, mac_address: str) -> Optional[Satellite]: """Get satellite by MAC address.""" with self._lock: return self._satellites.get(mac_address.lower()) def get_satellite_by_id(self, satellite_id: str) -> Optional[Satellite]: """Get satellite by satellite ID.""" with self._lock: return self._satellites_by_id.get(satellite_id) def get_satellite_by_alias(self, alias_name: str) -> Optional[Satellite]: """Get satellite by alias name.""" with self._lock: for satellite in self._satellites.values(): if satellite.alias_name.lower() == alias_name.lower(): return satellite return None def get_satellites_by_room(self, room_id: str) -> List[Satellite]: """Get all satellites in a specific room.""" with self._lock: return [s for s in self._satellites.values() if s.room_id.lower() == room_id.lower()] def get_connected_satellites(self) -> List[Satellite]: """Get all currently connected satellites.""" with self._lock: return [s for s in self._satellites.values() if s.is_connected] def get_registered_satellites(self) -> List[Satellite]: """Get all registered satellites.""" with self._lock: return list(self._satellites.values()) # Bulk Operations def disconnect_all(self, query: Optional[str] = None, reason: str = "bulk_disconnect") -> int: """ Disconnect all satellites or satellites matching a query. Args: query: Optional query to filter satellites reason: Reason for disconnection Returns: int: Number of satellites disconnected """ with self._lock: if query: satellites = self[query] if not isinstance(satellites, list): satellites = [satellites] else: satellites = list(self._satellites.values()) disconnected = 0 for satellite in satellites: if satellite.is_connected: if satellite.disconnect(reason=reason): disconnected += 1 pprint(f"Bulk disconnect: {disconnected} satellites disconnected") return disconnected def reconnect_all(self, query: Optional[str] = None) -> int: """ Reconnect all satellites or satellites matching a query. Args: query: Optional query to filter satellites Returns: int: Number of satellites reconnected """ with self._lock: if query: satellites = self[query] if not isinstance(satellites, list): satellites = [satellites] else: satellites = list(self._satellites.values()) reconnected = 0 for satellite in satellites: if not satellite.is_connected and satellite.status != SatelliteStatus.BLACKLISTED: if satellite.reconnect(): reconnected += 1 pprint(f"Bulk reconnect: {reconnected} satellites reconnected") return reconnected def say_all( self, text: str, query: Optional[str] = None, voice_settings: Optional[Dict[str, Any]] = None ) -> int: """ Send TTS message to all satellites or satellites matching a query. Args: text: Text to speak query: Optional query to filter satellites voice_settings: Optional voice configuration Returns: int: Number of satellites that received the message """ with self._lock: if query: satellites = self[query] if not isinstance(satellites, list): satellites = [satellites] else: satellites = self.get_connected_satellites() sent = 0 for satellite in satellites: if satellite.is_connected: if satellite.say(text, voice_settings): sent += 1 pprint(f"Bulk TTS: '{text[:50]}...' sent to {sent} satellites") return sent def update_all_capabilities( self, capability: SatelliteCapability, add: bool = True, query: Optional[str] = None ) -> int: """ Add or remove a capability from satellites. Args: capability: Capability to add/remove add: True to add, False to remove query: Optional query to filter satellites Returns: int: Number of satellites updated """ with self._lock: if query: satellites = self[query] if not isinstance(satellites, list): satellites = [satellites] else: satellites = list(self._satellites.values()) updated = 0 for satellite in satellites: if add: if not satellite.has_capability(capability): satellite.add_capability(capability) updated += 1 else: if satellite.has_capability(capability): satellite.remove_capability(capability) updated += 1 action = "added" if add else "removed" pprint(f"Capability {capability.value} {action} for {updated} satellites") return updated # Registration Management def enter_registration_mode(self, timeout: Optional[float] = None) -> bool: """ Enter registration mode to allow new satellites to register. Args: timeout: Optional timeout in seconds (uses default if None) Returns: bool: True if registration mode was activated """ with self._lock: if self._registration_mode: pprint("Registration mode is already active") return True timeout = timeout or self._registration_timeout self._registration_mode = True # Set up automatic exit timer if self._registration_timer: self._registration_timer.cancel() self._registration_timer = threading.Timer( timeout, self._exit_registration_mode_auto ) self._registration_timer.start() pprint(f"Registration mode activated for {timeout} seconds") # Trigger event self._trigger_registration_mode_event(True, timeout) return True def exit_registration_mode(self) -> bool: """ Exit registration mode. Returns: bool: True if registration mode was deactivated """ with self._lock: if not self._registration_mode: return True self._registration_mode = False if self._registration_timer: self._registration_timer.cancel() self._registration_timer = None pprint("Registration mode deactivated") # Trigger event self._trigger_registration_mode_event(False, 0) return True def is_registration_mode_active(self) -> bool: """Check if registration mode is active.""" return self._registration_mode def can_register_satellite(self, mac_address: str) -> bool: """ Check if a satellite can be registered. Args: mac_address: MAC address to check Returns: bool: True if registration is allowed """ with self._lock: mac_address = mac_address.lower() # Check if at capacity if len(self._satellites) >= self._max_satellites: return False # Check if blacklisted if mac_address in self._blacklisted_macs: return False # Check if already registered if mac_address in self._satellites: return True # Can update existing registration # Check if registration mode is active or auto-registration is enabled return self._registration_mode or self._enable_auto_registration # Blacklist Management def add_to_blacklist(self, mac_address: str, reason: str = "manual_blacklist") -> bool: """ Add a MAC address to the blacklist. Args: mac_address: MAC address to blacklist reason: Reason for blacklisting Returns: bool: True if added to blacklist """ with self._lock: mac_address = mac_address.lower() if mac_address in self._blacklisted_macs: return True self._blacklisted_macs.add(mac_address) # Disconnect if currently connected if mac_address in self._satellites: satellite = self._satellites[mac_address] if satellite.is_connected: satellite.disconnect(reason=f"blacklisted: {reason}") satellite.set_error(f"Blacklisted: {reason}") # Save blacklist self._save_blacklist() pprint(f"MAC address blacklisted: {mac_address} (reason: {reason})") return True def remove_from_blacklist(self, mac_address: str) -> bool: """ Remove a MAC address from the blacklist. Args: mac_address: MAC address to remove from blacklist Returns: bool: True if removed from blacklist """ with self._lock: mac_address = mac_address.lower() if mac_address not in self._blacklisted_macs: return True self._blacklisted_macs.remove(mac_address) # Save blacklist self._save_blacklist() pprint(f"MAC address removed from blacklist: {mac_address}") return True def is_blacklisted(self, mac_address: str) -> bool: """Check if a MAC address is blacklisted.""" return mac_address.lower() in self._blacklisted_macs def get_blacklisted_macs(self) -> List[str]: """Get all blacklisted MAC addresses.""" with self._lock: return list(self._blacklisted_macs) # Connection Management def handle_satellite_connection( self, mac_address: str, ip_address: str, room_id: str, alias_name: str, version: str = "1.0.0", capabilities: Optional[List[str]] = None, audio_ports: Optional[Dict[str, int]] = None ) -> Tuple[bool, str, Optional[Satellite]]: """ Handle a satellite connection attempt. Args: mac_address: Satellite MAC address ip_address: Satellite IP address room_id: Room identifier alias_name: Human-readable name version: Software version capabilities: List of capability names audio_ports: Audio port information Returns: Tuple[bool, str, Optional[Satellite]]: (success, message, satellite) """ with self._lock: mac_address = mac_address.lower() # Check if can register/connect if not self.can_register_satellite(mac_address): if mac_address in self._blacklisted_macs: return False, "MAC address is blacklisted", None elif len(self._satellites) >= self._max_satellites: return False, "Maximum satellites limit reached", None else: return False, "Registration mode is not active", None # Parse capabilities parsed_capabilities = [] if capabilities: for cap_name in capabilities: try: cap = SatelliteCapability(cap_name.lower()) parsed_capabilities.append(cap) except ValueError: pprint(f"Unknown capability: {cap_name}") # Parse audio ports audio_port_info = AudioPortInfo() if audio_ports: audio_port_info = AudioPortInfo.from_dict(audio_ports) # Get or create satellite satellite = self._satellites.get(mac_address) if satellite is None: # Create new satellite satellite_info = create_satellite_info( satellite_id=str(uuid.uuid4()), mac_address=mac_address, room_id=room_id, alias_name=alias_name, ip_address=ip_address, version=version, capabilities=parsed_capabilities ) satellite = self.add_satellite(satellite_info) else: # Update existing satellite satellite._info.room_id = room_id satellite._info.alias_name = alias_name satellite._info.version = version satellite._info.capabilities = parsed_capabilities satellite.update_ip_address(ip_address) # Mark as connected satellite.set_connected(ip_address, audio_port_info) satellite.update_last_seen() # Update statistics self._update_stats() pprint(f"Satellite connected: {alias_name} ({mac_address}) from {ip_address}") return True, "Connection accepted", satellite def handle_satellite_disconnection( self, mac_address: str, reason: str = "normal_disconnect" ) -> bool: """ Handle a satellite disconnection. Args: mac_address: Satellite MAC address reason: Reason for disconnection Returns: bool: True if handled successfully """ with self._lock: mac_address = mac_address.lower() satellite = self._satellites.get(mac_address) if satellite is None: return True satellite.set_disconnected(reason) # Update statistics self._update_stats() pprint(f"Satellite disconnected: {satellite.alias_name} ({mac_address}) - {reason}") return True # Statistics and Monitoring def get_stats(self) -> SatelliteStats: """Get comprehensive satellite statistics.""" with self._lock: self._update_stats() return self._stats def get_status(self) -> Dict[str, Any]: """Get comprehensive status information.""" with self._lock: stats = self.get_stats() return { "statistics": stats.to_dict(), "configuration": { "max_satellites": self._max_satellites, "auto_registration": self._enable_auto_registration, "registration_timeout": self._registration_timeout, "connection_timeout": self._connection_timeout, "heartbeat_interval": self._heartbeat_interval, "registration_dir": str(self._registration_dir), "blacklist_file": str(self._blacklist_file), }, "registration": { "mode_active": self._registration_mode, "can_register": len(self._satellites) < self._max_satellites, }, "satellites_by_status": { status.value: len([ s for s in self._satellites.values() if s.status == status ]) for status in SatelliteStatus }, "satellites_by_room": self._get_satellites_by_room_stats(), "blacklist": { "count": len(self._blacklisted_macs), "mac_addresses": list(self._blacklisted_macs), }, "capabilities": self._get_capabilities_stats(), "timestamp": datetime.now(timezone.utc).isoformat(), } def _update_stats(self) -> None: """Update internal statistics.""" now = datetime.now(timezone.utc) total_satellites = len(self._satellites) connected_count = 0 registered_count = 0 blacklisted_count = 0 error_count = 0 total_connections = 0 total_messages = 0 total_errors = 0 for satellite in self._satellites.values(): if satellite.status == SatelliteStatus.CONNECTED: connected_count += 1 elif satellite.status == SatelliteStatus.REGISTERED: registered_count += 1 elif satellite.status == SatelliteStatus.BLACKLISTED: blacklisted_count += 1 elif satellite.status == SatelliteStatus.ERROR: error_count += 1 # Get satellite stats sat_status = satellite.get_status() stats = sat_status["statistics"] total_connections += stats.get("connection_count", 0) total_messages += stats.get("message_count", 0) total_errors += stats.get("error_count", 0) blacklisted_count += len(self._blacklisted_macs) uptime = (now - self._start_time).total_seconds() self._stats = SatelliteStats( total_satellites=total_satellites, connected_satellites=connected_count, registered_satellites=registered_count, blacklisted_satellites=blacklisted_count, error_satellites=error_count, total_connections=total_connections, total_messages=total_messages, total_errors=total_errors, uptime_seconds=uptime, last_updated=now ) def _get_satellites_by_room_stats(self) -> Dict[str, int]: """Get satellite count by room.""" room_stats = {} for satellite in self._satellites.values(): room = satellite.room_id room_stats[room] = room_stats.get(room, 0) + 1 return room_stats def _get_capabilities_stats(self) -> Dict[str, int]: """Get capability usage statistics.""" capability_stats = {} for satellite in self._satellites.values(): for capability in satellite.capabilities: cap_name = capability.value capability_stats[cap_name] = capability_stats.get(cap_name, 0) + 1 return capability_stats # File Management def _load_registered_satellites(self) -> None: """Load registered satellites from files.""" if not self._registration_dir.exists(): return loaded_count = 0 for file_path in self._registration_dir.glob("*.json"): try: with open(file_path, 'r') as f: data = json.load(f) satellite_info = SatelliteInfo.from_dict(data) satellite = create_satellite( satellite_info, self._application, self._network_manager ) mac_address = satellite_info.mac_address.lower() self._satellites[mac_address] = satellite self._satellites_by_id[satellite_info.satellite_id] = satellite loaded_count += 1 except Exception as e: pprint(f"Error loading satellite from {file_path}: {e}") if loaded_count > 0: pprint(f"Loaded {loaded_count} registered satellites") def _save_satellite_registration(self, satellite_info: SatelliteInfo) -> None: """Save satellite registration to file.""" filename = f"{satellite_info.mac_address.replace(':', '_')}.json" file_path = self._registration_dir / filename try: with open(file_path, 'w') as f: json.dump(satellite_info.to_dict(), f, indent=2) pprint(f"Satellite registration saved: {filename}") except Exception as e: pprint(f"Error saving satellite registration: {e}") def _remove_satellite_registration(self, mac_address: str) -> None: """Remove satellite registration file.""" filename = f"{mac_address.replace(':', '_')}.json" file_path = self._registration_dir / filename try: if file_path.exists(): file_path.unlink() pprint(f"Satellite registration removed: {filename}") except Exception as e: pprint(f"Error removing satellite registration: {e}") def _load_blacklist(self) -> None: """Load blacklisted MAC addresses.""" if not self._blacklist_file.exists(): return try: with open(self._blacklist_file, 'r') as f: data = json.load(f) blacklist = data.get("blacklisted_macs", []) self._blacklisted_macs = set(mac.lower() for mac in blacklist) if self._blacklisted_macs: pprint(f"Loaded {len(self._blacklisted_macs)} blacklisted MAC addresses") except Exception as e: pprint(f"Error loading blacklist: {e}") def _save_blacklist(self) -> None: """Save blacklisted MAC addresses.""" try: data = { "blacklisted_macs": list(self._blacklisted_macs), "updated_at": datetime.now(timezone.utc).isoformat() } with open(self._blacklist_file, 'w') as f: json.dump(data, f, indent=2) except Exception as e: pprint(f"Error saving blacklist: {e}") # Event Integration def set_event_handler(self, event_handler) -> None: """Set the event handler for satellite events.""" self._event_handler = event_handler def set_network_manager(self, network_manager) -> None: """Set the network manager for satellite communication.""" self._network_manager = network_manager # Update all satellites with network manager with self._lock: for satellite in self._satellites.values(): satellite._network_manager = network_manager def _trigger_satellite_registered_event(self, satellite: Satellite) -> None: """Trigger satellite registered event.""" try: if self._application_ref and self._application_ref(): app = self._application_ref() event_handler = app.get_event_handler() from ..events import EventDataFactory register_data = EventDataFactory.create_event_data( "satellite_registered", satellite_id=satellite.satellite_id, mac_address=satellite.mac_address, room_id=satellite.room_id, alias_name=satellite.alias_name, registration_time=datetime.now(timezone.utc).isoformat() ) event_handler.trigger_event("satellite_registered", register_data) except Exception as e: pprint(f"Error triggering satellite registered event: {e}") def _trigger_registration_mode_event(self, active: bool, timeout: float) -> None: """Trigger registration mode change event.""" try: if self._application_ref and self._application_ref(): app = self._application_ref() event_handler = app.get_event_handler() from ..events import EventDataFactory mode_data = EventDataFactory.create_event_data( "registration_mode_changed", active=active, timeout=timeout, timestamp=datetime.now(timezone.utc).isoformat() ) event_handler.trigger_event("registration_mode_changed", mode_data) except Exception as e: pprint(f"Error triggering registration mode event: {e}") def _exit_registration_mode_auto(self) -> None: """Automatically exit registration mode (timer callback).""" self.exit_registration_mode() pprint("Registration mode automatically deactivated (timeout)") # Cleanup and Maintenance def cleanup_disconnected_satellites(self, max_age_hours: float = 24.0) -> int: """ Remove satellites that have been disconnected for too long. Args: max_age_hours: Maximum age in hours for disconnected satellites Returns: int: Number of satellites cleaned up """ with self._lock: cutoff_time = datetime.now(timezone.utc) - timedelta(hours=max_age_hours) to_remove = [] for mac_address, satellite in self._satellites.items(): if (not satellite.is_connected and satellite._info.last_seen and satellite._info.last_seen < cutoff_time): to_remove.append(mac_address) for mac_address in to_remove: self.remove_satellite(mac_address, "cleanup_old") if to_remove: pprint(f"Cleaned up {len(to_remove)} old disconnected satellites") return len(to_remove) def validate_satellite_states(self) -> int: """ Validate and fix inconsistent satellite states. Returns: int: Number of satellites with fixed states """ with self._lock: fixed_count = 0 for satellite in self._satellites.values(): # Check for inconsistent states if satellite.status == SatelliteStatus.CONNECTED and not satellite._sockets: satellite.set_disconnected("state_validation") fixed_count += 1 elif satellite.status != SatelliteStatus.CONNECTED and satellite._sockets: satellite._sockets.clear() fixed_count += 1 if fixed_count > 0: pprint(f"Fixed {fixed_count} inconsistent satellite states") return fixed_count def shutdown(self) -> None: """Shutdown the satellite manager gracefully.""" with self._lock: pprint("Shutting down satellite manager...") # Exit registration mode self.exit_registration_mode() # Disconnect all satellites disconnected = self.disconnect_all(reason="system_shutdown") # Save final state self._save_blacklist() for satellite in self._satellites.values(): self._save_satellite_registration(satellite._info) pprint(f"Satellite manager shutdown complete ({disconnected} satellites disconnected)") def __str__(self) -> str: """String representation of the satellite manager.""" stats = self.get_stats() return (f"SatelliteManager(total={stats.total_satellites}, " f"connected={stats.connected_satellites}, " f"reg_mode={self._registration_mode})") def __repr__(self) -> str: """Detailed representation of the satellite manager.""" return (f"SatelliteManager(satellites={len(self._satellites)}, " f"max={self._max_satellites}, " f"auto_reg={self._enable_auto_registration}, " f"reg_mode={self._registration_mode})") # Factory function def create_satellite_manager( application, **kwargs ) -> SatelliteManager: """ Create a SatelliteManager instance with default configuration. Args: application: Application container reference **kwargs: Additional configuration parameters Returns: SatelliteManager: Configured satellite manager """ return SatelliteManager(application, **kwargs) # Module exports __all__ = [ "SatelliteManager", "SatelliteManagerError", "SatelliteNotFoundError", "SatelliteRegistrationError", "SatelliteConnectionError", "ConnectionState", "SatelliteStats", "create_satellite_manager", ]