""" Satellite Class Implementation This module implements the Satellite class that represents individual satellites (clients) in the Trixy system. Each satellite instance provides methods for communication, status monitoring, and connection management. The Satellite class implements the interface specified in CLAUDE.md: - Properties: room_id, mac_address, alias_name, IP, sockets, last_conversation - Methods: say(), disconnect(), reconnect(), get_status() - Integration with network system for socket management - Integration with event system for satellite events - Thread-safe operations """ import time import threading import socket from typing import Optional, Dict, Any, List, Union, Tuple from dataclasses import dataclass, field from datetime import datetime, timezone from enum import Enum import json import uuid import weakref # Import network commands for satellite communication try: from ..network.cmd.satellite import ( SatelliteDisconnectCommand, SatelliteStatusCommand, create_satellite_disconnect ) from ..network.cmd.audio import ( TTSPlaybackCommand, AudioStreamCommand ) except ImportError: # Fallback for development SatelliteDisconnectCommand = None SatelliteStatusCommand = None TTSPlaybackCommand = None AudioStreamCommand = None def pprint(message: str) -> None: """Satellite logging function.""" print(f"[SATELLITE] {message}") class SatelliteStatus(Enum): """Status of a satellite connection.""" REGISTERED = "registered" NOT_CONNECTED = "not_connected" CONNECTING = "connecting" CONNECTED = "connected" DISCONNECTING = "disconnecting" ERROR = "error" BLACKLISTED = "blacklisted" class SatelliteCapability(Enum): """Capabilities that a satellite can support.""" WAKEWORD_DETECTION = "wakeword_detection" VOICE_RECOGNITION = "voice_recognition" AUDIO_INPUT = "audio_input" AUDIO_OUTPUT = "audio_output" MUSIC_PLAYBACK = "music_playback" TEXT_TO_SPEECH = "text_to_speech" LED_CONTROL = "led_control" BUTTON_INPUT = "button_input" SENSOR_DATA = "sensor_data" @dataclass class AudioPortInfo: """Information about audio ports for a satellite.""" raw_input_port: Optional[int] = None raw_output_port: Optional[int] = None music_output_port: Optional[int] = None control_port: Optional[int] = None def to_dict(self) -> Dict[str, Any]: """Convert to dictionary representation.""" return { "raw_input_port": self.raw_input_port, "raw_output_port": self.raw_output_port, "music_output_port": self.music_output_port, "control_port": self.control_port, } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'AudioPortInfo': """Create from dictionary representation.""" return cls( raw_input_port=data.get("raw_input_port"), raw_output_port=data.get("raw_output_port"), music_output_port=data.get("music_output_port"), control_port=data.get("control_port"), ) @dataclass class SatelliteInfo: """Complete information about a satellite.""" satellite_id: str mac_address: str room_id: str alias_name: str ip_address: Optional[str] = None version: str = "1.0.0" capabilities: List[SatelliteCapability] = field(default_factory=list) device_info: Dict[str, Any] = field(default_factory=dict) registered_at: Optional[datetime] = None last_seen: Optional[datetime] = None def to_dict(self) -> Dict[str, Any]: """Convert to dictionary representation.""" return { "satellite_id": self.satellite_id, "mac_address": self.mac_address, "room_id": self.room_id, "alias_name": self.alias_name, "ip_address": self.ip_address, "version": self.version, "capabilities": [cap.value for cap in self.capabilities], "device_info": self.device_info, "registered_at": self.registered_at.isoformat() if self.registered_at else None, "last_seen": self.last_seen.isoformat() if self.last_seen else None, } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'SatelliteInfo': """Create from dictionary representation.""" return cls( satellite_id=data["satellite_id"], mac_address=data["mac_address"], room_id=data["room_id"], alias_name=data["alias_name"], ip_address=data.get("ip_address"), version=data.get("version", "1.0.0"), capabilities=[SatelliteCapability(cap) for cap in data.get("capabilities", [])], device_info=data.get("device_info", {}), registered_at=datetime.fromisoformat(data["registered_at"]) if data.get("registered_at") else None, last_seen=datetime.fromisoformat(data["last_seen"]) if data.get("last_seen") else None, ) class SatelliteError(Exception): """Base exception for satellite-related errors.""" pass class SatelliteConnectionError(SatelliteError): """Raised when satellite connection operations fail.""" pass class SatelliteCommunicationError(SatelliteError): """Raised when satellite communication fails.""" pass class Satellite: """ Represents an individual satellite (client) in the Trixy system. This class provides the interface for interacting with a single satellite, including communication, status monitoring, and connection management. The satellite maintains its own connection state and provides thread-safe operations for multi-threaded environments. """ def __init__( self, satellite_info: SatelliteInfo, application, network_manager=None ): """ Initialize a satellite instance. Args: satellite_info: Complete satellite information application: Application container reference network_manager: Network manager for communication """ self._info = satellite_info self._application = application self._application_ref = weakref.ref(application) if application else None self._network_manager = network_manager # Connection state self._status = SatelliteStatus.REGISTERED self._connected_at: Optional[datetime] = None self._last_heartbeat: Optional[datetime] = None self._reconnect_attempts = 0 self._max_reconnect_attempts = 3 # Socket information self._sockets: Dict[str, Any] = {} self._audio_ports = AudioPortInfo() # Conversation tracking self._last_conversation: Optional[Dict[str, Any]] = None self._conversation_history: List[Dict[str, Any]] = [] # Thread safety self._lock = threading.RLock() # Statistics self._stats = { "connection_count": 0, "message_count": 0, "error_count": 0, "last_error": None, "uptime": 0.0, } pprint(f"Satellite created: {self.alias_name} ({self.mac_address})") # Properties as specified in CLAUDE.md @property def room_id(self) -> str: """Get the room ID where this satellite is located.""" return self._info.room_id @room_id.setter def room_id(self, value: str) -> None: """Set the room ID (requires re-registration).""" with self._lock: old_room = self._info.room_id self._info.room_id = value pprint(f"Satellite {self.alias_name} moved from {old_room} to {value}") @property def mac_address(self) -> str: """Get the MAC address of this satellite.""" return self._info.mac_address @property def alias_name(self) -> str: """Get the alias name of this satellite.""" return self._info.alias_name @alias_name.setter def alias_name(self, value: str) -> None: """Set the alias name.""" with self._lock: old_alias = self._info.alias_name self._info.alias_name = value pprint(f"Satellite {old_alias} renamed to {value}") @property def IP(self) -> Optional[str]: """Get the IP address of this satellite.""" return self._info.ip_address @property def ip_address(self) -> Optional[str]: """Get the IP address of this satellite (alternative property name).""" return self._info.ip_address @property def sockets(self) -> Dict[str, Any]: """Get the socket connections for this satellite.""" with self._lock: return self._sockets.copy() @property def last_conversation(self) -> Optional[Dict[str, Any]]: """Get the last conversation data.""" return self._last_conversation @property def satellite_id(self) -> str: """Get the unique satellite ID.""" return self._info.satellite_id @property def version(self) -> str: """Get the satellite software version.""" return self._info.version @property def capabilities(self) -> List[SatelliteCapability]: """Get the satellite capabilities.""" return self._info.capabilities.copy() @property def status(self) -> SatelliteStatus: """Get the current satellite status.""" return self._status @property def is_connected(self) -> bool: """Check if the satellite is currently connected.""" return self._status == SatelliteStatus.CONNECTED @property def is_registered(self) -> bool: """Check if the satellite is registered.""" return self._status != SatelliteStatus.NOT_CONNECTED @property def uptime(self) -> float: """Get the uptime in seconds since connection.""" if not self._connected_at: return 0.0 return (datetime.now(timezone.utc) - self._connected_at).total_seconds() @property def audio_ports(self) -> AudioPortInfo: """Get the audio port information.""" return self._audio_ports # Methods as specified in CLAUDE.md def say(self, text: str, voice_settings: Optional[Dict[str, Any]] = None) -> bool: """ Send text-to-speech audio to this satellite. Args: text: Text to speak voice_settings: Optional voice configuration Returns: bool: True if message was sent successfully """ if not self.is_connected: pprint(f"Cannot send TTS to {self.alias_name}: not connected") return False try: with self._lock: # Create TTS command if available if TTSPlaybackCommand is not None: tts_command = TTSPlaybackCommand( text=text, satellite_id=self.satellite_id, voice_settings=voice_settings or {} ) # Send via network manager if self._network_manager: success = self._network_manager.send_command( self.ip_address, tts_command ) if success: self._stats["message_count"] += 1 pprint(f"TTS sent to {self.alias_name}: '{text[:50]}...'") return True # Fallback: trigger event for TTS processing if self._application_ref and self._application_ref(): app = self._application_ref() event_handler = app.get_event_handler() # Import event data factory try: from ..events import EventDataFactory tts_data = EventDataFactory.create_event_data( "tts_received", conversation_id=str(uuid.uuid4()), text=text, satellite_id=self.satellite_id, voice_settings=voice_settings or {}, audio_data=b"", # Will be filled by TTS plugin ) event_handler.trigger_event("tts_received", tts_data) self._stats["message_count"] += 1 return True except ImportError: pass pprint(f"Failed to send TTS to {self.alias_name}: no communication method available") return False except Exception as e: self._stats["error_count"] += 1 self._stats["last_error"] = str(e) pprint(f"Error sending TTS to {self.alias_name}: {e}") return False def disconnect(self, reason: str = "normal_disconnect", graceful: bool = True) -> bool: """ Disconnect this satellite from the server. Args: reason: Reason for disconnection graceful: Whether to perform graceful disconnect Returns: bool: True if disconnect was initiated successfully """ if not self.is_connected: pprint(f"Satellite {self.alias_name} is not connected") return True try: with self._lock: pprint(f"Disconnecting satellite {self.alias_name} (reason: {reason})") self._status = SatelliteStatus.DISCONNECTING # Send disconnect command if available if SatelliteDisconnectCommand is not None and self._network_manager: disconnect_cmd = create_satellite_disconnect( self.satellite_id, reason=reason, graceful=graceful ) self._network_manager.send_command( self.ip_address, disconnect_cmd ) # Close sockets for socket_name, socket_obj in self._sockets.items(): try: if hasattr(socket_obj, 'close'): socket_obj.close() pprint(f"Closed {socket_name} socket for {self.alias_name}") except Exception as e: pprint(f"Error closing {socket_name} socket: {e}") self._sockets.clear() self._status = SatelliteStatus.NOT_CONNECTED self._connected_at = None self._last_heartbeat = None # Trigger disconnect event self._trigger_disconnect_event(reason, graceful) pprint(f"Satellite {self.alias_name} disconnected successfully") return True except Exception as e: self._status = SatelliteStatus.ERROR self._stats["error_count"] += 1 self._stats["last_error"] = str(e) pprint(f"Error disconnecting satellite {self.alias_name}: {e}") return False def reconnect(self) -> bool: """ Attempt to reconnect this satellite. Returns: bool: True if reconnection was initiated successfully """ if self.is_connected: pprint(f"Satellite {self.alias_name} is already connected") return True if self._reconnect_attempts >= self._max_reconnect_attempts: pprint(f"Max reconnection attempts reached for {self.alias_name}") return False try: with self._lock: pprint(f"Reconnecting satellite {self.alias_name} (attempt {self._reconnect_attempts + 1})") self._status = SatelliteStatus.CONNECTING self._reconnect_attempts += 1 # In a real implementation, this would establish new socket connections # For now, we simulate reconnection by updating status # The actual reconnection logic would be handled by the network manager if self._network_manager: # Attempt to re-establish connection success = self._network_manager.reconnect_satellite(self) if success: self._status = SatelliteStatus.CONNECTED self._connected_at = datetime.now(timezone.utc) self._reconnect_attempts = 0 self._stats["connection_count"] += 1 # Trigger connect event self._trigger_connect_event() pprint(f"Satellite {self.alias_name} reconnected successfully") return True self._status = SatelliteStatus.ERROR pprint(f"Failed to reconnect satellite {self.alias_name}") return False except Exception as e: self._status = SatelliteStatus.ERROR self._stats["error_count"] += 1 self._stats["last_error"] = str(e) pprint(f"Error reconnecting satellite {self.alias_name}: {e}") return False def get_status(self) -> Dict[str, Any]: """ Get comprehensive status information for this satellite. Returns: Dict containing all status information """ with self._lock: return { "satellite_info": self._info.to_dict(), "connection": { "status": self._status.value, "connected_at": self._connected_at.isoformat() if self._connected_at else None, "last_heartbeat": self._last_heartbeat.isoformat() if self._last_heartbeat else None, "uptime_seconds": self.uptime, "reconnect_attempts": self._reconnect_attempts, "ip_address": self.ip_address, }, "sockets": { name: { "connected": bool(sock), "type": type(sock).__name__ if sock else None } for name, sock in self._sockets.items() }, "audio_ports": self._audio_ports.to_dict(), "conversation": { "last_conversation": self._last_conversation, "conversation_count": len(self._conversation_history), }, "statistics": self._stats.copy(), "capabilities": [cap.value for cap in self.capabilities], "timestamps": { "registered_at": self._info.registered_at.isoformat() if self._info.registered_at else None, "last_seen": self._info.last_seen.isoformat() if self._info.last_seen else None, "status_checked_at": datetime.now(timezone.utc).isoformat(), } } # Additional utility methods def update_last_seen(self) -> None: """Update the last seen timestamp.""" with self._lock: self._info.last_seen = datetime.now(timezone.utc) self._last_heartbeat = self._info.last_seen def update_ip_address(self, ip_address: str) -> None: """Update the IP address.""" with self._lock: old_ip = self._info.ip_address self._info.ip_address = ip_address if old_ip != ip_address: pprint(f"IP address updated for {self.alias_name}: {old_ip} -> {ip_address}") def add_socket(self, socket_name: str, socket_obj: Any) -> None: """Add a socket connection.""" with self._lock: self._sockets[socket_name] = socket_obj pprint(f"Socket '{socket_name}' added for {self.alias_name}") def remove_socket(self, socket_name: str) -> None: """Remove a socket connection.""" with self._lock: if socket_name in self._sockets: del self._sockets[socket_name] pprint(f"Socket '{socket_name}' removed for {self.alias_name}") def update_audio_ports(self, audio_ports: AudioPortInfo) -> None: """Update the audio port information.""" with self._lock: self._audio_ports = audio_ports pprint(f"Audio ports updated for {self.alias_name}") def record_conversation(self, conversation_data: Dict[str, Any]) -> None: """Record a conversation for this satellite.""" with self._lock: self._last_conversation = conversation_data self._conversation_history.append(conversation_data) # Keep only recent conversations if len(self._conversation_history) > 100: self._conversation_history = self._conversation_history[-100:] def set_connected(self, ip_address: str, audio_ports: AudioPortInfo) -> None: """Mark satellite as connected.""" with self._lock: self._status = SatelliteStatus.CONNECTED self._info.ip_address = ip_address self._audio_ports = audio_ports self._connected_at = datetime.now(timezone.utc) self._reconnect_attempts = 0 self._stats["connection_count"] += 1 self._trigger_connect_event() def set_disconnected(self, reason: str = "unknown") -> None: """Mark satellite as disconnected.""" with self._lock: was_connected = self._status == SatelliteStatus.CONNECTED self._status = SatelliteStatus.NOT_CONNECTED self._connected_at = None self._sockets.clear() if was_connected: self._trigger_disconnect_event(reason, False) def set_error(self, error_message: str) -> None: """Mark satellite as in error state.""" with self._lock: self._status = SatelliteStatus.ERROR self._stats["error_count"] += 1 self._stats["last_error"] = error_message pprint(f"Satellite {self.alias_name} entered error state: {error_message}") def has_capability(self, capability: SatelliteCapability) -> bool: """Check if satellite has a specific capability.""" return capability in self._info.capabilities def add_capability(self, capability: SatelliteCapability) -> None: """Add a capability to this satellite.""" with self._lock: if capability not in self._info.capabilities: self._info.capabilities.append(capability) pprint(f"Capability {capability.value} added to {self.alias_name}") def remove_capability(self, capability: SatelliteCapability) -> None: """Remove a capability from this satellite.""" with self._lock: if capability in self._info.capabilities: self._info.capabilities.remove(capability) pprint(f"Capability {capability.value} removed from {self.alias_name}") def _trigger_connect_event(self) -> None: """Trigger satellite connected event.""" try: if self._application_ref and self._application_ref(): app = self._application_ref() event_handler = app.get_event_handler() from ..events import EventDataFactory connect_data = EventDataFactory.create_event_data( "satellite_connected", satellite_id=self.satellite_id, mac_address=self.mac_address, room_id=self.room_id, alias_name=self.alias_name, ip_address=self.ip_address, connection_time=datetime.now(timezone.utc).isoformat() ) event_handler.trigger_event("satellite_connected", connect_data) except Exception as e: pprint(f"Error triggering connect event for {self.alias_name}: {e}") def _trigger_disconnect_event(self, reason: str, graceful: bool) -> None: """Trigger satellite disconnected event.""" try: if self._application_ref and self._application_ref(): app = self._application_ref() event_handler = app.get_event_handler() from ..events import EventDataFactory disconnect_data = EventDataFactory.create_event_data( "satellite_disconnected", satellite_id=self.satellite_id, mac_address=self.mac_address, room_id=self.room_id, alias_name=self.alias_name, reason=reason, graceful_disconnect=graceful, disconnection_time=datetime.now(timezone.utc).isoformat() ) event_handler.trigger_event("satellite_disconnected", disconnect_data) except Exception as e: pprint(f"Error triggering disconnect event for {self.alias_name}: {e}") def to_dict(self) -> Dict[str, Any]: """Convert satellite to dictionary representation.""" return { "info": self._info.to_dict(), "status": self._status.value, "audio_ports": self._audio_ports.to_dict(), "statistics": self._stats.copy(), "connected_at": self._connected_at.isoformat() if self._connected_at else None, } def __str__(self) -> str: """String representation of the satellite.""" return f"Satellite(alias='{self.alias_name}', room='{self.room_id}', status='{self._status.value}')" def __repr__(self) -> str: """Detailed representation of the satellite.""" return (f"Satellite(id='{self.satellite_id}', alias='{self.alias_name}', " f"room='{self.room_id}', mac='{self.mac_address}', status='{self._status.value}')") # Factory functions def create_satellite_info( satellite_id: str, mac_address: str, room_id: str, alias_name: str, ip_address: Optional[str] = None, version: str = "1.0.0", capabilities: Optional[List[SatelliteCapability]] = None, device_info: Optional[Dict[str, Any]] = None ) -> SatelliteInfo: """ Create a SatelliteInfo instance. Args: satellite_id: Unique satellite identifier mac_address: MAC address room_id: Room identifier alias_name: Human-readable alias ip_address: IP address version: Software version capabilities: List of capabilities device_info: Additional device information Returns: SatelliteInfo: Configured satellite information """ return SatelliteInfo( satellite_id=satellite_id, mac_address=mac_address, room_id=room_id, alias_name=alias_name, ip_address=ip_address, version=version, capabilities=capabilities or [], device_info=device_info or {}, registered_at=datetime.now(timezone.utc) ) def create_satellite( satellite_info: SatelliteInfo, application, network_manager=None ) -> Satellite: """ Create a Satellite instance. Args: satellite_info: Satellite information application: Application container network_manager: Network manager for communication Returns: Satellite: Configured satellite instance """ return Satellite(satellite_info, application, network_manager) # Module exports __all__ = [ "Satellite", "SatelliteError", "SatelliteConnectionError", "SatelliteCommunicationError", "SatelliteStatus", "SatelliteCapability", "AudioPortInfo", "SatelliteInfo", "create_satellite_info", "create_satellite", ]