Trixy7/trixy_core/music/stream/config.py

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

Einstellungen für Audio-Streaming.
"""

from dataclasses import dataclass, field
from typing import Any


@dataclass
class StreamConfig:
    """
    Konfiguration für Audio-Streaming.

    Definiert Audio-Format, Puffergrößen und Streaming-Verhalten.
    """

    # Audio-Format
    sample_rate: int = 44100      # Hz (44.1kHz - Standard für die meisten Geräte)
    channels: int = 2              # Stereo
    bit_depth: int = 16            # 16-bit

    # Pufferung
    buffer_size_ms: int = 100      # Puffergröße in Millisekunden
    prebuffer_ms: int = 500        # Vor-Pufferung bevor Start
    max_buffer_ms: int = 5000      # Maximale Puffergröße

    # Streaming
    chunk_size: int = 4096         # Bytes pro Chunk
    send_interval_ms: int = 20     # Intervall zwischen Paketen

    # Synchronisation
    sync_enabled: bool = True      # Zeit-Synchronisation
    sync_interval_ms: int = 1000   # Sync-Intervall
    max_drift_ms: int = 50         # Maximale erlaubte Drift

    # Fehlerbehandlung
    retry_count: int = 3           # Wiederholungsversuche
    retry_delay_ms: int = 100      # Verzögerung zwischen Versuchen
    disconnect_timeout_ms: int = 5000  # Timeout für Disconnect

    # Kompression (optional)
    compression_enabled: bool = False
    compression_level: int = 6     # 0-9 (0=keine, 9=max)

    # Metadaten
    send_metadata: bool = True     # Track-Info mitsenden
    metadata_interval_ms: int = 5000  # Metadaten-Intervall

    # Zusätzliche Einstellungen
    extra: dict[str, Any] = field(default_factory=dict)

    @property
    def bytes_per_sample(self) -> int:
        """Bytes pro Sample."""
        return self.bit_depth // 8

    @property
    def bytes_per_frame(self) -> int:
        """Bytes pro Frame (alle Kanäle)."""
        return self.bytes_per_sample * self.channels

    @property
    def bytes_per_second(self) -> int:
        """Bytes pro Sekunde."""
        return self.sample_rate * self.bytes_per_frame

    @property
    def buffer_size_bytes(self) -> int:
        """Puffergröße in Bytes."""
        return int(self.bytes_per_second * self.buffer_size_ms / 1000)

    @property
    def prebuffer_bytes(self) -> int:
        """Vor-Puffer in Bytes."""
        return int(self.bytes_per_second * self.prebuffer_ms / 1000)

    def validate(self) -> list[str]:
        """
        Validiert die Konfiguration.

        Returns:
            Liste von Fehlermeldungen (leer = valide)
        """
        errors = []

        if self.sample_rate not in (8000, 11025, 16000, 22050, 44100, 48000, 96000):
            errors.append(f"Ungültige Sample-Rate: {self.sample_rate}")

        if self.channels not in (1, 2):
            errors.append(f"Ungültige Kanalzahl: {self.channels}")

        if self.bit_depth not in (8, 16, 24, 32):
            errors.append(f"Ungültige Bit-Tiefe: {self.bit_depth}")

        if self.buffer_size_ms < 10:
            errors.append("Buffer zu klein (min 10ms)")

        if self.prebuffer_ms > self.max_buffer_ms:
            errors.append("Prebuffer größer als max_buffer")

        return errors

    @classmethod
    def for_voice(cls) -> "StreamConfig":
        """Konfiguration für Sprach-Audio (16kHz mono)."""
        return cls(
            sample_rate=16000,
            channels=1,
            bit_depth=16,
            buffer_size_ms=50,
            prebuffer_ms=200,
        )

    @classmethod
    def for_music(cls) -> "StreamConfig":
        """Konfiguration für Musik (48kHz stereo für beste Kompatibilität)."""
        return cls(
            sample_rate=48000,
            channels=2,
            bit_depth=16,
            buffer_size_ms=100,
            prebuffer_ms=500,
        )

    @classmethod
    def for_low_latency(cls) -> "StreamConfig":
        """Konfiguration für geringe Latenz."""
        return cls(
            sample_rate=48000,
            channels=2,
            bit_depth=16,
            buffer_size_ms=20,
            prebuffer_ms=100,
            max_buffer_ms=1000,
            send_interval_ms=10,
        )

    def to_dict(self) -> dict[str, Any]:
        """Konvertiert zu Dictionary."""
        return {
            "sample_rate": self.sample_rate,
            "channels": self.channels,
            "bit_depth": self.bit_depth,
            "buffer_size_ms": self.buffer_size_ms,
            "prebuffer_ms": self.prebuffer_ms,
            "max_buffer_ms": self.max_buffer_ms,
            "chunk_size": self.chunk_size,
            "send_interval_ms": self.send_interval_ms,
            "sync_enabled": self.sync_enabled,
            "compression_enabled": self.compression_enabled,
        }