Feat: Hermes Voice Client (Wake Word + STT + Vision + TTS)
Windows-Desktop-Script: Energy VAD + Whisper Wake Detection, faster-whisper STT, mss Screen Capture, MC2 Vision/Chat API, Edge TTS Ausgabe, pystray System Tray. Kein Account nötig — Wake Word frei konfigurierbar via WAKE_WORDS in config.py. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,154 @@
|
||||
"""
|
||||
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)
|
||||
Reference in New Issue
Block a user