my_voice_assistant/transcribe.py
dschlueter 53b0dd4055 feat: initial voice assistant implementation
5-state machine (LAUSCH/AUFNAHME/GENERATOR/VORLESE/DIALOG) with:
- Wake-word detection via openwakeword (hey_jarvis)
- Speech-to-text via faster-whisper large-v3 (GPU)
- LLM streaming via OpenAI-compatible backends (llama.cpp/ollama/openai)
- TTS via piper with automatic language detection and voice selection
- Selectable mic input (ReSpeaker, MOTU M2, camera, bluetooth, index)
- Selectable audio output sink
- Control words during playback (Stopp/Pause/Weiter/Noch einmal/…)
- "Stopp" abort in all active states → LAUSCH

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-16 00:43:02 +02:00

115 lines
4 KiB
Python
Executable file

#!/usr/bin/env python3
"""
Live-Transkription via faster-whisper (large-v3, GPU).
Aufnahme-Logik:
- Puffert Audio in CHUNK_SEC-Sekunden-Blöcken
- Faster-whisper VAD filtert Stille heraus
- Ausgabe: fortlaufender Text auf stdout
Verwendung:
python3 transcribe.py # ReSpeaker, GPU 2
python3 transcribe.py --mic motu # MOTU M2
python3 transcribe.py --mic camera # Kamera-Mikrofon
python3 transcribe.py --mic 17 # sounddevice-Index direkt
python3 transcribe.py --list-mics # alle Eingabegeräte anzeigen
python3 transcribe.py [--lang de] [--model large-v3] [--gpu 1]
GPU-Belegung:
GPU 0 NVIDIA T600 (4 GB) — reserviert
GPU 1 RTX 3090 (24 GB) — llama.cpp / Port 8001
GPU 2 RTX 3090 (24 GB) — Whisper (default)
"""
import argparse
import queue
import sys
import numpy as np
import sounddevice as sd
from faster_whisper import WhisperModel
from scipy.signal import resample_poly
WHISPER_RATE = 16000
CHUNK_SEC = 5 # Sekunden pro Transkriptions-Durchlauf
CHANNELS = 1
sys.path.insert(0, __import__("os").path.dirname(__file__))
from mic import resolve_mic, print_mic_list, DEFAULT_MIC # noqa: E402
def main():
parser = argparse.ArgumentParser(
formatter_class=argparse.ArgumentDefaultsHelpFormatter
)
parser.add_argument("--mic", default=DEFAULT_MIC,
help="Eingabegerät: respeaker | motu | camera | default | "
"bluetooth | <index> | <substring>")
parser.add_argument("--list-mics", action="store_true",
help="Verfügbare Mikrofone anzeigen und beenden")
parser.add_argument("--lang", default="de", help="Sprache (de/en/…)")
parser.add_argument("--model", default="large-v3", help="Whisper-Modell")
parser.add_argument("--gpu", type=int, default=1,
help="CUDA-Index innerhalb CUDA_VISIBLE_DEVICES (default: 1 = phys. GPU 2)")
args = parser.parse_args()
if args.list_mics:
print_mic_list()
return
device_idx, capture_rate, device_label = resolve_mic(args.mic)
chunk_samples = CHUNK_SEC * capture_rate
print(f"Mikrofon : {device_label} (Index {device_idx}, {capture_rate} Hz → resample → {WHISPER_RATE} Hz)")
print(f"Modell : {args.model} | Sprache: {args.lang} | GPU: {args.gpu}")
print("Lade Modell …", flush=True)
model = WhisperModel(
args.model,
device="cuda",
device_index=args.gpu,
compute_type="float16",
)
print("Modell geladen. Spreche …\n", flush=True)
audio_q: "queue.Queue[np.ndarray]" = queue.Queue()
def callback(indata, frames, time_info, status):
if status:
print(f"[audio status] {status}", file=sys.stderr)
audio_q.put(indata[:, 0].copy())
try:
with sd.InputStream(
device=device_idx,
channels=CHANNELS,
samplerate=capture_rate,
dtype="float32",
blocksize=capture_rate // 4, # 250 ms Blöcke
callback=callback,
):
buffer = np.empty(0, dtype="float32")
while True:
while buffer.shape[0] < chunk_samples:
buffer = np.concatenate([buffer, audio_q.get()])
chunk = buffer[:chunk_samples].copy()
buffer = buffer[chunk_samples:]
chunk_16k = resample_poly(chunk, WHISPER_RATE, capture_rate)
segments, _ = model.transcribe(
chunk_16k,
language=args.lang,
vad_filter=True,
vad_parameters={"min_silence_duration_ms": 500},
beam_size=5,
)
text = " ".join(s.text.strip() for s in segments).strip()
if text:
print(text, flush=True)
except KeyboardInterrupt:
print("\nBeendet.")
if __name__ == "__main__":
main()