Trixy7/trixy_core/music/formats/ogg.py

# -*- coding: utf-8 -*-
"""
OGG/Vorbis-Decoder.

Dekodiert OGG Vorbis-Dateien zu PCM.
"""

import asyncio
import io
import logging
from pathlib import Path
from typing import AsyncIterator

from trixy_core.music.formats.base import (
    AudioDecoder,
    AudioFormat,
    AudioInfo,
    DecodedAudio,
    DecoderError,
)


class OGGDecoder(AudioDecoder):
    """
    OGG Vorbis-Decoder.

    Verwendet pydub (mit ffmpeg Backend) oder pyogg für das Dekodieren.
    """

    FORMAT = AudioFormat.OGG
    EXTENSIONS = {"ogg", "oga", "opus"}
    MIME_TYPES = {"audio/ogg", "audio/vorbis", "application/ogg", "audio/opus"}

    def __init__(self) -> None:
        super().__init__()
        self.logger = logging.getLogger(__name__)
        self._use_pydub = False
        self._use_ffmpeg = False

    def _check_dependencies(self) -> bool:
        """Prüft verfügbare Decoder-Backends."""
        # Versuche pydub
        try:
            from pydub import AudioSegment
            self._use_pydub = True
            return True
        except ImportError:
            pass

        # Versuche ffmpeg direkt
        import shutil
        if shutil.which("ffmpeg"):
            self._use_ffmpeg = True
            return True

        return False

    async def get_info(self, source: Path | bytes) -> AudioInfo:
        """Liest OGG-Informationen."""
        if isinstance(source, bytes):
            return await self._get_info_from_bytes(source)
        return await self._get_info_from_file(source)

    async def _get_info_from_file(self, path: Path) -> AudioInfo:
        """Liest Informationen aus OGG-Datei."""
        if not path.exists():
            raise DecoderError(f"Datei nicht gefunden: {path}")

        file_size = path.stat().st_size

        # Versuche Mutagen
        try:
            import mutagen.oggvorbis
            audio = mutagen.oggvorbis.OggVorbis(path)

            return AudioInfo(
                format=AudioFormat.OGG,
                duration_ms=int(audio.info.length * 1000),
                sample_rate=audio.info.sample_rate,
                channels=audio.info.channels,
                bit_rate=audio.info.bitrate,
                file_size=file_size,
            )
        except ImportError:
            pass
        except Exception:
            # Versuche Opus
            try:
                import mutagen.oggopus
                audio = mutagen.oggopus.OggOpus(path)

                return AudioInfo(
                    format=AudioFormat.OGG,
                    duration_ms=int(audio.info.length * 1000),
                    sample_rate=48000,  # Opus immer 48kHz
                    channels=audio.info.channels,
                    file_size=file_size,
                )
            except Exception:
                pass

        # Fallback: ffprobe
        if self._use_ffmpeg:
            return await self._get_info_ffprobe(path)

        return AudioInfo(
            format=AudioFormat.OGG,
            file_size=file_size,
        )

    async def _get_info_from_bytes(self, data: bytes) -> AudioInfo:
        """Liest Informationen aus OGG-Bytes."""
        # OGG Header parsen für grundlegende Info
        info = AudioInfo(
            format=AudioFormat.OGG,
            file_size=len(data),
        )

        # OGG Magic: "OggS"
        if data[:4] != b"OggS":
            raise DecoderError("Keine gültige OGG-Datei")

        return info

    async def _get_info_ffprobe(self, path: Path) -> AudioInfo:
        """Liest Info mit ffprobe."""
        try:
            cmd = [
                "ffprobe", "-v", "quiet",
                "-print_format", "json",
                "-show_format", "-show_streams",
                str(path)
            ]

            process = await asyncio.create_subprocess_exec(
                *cmd,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
            )
            stdout, _ = await process.communicate()

            import json
            data = json.loads(stdout)

            stream = next((s for s in data.get("streams", []) if s.get("codec_type") == "audio"), {})
            format_info = data.get("format", {})

            return AudioInfo(
                format=AudioFormat.OGG,
                duration_ms=int(float(format_info.get("duration", 0)) * 1000),
                sample_rate=int(stream.get("sample_rate", 44100)),
                channels=int(stream.get("channels", 2)),
                bit_rate=int(format_info.get("bit_rate", 0)),
                file_size=int(format_info.get("size", 0)),
            )

        except Exception as e:
            self.logger.debug(f"ffprobe Fehler: {e}")
            return AudioInfo(format=AudioFormat.OGG)

    async def decode(
        self,
        source: Path | bytes,
        target_sample_rate: int | None = None,
        target_channels: int | None = None,
    ) -> DecodedAudio:
        """Dekodiert OGG vollständig zu PCM."""
        target_sr = target_sample_rate or self.OUTPUT_SAMPLE_RATE
        target_ch = target_channels or self.OUTPUT_CHANNELS

        if self._use_pydub:
            return await self._decode_pydub(source, target_sr, target_ch)
        elif self._use_ffmpeg:
            return await self._decode_ffmpeg(source, target_sr, target_ch)
        else:
            raise DecoderError("Kein OGG-Decoder verfügbar")

    async def _decode_pydub(
        self,
        source: Path | bytes,
        target_sr: int,
        target_ch: int,
    ) -> DecodedAudio:
        """Dekodiert mit pydub."""
        try:
            from pydub import AudioSegment

            def decode():
                if isinstance(source, bytes):
                    audio = AudioSegment.from_ogg(io.BytesIO(source))
                else:
                    audio = AudioSegment.from_ogg(str(source))

                # Resampling falls nötig
                if audio.frame_rate != target_sr:
                    audio = audio.set_frame_rate(target_sr)

                # Channel-Anpassung
                if audio.channels != target_ch:
                    audio = audio.set_channels(target_ch)

                return DecodedAudio(
                    pcm_data=audio.raw_data,
                    sample_rate=audio.frame_rate,
                    channels=audio.channels,
                    bit_depth=audio.sample_width * 8,
                )

            return await asyncio.to_thread(decode)

        except Exception as e:
            raise DecoderError(f"pydub OGG-Dekodierung fehlgeschlagen: {e}")

    async def _decode_ffmpeg(
        self,
        source: Path | bytes,
        target_sr: int,
        target_ch: int,
    ) -> DecodedAudio:
        """Dekodiert mit ffmpeg."""
        try:
            if isinstance(source, bytes):
                cmd = [
                    "ffmpeg", "-y", "-hide_banner", "-loglevel", "error",
                    "-i", "pipe:0",
                    "-f", "s16le",
                    "-acodec", "pcm_s16le",
                    "-ar", str(target_sr),
                    "-ac", str(target_ch),
                    "pipe:1"
                ]
                process = await asyncio.create_subprocess_exec(
                    *cmd,
                    stdin=asyncio.subprocess.PIPE,
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.PIPE,
                )
                stdout, stderr = await process.communicate(input=source)
            else:
                cmd = [
                    "ffmpeg", "-y", "-hide_banner", "-loglevel", "error",
                    "-i", str(source),
                    "-f", "s16le",
                    "-acodec", "pcm_s16le",
                    "-ar", str(target_sr),
                    "-ac", str(target_ch),
                    "pipe:1"
                ]
                process = await asyncio.create_subprocess_exec(
                    *cmd,
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.PIPE,
                )
                stdout, stderr = await process.communicate()

            if process.returncode != 0:
                raise DecoderError(f"ffmpeg Fehler: {stderr.decode()}")

            return DecodedAudio(
                pcm_data=stdout,
                sample_rate=target_sr,
                channels=target_ch,
                bit_depth=16,
            )

        except DecoderError:
            raise
        except Exception as e:
            raise DecoderError(f"ffmpeg OGG-Dekodierung fehlgeschlagen: {e}")

    async def decode_stream(
        self,
        source: Path | bytes,
        chunk_duration_ms: int = 100,
    ) -> AsyncIterator[DecodedAudio]:
        """Dekodiert OGG als Stream."""
        bytes_per_sample = 2
        samples_per_chunk = int(self.OUTPUT_SAMPLE_RATE * chunk_duration_ms / 1000)
        chunk_size = samples_per_chunk * bytes_per_sample * self.OUTPUT_CHANNELS

        if self._use_pydub:
            decoded = await self.decode(source)
            data = decoded.pcm_data

            for i in range(0, len(data), chunk_size):
                yield DecodedAudio(
                    pcm_data=data[i:i + chunk_size],
                    sample_rate=decoded.sample_rate,
                    channels=decoded.channels,
                    bit_depth=decoded.bit_depth,
                )

        elif self._use_ffmpeg:
            async for chunk in self._stream_ffmpeg(source, chunk_size):
                yield chunk

    async def _stream_ffmpeg(
        self,
        source: Path | bytes,
        chunk_size: int,
    ) -> AsyncIterator[DecodedAudio]:
        """Streaming mit ffmpeg."""
        try:
            if isinstance(source, bytes):
                cmd = [
                    "ffmpeg", "-y", "-hide_banner", "-loglevel", "error",
                    "-i", "pipe:0",
                    "-f", "s16le",
                    "-acodec", "pcm_s16le",
                    "-ar", str(self.OUTPUT_SAMPLE_RATE),
                    "-ac", str(self.OUTPUT_CHANNELS),
                    "pipe:1"
                ]
                process = await asyncio.create_subprocess_exec(
                    *cmd,
                    stdin=asyncio.subprocess.PIPE,
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.PIPE,
                )
                if process.stdin:
                    process.stdin.write(source)
                    await process.stdin.drain()
                    process.stdin.close()
            else:
                cmd = [
                    "ffmpeg", "-y", "-hide_banner", "-loglevel", "error",
                    "-i", str(source),
                    "-f", "s16le",
                    "-acodec", "pcm_s16le",
                    "-ar", str(self.OUTPUT_SAMPLE_RATE),
                    "-ac", str(self.OUTPUT_CHANNELS),
                    "pipe:1"
                ]
                process = await asyncio.create_subprocess_exec(
                    *cmd,
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.PIPE,
                )

            while True:
                chunk = await process.stdout.read(chunk_size)
                if not chunk:
                    break

                yield DecodedAudio(
                    pcm_data=chunk,
                    sample_rate=self.OUTPUT_SAMPLE_RATE,
                    channels=self.OUTPUT_CHANNELS,
                    bit_depth=16,
                )

            await process.wait()

        except Exception as e:
            raise DecoderError(f"OGG Streaming fehlgeschlagen: {e}")