my_voice_assistant_v2/app/audio/vad.py

"""Einfache energie-basierte Sprachaktivitaetserkennung (VAD).

Reines Python (stdlib `array`), arbeitet auf s16le-PCM (mono). Erkennt das Ende
einer Aeusserung anhand andauernder Stille nach erkannter Sprache. Damit kann der
Server in /ws/voice Aeusserungen automatisch segmentieren, ohne dass der Client
ein explizites Ende-Signal schickt.

Hinweis: Das ersetzt keinen echten Streaming-STT-Dienst (keine wortweisen
Teil-Transkripte) - es bestimmt nur die Aeusserungsgrenzen.
"""

import array
import math


def rms(pcm: bytes) -> float:
    """Lautstaerke (RMS) eines s16le-PCM-Puffers; 0.0 bei leerem Puffer."""
    usable = len(pcm) - (len(pcm) % 2)
    if usable <= 0:
        return 0.0
    samples = array.array("h")
    samples.frombytes(pcm[:usable])
    if not samples:
        return 0.0
    return math.sqrt(sum(s * s for s in samples) / len(samples))


class EnergyVAD:
    def __init__(self, sample_rate: int = 16000, threshold: float = 500.0, silence_ms: float = 700.0):
        self.sample_rate = sample_rate
        self.threshold = threshold
        self.silence_ms = silence_ms
        self._speech_started = False
        self._silence_ms = 0.0

    def feed(self, pcm: bytes) -> bool:
        """Verarbeitet einen Audio-Chunk.

        Liefert True, sobald nach erkannter Sprache genug Stille (silence_ms)
        vergangen ist - die Aeusserung gilt dann als beendet.
        """
        level = rms(pcm)
        n_samples = len(pcm) // 2
        chunk_ms = (n_samples / self.sample_rate) * 1000.0 if self.sample_rate else 0.0

        if level >= self.threshold:
            self._speech_started = True
            self._silence_ms = 0.0
            return False

        if self._speech_started:
            self._silence_ms += chunk_ms
            if self._silence_ms >= self.silence_ms:
                return True
        return False

    def reset(self) -> None:
        self._speech_started = False
        self._silence_ms = 0.0