Trixy7/trixy_core/music/stream/session.py

# -*- coding: utf-8 -*-
"""
Stream-Session.

Verwaltet eine einzelne Streaming-Verbindung zu einem Satellite.
"""

import asyncio
import logging
import time
import uuid
from collections import deque
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any, Callable, Awaitable

from trixy_core.music.stream.config import StreamConfig
from trixy_core.music.track import Track


class StreamState(Enum):
    """Zustand einer Stream-Session."""
    IDLE = "idle"              # Keine aktive Session
    CONNECTING = "connecting"  # Verbindung wird aufgebaut
    BUFFERING = "buffering"    # Puffert Daten
    STREAMING = "streaming"    # Aktives Streaming
    PAUSED = "paused"          # Pausiert
    SEEKING = "seeking"        # Sucht neue Position
    ERROR = "error"            # Fehler
    DISCONNECTED = "disconnected"  # Verbindung getrennt


@dataclass
class StreamStats:
    """Statistiken einer Stream-Session."""
    bytes_sent: int = 0
    packets_sent: int = 0
    packets_lost: int = 0
    buffer_underruns: int = 0
    buffer_overruns: int = 0

    # Timing
    start_time: datetime | None = None
    total_duration_ms: int = 0
    current_position_ms: int = 0

    # Latenz
    min_latency_ms: float = 0.0
    max_latency_ms: float = 0.0
    avg_latency_ms: float = 0.0
    _latency_samples: list[float] = field(default_factory=list)

    def add_latency_sample(self, latency_ms: float) -> None:
        """Fügt Latenz-Messwert hinzu."""
        self._latency_samples.append(latency_ms)

        # Nur letzte 100 Samples behalten
        if len(self._latency_samples) > 100:
            self._latency_samples.pop(0)

        # Statistiken aktualisieren
        self.min_latency_ms = min(self._latency_samples)
        self.max_latency_ms = max(self._latency_samples)
        self.avg_latency_ms = sum(self._latency_samples) / len(self._latency_samples)

    @property
    def loss_rate(self) -> float:
        """Paketverlust-Rate."""
        total = self.packets_sent + self.packets_lost
        return self.packets_lost / total if total > 0 else 0.0

    def to_dict(self) -> dict[str, Any]:
        """Konvertiert zu Dictionary."""
        return {
            "bytes_sent": self.bytes_sent,
            "packets_sent": self.packets_sent,
            "packets_lost": self.packets_lost,
            "buffer_underruns": self.buffer_underruns,
            "loss_rate": self.loss_rate,
            "current_position_ms": self.current_position_ms,
            "latency": {
                "min_ms": self.min_latency_ms,
                "max_ms": self.max_latency_ms,
                "avg_ms": self.avg_latency_ms,
            },
        }


class StreamSession:
    """
    Eine Streaming-Session zu einem Satellite.

    Verwaltet Pufferung, Timing und Synchronisation.
    """

    def __init__(
        self,
        satellite_id: str,
        config: StreamConfig | None = None,
        send_callback: Callable[[bytes], Awaitable[bool]] | None = None,
    ) -> None:
        """
        Initialisiert die Stream-Session.

        Args:
            satellite_id: ID des Ziel-Satellites
            config: Stream-Konfiguration
            send_callback: Async Callback zum Senden von Daten
        """
        self.id = f"session_{uuid.uuid4().hex[:8]}"
        self.satellite_id = satellite_id
        self.config = config or StreamConfig.for_music()
        self._send_callback = send_callback

        self.logger = logging.getLogger(__name__)

        # Zustand
        self._state = StreamState.IDLE
        self._error_message = ""

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

        # Puffer
        self._buffer: deque[bytes] = deque()
        self._buffer_size = 0
        self._buffer_lock = asyncio.Lock()

        # Timing
        self._start_time: float = 0.0
        self._position_ms: int = 0
        self._last_sync_time: float = 0.0

        # Statistiken
        self.stats = StreamStats()

        # Streaming-Task
        self._stream_task: asyncio.Task | None = None
        self._stop_event = asyncio.Event()

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

    @property
    def state(self) -> StreamState:
        """Aktueller Zustand."""
        return self._state

    @property
    def is_active(self) -> bool:
        """Prüft ob Session aktiv ist."""
        return self._state in (StreamState.BUFFERING, StreamState.STREAMING)

    @property
    def is_streaming(self) -> bool:
        """Prüft ob aktiv gestreamt wird."""
        return self._state == StreamState.STREAMING

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

    @property
    def position_ms(self) -> int:
        """Aktuelle Position in Millisekunden."""
        return self._position_ms

    @property
    def buffer_level(self) -> float:
        """Puffer-Füllstand (0.0 - 1.0)."""
        max_buffer = self.config.max_buffer_ms * self.config.bytes_per_second // 1000
        return min(1.0, self._buffer_size / max_buffer) if max_buffer > 0 else 0.0

    @property
    def buffer_ms(self) -> int:
        """Puffer-Inhalt in Millisekunden."""
        return int(self._buffer_size * 1000 / self.config.bytes_per_second)

    # ==========================================================================
    # Lifecycle
    # ==========================================================================

    async def start(self, track: Track) -> bool:
        """
        Startet Streaming für einen Track.

        Args:
            track: Zu streamender Track

        Returns:
            True wenn erfolgreich gestartet
        """
        if self._state not in (StreamState.IDLE, StreamState.DISCONNECTED, StreamState.ERROR):
            await self.stop()

        self._current_track = track
        self._state = StreamState.CONNECTING
        self._stop_event.clear()

        self.logger.debug(f"Session {self.id}: Starte Streaming für {track.title}")

        try:
            self._state = StreamState.BUFFERING
            self.stats = StreamStats()
            self.stats.start_time = datetime.now()
            self.stats.total_duration_ms = track.duration_ms

            # Streaming-Task starten
            self._stream_task = asyncio.create_task(self._streaming_loop())

            return True

        except Exception as e:
            self.logger.error(f"Session {self.id}: Start fehlgeschlagen: {e}")
            self._state = StreamState.ERROR
            self._error_message = str(e)
            return False

    async def stop(self) -> None:
        """Stoppt das Streaming."""
        self.logger.debug(f"Session {self.id}: Stoppe Streaming")

        self._stop_event.set()

        if self._stream_task and not self._stream_task.done():
            self._stream_task.cancel()
            try:
                await self._stream_task
            except asyncio.CancelledError:
                pass

        # Puffer leeren
        async with self._buffer_lock:
            self._buffer.clear()
            self._buffer_size = 0

        self._state = StreamState.IDLE
        self._current_track = None

    async def pause(self) -> None:
        """Pausiert das Streaming."""
        if self._state == StreamState.STREAMING:
            self._state = StreamState.PAUSED
            self.logger.debug(f"Session {self.id}: Pausiert")

    async def resume(self) -> None:
        """Setzt das Streaming fort."""
        if self._state == StreamState.PAUSED:
            self._state = StreamState.STREAMING
            self.logger.debug(f"Session {self.id}: Fortgesetzt")

    async def seek(self, position_ms: int) -> bool:
        """
        Springt zu einer Position.

        Args:
            position_ms: Ziel-Position in Millisekunden

        Returns:
            True wenn erfolgreich
        """
        if not self._current_track:
            return False

        self._state = StreamState.SEEKING

        # Puffer leeren
        async with self._buffer_lock:
            self._buffer.clear()
            self._buffer_size = 0

        self._position_ms = position_ms
        self._state = StreamState.BUFFERING

        self.logger.debug(f"Session {self.id}: Seek zu {position_ms}ms")
        return True

    # ==========================================================================
    # Puffer-Management
    # ==========================================================================

    async def feed(self, data: bytes) -> bool:
        """
        Füttert den Puffer mit Audio-Daten.

        Args:
            data: PCM-Audio-Daten

        Returns:
            True wenn akzeptiert
        """
        async with self._buffer_lock:
            # Puffer-Limit prüfen
            max_bytes = self.config.max_buffer_ms * self.config.bytes_per_second // 1000

            if self._buffer_size + len(data) > max_bytes:
                self.stats.buffer_overruns += 1
                return False

            self._buffer.append(data)
            self._buffer_size += len(data)

            # Zustand aktualisieren
            if self._state == StreamState.BUFFERING:
                if self._buffer_size >= self.config.prebuffer_bytes:
                    self._state = StreamState.STREAMING
                    self._start_time = time.monotonic()

            return True

    async def _get_chunk(self) -> bytes | None:
        """Holt Chunk aus Puffer."""
        async with self._buffer_lock:
            if not self._buffer:
                return None

            chunk = self._buffer.popleft()
            self._buffer_size -= len(chunk)

            return chunk

    # ==========================================================================
    # Streaming
    # ==========================================================================

    async def _streaming_loop(self) -> None:
        """Haupt-Streaming-Loop."""
        interval = self.config.send_interval_ms / 1000.0

        while not self._stop_event.is_set():
            try:
                # Auf aktiven Zustand warten
                if self._state == StreamState.PAUSED:
                    await asyncio.sleep(0.1)
                    continue

                if self._state == StreamState.BUFFERING:
                    await asyncio.sleep(0.01)
                    continue

                if self._state != StreamState.STREAMING:
                    await asyncio.sleep(0.1)
                    continue

                # Chunk holen
                chunk = await self._get_chunk()

                if chunk is None:
                    # Buffer Underrun
                    self.stats.buffer_underruns += 1
                    self._state = StreamState.BUFFERING
                    self.logger.debug(f"Session {self.id}: Buffer Underrun")
                    continue

                # Senden
                if self._send_callback:
                    send_start = time.monotonic()
                    success = await self._send_callback(chunk)
                    send_duration = (time.monotonic() - send_start) * 1000

                    if success:
                        self.stats.packets_sent += 1
                        self.stats.bytes_sent += len(chunk)
                        self.stats.add_latency_sample(send_duration)
                    else:
                        self.stats.packets_lost += 1

                # Position aktualisieren
                chunk_duration_ms = int(
                    len(chunk) * 1000 / self.config.bytes_per_second
                )
                self._position_ms += chunk_duration_ms
                self.stats.current_position_ms = self._position_ms

                # Timing
                await asyncio.sleep(interval)

            except asyncio.CancelledError:
                break
            except Exception as e:
                self.logger.error(f"Session {self.id}: Streaming-Fehler: {e}")
                self._state = StreamState.ERROR
                self._error_message = str(e)
                break

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

    async def sync(self, client_time_ms: int) -> dict[str, Any]:
        """
        Synchronisiert mit Client.

        Args:
            client_time_ms: Client-Zeit in Millisekunden

        Returns:
            Sync-Informationen
        """
        server_time_ms = self._position_ms
        drift_ms = server_time_ms - client_time_ms

        self._last_sync_time = time.monotonic()

        return {
            "server_time_ms": server_time_ms,
            "client_time_ms": client_time_ms,
            "drift_ms": drift_ms,
            "should_skip": drift_ms > self.config.max_drift_ms,
            "should_wait": drift_ms < -self.config.max_drift_ms,
        }

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

    def get_status(self) -> dict[str, Any]:
        """Liefert Session-Status."""
        return {
            "id": self.id,
            "satellite_id": self.satellite_id,
            "state": self._state.value,
            "track": self._current_track.to_dict() if self._current_track else None,
            "position_ms": self._position_ms,
            "buffer_level": self.buffer_level,
            "buffer_ms": self.buffer_ms,
            "stats": self.stats.to_dict(),
            "error": self._error_message if self._state == StreamState.ERROR else None,
        }

    def __repr__(self) -> str:
        return (
            f"StreamSession(id={self.id!r}, "
            f"satellite={self.satellite_id!r}, "
            f"state={self._state.value!r})"
        )