# -*- coding: utf-8 -*- """ SatelliteManager für die Verwaltung aller verbundenen Satellites. """ import asyncio from typing import Iterator, overload, TYPE_CHECKING from trixy_core.satellite.satellite import Satellite, ConnectionState from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn if TYPE_CHECKING: from trixy_core.application import IApplication class SatelliteManager: """ Verwaltet alle registrierten und verbundenen Satellites. Bietet: - Registry aller Satellites - Suche nach Room, MAC, Verbindungsstatus - Index-Zugriff mit __getitem__ - Verbindungsmanagement """ def __init__(self, application: "IApplication") -> None: """ Initialisiert den SatelliteManager. Args: application: Referenz zur Hauptanwendung """ self._application = application self._satellites: dict[str, Satellite] = {} self._mac_index: dict[str, str] = {} # MAC -> Satellite-ID self._room_index: dict[str, set[str]] = {} # Room -> Set von Satellite-IDs @property def count(self) -> int: """Anzahl registrierter Satellites.""" return len(self._satellites) @property def connected_count(self) -> int: """Anzahl verbundener Satellites.""" return sum(1 for s in self._satellites.values() if s.is_connected) def add(self, satellite: Satellite) -> None: """ Fügt einen Satellite hinzu. Args: satellite: Der hinzuzufügende Satellite """ # Application-Referenz setzen für Event-Zugriff satellite.application = self._application self._satellites[satellite.id] = satellite # MAC-Index aktualisieren if satellite.mac_address: self._mac_index[satellite.mac_address] = satellite.id # Room-Index aktualisieren if satellite.room_id: if satellite.room_id not in self._room_index: self._room_index[satellite.room_id] = set() self._room_index[satellite.room_id].add(satellite.id) pdebug(f"Satellite hinzugefügt: {satellite.alias} ({satellite.id})") def remove(self, satellite_id: str) -> Satellite | None: """ Entfernt einen Satellite. Args: satellite_id: ID des zu entfernenden Satellites Returns: Der entfernte Satellite oder None """ satellite = self._satellites.pop(satellite_id, None) if satellite: # Indizes bereinigen if satellite.mac_address in self._mac_index: del self._mac_index[satellite.mac_address] if satellite.room_id in self._room_index: self._room_index[satellite.room_id].discard(satellite_id) if not self._room_index[satellite.room_id]: del self._room_index[satellite.room_id] pdebug(f"Satellite entfernt: {satellite.alias} ({satellite_id})") return satellite def get(self, satellite_id: str) -> Satellite | None: """ Gibt einen Satellite nach ID zurück. Args: satellite_id: ID des Satellites Returns: Satellite oder None """ return self._satellites.get(satellite_id) def get_by_mac(self, mac_address: str) -> Satellite | None: """ Gibt einen Satellite nach MAC-Adresse zurück. Args: mac_address: MAC-Adresse Returns: Satellite oder None """ satellite_id = self._mac_index.get(mac_address) if satellite_id: return self._satellites.get(satellite_id) return None def get_by_room(self, room_id: str) -> list[Satellite]: """ Gibt alle Satellites in einem Raum zurück. Args: room_id: Raum-Kennung Returns: Liste der Satellites im Raum """ satellite_ids = self._room_index.get(room_id, set()) return [self._satellites[sid] for sid in satellite_ids if sid in self._satellites] def get_connected(self) -> list[Satellite]: """Gibt alle verbundenen Satellites zurück.""" return [s for s in self._satellites.values() if s.is_connected] def get_disconnected(self) -> list[Satellite]: """Gibt alle getrennten Satellites zurück.""" return [s for s in self._satellites.values() if not s.is_connected] def get_all(self) -> list[Satellite]: """Gibt alle Satellites zurück.""" return list(self._satellites.values()) @overload def __getitem__(self, key: int) -> Satellite: ... @overload def __getitem__(self, key: str) -> Satellite | list[Satellite]: ... def __getitem__(self, key: int | str) -> Satellite | list[Satellite]: """ Index-Zugriff auf Satellites. Unterstützt: - Integer: Zugriff nach Position - String mit ID: Direkter ID-Zugriff - String mit Selektor: "room:wohnzimmer", "mac:AA:BB:CC:..." Args: key: Index, ID oder Selektor Returns: Einzelner Satellite oder Liste Raises: KeyError: Wenn nicht gefunden IndexError: Bei ungültigem Index """ if isinstance(key, int): satellites = list(self._satellites.values()) if key < 0 or key >= len(satellites): raise IndexError(f"Satellite-Index außerhalb des Bereichs: {key}") return satellites[key] # String-Selektoren if ":" in key: selector, value = key.split(":", 1) selector = selector.lower() if selector == "room": result = self.get_by_room(value) if not result: raise KeyError(f"Keine Satellites im Raum: {value}") return result elif selector == "mac": satellite = self.get_by_mac(value) if satellite is None: raise KeyError(f"Kein Satellite mit MAC: {value}") return satellite elif selector == "id": satellite = self._satellites.get(value) if satellite is None: raise KeyError(f"Kein Satellite mit ID: {value}") return satellite # Direkter ID-Zugriff if key in self._satellites: return self._satellites[key] raise KeyError(f"Satellite nicht gefunden: {key}") def __contains__(self, key: str) -> bool: """Prüft, ob ein Satellite existiert.""" return key in self._satellites or key in self._mac_index def __iter__(self) -> Iterator[Satellite]: """Iterator über alle Satellites.""" return iter(self._satellites.values()) def __len__(self) -> int: """Anzahl der Satellites.""" return len(self._satellites) async def disconnect(self, satellite_id: str, reason: str = "") -> bool: """ Trennt einen einzelnen Satellite. Args: satellite_id: ID des Satellites reason: Grund für die Trennung Returns: True bei Erfolg """ satellite = self._satellites.get(satellite_id) if satellite is None: return False await satellite.disconnect(reason) pinfo(f"Satellite getrennt: {satellite.alias} - {reason}") return True async def disconnect_all(self, reason: str = "Server shutdown") -> int: """ Trennt alle verbundenen Satellites. Args: reason: Grund für die Trennung Returns: Anzahl getrennter Satellites """ count = 0 for satellite in self.get_connected(): await satellite.disconnect(reason) count += 1 pinfo(f"{count} Satellites getrennt") return count async def reconnect(self, satellite_id: str) -> bool: """ Fordert einen Satellite zur Wiederverbindung auf. Args: satellite_id: ID des Satellites Returns: True bei Erfolg """ satellite = self._satellites.get(satellite_id) if satellite is None: return False # Implementierung abhängig von der Netzwerk-Schicht # Hier nur Zustandsänderung satellite.state = ConnectionState.CONNECTING pwarn(f"Wiederverbindung angefordert: {satellite.alias}") return True async def reconnect_all(self) -> int: """ Fordert alle getrennten Satellites zur Wiederverbindung auf. Returns: Anzahl angeforderter Wiederverbindungen """ count = 0 for satellite in self.get_disconnected(): await self.reconnect(satellite.id) count += 1 return count def find( self, room: str | None = None, alias: str | None = None, connected: bool | None = None ) -> list[Satellite]: """ Sucht Satellites nach Kriterien. Args: room: Filter nach Raum alias: Filter nach Alias (Teilübereinstimmung) connected: Filter nach Verbindungsstatus Returns: Liste passender Satellites """ result: list[Satellite] = [] for satellite in self._satellites.values(): if room is not None and satellite.room_id != room: continue if alias is not None and alias.lower() not in satellite.alias.lower(): continue if connected is not None and satellite.is_connected != connected: continue result.append(satellite) return result async def say( self, text: str, room: str | None = None, voice: str = "default", ) -> bool: """ Spricht Text auf allen (oder bestimmten) Satellites. Löst das 'tts_request' Event aus, das von einem TTS-Plugin verarbeitet wird. Args: text: Der zu sprechende Text. room: Nur Satellites in diesem Raum (optional). voice: Stimme/Profil für TTS. Returns: True wenn Event ausgelöst wurde. Example: # Auf allen Satellites await satellite_manager.say("Guten Morgen") # Nur im Wohnzimmer await satellite_manager.say("Willkommen", room="wohnzimmer") """ from trixy_core.satellite.api import get_satellite_api api = get_satellite_api(self._application) return await api.say(text, room=room, voice=voice) async def say_raw( self, audio_data: bytes, room: str | None = None, ) -> int: """ Sendet rohe Audio-Daten an Satellites. Args: audio_data: Audio-Bytes (16KHz, 16-bit, mono). room: Nur Satellites in diesem Raum (optional). Returns: Anzahl erfolgreich gesendeter Satellites. Example: count = await satellite_manager.say_raw(audio_bytes) pinfo(f"Audio an {count} Satellites gesendet") """ targets = self.get_by_room(room) if room else self.get_connected() count = 0 for satellite in targets: if satellite.is_connected: if await satellite.say(audio_data): count += 1 return count async def play( self, source: str, room: str | None = None, volume: float = 1.0, ): """ Startet Audio/Musik-Streaming auf Satellites. Args: source: Audio-Quelle (URL oder Dateipfad). room: Nur Satellites in diesem Raum (optional). volume: Lautstärke (0.0-1.0). Returns: StreamSession für Kontrolle des Streams. Example: # Musik auf allen Satellites stream = await satellite_manager.play("http://radio.mp3") # Später: Satellite hinzufügen await stream.add_satellite("küche") """ from trixy_core.satellite.api import get_satellite_api api = get_satellite_api(self._application) return await api.play(source, room=room, volume=volume)