mission-control-v2/client/hermes-voice/audio_input.py

"""
Audio-Input: Energy VAD → Whisper Wake Detection → Whisper STT

Ablauf:
  1. Energie-VAD erkennt Sprache im Mikrofon
  2. Whisper tiny transkribiert den Clip (schnell)
  3. Enthält das Transcript ein Wake Word? → Ding + Befehl aufnehmen
  4. Enthält es Wake Word + Befehl in einem Atemzug? → direkt zurückgeben
"""

import os
import threading
import tempfile
import numpy as np
import sounddevice as sd
import scipy.io.wavfile as wav
from faster_whisper import WhisperModel
from config import (
    WAKE_WORDS, WAKE_WHISPER_MODEL,
    WHISPER_MODEL_SIZE, WHISPER_LANGUAGE,
    SILENCE_TIMEOUT, MAX_RECORD_SECONDS,
)

SAMPLE_RATE = 16000
CHUNK = 1024

# Energie-Schwelle: unter diesem RMS = Stille
# Ggf. anpassen wenn zu sensitiv (höher) oder zu träge (niedriger)
ENERGY_THRESHOLD = 0.008

_whisper_tiny: WhisperModel | None = None
_whisper_main: WhisperModel | None = None


def _get_whisper(size: str) -> WhisperModel:
    global _whisper_tiny, _whisper_main
    if size == WAKE_WHISPER_MODEL:
        if _whisper_tiny is None:
            print(f"[STT] Lade Whisper {size} (Wake Detection)…")
            _whisper_tiny = WhisperModel(size, device="cpu", compute_type="int8")
        return _whisper_tiny
    else:
        if _whisper_main is None:
            print(f"[STT] Lade Whisper {size} (Befehl-Transkription)…")
            _whisper_main = WhisperModel(size, device="cpu", compute_type="int8")
        return _whisper_main


def _record_until_silence(
    stop_event: threading.Event,
    max_seconds: float = 6.0,
    silence_timeout: float = 1.5,
) -> np.ndarray | None:
    """Nimmt Audio auf bis zur Stille oder Timeout. Gibt None zurück wenn gestoppt."""
    frames: list[np.ndarray] = []
    speech_chunks = 0
    silence_chunks = 0
    silence_limit = int(silence_timeout * SAMPLE_RATE / CHUNK)
    max_chunks = int(max_seconds * SAMPLE_RATE / CHUNK)

    with sd.InputStream(samplerate=SAMPLE_RATE, channels=1,
                        dtype="float32", blocksize=CHUNK) as stream:
        for _ in range(max_chunks):
            if stop_event.is_set():
                return None
            data, _ = stream.read(CHUNK)
            chunk = data[:, 0]
            frames.append(chunk.copy())
            energy = float(np.sqrt(np.mean(chunk ** 2)))

            if energy > ENERGY_THRESHOLD:
                speech_chunks += 1
                silence_chunks = 0
            else:
                silence_chunks += 1

            if speech_chunks > 2 and silence_chunks >= silence_limit:
                break

    if speech_chunks < 2:
        return None  # nur Rauschen, kein echter Sprachinhalt

    return np.concatenate(frames)


def _transcribe(audio: np.ndarray, model_size: str) -> str:
    """Schreibt Audio-Array als WAV, transkribiert mit Whisper."""
    with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
        tmp = f.name
    try:
        wav.write(tmp, SAMPLE_RATE, (audio * 32767).astype(np.int16))
        model = _get_whisper(model_size)
        segments, _ = model.transcribe(
            tmp,
            language=WHISPER_LANGUAGE,
            beam_size=3,
            vad_filter=True,
        )
        return " ".join(s.text for s in segments).strip()
    finally:
        os.unlink(tmp)


def _strip_wake_word(text: str) -> str:
    """Entfernt das Wake Word vom Anfang des Textes."""
    lower = text.lower()
    for ww in sorted(WAKE_WORDS, key=len, reverse=True):
        idx = lower.find(ww)
        if idx != -1:
            rest = text[idx + len(ww):].lstrip(" ,.")
            return rest
    return text


def wait_for_wake_word(stop_event: threading.Event) -> bool:
    """
    Lauscht kontinuierlich. Gibt True zurück wenn Wake Word erkannt, False wenn gestoppt.
    Initialisiert Whisper-Modelle beim ersten Aufruf.
    """
    _get_whisper(WAKE_WHISPER_MODEL)  # Modell vorladen
    print(f"[Wake] Höre auf: {WAKE_WORDS}")

    while not stop_event.is_set():
        audio = _record_until_silence(stop_event, max_seconds=6.0, silence_timeout=1.0)
        if audio is None:
            continue

        text = _transcribe(audio, WAKE_WHISPER_MODEL)
        if not text:
            continue

        lower = text.lower()
        if any(ww in lower for ww in WAKE_WORDS):
            return True

    return False


def record_speech() -> np.ndarray:
    """Nimmt den eigentlichen Befehl nach dem Wake Word auf."""
    stop = threading.Event()  # separater Event, läuft immer durch
    audio = _record_until_silence(
        stop,
        max_seconds=MAX_RECORD_SECONDS,
        silence_timeout=SILENCE_TIMEOUT,
    )
    return audio if audio is not None else np.array([], dtype=np.float32)


def transcribe(audio: np.ndarray) -> str:
    """Transkribiert Befehl-Audio mit dem größeren Hauptmodell."""
    if len(audio) < SAMPLE_RATE * 0.3:
        return ""
    return _transcribe(audio, WHISPER_MODEL_SIZE)