Feat: Hermes Voice Client — CustomTkinter GUI + Push-to-Talk + .exe Build
- main.py: CustomTkinter GUI (400x520) mit Status-Indikator, Conversation-Log, Screenshot-Toggle und Settings-Dialog. Thread-sicherer UI-Queue für Hotkey-Callbacks. - hotkey_listener.py: pynput globaler Hotkey (press/release) + KeyCapturer für interaktive Hotkey-Konfiguration im Settings-Dialog. - config_manager.py: JSON-Settings in ~/.hermes-voice/settings.json mit DEFAULTS, load() merged gespeicherte mit Default-Werten. - audio_input.py: Vereinfacht auf start_recording/stop_recording/transcribe (kein Wake-Word-Loop mehr, direkt hotkey-gesteuert). - ai_client.py: Settings-dict statt config.py-Imports, MC2-URL/Modelle konfigurierbar. - requirements.txt: customtkinter>=5.2.0 + pynput>=1.7.6 hinzugefügt, pystray entfernt. - build.bat: PyInstaller --onefile --windowed → dist/HermesVoice.exe. - setup.bat: Veraltete Wake-Word-Hinweise entfernt, GUI-Check angepasst. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -1,154 +1,91 @@
|
||||
"""
|
||||
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
|
||||
Audio-Input: Aufnahme + Whisper STT.
|
||||
Kein Wake-Word-Loop — Aufnahme startet direkt auf Hotkey-Signal.
|
||||
"""
|
||||
|
||||
import os
|
||||
import threading
|
||||
import tempfile
|
||||
import threading
|
||||
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,
|
||||
)
|
||||
from pathlib import Path
|
||||
|
||||
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
|
||||
_model: WhisperModel | None = None
|
||||
_model_size: str = ""
|
||||
_recording = False
|
||||
_frames: list[np.ndarray] = []
|
||||
_stream: sd.InputStream | None = None
|
||||
_lock = threading.Lock()
|
||||
|
||||
|
||||
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 load_model(model_size: str):
|
||||
global _model, _model_size
|
||||
if _model is None or _model_size != model_size:
|
||||
model_dir = Path.home() / ".hermes-voice" / "whisper-models"
|
||||
model_dir.mkdir(parents=True, exist_ok=True)
|
||||
_model = WhisperModel(
|
||||
model_size,
|
||||
device="cpu",
|
||||
compute_type="int8",
|
||||
download_root=str(model_dir),
|
||||
)
|
||||
_model_size = model_size
|
||||
|
||||
|
||||
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)
|
||||
def start_recording():
|
||||
global _recording, _frames, _stream
|
||||
with _lock:
|
||||
_recording = True
|
||||
_frames = []
|
||||
|
||||
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)))
|
||||
def callback(indata, frames, time, status):
|
||||
if _recording:
|
||||
_frames.append(indata[:, 0].copy())
|
||||
|
||||
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)
|
||||
_stream = sd.InputStream(
|
||||
samplerate=SAMPLE_RATE,
|
||||
channels=1,
|
||||
dtype="float32",
|
||||
blocksize=CHUNK,
|
||||
callback=callback,
|
||||
)
|
||||
_stream.start()
|
||||
|
||||
|
||||
def _transcribe(audio: np.ndarray, model_size: str) -> str:
|
||||
"""Schreibt Audio-Array als WAV, transkribiert mit Whisper."""
|
||||
def stop_recording() -> np.ndarray:
|
||||
global _recording, _stream
|
||||
with _lock:
|
||||
_recording = False
|
||||
if _stream:
|
||||
_stream.stop()
|
||||
_stream.close()
|
||||
_stream = None
|
||||
if not _frames:
|
||||
return np.array([], dtype=np.float32)
|
||||
return np.concatenate(_frames)
|
||||
|
||||
|
||||
def transcribe(audio: np.ndarray, model_size: str, language: str) -> str:
|
||||
if len(audio) < SAMPLE_RATE * 0.3:
|
||||
return ""
|
||||
load_model(model_size)
|
||||
|
||||
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(
|
||||
lang = language if language != "auto" else None
|
||||
segments, _ = _model.transcribe(
|
||||
tmp,
|
||||
language=WHISPER_LANGUAGE,
|
||||
language=lang,
|
||||
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