Trixy7/trixy_core/audio/player.py

# -*- coding: utf-8 -*-
"""
Streaming Audio Player für Trixy Satellites.

Einfacher Thread-basierter Player mit sounddevice.
"""

import threading
import queue
import time
from dataclasses import dataclass
from enum import Enum, auto
from typing import Callable

from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn

# Backend
try:
    import sounddevice as sd
    _HAS_SOUNDDEVICE = True
except ImportError:
    _HAS_SOUNDDEVICE = False
    pwarn("[AUDIO] sounddevice nicht verfügbar")


class PlayerState(Enum):
    """Zustand des Audio-Players."""
    STOPPED = auto()
    BUFFERING = auto()
    PLAYING = auto()
    PAUSED = auto()


@dataclass
class PlayerConfig:
    """Konfiguration für den Audio-Player."""
    sample_rate: int = 44100
    channels: int = 2
    sample_width: int = 2  # 16-bit
    prebuffer_ms: int = 1000

    @property
    def bytes_per_second(self) -> int:
        return self.sample_rate * self.channels * self.sample_width

    @property
    def prebuffer_bytes(self) -> int:
        return (self.bytes_per_second * self.prebuffer_ms) // 1000

    @property
    def frame_size(self) -> int:
        return self.channels * self.sample_width


class BlockingStreamPlayer:
    """Thread-basierter Streaming Player."""

    def __init__(self, config: PlayerConfig | None = None):
        self._config = config or PlayerConfig()
        self._state = PlayerState.STOPPED
        self._lock = threading.Lock()
        self._queue: queue.Queue[bytes | None] = queue.Queue()
        self._prebuffer: list[bytes] = []
        self._prebuffer_size = 0
        self._thread: threading.Thread | None = None
        self._stop_event = threading.Event()
        self._pause_event = threading.Event()  # Für Pause-Funktionalität
        self._stream: "sd.RawOutputStream | None" = None  # Referenz für abort() bei Stop
        self._bytes_played = 0
        self._on_finished: Callable[[], None] | None = None

    @property
    def state(self) -> PlayerState:
        with self._lock:
            return self._state

    @property
    def is_playing(self) -> bool:
        with self._lock:
            return self._state == PlayerState.PLAYING

    @property
    def buffer_ms(self) -> int:
        with self._lock:
            queue_bytes = self._queue.qsize() * 19200
            return ((self._prebuffer_size + queue_bytes) * 1000) // self._config.bytes_per_second

    @property
    def stats(self) -> dict:
        return {
            "state": self.state.name,
            "bytes_played": self._bytes_played,
            "queue_size": self._queue.qsize(),
        }

    def on_finished(self, callback: Callable[[], None]) -> None:
        self._on_finished = callback

    def add_chunk(self, audio_data: bytes) -> bool:
        if not audio_data:
            return False

        should_start = False

        with self._lock:
            if self._state == PlayerState.STOPPED:
                self._state = PlayerState.BUFFERING

            if self._state == PlayerState.BUFFERING:
                self._prebuffer.append(audio_data)
                self._prebuffer_size += len(audio_data)

                if self._prebuffer_size >= self._config.prebuffer_bytes:
                    for chunk in self._prebuffer:
                        self._queue.put(chunk)
                    self._prebuffer.clear()
                    self._prebuffer_size = 0
                    should_start = True
            else:
                self._queue.put(audio_data)

        if should_start:
            self._start_playback_thread()

        return True

    def finish(self) -> None:
        should_start = False

        with self._lock:
            if self._state == PlayerState.BUFFERING and self._prebuffer_size > 0:
                for chunk in self._prebuffer:
                    self._queue.put(chunk)
                self._prebuffer.clear()
                self._prebuffer_size = 0
                should_start = True

        if should_start:
            self._start_playback_thread()

        self._queue.put(None)

    def stop(self) -> None:
        self._stop_event.set()
        self._pause_event.clear()  # Pause aufheben falls aktiv

        # Stream sofort abbrechen - befreit blockierende stream.write() Aufrufe
        stream = self._stream
        if stream is not None:
            try:
                stream.abort()
            except Exception:
                pass

        # Queue leeren und None einfügen um queue.get() aufzuwecken
        try:
            while True:
                self._queue.get_nowait()
        except queue.Empty:
            pass
        self._queue.put(None)

        if self._thread and self._thread.is_alive():
            self._thread.join(timeout=2.0)
            if self._thread.is_alive():
                pdebug("[PLAYER] Thread reagiert nicht nach abort()")
                self._thread.join(timeout=3.0)

        with self._lock:
            self._state = PlayerState.STOPPED
            self._prebuffer.clear()
            self._prebuffer_size = 0

        self._stop_event.clear()

    def pause(self) -> None:
        """Pausiert die Wiedergabe."""
        with self._lock:
            if self._state == PlayerState.PLAYING:
                self._state = PlayerState.PAUSED
                self._pause_event.set()
                pdebug("[PLAYER] Wiedergabe pausiert")

    def resume(self) -> None:
        """Setzt die Wiedergabe fort."""
        with self._lock:
            if self._state == PlayerState.PAUSED:
                self._state = PlayerState.PLAYING
                self._pause_event.clear()
                pdebug("[PLAYER] Wiedergabe fortgesetzt")

    def toggle_pause(self) -> bool:
        """Wechselt zwischen Play und Pause. Gibt True zurück wenn jetzt pausiert."""
        with self._lock:
            if self._state == PlayerState.PLAYING:
                self._state = PlayerState.PAUSED
                self._pause_event.set()
                pdebug("[PLAYER] Wiedergabe pausiert")
                return True
            elif self._state == PlayerState.PAUSED:
                self._state = PlayerState.PLAYING
                self._pause_event.clear()
                pdebug("[PLAYER] Wiedergabe fortgesetzt")
                return False
        return False

    @property
    def is_paused(self) -> bool:
        with self._lock:
            return self._state == PlayerState.PAUSED

    def _start_playback_thread(self) -> None:
        with self._lock:
            if self._state == PlayerState.PLAYING:
                return
            self._state = PlayerState.PLAYING

        self._stop_event.clear()
        self._thread = threading.Thread(target=self._playback_loop, daemon=True)
        self._thread.start()

    def _playback_loop(self) -> None:
        if not _HAS_SOUNDDEVICE:
            perror("[PLAYER] sounddevice nicht verfügbar")
            return

        stream = None
        try:
            stream = sd.RawOutputStream(
                samplerate=self._config.sample_rate,
                channels=self._config.channels,
                dtype='int16',
                latency='high',
            )
            self._stream = stream  # Referenz für stop() -> abort()
            stream.start()

            frame_size = self._config.frame_size
            pending_data = b''

            while not self._stop_event.is_set():
                # Pause-Handling: Warten während pausiert
                while self._pause_event.is_set() and not self._stop_event.is_set():
                    stream.stop()  # Stream pausieren
                    time.sleep(0.1)
                    if not self._pause_event.is_set():
                        stream.start()  # Stream fortsetzen
                        break

                if self._stop_event.is_set():
                    break

                try:
                    chunk = self._queue.get(timeout=0.1)

                    if chunk is None:
                        if pending_data:
                            padding = frame_size - (len(pending_data) % frame_size)
                            if padding < frame_size:
                                pending_data += b'\x00' * padding
                            stream.write(pending_data)
                            self._bytes_played += len(pending_data)
                        break

                    data = pending_data + chunk
                    aligned_len = (len(data) // frame_size) * frame_size

                    if aligned_len > 0:
                        stream.write(data[:aligned_len])
                        self._bytes_played += aligned_len

                    pending_data = data[aligned_len:]

                except queue.Empty:
                    continue
                except sd.PortAudioError:
                    # stream.abort() wurde aufgerufen - normaler Shutdown
                    break
                except Exception as e:
                    perror(f"[PLAYER] Fehler: {e}")
                    break

        except Exception as e:
            perror(f"[PLAYER] Stream-Fehler: {e}")
        finally:
            self._stream = None
            if stream is not None:
                try:
                    stream.close()
                except Exception:
                    pass

        with self._lock:
            self._state = PlayerState.STOPPED

        if self._on_finished:
            try:
                self._on_finished()
            except Exception:
                pass


# Aliase für Kompatibilität
SimpleStreamingPlayer = BlockingStreamPlayer
RobustStreamingPlayer = BlockingStreamPlayer


class AudioOutputManager:
    """Manager für Audio-Ausgabe."""

    def __init__(
        self,
        tts_sample_rate: int = 22050,
        music_sample_rate: int = 44100,
    ):
        self._tts_sample_rate = tts_sample_rate
        self._music_sample_rate = music_sample_rate

        self._tts_player = BlockingStreamPlayer(PlayerConfig(
            sample_rate=tts_sample_rate,
            channels=1,
            sample_width=2,
            prebuffer_ms=500,
        ))

        self._music_player = BlockingStreamPlayer(PlayerConfig(
            sample_rate=music_sample_rate,
            channels=2,
            sample_width=2,
            prebuffer_ms=2000,
        ))

        self._initialized = False

    async def initialize(self) -> bool:
        self._initialized = True
        return True

    async def shutdown(self) -> None:
        pdebug("[AUDIO] AudioOutputManager wird heruntergefahren...")
        self._tts_player.stop()
        self._music_player.stop()
        self._initialized = False
        pdebug("[AUDIO] AudioOutputManager heruntergefahren")

    # TTS
    def play_tts_chunk(self, audio_data: bytes) -> bool:
        return self._tts_player.add_chunk(audio_data)

    def finish_tts(self) -> None:
        self._tts_player.finish()

    def stop_tts(self) -> None:
        self._tts_player.stop()

    @property
    def tts_playing(self) -> bool:
        return self._tts_player.is_playing

    # Musik
    def play_music_chunk(self, audio_data: bytes) -> bool:
        return self._music_player.add_chunk(audio_data)

    def finish_music(self) -> None:
        self._music_player.finish()

    def stop_music(self) -> None:
        self._music_player.stop()

    @property
    def music_playing(self) -> bool:
        return self._music_player.is_playing

    @property
    def music_paused(self) -> bool:
        return self._music_player.is_paused

    def pause_music(self) -> None:
        """Pausiert die Musik-Wiedergabe."""
        self._music_player.pause()

    def resume_music(self) -> None:
        """Setzt die Musik-Wiedergabe fort."""
        self._music_player.resume()

    def toggle_music_pause(self) -> bool:
        """Wechselt zwischen Play und Pause. Gibt True zurück wenn jetzt pausiert."""
        return self._music_player.toggle_pause()

    # Gemeinsam
    def stop_all(self) -> None:
        self._tts_player.stop()
        self._music_player.stop()

    @property
    def stats(self) -> dict:
        return {
            "tts": self._tts_player.stats,
            "music": self._music_player.stats,
        }

    @property
    def music_buffer_full(self) -> bool:
        return self._music_player.buffer_ms > 5000

    @property
    def music_buffer_low(self) -> bool:
        return self._music_player.buffer_ms < 500

    @property
    def music_buffer_ms(self) -> int:
        return self._music_player.buffer_ms

    @property
    def tts_buffer_full(self) -> bool:
        return self._tts_player.buffer_ms > 5000

    def can_accept_music_chunk(self) -> bool:
        return not self.music_buffer_full

    def can_accept_tts_chunk(self) -> bool:
        return not self.tts_buffer_full

    # Format-Update
    async def update_tts_format(
        self,
        sample_rate: int,
        channels: int = 1,
        sample_width: int = 2
    ) -> None:
        self._tts_player.stop()
        self._tts_sample_rate = sample_rate
        self._tts_player = BlockingStreamPlayer(PlayerConfig(
            sample_rate=sample_rate,
            channels=channels,
            sample_width=sample_width,
            prebuffer_ms=500,
        ))

    async def update_music_format(
        self,
        sample_rate: int,
        channels: int = 2,
        sample_width: int = 2
    ) -> None:
        self._music_player.stop()
        self._music_sample_rate = sample_rate
        self._music_player = BlockingStreamPlayer(PlayerConfig(
            sample_rate=sample_rate,
            channels=channels,
            sample_width=sample_width,
            prebuffer_ms=2000,
        ))

    @property
    def tts_format(self) -> dict:
        cfg = self._tts_player._config
        return {"sample_rate": cfg.sample_rate, "channels": cfg.channels, "sample_width": cfg.sample_width}

    @property
    def music_format(self) -> dict:
        cfg = self._music_player._config
        return {"sample_rate": cfg.sample_rate, "channels": cfg.channels, "sample_width": cfg.sample_width}