# -*- 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})" )