Trixy7/trixy_core/music/stream/streamer.py

# -*- coding: utf-8 -*-
"""
Audio-Streamer.

Verwaltet Audio-Streaming an mehrere Satellites gleichzeitig.
"""

import asyncio
import logging
from typing import Any, Callable, Awaitable

from trixy_core.music.stream.config import StreamConfig
from trixy_core.music.stream.session import StreamSession, StreamState
from trixy_core.music.track import Track
from trixy_core.music.formats.base import DecodedAudio


class AudioStreamer:
    """
    Zentrale Streaming-Verwaltung.

    Streamt Audio an einen oder mehrere Satellites gleichzeitig,
    mit Synchronisation und Pufferung.
    """

    def __init__(
        self,
        config: StreamConfig | None = None,
    ) -> None:
        """
        Initialisiert den AudioStreamer.

        Args:
            config: Standard-Stream-Konfiguration
        """
        self.default_config = config or StreamConfig.for_music()
        self.logger = logging.getLogger(__name__)

        # Sessions pro Satellite
        self._sessions: dict[str, StreamSession] = {}

        # Aktueller Track
        self._current_track: Track | None = None

        # Callbacks für Satellite-Kommunikation
        self._send_callbacks: dict[str, Callable[[bytes], Awaitable[bool]]] = {}

        # Decoder-Task
        self._decoder_task: asyncio.Task | None = None
        self._stop_event = asyncio.Event()

    # ==========================================================================
    # Properties
    # ==========================================================================

    @property
    def sessions(self) -> list[StreamSession]:
        """Alle aktiven Sessions."""
        return list(self._sessions.values())

    @property
    def session_count(self) -> int:
        """Anzahl aktiver Sessions."""
        return len(self._sessions)

    @property
    def is_streaming(self) -> bool:
        """Prüft ob aktiv gestreamt wird."""
        return any(s.is_streaming for s in self._sessions.values())

    @property
    def current_track(self) -> Track | None:
        """Aktuell gestreamter Track."""
        return self._current_track

    # ==========================================================================
    # Session-Management
    # ==========================================================================

    def add_satellite(
        self,
        satellite_id: str,
        send_callback: Callable[[bytes], Awaitable[bool]],
        config: StreamConfig | None = None,
    ) -> StreamSession:
        """
        Fügt einen Satellite als Streaming-Ziel hinzu.

        Args:
            satellite_id: ID des Satellites
            send_callback: Async Callback zum Senden von Daten
            config: Optionale Konfiguration

        Returns:
            Erstellte Session
        """
        # Alte Session entfernen
        if satellite_id in self._sessions:
            asyncio.create_task(self.remove_satellite(satellite_id))

        self._send_callbacks[satellite_id] = send_callback

        session = StreamSession(
            satellite_id=satellite_id,
            config=config or self.default_config,
            send_callback=send_callback,
        )

        self._sessions[satellite_id] = session
        self.logger.debug(f"Satellite hinzugefügt: {satellite_id}")

        return session

    async def remove_satellite(self, satellite_id: str) -> bool:
        """
        Entfernt einen Satellite.

        Args:
            satellite_id: ID des Satellites

        Returns:
            True wenn entfernt
        """
        if satellite_id in self._sessions:
            session = self._sessions.pop(satellite_id)
            await session.stop()
            self._send_callbacks.pop(satellite_id, None)
            self.logger.debug(f"Satellite entfernt: {satellite_id}")
            return True
        return False

    def get_session(self, satellite_id: str) -> StreamSession | None:
        """
        Holt Session für Satellite.

        Args:
            satellite_id: ID des Satellites

        Returns:
            Session oder None
        """
        return self._sessions.get(satellite_id)

    # ==========================================================================
    # Streaming-Steuerung
    # ==========================================================================

    async def start(
        self,
        track: Track,
        satellite_ids: list[str] | None = None,
    ) -> dict[str, bool]:
        """
        Startet Streaming für Track an ausgewählte Satellites.

        Args:
            track: Zu streamender Track
            satellite_ids: Ziel-Satellites (None = alle)

        Returns:
            Dict Satellite-ID → Erfolg
        """
        self._current_track = track
        self._stop_event.clear()

        # Ziel-Sessions bestimmen
        if satellite_ids:
            sessions = [self._sessions[sid] for sid in satellite_ids if sid in self._sessions]
        else:
            sessions = list(self._sessions.values())

        if not sessions:
            self.logger.warning("Keine Satellites für Streaming verfügbar")
            return {}

        # Alle Sessions starten
        results = {}
        for session in sessions:
            try:
                success = await session.start(track)
                results[session.satellite_id] = success
            except Exception as e:
                self.logger.error(f"Session-Start fehlgeschlagen: {e}")
                results[session.satellite_id] = False

        active_count = sum(1 for v in results.values() if v)
        self.logger.info(f"Streaming gestartet: {active_count}/{len(sessions)} Sessions")

        return results

    async def stop(self) -> None:
        """Stoppt alle Streaming-Sessions."""
        self._stop_event.set()

        # Decoder-Task stoppen
        if self._decoder_task and not self._decoder_task.done():
            self._decoder_task.cancel()
            try:
                await self._decoder_task
            except asyncio.CancelledError:
                pass

        # Alle Sessions stoppen
        for session in self._sessions.values():
            await session.stop()

        self._current_track = None
        self.logger.info("Alle Streams gestoppt")

    async def pause(self, satellite_ids: list[str] | None = None) -> None:
        """
        Pausiert Streaming.

        Args:
            satellite_ids: Betroffene Satellites (None = alle)
        """
        sessions = self._get_sessions(satellite_ids)
        for session in sessions:
            await session.pause()

    async def resume(self, satellite_ids: list[str] | None = None) -> None:
        """
        Setzt Streaming fort.

        Args:
            satellite_ids: Betroffene Satellites (None = alle)
        """
        sessions = self._get_sessions(satellite_ids)
        for session in sessions:
            await session.resume()

    async def seek(
        self,
        position_ms: int,
        satellite_ids: list[str] | None = None,
    ) -> dict[str, bool]:
        """
        Springt zu Position.

        Args:
            position_ms: Ziel-Position
            satellite_ids: Betroffene Satellites (None = alle)

        Returns:
            Dict Satellite-ID → Erfolg
        """
        sessions = self._get_sessions(satellite_ids)
        results = {}

        for session in sessions:
            results[session.satellite_id] = await session.seek(position_ms)

        return results

    # ==========================================================================
    # Audio-Daten
    # ==========================================================================

    async def feed(
        self,
        data: bytes,
        satellite_ids: list[str] | None = None,
    ) -> dict[str, bool]:
        """
        Füttert alle Sessions mit Audio-Daten.

        Args:
            data: PCM-Audio-Daten
            satellite_ids: Ziel-Satellites (None = alle)

        Returns:
            Dict Satellite-ID → Akzeptiert
        """
        sessions = self._get_sessions(satellite_ids)
        results = {}

        for session in sessions:
            results[session.satellite_id] = await session.feed(data)

        return results

    async def feed_decoded(
        self,
        audio: DecodedAudio,
        satellite_ids: list[str] | None = None,
    ) -> dict[str, bool]:
        """
        Füttert Sessions mit dekodierten Audio-Daten.

        Args:
            audio: Dekodierte Audio-Daten
            satellite_ids: Ziel-Satellites (None = alle)

        Returns:
            Dict Satellite-ID → Akzeptiert
        """
        return await self.feed(audio.pcm_data, satellite_ids)

    # ==========================================================================
    # Synchronisation
    # ==========================================================================

    async def sync_all(self) -> dict[str, dict[str, Any]]:
        """
        Synchronisiert alle Sessions.

        Returns:
            Dict Satellite-ID → Sync-Info
        """
        results = {}

        for session in self._sessions.values():
            if session.is_active:
                # Hier würde normalerweise der Client-Time-Stempel
                # vom Satellite kommen. Für Demo: eigene Zeit verwenden.
                results[session.satellite_id] = await session.sync(session.position_ms)

        return results

    async def broadcast_sync(self, target_position_ms: int) -> None:
        """
        Synchronisiert alle Sessions auf Ziel-Position.

        Args:
            target_position_ms: Ziel-Position
        """
        for session in self._sessions.values():
            if session.is_active:
                drift = session.position_ms - target_position_ms

                if abs(drift) > session.config.max_drift_ms:
                    await session.seek(target_position_ms)

    # ==========================================================================
    # Hilfsmethoden
    # ==========================================================================

    def _get_sessions(self, satellite_ids: list[str] | None) -> list[StreamSession]:
        """Holt Sessions für Satellite-IDs."""
        if satellite_ids:
            return [self._sessions[sid] for sid in satellite_ids if sid in self._sessions]
        return list(self._sessions.values())

    # ==========================================================================
    # Status
    # ==========================================================================

    def get_status(self) -> dict[str, Any]:
        """Liefert Gesamtstatus."""
        return {
            "session_count": self.session_count,
            "is_streaming": self.is_streaming,
            "current_track": self._current_track.to_dict() if self._current_track else None,
            "sessions": {
                sid: session.get_status()
                for sid, session in self._sessions.items()
            },
        }

    def get_session_status(self, satellite_id: str) -> dict[str, Any] | None:
        """Liefert Status einer Session."""
        session = self._sessions.get(satellite_id)
        return session.get_status() if session else None

    def __len__(self) -> int:
        return len(self._sessions)

    def __contains__(self, satellite_id: str) -> bool:
        return satellite_id in self._sessions

    def __repr__(self) -> str:
        return (
            f"AudioStreamer(sessions={self.session_count}, "
            f"streaming={self.is_streaming})"
        )