Trixy7/trixy_core/satellite/satellite.py

# -*- coding: utf-8 -*-
"""
Satellite-Klasse für die Verwaltung einzelner Satellites.
"""

import asyncio
import time
import uuid
import wave
from dataclasses import dataclass, field
from datetime import datetime
from enum import IntEnum, auto
from typing import Any, TYPE_CHECKING

from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn

if TYPE_CHECKING:
    from asyncio import StreamWriter
    from trixy_core.application import IApplication


class ConnectionState(IntEnum):
    """Verbindungszustand eines Satellites."""
    DISCONNECTED = auto()
    CONNECTING = auto()
    CONNECTED = auto()
    AUTHENTICATED = auto()
    DISCONNECTING = auto()
    ERROR = auto()


@dataclass
class SatelliteSocket:
    """Referenzen zu den Socket-Verbindungen."""
    command: Any = None  # StreamWriter für Command-Socket
    audio_in: Any = None  # StreamWriter für Audio-Input
    audio_out: Any = None  # StreamWriter für Audio-Output
    music_out: Any = None  # StreamWriter für Music-Output


class Satellite:
    """
    Repräsentiert einen verbundenen Satellite.

    Speichert alle Informationen über einen Satellite und
    bietet Methoden zur Kommunikation.
    """

    def __init__(
        self,
        satellite_id: str | None = None,
        room_id: str = "",
        mac_address: str = "",
        alias: str = "",
        ip_address: str = "",
        application: "IApplication | None" = None
    ) -> None:
        """
        Initialisiert einen Satellite.

        Args:
            satellite_id: Eindeutige ID (wird generiert wenn None)
            room_id: Raum-Kennung
            mac_address: MAC-Adresse des Satellites
            alias: Anzeigename
            ip_address: IP-Adresse
            application: Referenz zur Hauptanwendung (für Event-Zugriff)
        """
        self._id = satellite_id or str(uuid.uuid4())
        self._room_id = room_id
        self._mac_address = mac_address
        self._alias = alias
        self._ip_address = ip_address
        self._application = application

        self._state = ConnectionState.DISCONNECTED
        self._sockets = SatelliteSocket()
        self._conversation_id: str | None = None

        self._connected_at: datetime | None = None
        self._last_heartbeat: datetime | None = None
        self._version: str = ""
        self._metadata: dict[str, Any] = {}

        # Request/Response-Korrelation (fuer Config-Proxy)
        self._pending_responses: dict[str, asyncio.Future] = {}

        # Wakeword-Steuerung
        self._wakeword_paused: bool = False
        self._wakeword_active: bool = True

        # Live-Audio-Streaming
        self._streaming_active: bool = False
        self._stream_id: str = ""
        self._stream_target: "Satellite | None" = None
        self._stream_start_time: float = 0.0
        self._stream_bytes_forwarded: int = 0
        self._stream_timeout_handle: asyncio.TimerHandle | None = None
        self._stream_listener_registered: bool = False

        # Stream-Aufnahme (Disk-basiert, nicht RAM)
        self._stream_recording_active: bool = False
        self._stream_recording_filename: str = ""
        self._stream_recording_wav: Any = None  # wave.Wave_write
        self._stream_recording_start_time: float = 0.0
        self._stream_recording_bytes: int = 0

        # Server-gesteuerte Aufnahme
        self._recording_active: bool = False
        self._recording_id: str = ""
        self._recording_buffer: bytearray = bytearray()
        self._recording_start_time: float = 0.0
        self._recording_filename: str | None = None
        self._recording_timeout_handle: asyncio.TimerHandle | None = None
        self._recording_listener_registered: bool = False

    @property
    def id(self) -> str:
        """Eindeutige Satellite-ID."""
        return self._id

    @property
    def application(self) -> "IApplication | None":
        """Referenz zur Hauptanwendung."""
        return self._application

    @application.setter
    def application(self, value: "IApplication | None") -> None:
        """Setzt die Application-Referenz."""
        self._application = value

    @property
    def room_id(self) -> str:
        """Raum-Kennung."""
        return self._room_id

    @room_id.setter
    def room_id(self, value: str) -> None:
        self._room_id = value

    @property
    def mac_address(self) -> str:
        """MAC-Adresse."""
        return self._mac_address

    @property
    def alias(self) -> str:
        """Anzeigename."""
        return self._alias

    @alias.setter
    def alias(self, value: str) -> None:
        self._alias = value

    @property
    def ip_address(self) -> str:
        """IP-Adresse."""
        return self._ip_address

    @ip_address.setter
    def ip_address(self, value: str) -> None:
        self._ip_address = value

    @property
    def state(self) -> ConnectionState:
        """Aktueller Verbindungszustand."""
        return self._state

    @state.setter
    def state(self, value: ConnectionState) -> None:
        self._state = value

    @property
    def is_connected(self) -> bool:
        """Prüft, ob der Satellite verbunden ist."""
        return self._state in (ConnectionState.CONNECTED, ConnectionState.AUTHENTICATED)

    @property
    def sockets(self) -> SatelliteSocket:
        """Socket-Referenzen."""
        return self._sockets

    @property
    def conversation_id(self) -> str | None:
        """Aktuelle Konversations-ID."""
        return self._conversation_id

    @conversation_id.setter
    def conversation_id(self, value: str | None) -> None:
        self._conversation_id = value

    @property
    def has_active_conversation(self) -> bool:
        """Prüft, ob eine aktive Konversation läuft."""
        return self._conversation_id is not None

    @property
    def connected_at(self) -> datetime | None:
        """Zeitpunkt der Verbindung."""
        return self._connected_at

    @property
    def last_heartbeat(self) -> datetime | None:
        """Letzter Heartbeat."""
        return self._last_heartbeat

    @property
    def version(self) -> str:
        """Client-Version."""
        return self._version

    @version.setter
    def version(self, value: str) -> None:
        self._version = value

    @property
    def metadata(self) -> dict[str, Any]:
        """Zusätzliche Metadaten."""
        return self._metadata

    @property
    def wakeword_paused(self) -> bool:
        """Wakeword-Erkennung extern pausiert."""
        return self._wakeword_paused

    @wakeword_paused.setter
    def wakeword_paused(self, value: bool) -> None:
        """Setzt externen Pause-Zustand. Sendet automatisch Command."""
        if self._wakeword_paused == value:
            return
        self._wakeword_paused = value
        if value:
            from trixy_core.network.cmd import WakewordPause
            cmd = WakewordPause(reason="server-controlled")
        else:
            from trixy_core.network.cmd import WakewordResume
            cmd = WakewordResume()
        asyncio.ensure_future(self.send_command(cmd))

    @property
    def wakeword_active(self) -> bool:
        """Wakeword-Erkennung aktiv (Modelle geladen)."""
        return self._wakeword_active

    @wakeword_active.setter
    def wakeword_active(self, value: bool) -> None:
        """Aktiviert/Deaktiviert Wakeword-Erkennung. Sendet automatisch Command."""
        if self._wakeword_active == value:
            return
        self._wakeword_active = value
        if value:
            from trixy_core.network.cmd import WakewordControlStart
            cmd = WakewordControlStart()
        else:
            from trixy_core.network.cmd import WakewordControlStop
            cmd = WakewordControlStop(reason="server-controlled")
        asyncio.ensure_future(self.send_command(cmd))

    def set_connected(self) -> None:
        """Markiert den Satellite als verbunden."""
        self._state = ConnectionState.CONNECTED
        self._connected_at = datetime.now()
        self.update_heartbeat()

    def update_heartbeat(self) -> None:
        """Aktualisiert den Heartbeat-Zeitstempel."""
        self._last_heartbeat = datetime.now()

    async def say(self, audio_data: bytes, chunk_size: int = 32768) -> bool:
        """
        Sendet rohe Audio-Daten zum Satellite in Chunks.

        Args:
            audio_data: Audio-Bytes (16KHz, 16-bit, mono)
            chunk_size: Größe der Chunks in Bytes (default: 32KB)

        Returns:
            True bei Erfolg
        """
        import asyncio
        from trixy_core.utils.debug import pdebug

        if not self.is_connected or self._sockets.audio_out is None:
            return False

        try:
            writer: "StreamWriter" = self._sockets.audio_out
            offset = 0
            chunk_count = 0

            while offset < len(audio_data):
                chunk = audio_data[offset:offset + chunk_size]
                writer.write(chunk)
                await writer.drain()
                chunk_count += 1
                offset += chunk_size
                # Minimale Pause für Flow-Control (nicht zu lang!)
                if offset < len(audio_data):
                    await asyncio.sleep(0.001)

            pdebug(f"[SERVER] Audio gesendet: {len(audio_data)} bytes in {chunk_count} Chunks")
            return True
        except Exception as e:
            from trixy_core.utils.debug import perror
            perror(f"[SERVER] Audio senden fehlgeschlagen: {e}")
            return False

    async def speak(
        self,
        text: str,
        voice: str = "default",
        speed: float = 1.0,
        volume: float = 1.0,
    ) -> bool:
        """
        Spricht Text über TTS auf diesem Satellite.

        Löst das 'tts_request' Event aus, das von einem TTS-Plugin
        verarbeitet wird.

        Args:
            text: Der zu sprechende Text.
            voice: Stimme/Profil für TTS.
            speed: Sprechgeschwindigkeit.
            volume: Lautstärke.

        Returns:
            True wenn Event ausgelöst wurde.

        Example:
            satellite = satellite_manager[0]
            await satellite.speak("Hallo Welt")
        """
        if not self._application:
            return False

        from trixy_core.events.event_data.basic import TTSRequest
        from trixy_core.utils.template_formatter import format_template

        # Template-Platzhalter aufloesen
        text = format_template(text, application=self._application)

        request_id = str(uuid.uuid4())
        event_data = TTSRequest(
            request_id=request_id,
            satellite_id=self._id,
            text=text,
            voice=voice,
            speed=speed,
            volume=volume,
            source="satellite.speak"
        )

        await self._application.events.trigger("tts_request", event_data)
        return not event_data.is_cancelled()

    async def play(
        self,
        source: str,
        volume: float = 1.0,
    ) -> bool:
        """
        Startet Audio/Musik-Wiedergabe auf diesem Satellite.

        Args:
            source: Audio-Quelle (URL oder Dateipfad).
            volume: Lautstärke.

        Returns:
            True wenn gestartet.

        Example:
            await satellite.play("http://radio.mp3")
        """
        if not self._application:
            return False

        from trixy_core.events.event_data.basic import StreamStart

        event_data = StreamStart(
            satellite_id=self._id,
            source=source,
            volume=volume,
            stream_type="music"
        )
        event_data.metadata["origin"] = "satellite.play"

        await self._application.events.trigger("stream_start", event_data)
        return not event_data.is_cancelled()

    async def stop_playback(self) -> bool:
        """
        Stoppt alle Wiedergabe auf diesem Satellite.

        Returns:
            True bei Erfolg.
        """
        if not self._application:
            return False

        from trixy_core.events.event_data.basic import StreamStop

        event_data = StreamStop(
            satellite_id=self._id,
            stream_type="all",
            source="satellite.stop_playback"
        )

        await self._application.events.trigger("stream_stop", event_data)
        return not event_data.is_cancelled()

    async def send_command(self, message: Any) -> bool:
        """
        Sendet einen Befehl zum Satellite.

        Args:
            message: Das zu sendende Command-Objekt

        Returns:
            True bei Erfolg
        """
        if not self.is_connected or self._sockets.command is None:
            return False

        try:
            from trixy_core.network.protocol import TrixyProtocol, ProtocolFlags
            from trixy_core.utils.debug import pdebug

            protocol = TrixyProtocol()
            data = protocol.serialize(message, ProtocolFlags.PICKLE)

            writer: "StreamWriter" = self._sockets.command
            writer.write(data)
            await writer.drain()

            pdebug(f"Command gesendet an {self._alias}: {type(message).__name__}")
            return True
        except Exception as e:
            from trixy_core.utils.debug import perror
            perror(f"Command senden fehlgeschlagen: {e}")
            return False

    async def request_and_wait(
        self, message: Any, timeout: float = 10.0
    ) -> Any | None:
        """
        Sendet eine Anfrage und wartet auf die Antwort.

        Nutzt message_id fuer Request/Response-Korrelation.

        Args:
            message: Das zu sendende Request-Objekt (muss message_id haben)
            timeout: Maximale Wartezeit in Sekunden

        Returns:
            Die Antwort-Nachricht oder None bei Timeout/Fehler
        """
        message_id = getattr(message, "message_id", "")
        if not message_id:
            return None

        future: asyncio.Future = asyncio.get_event_loop().create_future()
        self._pending_responses[message_id] = future

        try:
            success = await self.send_command(message)
            if not success:
                return None
            return await asyncio.wait_for(future, timeout=timeout)
        except asyncio.TimeoutError:
            pdebug(f"Request-Timeout fuer {self._alias}: {type(message).__name__}")
            return None
        except Exception as e:
            perror(f"Request fehlgeschlagen fuer {self._alias}: {e}")
            return None
        finally:
            self._pending_responses.pop(message_id, None)

    def resolve_pending(self, message_id: str, response: Any) -> bool:
        """
        Loest eine wartende Anfrage mit der empfangenen Antwort auf.

        Args:
            message_id: Die korrelierte message_id
            response: Die Antwort-Nachricht

        Returns:
            True wenn eine wartende Anfrage aufgeloest wurde
        """
        future = self._pending_responses.get(message_id)
        if future and not future.done():
            future.set_result(response)
            return True
        return False

    # === Server-gesteuerte Aufnahme ===

    @property
    def is_recording(self) -> bool:
        """Prüft ob eine server-gesteuerte Aufnahme aktiv ist."""
        return self._recording_active

    @property
    def recording_id(self) -> str:
        """Aktuelle Recording-ID."""
        return self._recording_id

    async def start_recording(
        self, timeout: float = 300.0, filename: str | None = None
    ) -> bool:
        """
        Startet eine server-gesteuerte Aufnahme auf diesem Satellite.

        Args:
            timeout: Maximale Aufnahmedauer in Sekunden
            filename: Optionaler Dateiname für WAV-Export

        Returns:
            True wenn Aufnahme gestartet wurde
        """
        if not self._application:
            perror("Keine Application-Referenz für Recording")
            return False

        if not self.is_connected:
            perror(f"Satellite {self._alias} nicht verbunden")
            return False

        if self._recording_active:
            pwarn(f"Satellite {self._alias} nimmt bereits auf")
            return False

        if self._streaming_active:
            pwarn(f"Satellite {self._alias} hat aktiven Stream — Aufnahme nicht möglich")
            return False

        if self.has_active_conversation:
            pwarn(f"Satellite {self._alias} hat aktive Konversation")
            return False

        # Recording-ID generieren
        self._recording_id = f"rec-{uuid.uuid4().hex[:12]}"
        self._recording_active = True
        self._recording_buffer = bytearray()
        self._recording_start_time = time.monotonic()
        self._recording_filename = filename

        # Event-Listener für audio_input_received registrieren
        self._application.events.register(
            "audio_input_received", self._on_recording_audio_received
        )

        # Event-Listener für satellite_record_stopped (Client-seitige Beendigung)
        self._application.events.register(
            "satellite_record_stopped", self._on_satellite_record_stopped_event
        )
        self._recording_listener_registered = True

        # Timeout-Timer starten
        loop = asyncio.get_event_loop()
        self._recording_timeout_handle = loop.call_later(
            timeout, lambda: asyncio.ensure_future(self._on_recording_timeout())
        )

        # SatelliteRecordStart Command an Satellite senden
        from trixy_core.network.cmd import SatelliteRecordStart
        cmd = SatelliteRecordStart(
            recording_id=self._recording_id,
            timeout_seconds=timeout,
        )
        await self.send_command(cmd)

        # Event triggern
        from trixy_core.events.event_data import SatelliteRecordingStarted
        await self._application.events.trigger(
            "satellite_recording_started",
            SatelliteRecordingStarted(
                satellite_id=self._id,
                recording_id=self._recording_id,
                timeout_seconds=timeout,
                source="satellite.start_recording",
            ),
        )

        pinfo(f"Server-Recording gestartet: {self._alias} ({self._recording_id}, timeout={timeout}s)")
        return True

    async def stop_recording(self, reason: str = "manual") -> bool:
        """
        Stoppt eine laufende server-gesteuerte Aufnahme.

        Args:
            reason: Grund für das Stoppen ("manual", "timeout", "wakeword", "disconnect")

        Returns:
            True wenn Aufnahme gestoppt wurde
        """
        if not self._recording_active:
            return False

        # Timeout-Timer abbrechen
        if self._recording_timeout_handle:
            self._recording_timeout_handle.cancel()
            self._recording_timeout_handle = None

        # Event-Listener entfernen
        if self._recording_listener_registered and self._application:
            self._application.events.unregister(
                "audio_input_received", self._on_recording_audio_received
            )
            self._application.events.unregister(
                "satellite_record_stopped", self._on_satellite_record_stopped_event
            )
            self._recording_listener_registered = False

        # Dauer berechnen
        duration = time.monotonic() - self._recording_start_time
        audio_size = len(self._recording_buffer)

        # WAV speichern falls Dateiname gesetzt
        saved_filename = ""
        if self._recording_filename and self._recording_buffer:
            self._save_recording_wav(self._recording_filename, bytes(self._recording_buffer))
            saved_filename = self._recording_filename

        recording_id = self._recording_id
        self._recording_active = False

        # SatelliteRecordStop Command an Satellite senden
        from trixy_core.network.cmd import SatelliteRecordStop
        cmd = SatelliteRecordStop(
            recording_id=recording_id,
            reason=reason,
        )
        await self.send_command(cmd)

        # Event triggern
        if self._application:
            from trixy_core.events.event_data import SatelliteRecordingStopped
            await self._application.events.trigger(
                "satellite_recording_stopped",
                SatelliteRecordingStopped(
                    satellite_id=self._id,
                    recording_id=recording_id,
                    reason=reason,
                    duration_seconds=duration,
                    audio_size_bytes=audio_size,
                    filename=saved_filename,
                    source="satellite.stop_recording",
                ),
            )

        pinfo(
            f"Server-Recording gestoppt: {self._alias} ({recording_id}, "
            f"reason={reason}, {duration:.1f}s, {audio_size} bytes)"
        )
        return True

    def get_recording(self) -> bytes:
        """Gibt die aufgenommenen Audio-Daten zurück."""
        return bytes(self._recording_buffer)

    def clear_recording(self) -> None:
        """Leert den Recording-Buffer und setzt Status zurück."""
        cleared_bytes = len(self._recording_buffer)
        recording_id = self._recording_id

        self._recording_buffer = bytearray()
        self._recording_id = ""
        self._recording_filename = None

        # Event triggern
        if self._application and recording_id:
            from trixy_core.events.event_data import SatelliteRecordingCleared
            asyncio.ensure_future(
                self._application.events.trigger(
                    "satellite_recording_cleared",
                    SatelliteRecordingCleared(
                        satellite_id=self._id,
                        recording_id=recording_id,
                        cleared_bytes=cleared_bytes,
                        source="satellite.clear_recording",
                    ),
                )
            )

        pdebug(f"Recording-Buffer geleert: {cleared_bytes} bytes")

    async def _on_recording_audio_received(self, event_name: str, event_data: Any) -> None:
        """Handler für audio_input_received Events während Recording."""
        # satellite_id filtern
        satellite_id = getattr(event_data, "satellite_id", "") or ""
        if isinstance(event_data, dict):
            satellite_id = event_data.get("satellite_id", "")

        if satellite_id != self._id:
            return

        if not self._recording_active:
            return

        # Audio-Daten extrahieren
        audio_data = getattr(event_data, "audio_data", b"") or b""
        if isinstance(event_data, dict):
            audio_data = event_data.get("audio_data", b"")

        if audio_data:
            self._recording_buffer.extend(audio_data)

    async def _on_recording_timeout(self) -> None:
        """Timeout-Handler für server-gesteuerte Aufnahme."""
        if self._recording_active:
            pwarn(f"Recording-Timeout erreicht: {self._alias} ({self._recording_id})")
            await self.stop_recording(reason="timeout")

    async def _on_satellite_record_stopped_event(self, event_name: str, event_data: Any) -> None:
        """Handler für satellite_record_stopped Events (Client-seitig durch Wakeword gestoppt)."""
        satellite_id = getattr(event_data, "satellite_id", "") or ""
        if isinstance(event_data, dict):
            satellite_id = event_data.get("satellite_id", "")

        if satellite_id != self._id:
            return

        if not self._recording_active:
            return

        reason = getattr(event_data, "reason", "wakeword") or "wakeword"
        if isinstance(event_data, dict):
            reason = event_data.get("reason", "wakeword")

        pdebug(f"Recording vom Client gestoppt: {self._alias} (reason={reason})")
        await self.stop_recording(reason=reason)

    def _save_recording_wav(self, filename: str, audio_data: bytes) -> None:
        """Speichert Audio-Daten als WAV-Datei (16KHz, 16-bit, mono)."""
        try:
            with wave.open(filename, "wb") as wf:
                wf.setnchannels(1)
                wf.setsampwidth(2)
                wf.setframerate(16000)
                wf.writeframes(audio_data)
            pinfo(f"Recording gespeichert: {filename} ({len(audio_data)} bytes)")
        except Exception as e:
            perror(f"Recording speichern fehlgeschlagen: {e}")

    # === Live-Audio-Streaming ===

    @property
    def is_streaming(self) -> bool:
        """Prüft ob ein Live-Audio-Stream aktiv ist."""
        return self._streaming_active

    @property
    def stream_id(self) -> str:
        """Aktuelle Stream-ID."""
        return self._stream_id

    @property
    def stream_target(self) -> "Satellite | None":
        """Ziel-Satellite des Streams."""
        return self._stream_target

    @property
    def is_stream_recording(self) -> bool:
        """Prüft ob der Stream parallel auf Disk aufgenommen wird."""
        return self._stream_recording_active

    async def start_stream(
        self, target_satellite: "Satellite", timeout: float = 0.0
    ) -> bool:
        """
        Startet Live-Audio-Streaming von diesem Satellite zu einem Ziel-Satellite.

        Audio wird in Echtzeit vom Mikrofon (Port 2102) zum Lautsprecher
        des Ziel-Satellites (Port 2103) weitergeleitet.

        Args:
            target_satellite: Ziel-Satellite (Lautsprecher)
            timeout: Maximale Stream-Dauer in Sekunden (0 = unbegrenzt)

        Returns:
            True wenn Stream gestartet wurde
        """
        if not self._application:
            perror("Keine Application-Referenz für Streaming")
            return False

        if not self.is_connected:
            perror(f"Satellite {self._alias} nicht verbunden")
            return False

        if self._streaming_active:
            pwarn(f"Satellite {self._alias} streamt bereits")
            return False

        if self.has_active_conversation:
            pwarn(f"Satellite {self._alias} hat aktive Konversation")
            return False

        if target_satellite is self:
            perror(f"Satellite {self._alias} kann nicht zu sich selbst streamen")
            return False

        # Laufende Aufnahme stoppen (Mutual Exclusion)
        if self._recording_active:
            pwarn(f"Satellite {self._alias} hat aktive Aufnahme — wird gestoppt für Stream")
            await self.stop_recording(reason="stream_start")

        # Ziel prüfen
        if not target_satellite.is_connected:
            perror(f"Ziel-Satellite {target_satellite.alias} nicht verbunden")
            return False

        if target_satellite.sockets.audio_out is None:
            perror(f"Ziel-Satellite {target_satellite.alias} hat keinen audio_out Socket")
            return False

        # Stream-State setzen
        self._stream_id = f"stream-{uuid.uuid4().hex[:12]}"
        self._streaming_active = True
        self._stream_target = target_satellite
        self._stream_start_time = time.monotonic()
        self._stream_bytes_forwarded = 0

        # Event-Listener registrieren
        self._application.events.register(
            "audio_input_received", self._on_stream_audio_received
        )
        self._application.events.register(
            "satellite_record_stopped", self._on_stream_record_stopped_event
        )
        self._stream_listener_registered = True

        # Timeout-Timer (wenn > 0)
        if timeout > 0:
            loop = asyncio.get_event_loop()
            self._stream_timeout_handle = loop.call_later(
                timeout, lambda: asyncio.ensure_future(self._on_stream_timeout())
            )

        # SatelliteRecordStart Command an Quell-Satellite senden
        from trixy_core.network.cmd import SatelliteRecordStart
        cmd = SatelliteRecordStart(
            recording_id=self._stream_id,
            timeout_seconds=timeout if timeout > 0 else 0,
        )
        await self.send_command(cmd)

        # Event triggern
        from trixy_core.events.event_data import SatelliteStreamStarted
        await self._application.events.trigger(
            "satellite_stream_started",
            SatelliteStreamStarted(
                satellite_id=self._id,
                target_satellite_id=target_satellite.id,
                stream_id=self._stream_id,
                source="satellite.start_stream",
            ),
        )

        pinfo(
            f"Live-Stream gestartet: {self._alias} → {target_satellite.alias} "
            f"({self._stream_id}, timeout={timeout}s)"
        )
        return True

    async def stop_stream(self, reason: str = "manual") -> bool:
        """
        Stoppt einen laufenden Live-Audio-Stream.

        Args:
            reason: Grund für das Stoppen ("manual", "timeout", "wakeword",
                    "disconnect", "target_disconnect")

        Returns:
            True wenn Stream gestoppt wurde
        """
        if not self._streaming_active:
            return False

        # Stream-Aufnahme stoppen falls aktiv
        if self._stream_recording_active:
            await self.stop_record_stream(reason=reason)

        # Timeout-Timer abbrechen
        if self._stream_timeout_handle:
            self._stream_timeout_handle.cancel()
            self._stream_timeout_handle = None

        # Event-Listener entfernen
        if self._stream_listener_registered and self._application:
            self._application.events.unregister(
                "audio_input_received", self._on_stream_audio_received
            )
            self._application.events.unregister(
                "satellite_record_stopped", self._on_stream_record_stopped_event
            )
            self._stream_listener_registered = False

        # Dauer und Stats berechnen
        duration = time.monotonic() - self._stream_start_time
        bytes_forwarded = self._stream_bytes_forwarded
        stream_id = self._stream_id
        target_id = self._stream_target.id if self._stream_target else ""
        target_alias = self._stream_target.alias if self._stream_target else ""

        # State zurücksetzen
        self._streaming_active = False
        self._stream_id = ""
        self._stream_target = None
        self._stream_bytes_forwarded = 0

        # SatelliteRecordStop Command an Quell-Satellite senden
        from trixy_core.network.cmd import SatelliteRecordStop
        cmd = SatelliteRecordStop(
            recording_id=stream_id,
            reason=reason,
        )
        await self.send_command(cmd)

        # Event triggern
        if self._application:
            from trixy_core.events.event_data import SatelliteStreamStopped
            await self._application.events.trigger(
                "satellite_stream_stopped",
                SatelliteStreamStopped(
                    satellite_id=self._id,
                    target_satellite_id=target_id,
                    stream_id=stream_id,
                    reason=reason,
                    duration_seconds=duration,
                    bytes_forwarded=bytes_forwarded,
                    source="satellite.stop_stream",
                ),
            )

        pinfo(
            f"Live-Stream gestoppt: {self._alias} → {target_alias} "
            f"({stream_id}, reason={reason}, {duration:.1f}s, {bytes_forwarded} bytes)"
        )
        return True

    async def start_record_stream(self, filename: str) -> bool:
        """
        Startet parallele Aufnahme des laufenden Streams in eine WAV-Datei.

        Args:
            filename: Pfad zur WAV-Datei

        Returns:
            True wenn Aufnahme gestartet wurde
        """
        if not self._streaming_active:
            perror("Kein aktiver Stream — Stream-Aufnahme nicht möglich")
            return False

        if self._stream_recording_active:
            pwarn("Stream-Aufnahme läuft bereits")
            return False

        try:
            wf = wave.open(filename, "wb")
            wf.setnchannels(1)
            wf.setsampwidth(2)
            wf.setframerate(16000)
        except Exception as e:
            perror(f"WAV-Datei öffnen fehlgeschlagen: {e}")
            return False

        self._stream_recording_active = True
        self._stream_recording_filename = filename
        self._stream_recording_wav = wf
        self._stream_recording_start_time = time.monotonic()
        self._stream_recording_bytes = 0

        # Event triggern
        if self._application:
            from trixy_core.events.event_data import SatelliteStreamRecordStarted
            await self._application.events.trigger(
                "satellite_stream_record_started",
                SatelliteStreamRecordStarted(
                    satellite_id=self._id,
                    stream_id=self._stream_id,
                    filename=filename,
                    source="satellite.start_record_stream",
                ),
            )

        pinfo(f"Stream-Aufnahme gestartet: {filename} ({self._stream_id})")
        return True

    async def stop_record_stream(self, reason: str = "manual") -> bool:
        """
        Stoppt die parallele Stream-Aufnahme.

        Args:
            reason: Grund für das Stoppen

        Returns:
            True wenn Aufnahme gestoppt wurde
        """
        if not self._stream_recording_active:
            return False

        # WAV-Datei schließen
        duration = time.monotonic() - self._stream_recording_start_time
        filename = self._stream_recording_filename
        file_size = self._stream_recording_bytes

        if self._stream_recording_wav:
            try:
                self._stream_recording_wav.close()
            except Exception as e:
                perror(f"WAV-Datei schließen fehlgeschlagen: {e}")
            self._stream_recording_wav = None

        # State zurücksetzen
        self._stream_recording_active = False
        self._stream_recording_filename = ""
        self._stream_recording_bytes = 0

        # Event triggern
        if self._application:
            from trixy_core.events.event_data import SatelliteStreamRecordStopped
            await self._application.events.trigger(
                "satellite_stream_record_stopped",
                SatelliteStreamRecordStopped(
                    satellite_id=self._id,
                    stream_id=self._stream_id,
                    filename=filename,
                    reason=reason,
                    duration_seconds=duration,
                    file_size_bytes=file_size,
                    source="satellite.stop_record_stream",
                ),
            )

        pinfo(f"Stream-Aufnahme gestoppt: {filename} (reason={reason}, {duration:.1f}s, {file_size} bytes)")
        return True

    async def _on_stream_audio_received(self, event_name: str, event_data: Any) -> None:
        """Handler für audio_input_received Events während Streaming."""
        # satellite_id filtern
        satellite_id = getattr(event_data, "satellite_id", "") or ""
        if isinstance(event_data, dict):
            satellite_id = event_data.get("satellite_id", "")

        if satellite_id != self._id:
            return

        if not self._streaming_active or not self._stream_target:
            return

        # Audio-Daten extrahieren
        audio_data = getattr(event_data, "audio_data", b"") or b""
        if isinstance(event_data, dict):
            audio_data = event_data.get("audio_data", b"")

        if not audio_data:
            return

        # Direkter Socket-Write zum Ziel-Satellite
        target = self._stream_target
        if target.sockets.audio_out is None or not target.is_connected:
            pwarn(f"Ziel-Satellite {target.alias} nicht mehr erreichbar")
            await self.stop_stream(reason="target_disconnect")
            return

        try:
            writer: "StreamWriter" = target.sockets.audio_out
            writer.write(audio_data)
            await writer.drain()
            self._stream_bytes_forwarded += len(audio_data)
        except Exception as e:
            perror(f"Stream-Weiterleitung fehlgeschlagen: {e}")
            await self.stop_stream(reason="target_disconnect")
            return

        # Parallel auf Disk schreiben falls Stream-Aufnahme aktiv
        if self._stream_recording_active and self._stream_recording_wav:
            try:
                self._stream_recording_wav.writeframes(audio_data)
                self._stream_recording_bytes += len(audio_data)
            except Exception as e:
                perror(f"Stream-Aufnahme schreiben fehlgeschlagen: {e}")
                await self.stop_record_stream(reason="write_error")

    async def _on_stream_record_stopped_event(self, event_name: str, event_data: Any) -> None:
        """Handler für satellite_record_stopped Events (Client-seitig durch Wakeword gestoppt)."""
        satellite_id = getattr(event_data, "satellite_id", "") or ""
        if isinstance(event_data, dict):
            satellite_id = event_data.get("satellite_id", "")

        if satellite_id != self._id:
            return

        if not self._streaming_active:
            return

        reason = getattr(event_data, "reason", "wakeword") or "wakeword"
        if isinstance(event_data, dict):
            reason = event_data.get("reason", "wakeword")

        pdebug(f"Stream vom Client gestoppt: {self._alias} (reason={reason})")
        await self.stop_stream(reason=reason)

    async def _on_stream_timeout(self) -> None:
        """Timeout-Handler für Live-Audio-Stream."""
        if self._streaming_active:
            pwarn(f"Stream-Timeout erreicht: {self._alias} ({self._stream_id})")
            await self.stop_stream(reason="timeout")

    async def disconnect(self, reason: str = "") -> None:
        """
        Trennt die Verbindung zum Satellite.

        Sendet zuerst eine Benachrichtigung an den Client,
        dann werden alle Sockets geschlossen.

        Args:
            reason: Grund für die Trennung
        """
        # Laufenden Stream beenden
        if self._streaming_active:
            await self.stop_stream(reason="disconnect")

        # Laufende Aufnahme beenden
        if self._recording_active:
            await self.stop_recording(reason="disconnect")

        # Benachrichtige Client vor dem Ändern des States
        # (send_command prüft is_connected, daher erst senden)
        if self._sockets.command is not None and self.is_connected:
            try:
                from trixy_core.network.cmd import ServerShutdown

                shutdown_msg = ServerShutdown(
                    reason=reason,
                    reconnect_after_seconds=0,
                )
                await self.send_command(shutdown_msg)

                # Kurz warten, damit die Nachricht ankommt
                await asyncio.sleep(0.1)
            except Exception:
                pass  # Fehler ignorieren beim Shutdown

        self._state = ConnectionState.DISCONNECTING

        # Schließe alle Sockets
        for socket_name in ("command", "audio_in", "audio_out", "music_out"):
            socket = getattr(self._sockets, socket_name)
            if socket is not None:
                try:
                    socket.close()
                    await socket.wait_closed()
                except Exception:
                    pass
                setattr(self._sockets, socket_name, None)

        self._state = ConnectionState.DISCONNECTED
        self._conversation_id = None

    def to_dict(self) -> dict[str, Any]:
        """Konvertiert den Satellite in ein Dictionary."""
        return {
            "id": self._id,
            "room_id": self._room_id,
            "mac_address": self._mac_address,
            "alias": self._alias,
            "ip_address": self._ip_address,
            "state": self._state.name,
            "connected_at": self._connected_at.isoformat() if self._connected_at else None,
            "last_heartbeat": self._last_heartbeat.isoformat() if self._last_heartbeat else None,
            "version": self._version,
            "has_active_conversation": self.has_active_conversation,
        }

    def __repr__(self) -> str:
        return (
            f"Satellite(id={self._id!r}, room={self._room_id!r}, "
            f"alias={self._alias!r}, state={self._state.name})"
        )