Trixy7/trixy_core/audio/stt.py

# -*- coding: utf-8 -*-
"""
STT (Speech-to-Text) Interface.

Definiert die Schnittstelle für STT-Provider (Plugins).
"""

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any, AsyncIterator


class STTState(Enum):
    """Zustand des STT-Providers."""
    UNINITIALIZED = "uninitialized"
    READY = "ready"
    PROCESSING = "processing"
    ERROR = "error"


@dataclass
class STTConfig:
    """Konfiguration für STT-Provider."""

    # Audio-Format
    sample_rate: int = 16000
    channels: int = 1
    sample_width: int = 2  # 16-bit = 2 bytes

    # Sprache
    language: str = "de-DE"
    alternative_languages: list[str] = field(default_factory=list)

    # Verhalten
    enable_punctuation: bool = True
    enable_profanity_filter: bool = False
    max_alternatives: int = 1

    # Performance
    use_gpu: bool = False
    num_threads: int = 4

    # Modell
    model_path: str | None = None
    model_name: str | None = None


@dataclass
class WordTiming:
    """Timing-Information für ein Wort."""
    word: str
    start_time: float  # Sekunden
    end_time: float
    confidence: float = 1.0


@dataclass
class STTResult:
    """Ergebnis einer STT-Transkription."""

    # Hauptergebnis
    text: str
    confidence: float = 1.0

    # Alternativen
    alternatives: list[str] = field(default_factory=list)

    # Metadaten
    language: str = "de-DE"
    duration_seconds: float = 0.0
    processing_time_ms: float = 0.0

    # Detaillierte Informationen
    word_timings: list[WordTiming] = field(default_factory=list)
    is_final: bool = True

    # Provider-Info
    provider: str = ""
    model: str = ""

    # Zusätzliche Daten
    raw_result: Any = None

    def to_dict(self) -> dict:
        """Konvertiert zu Dictionary."""
        return {
            "text": self.text,
            "confidence": self.confidence,
            "alternatives": self.alternatives,
            "language": self.language,
            "duration_seconds": self.duration_seconds,
            "processing_time_ms": self.processing_time_ms,
            "is_final": self.is_final,
            "provider": self.provider,
            "model": self.model,
            "word_timings": [
                {
                    "word": w.word,
                    "start": w.start_time,
                    "end": w.end_time,
                    "confidence": w.confidence,
                }
                for w in self.word_timings
            ],
        }


class STTProvider(ABC):
    """
    Abstrakte Basisklasse für STT-Provider.

    Plugins implementieren diese Klasse um Speech-to-Text
    Funktionalität bereitzustellen.

    Events:
        - stt_started: STT-Verarbeitung gestartet
        - stt_completed: STT erfolgreich abgeschlossen
        - stt_error: Fehler bei STT
        - stt_partial: Teilergebnis (bei Streaming)
    """

    def __init__(self, config: STTConfig | None = None):
        """
        Initialisiert den Provider.

        Args:
            config: STT-Konfiguration
        """
        self._config = config or STTConfig()
        self._state = STTState.UNINITIALIZED
        self._model_loaded = False

    @property
    def config(self) -> STTConfig:
        """Aktuelle Konfiguration."""
        return self._config

    @property
    def state(self) -> STTState:
        """Aktueller Zustand."""
        return self._state

    @property
    def is_ready(self) -> bool:
        """Prüft ob Provider bereit ist."""
        return self._state == STTState.READY

    @property
    @abstractmethod
    def name(self) -> str:
        """Name des Providers (z.B. 'whisper', 'vosk')."""
        pass

    @property
    @abstractmethod
    def supported_languages(self) -> list[str]:
        """Liste unterstützter Sprachen."""
        pass

    @property
    def supports_streaming(self) -> bool:
        """Prüft ob Streaming unterstützt wird."""
        return False

    @property
    def supports_word_timings(self) -> bool:
        """Prüft ob Wort-Timings unterstützt werden."""
        return False

    # === Lifecycle ===

    @abstractmethod
    async def initialize(self) -> None:
        """
        Initialisiert den Provider (lädt Modell, etc.).

        Raises:
            RuntimeError: Bei Initialisierungsfehler
        """
        pass

    @abstractmethod
    async def shutdown(self) -> None:
        """Fährt den Provider herunter und gibt Ressourcen frei."""
        pass

    # === Transkription ===

    @abstractmethod
    async def transcribe(
        self,
        audio_data: bytes,
        language: str | None = None,
    ) -> STTResult:
        """
        Transkribiert Audio-Daten zu Text.

        Args:
            audio_data: Raw PCM Audio-Daten (16kHz, 16-bit, mono)
            language: Sprache (optional, default aus Config)

        Returns:
            STTResult mit Transkription

        Raises:
            RuntimeError: Bei Transkriptionsfehler
        """
        pass

    async def transcribe_file(
        self,
        file_path: str,
        language: str | None = None,
    ) -> STTResult:
        """
        Transkribiert eine Audio-Datei.

        Args:
            file_path: Pfad zur Audio-Datei
            language: Sprache (optional)

        Returns:
            STTResult mit Transkription
        """
        import wave

        with wave.open(file_path, "rb") as wf:
            audio_data = wf.readframes(wf.getnframes())

        return await self.transcribe(audio_data, language)

    # === Streaming (optional) ===

    async def transcribe_stream(
        self,
        audio_stream: AsyncIterator[bytes],
        language: str | None = None,
    ) -> AsyncIterator[STTResult]:
        """
        Transkribiert einen Audio-Stream (Streaming-Modus).

        Args:
            audio_stream: Async Iterator von Audio-Chunks
            language: Sprache (optional)

        Yields:
            STTResult mit Teilergebnissen und finalem Ergebnis
        """
        # Default-Implementierung: Sammle alle Daten und transkribiere
        chunks = []
        async for chunk in audio_stream:
            chunks.append(chunk)

        audio_data = b"".join(chunks)
        result = await self.transcribe(audio_data, language)
        yield result

    # === Utilities ===

    def validate_audio(self, audio_data: bytes) -> bool:
        """
        Validiert Audio-Daten.

        Args:
            audio_data: Raw Audio-Daten

        Returns:
            True wenn valide
        """
        if not audio_data:
            return False

        # Mindestlänge prüfen (0.1 Sekunden)
        min_samples = int(self._config.sample_rate * 0.1)
        min_bytes = min_samples * self._config.sample_width * self._config.channels

        return len(audio_data) >= min_bytes

    def get_audio_duration(self, audio_data: bytes) -> float:
        """
        Berechnet Audio-Dauer in Sekunden.

        Args:
            audio_data: Raw Audio-Daten

        Returns:
            Dauer in Sekunden
        """
        bytes_per_sample = self._config.sample_width * self._config.channels
        num_samples = len(audio_data) // bytes_per_sample
        return num_samples / self._config.sample_rate