Trixy7/trixy_core/wakeword/vad.py

# -*- coding: utf-8 -*-
"""
Voice Activity Detection (VAD) für Silence-Erkennung.
"""

from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from typing import Callable
import numpy as np


class VADState(Enum):
    """VAD-Zustand."""
    WAITING = "waiting"      # Warten auf Sprache
    SPEECH = "speech"        # Sprache erkannt
    SILENCE = "silence"      # Stille nach Sprache
    TIMEOUT = "timeout"      # Timeout erreicht
    NO_SPEECH = "no_speech"  # Keine Sprache innerhalb Timeout


@dataclass
class VADConfig:
    """Konfiguration für Voice Activity Detection."""

    # Schwellenwerte
    speech_threshold: float = 0.02     # RMS-Threshold für Sprache
    silence_threshold: float = 0.01    # RMS-Threshold für Stille

    # Zeitkonfiguration
    min_speech_ms: int = 200           # Mindestdauer für Sprache
    silence_duration_ms: int = 3000    # Stille-Dauer für Ende (3 Sekunden)
    max_duration_ms: int = 60000       # Maximale Aufnahmedauer (60 Sekunden)

    # Audio-Format
    sample_rate: int = 16000
    frame_length_ms: int = 80

    # No-Speech Timeout
    no_speech_timeout_ms: int = 5000   # 5 Sekunden ohne Sprache → Abbruch

    # Glättung
    smoothing_frames: int = 3          # Frames für Glättung

    @property
    def frame_length_samples(self) -> int:
        """Samples pro Frame."""
        return int(self.sample_rate * self.frame_length_ms / 1000)


@dataclass
class SpeechSegment:
    """Ein Sprach-Segment."""

    start_time: datetime
    end_time: datetime | None = None
    peak_level: float = 0.0
    avg_level: float = 0.0
    frame_count: int = 0

    @property
    def duration_ms(self) -> float:
        """Dauer in Millisekunden."""
        if self.end_time is None:
            return (datetime.now() - self.start_time).total_seconds() * 1000
        return (self.end_time - self.start_time).total_seconds() * 1000

    @property
    def is_complete(self) -> bool:
        """Prüft ob Segment abgeschlossen ist."""
        return self.end_time is not None


class VoiceActivityDetector:
    """
    Voice Activity Detector für Sprach-/Stille-Erkennung.

    Erkennt:
    - Beginn von Sprache (User hat angefangen zu sprechen)
    - Ende von Sprache (Stille nach Sprache)
    - Timeout (maximale Aufnahmedauer erreicht)
    """

    def __init__(self, config: VADConfig | None = None):
        """
        Initialisiert den VAD.

        Args:
            config: VAD-Konfiguration
        """
        self._config = config or VADConfig()

        # State
        self._state = VADState.WAITING
        self._start_time: datetime | None = None
        self._speech_detected = False
        self._speech_start_time: datetime | None = None
        self._last_speech_time: datetime | None = None

        # Level-Tracking
        self._level_history: list[float] = []
        self._peak_level = 0.0
        self._total_level = 0.0
        self._frame_count = 0

        # Segmente
        self._current_segment: SpeechSegment | None = None
        self._segments: list[SpeechSegment] = []

        # Callbacks
        self._on_speech_start: list[Callable[[], None]] = []
        self._on_speech_end: list[Callable[[float], None]] = []  # Dauer in ms
        self._on_silence: list[Callable[[], None]] = []
        self._on_timeout: list[Callable[[], None]] = []
        self._on_no_speech: list[Callable[[], None]] = []

    @property
    def config(self) -> VADConfig:
        """Gibt Konfiguration zurück."""
        return self._config

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

    @property
    def is_active(self) -> bool:
        """Prüft ob VAD aktiv ist."""
        return self._start_time is not None

    @property
    def has_speech(self) -> bool:
        """Prüft ob Sprache erkannt wurde."""
        return self._speech_detected

    @property
    def duration_ms(self) -> float:
        """Aktuelle Dauer in Millisekunden."""
        if self._start_time is None:
            return 0.0
        return (datetime.now() - self._start_time).total_seconds() * 1000

    @property
    def silence_duration_ms(self) -> float:
        """Aktuelle Stille-Dauer in Millisekunden."""
        if self._last_speech_time is None:
            return 0.0
        return (datetime.now() - self._last_speech_time).total_seconds() * 1000

    @property
    def segments(self) -> list[SpeechSegment]:
        """Alle erkannten Sprach-Segmente."""
        return list(self._segments)

    def start(self) -> None:
        """Startet die VAD-Analyse."""
        self._start_time = datetime.now()
        self._state = VADState.WAITING
        self._speech_detected = False
        self._speech_start_time = None
        self._last_speech_time = None
        self._level_history.clear()
        self._peak_level = 0.0
        self._total_level = 0.0
        self._frame_count = 0
        self._current_segment = None
        self._segments.clear()

    def stop(self) -> None:
        """Stoppt die VAD-Analyse."""
        # Schließe aktuelles Segment ab
        if self._current_segment and not self._current_segment.is_complete:
            self._current_segment.end_time = datetime.now()
            self._segments.append(self._current_segment)
            self._current_segment = None

        self._start_time = None

    def process_frame(self, audio_frame: bytes | np.ndarray) -> VADState:
        """
        Verarbeitet einen Audio-Frame.

        Args:
            audio_frame: Audio-Daten (16-bit PCM oder numpy array)

        Returns:
            Aktueller VAD-Zustand
        """
        if self._start_time is None:
            self.start()

        # Konvertiere zu numpy
        if isinstance(audio_frame, bytes):
            audio = np.frombuffer(audio_frame, dtype=np.int16)
        else:
            audio = audio_frame

        # Berechne RMS-Level
        level = self._calculate_rms(audio)
        self._update_level_tracking(level)

        # Prüfe Timeout
        if self.duration_ms >= self._config.max_duration_ms:
            self._state = VADState.TIMEOUT
            self._trigger_timeout()
            return self._state

        # Geglättetes Level für Entscheidungen
        smoothed_level = self._get_smoothed_level()

        # State-Machine
        if self._state == VADState.WAITING:
            self._handle_waiting_state(smoothed_level)

        elif self._state == VADState.SPEECH:
            self._handle_speech_state(smoothed_level)

        elif self._state == VADState.SILENCE:
            self._handle_silence_state(smoothed_level)

        return self._state

    def _calculate_rms(self, audio: np.ndarray) -> float:
        """Berechnet RMS-Level."""
        if len(audio) == 0:
            return 0.0

        audio_float = audio.astype(np.float32) / 32768.0
        rms = np.sqrt(np.mean(audio_float ** 2))

        return float(rms)

    def _update_level_tracking(self, level: float) -> None:
        """Aktualisiert Level-Tracking."""
        self._level_history.append(level)

        # Begrenze History
        max_history = self._config.smoothing_frames * 3
        while len(self._level_history) > max_history:
            self._level_history.pop(0)

        # Statistiken
        self._peak_level = max(self._peak_level, level)
        self._total_level += level
        self._frame_count += 1

    def _get_smoothed_level(self) -> float:
        """Gibt geglättetes Level zurück."""
        if len(self._level_history) < self._config.smoothing_frames:
            return sum(self._level_history) / len(self._level_history) if self._level_history else 0.0

        # Durchschnitt der letzten N Frames
        recent = self._level_history[-self._config.smoothing_frames:]
        return sum(recent) / len(recent)

    def _handle_waiting_state(self, level: float) -> None:
        """Behandelt WAITING-Zustand."""
        if level >= self._config.speech_threshold:
            # Sprache beginnt
            self._speech_start_time = datetime.now()
            self._last_speech_time = datetime.now()
            self._state = VADState.SPEECH

            # Starte neues Segment
            self._current_segment = SpeechSegment(
                start_time=datetime.now(),
                peak_level=level,
                avg_level=level,
                frame_count=1,
            )
        else:
            # No-Speech Timeout prüfen
            if (self._config.no_speech_timeout_ms > 0
                    and self._start_time
                    and self.duration_ms >= self._config.no_speech_timeout_ms):
                self._state = VADState.NO_SPEECH
                self._trigger_no_speech()

    def _handle_speech_state(self, level: float) -> None:
        """Behandelt SPEECH-Zustand."""
        if level >= self._config.silence_threshold:
            # Immer noch Sprache
            self._last_speech_time = datetime.now()

            # Update Segment
            if self._current_segment:
                self._current_segment.peak_level = max(self._current_segment.peak_level, level)
                self._current_segment.frame_count += 1
                self._current_segment.avg_level = (
                    (self._current_segment.avg_level * (self._current_segment.frame_count - 1) + level)
                    / self._current_segment.frame_count
                )

            # Prüfe ob Mindestdauer erreicht
            if self._speech_start_time:
                speech_duration = (datetime.now() - self._speech_start_time).total_seconds() * 1000
                if speech_duration >= self._config.min_speech_ms and not self._speech_detected:
                    self._speech_detected = True
                    self._trigger_speech_start()

        else:
            # Beginn von Stille
            self._state = VADState.SILENCE

    def _handle_silence_state(self, level: float) -> None:
        """Behandelt SILENCE-Zustand."""
        if level >= self._config.speech_threshold:
            # Sprache wieder da
            self._last_speech_time = datetime.now()
            self._state = VADState.SPEECH

            # Update Segment
            if self._current_segment:
                self._current_segment.peak_level = max(self._current_segment.peak_level, level)
                self._current_segment.frame_count += 1

        else:
            # Prüfe Stille-Dauer (nur wenn vorher Sprache war)
            if self._speech_detected and self._last_speech_time:
                silence_ms = (datetime.now() - self._last_speech_time).total_seconds() * 1000

                if silence_ms >= self._config.silence_duration_ms:
                    # Stille lange genug - Ende
                    self._complete_segment()
                    self._trigger_silence()

    def _complete_segment(self) -> None:
        """Schließt aktuelles Segment ab."""
        if self._current_segment:
            self._current_segment.end_time = datetime.now()
            self._segments.append(self._current_segment)
            self._current_segment = None

    def _trigger_speech_start(self) -> None:
        """Triggert Speech-Start Callbacks."""
        for callback in self._on_speech_start:
            try:
                callback()
            except Exception:
                pass

    def _trigger_silence(self) -> None:
        """Triggert Silence Callbacks."""
        speech_duration = 0.0
        if self._speech_start_time and self._last_speech_time:
            speech_duration = (self._last_speech_time - self._speech_start_time).total_seconds() * 1000

        for callback in self._on_speech_end:
            try:
                callback(speech_duration)
            except Exception:
                pass

        for callback in self._on_silence:
            try:
                callback()
            except Exception:
                pass

    def _trigger_no_speech(self) -> None:
        """Triggert No-Speech Callbacks."""
        for callback in self._on_no_speech:
            try:
                callback()
            except Exception:
                pass

    def _trigger_timeout(self) -> None:
        """Triggert Timeout Callbacks."""
        self._complete_segment()

        for callback in self._on_timeout:
            try:
                callback()
            except Exception:
                pass

    def on_speech_start(self, callback: Callable[[], None]) -> None:
        """Registriert Callback für Sprach-Beginn."""
        self._on_speech_start.append(callback)

    def on_speech_end(self, callback: Callable[[float], None]) -> None:
        """Registriert Callback für Sprach-Ende (mit Dauer in ms)."""
        self._on_speech_end.append(callback)

    def on_silence(self, callback: Callable[[], None]) -> None:
        """Registriert Callback für Stille-Erkennung."""
        self._on_silence.append(callback)

    def on_timeout(self, callback: Callable[[], None]) -> None:
        """Registriert Callback für Timeout."""
        self._on_timeout.append(callback)

    def on_no_speech(self, callback: Callable[[], None]) -> None:
        """Registriert Callback für No-Speech-Timeout."""
        self._on_no_speech.append(callback)

    def get_stats(self) -> dict:
        """Gibt Statistiken zurück."""
        avg_level = self._total_level / self._frame_count if self._frame_count > 0 else 0.0

        return {
            "state": self._state.value,
            "is_active": self.is_active,
            "has_speech": self._speech_detected,
            "duration_ms": self.duration_ms,
            "silence_duration_ms": self.silence_duration_ms,
            "peak_level": self._peak_level,
            "avg_level": avg_level,
            "frame_count": self._frame_count,
            "segment_count": len(self._segments),
        }

    def reset(self) -> None:
        """Setzt VAD zurück."""
        self._state = VADState.WAITING
        self._start_time = None
        self._speech_detected = False
        self._speech_start_time = None
        self._last_speech_time = None
        self._level_history.clear()
        self._peak_level = 0.0
        self._total_level = 0.0
        self._frame_count = 0
        self._current_segment = None
        self._segments.clear()