Trixy7/trixy_core/music/formats/base.py

# -*- coding: utf-8 -*-
"""
Basis-Klassen für Audio-Decoder.

Definiert die abstrakte Schnittstelle für alle Audio-Formate.
"""

from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from typing import AsyncIterator


class AudioFormat(Enum):
    """Unterstützte Audio-Formate."""
    MP3 = "mp3"
    WAV = "wav"
    OGG = "ogg"
    FLAC = "flac"
    AAC = "aac"
    M4A = "m4a"
    OPUS = "opus"
    WMA = "wma"
    AIFF = "aiff"
    UNKNOWN = "unknown"

    @classmethod
    def from_extension(cls, ext: str) -> "AudioFormat":
        """Bestimmt Format aus Dateiendung."""
        ext = ext.lower().lstrip(".")
        mapping = {
            "mp3": cls.MP3,
            "wav": cls.WAV,
            "wave": cls.WAV,
            "ogg": cls.OGG,
            "oga": cls.OGG,
            "flac": cls.FLAC,
            "aac": cls.AAC,
            "m4a": cls.M4A,
            "mp4": cls.M4A,
            "opus": cls.OPUS,
            "wma": cls.WMA,
            "aiff": cls.AIFF,
            "aif": cls.AIFF,
        }
        return mapping.get(ext, cls.UNKNOWN)

    @classmethod
    def from_mime(cls, mime: str) -> "AudioFormat":
        """Bestimmt Format aus MIME-Type."""
        mime = mime.lower().split(";")[0].strip()
        mapping = {
            "audio/mpeg": cls.MP3,
            "audio/mp3": cls.MP3,
            "audio/wav": cls.WAV,
            "audio/x-wav": cls.WAV,
            "audio/wave": cls.WAV,
            "audio/ogg": cls.OGG,
            "audio/vorbis": cls.OGG,
            "application/ogg": cls.OGG,
            "audio/flac": cls.FLAC,
            "audio/x-flac": cls.FLAC,
            "audio/aac": cls.AAC,
            "audio/mp4": cls.M4A,
            "audio/x-m4a": cls.M4A,
            "audio/opus": cls.OPUS,
            "audio/x-ms-wma": cls.WMA,
            "audio/aiff": cls.AIFF,
            "audio/x-aiff": cls.AIFF,
        }
        return mapping.get(mime, cls.UNKNOWN)


class DecoderError(Exception):
    """Fehler beim Decodieren."""
    pass


@dataclass
class AudioInfo:
    """Informationen über eine Audio-Datei."""
    format: AudioFormat
    duration_ms: int = 0
    sample_rate: int = 44100
    channels: int = 2
    bit_depth: int = 16
    bit_rate: int = 0
    file_size: int = 0

    @property
    def duration_seconds(self) -> float:
        """Dauer in Sekunden."""
        return self.duration_ms / 1000.0

    @property
    def is_stereo(self) -> bool:
        """Ist Stereo."""
        return self.channels == 2


@dataclass
class DecodedAudio:
    """Dekodierte Audio-Daten."""
    # PCM-Daten (16-bit signed, little-endian)
    pcm_data: bytes

    # Audio-Eigenschaften
    sample_rate: int = 44100
    channels: int = 2
    bit_depth: int = 16

    @property
    def duration_ms(self) -> int:
        """Dauer in Millisekunden."""
        bytes_per_sample = self.bit_depth // 8
        samples = len(self.pcm_data) // (bytes_per_sample * self.channels)
        return int((samples / self.sample_rate) * 1000)

    @property
    def sample_count(self) -> int:
        """Anzahl der Samples."""
        bytes_per_sample = self.bit_depth // 8
        return len(self.pcm_data) // (bytes_per_sample * self.channels)


class AudioDecoder(ABC):
    """
    Abstrakte Basis-Klasse für Audio-Decoder.

    Dekodiert komprimierte Audio-Formate zu PCM.
    """

    # Format das dieser Decoder unterstützt
    FORMAT: AudioFormat = AudioFormat.UNKNOWN

    # Unterstützte Dateiendungen
    EXTENSIONS: set[str] = set()

    # Unterstützte MIME-Types
    MIME_TYPES: set[str] = set()

    # Standard-Output-Eigenschaften
    OUTPUT_SAMPLE_RATE: int = 44100  # Hz
    OUTPUT_CHANNELS: int = 2         # Stereo
    OUTPUT_BIT_DEPTH: int = 16       # 16-bit

    def __init__(self) -> None:
        """Initialisiert den Decoder."""
        self._is_initialized = False

    @property
    def is_available(self) -> bool:
        """Prüft ob Decoder verfügbar ist (Dependencies installiert)."""
        return self._check_dependencies()

    @abstractmethod
    def _check_dependencies(self) -> bool:
        """Prüft ob alle Dependencies vorhanden sind."""
        pass

    @abstractmethod
    async def get_info(self, source: Path | bytes) -> AudioInfo:
        """
        Liest Audio-Informationen ohne vollständiges Dekodieren.

        Args:
            source: Dateipfad oder Rohdaten

        Returns:
            Audio-Informationen

        Raises:
            DecoderError: Bei Fehlern
        """
        pass

    @abstractmethod
    async def decode(
        self,
        source: Path | bytes,
        target_sample_rate: int | None = None,
        target_channels: int | None = None,
    ) -> DecodedAudio:
        """
        Dekodiert Audio vollständig zu PCM.

        Args:
            source: Dateipfad oder Rohdaten
            target_sample_rate: Ziel-Sample-Rate (None = Standard)
            target_channels: Ziel-Kanalzahl (None = Standard)

        Returns:
            Dekodierte Audio-Daten

        Raises:
            DecoderError: Bei Fehlern
        """
        pass

    @abstractmethod
    async def decode_stream(
        self,
        source: Path | bytes,
        chunk_duration_ms: int = 100,
    ) -> AsyncIterator[DecodedAudio]:
        """
        Dekodiert Audio als Stream (chunk-weise).

        Args:
            source: Dateipfad oder Rohdaten
            chunk_duration_ms: Chunk-Dauer in Millisekunden

        Yields:
            Dekodierte Audio-Chunks

        Raises:
            DecoderError: Bei Fehlern
        """
        pass

    def supports_extension(self, ext: str) -> bool:
        """Prüft ob Dateiendung unterstützt wird."""
        return ext.lower().lstrip(".") in self.EXTENSIONS

    def supports_mime(self, mime: str) -> bool:
        """Prüft ob MIME-Type unterstützt wird."""
        return mime.lower().split(";")[0].strip() in self.MIME_TYPES

    def __repr__(self) -> str:
        return (
            f"{self.__class__.__name__}("
            f"format={self.FORMAT.value}, "
            f"available={self.is_available})"
        )