From 6b23d8614af89551fb786317c9b408f31ac94887 Mon Sep 17 00:00:00 2001 From: dschlueter Date: Sun, 13 Sep 2026 21:05:18 +0200 Subject: [PATCH] Initiales Repository: stt.py, historisches stt_respeaker.py, CLAUDE.md, .gitignore Co-Authored-By: Claude Opus 5 --- .gitignore | 33 ++ CLAUDE.md | 56 +++ stt.py | 978 +++++++++++++++++++++++++++++++++++++++++++++++ stt_respeaker.py | 476 +++++++++++++++++++++++ 4 files changed, 1543 insertions(+) create mode 100644 .gitignore create mode 100644 CLAUDE.md create mode 100755 stt.py create mode 100644 stt_respeaker.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ac146a5 --- /dev/null +++ b/.gitignore @@ -0,0 +1,33 @@ +# Python +__pycache__/ +*.py[cod] +*.egg-info/ +.eggs/ +.venv/ +venv/ +.pytest_cache/ + +# Secrets / lokale Konfiguration +.env + +# Audio-Aufnahmen und Ausgaben (persönliche Daten) +*.wav +*.mp3 +*.ogg +*.flac + +# Whisper-/Modell-Caches +models/ + +# Logs +*.log + +# Editor / Tool-Konfiguration +.vscode/ +.idea/ +.claude/ + +# Editor-/Backup-Reste +*.bak +*.orig +*.patch diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..a1b8802 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,56 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +> Sprache: Code-Kommentare, Doku und Commits auf Deutsch (Umlaute in Code/Kommentaren bisher meist als ae/oe/ue). + +## Projektziel + +Lokale, schnelle, zuverlässige Spracherkennung (Sprache → Text), nicht blockierend. Soll später als +Baustein in Smarthome-Szenarien dienen (Home Assistant, Mikrocontroller wie ESP32/Arduino als +Audio-Clients) und das **STT-Pendant zum TTS-Projekt `chatterbox-tts-cli`** werden +(https://kitux.de/forgejo/dschlueter/chatterbox-tts-cli, lokaler Klon `~/chatterbox-tts-cli`: +CLI-Kern + FastAPI-Service mit Job-Queue/Worker-Thread + MCP-Adapter + systemd-User-Service). + +Referenz für bewährtes STT: `~/my_voice_assistant_v3` (https://kitux.de/forgejo/dschlueter/my_voice_assistant_v3), +v. a. `app/providers/stt/faster_whisper.py` (Modell einmal laden + cachen, Inferenz via `asyncio.to_thread`, +CPU-Fallback), `app/audio/vad.py` (Äußerungsende per Stille), `app/api/ws.py` (`/ws/voice`), `app/core/warmup.py`. + +## Dateien + +- `stt.py` — **aktueller Stand**, eigenständige CLI (einzige relevante Datei). +- `stt_respeaker.py` — **historisch**, nicht weiterentwickeln und nicht als Vorlage nutzen. + +Kein Build, keine Tests, keine Linter-Konfiguration. Git-Remote auf kitux.de (Forgejo) ist geplant, aber noch nicht eingerichtet. + +## Befehle + +```bash +pip install faster-whisper sounddevice numpy scipy requests # plus Systemtools: espeak-ng, ffmpeg, pactl/paplay, ollama + +python3 stt.py --list-devices # Geräte inkl. nativer Sampleraten aus /proc/asound +python3 stt.py # Live-Aufnahme, Sprache de, Stoppwort "finito" +python3 stt.py en -i respeaker -m large-v3 # Positional-Arg = Sprache +python3 stt.py --file audio.wav --no-llm # Datei-Modus (nur int16-Mono-WAV korrekt) +python3 -m py_compile stt.py # Syntax-Check +``` + +Rauchtest ohne Mikrofon/GPU: `python3 stt.py --file --no-llm -m tiny --device cpu`. + +## GPU-Umgebung + +T600 + 2× RTX 3090. `nvidia-smi` zählt in PCI-Reihenfolge, CUDA/torch/ctranslate2 standardmäßig „fastest first“ — +Indizes stimmen nicht überein. Andere Dienste (llama.cpp, chatterbox, open-notebook-stt auf :8902) pinnen GPUs per +`CUDA_VISIBLE_DEVICES=GPU-` (`nvidia-smi -L`); so auch hier verfahren statt per Index. + +## Architektur `stt.py` (Ist-Zustand) + +Ablauf in `main()`: `detect_compute()` → ein `WhisperModel` (für Stoppwort-Chunks **und** finale Transkription) → +`resolve_device()` (Default via `pactl` → `alsa.card` → sounddevice-Index `hw:N`) → `play_announcement()` +(espeak-ng → ffmpeg → `sd.play`, Fallback `paplay`) → `record_until_stopword()` (Callback-Stream schreibt +2-s-Blöcke in Queue; Hauptthread: Mono → 16 kHz → RMS-Gate → Whisper-Stoppwortprüfung) → Peak-Normalisierung → +WAV in `/tmp/stt_*.wav` → `transcribe()` → `llm_correct()` (Ollama `/api/chat`, `localhost:11434`). + +- `stereo_to_mono()`: ReSpeaker XVF3800 → nur rechter Kanal (ASR-Output); Mischen verschlechtert die Erkennung. +- Stoppwort-Entfernung aus dem Text übernimmt nur der LLM-Prompt (`LLM_SYSTEM_PROMPT`, nur de/en). +- Alle Ausgaben per `print`; das Endergebnis steht nach der Zeile `FINALES ERGEBNIS` auf stdout. diff --git a/stt.py b/stt.py new file mode 100755 index 0000000..16a1d78 --- /dev/null +++ b/stt.py @@ -0,0 +1,978 @@ +#!/usr/bin/env python3 +""" +STT-Tool: Nimmt Audio von einem Mikrofon auf und transkribiert es +mit faster-whisper (GPU/CPU) + optionale LLM-Korrektur via Ollama. + +Verwendung: + python3 stt.py # Default: System-Mic, medium, mistral-small3.2:24b + python3 stt.py --input-device respeaker # ReSpeaker XVF3800 verwenden + python3 stt.py --stt-model large-v3 # Groesseres Whisper-Modell + python3 stt.py --llm-model gemma3:latest # Anderes Ollama-Modell + python3 stt.py --no-llm # Ohne LLM-Korrektur + python3 stt.py --file audio.wav # WAV-Datei transkribieren + python3 stt.py --list-devices # Verfuegbare Geraete auflisten + python3 stt.py --list-models # Verfuegbare Ollama-Modelle auflisten + +Voraussetzungen: + pip install faster-whisper sounddevice numpy scipy requests + espeak-ng installiert (fuer TTS-Ansage) + ollama + gewaehltes LLM-Modell installiert +""" + +import sys +import os +import wave +import time +import tempfile +import argparse +import subprocess +import numpy as np +import sounddevice as sd +from scipy.signal import resample_poly +from faster_whisper import WhisperModel +import requests +import queue as queue_module + +sys.stdout.reconfigure(line_buffering=True) + +# === Defaults ================================================================ +WHISPER_SR = 16000 +DEFAULT_LANG = "de" +DEFAULT_STT_MODEL = "medium" +DEFAULT_LLM_MODEL = "mistral-small3.2:24b" +DEFAULT_MAX_DURATION = 120 +DEFAULT_VAD_THRESHOLD = 0.008 +DEFAULT_ANNOUNCEMENT_VOLUME = 3.0 +CHUNK_SEC = 2.0 + +INITIAL_PROMPT = "" +# Leer = kein Kontext-Hinweis. Kann per --initial-prompt gesetzt werden, +# z.B.: "Nachrichten, Politik, Friedrich Merz, CDU, CSU" + +STOPWORDS = { + "de": ["finito", "stop"], + "en": ["finito", "stop", "done", "finish"], + "fr": ["finito", "stop", "termine"], + "es": ["finito", "stop", "terminado"], +} + +LLM_SYSTEM_PROMPT = { + "de": ( + "Du bist ein Korrektor fuer deutsche Sprachtranskripte. " + "Korrigiere Grammatik, Rechtschreibung, korrigiere falsch verstandene " + "Woerter und ergaenze offensichtlich fehlende Woerter zu einem sinnvollen Text. " + "Achte auf korrekte Namen bekannter Personen und Orte. " + "Entferne das Stoppwort 'Finito' am Schluss falls vorhanden. " + "WICHTIG: Fuege KEINEN neuen Inhalt hinzu, der nicht im Transkript steht. " + "Ergaenze nur fehlende Woerter innerhalb vorhandener Saetze. " + "Erfinde keine neuen Saetze oder Fakten. " + "Antworte NUR mit dem korrigierten Text, ohne Einleitung oder Kommentar." + ), + "en": ( + "You are a proofreader for English speech transcripts. " + "Correct grammar, spelling, correct wrong understood words " + "and add missing words to make coherent text. " + "Remove the stopword 'finito' at the end. " + "IMPORTANT: Do NOT add any new content not present in the transcript. " + "Only fill in missing words within existing sentences. " + "Do not invent new sentences or facts. " + "Reply ONLY with the corrected text, no preamble." + ), +} + + +# === GPU/CPU Auto-Detection ================================================== + +def detect_compute(): + """Erkennt CUDA-Verfuegbarkeit und waehlt die GPU mit dem meisten freien VRAM.""" + try: + import ctranslate2 + cuda_count = ctranslate2.get_cuda_device_count() + if cuda_count > 0: + best_gpu = 0 + best_free = 0 + try: + import torch + if torch.cuda.is_available(): + for i in range(min(torch.cuda.device_count(), cuda_count)): + free, total = torch.cuda.mem_get_info(i) + if free > best_free: + best_free = free + best_gpu = i + except Exception: + pass + return "cuda", "float16", best_gpu + except Exception: + pass + return "cpu", "int8", None + + +# === Device Helpers ========================================================== + +def get_native_sample_rates(): + """Liest native Hardware-Sampleraten aus /proc/asound aus. + Gibt ein Dict {card_num: [rates]} zurueck. + + Quellen: + - USB-Audio: /proc/asound/cardN/stream0 (Zeile 'Rates: 44100, 48000, ...') + - HDA-Codec: /proc/asound/cardN/codec#0 (Zeile 'rates [0x7f0]: 32000 44100 ...') + """ + import re + rates_by_card = {} + proc_dir = '/proc/asound' + if not os.path.isdir(proc_dir): + return rates_by_card + for card_dir in sorted(os.listdir(proc_dir)): + if not card_dir.startswith('card'): + continue + card_num = card_dir.replace('card', '') + all_rates = set() + + # Quelle 1: stream0 (USB-Audio) + stream_file = os.path.join(proc_dir, card_dir, 'stream0') + if os.path.isfile(stream_file): + try: + with open(stream_file) as f: + content = f.read() + # 'Rates: 44100, 48000, 88200, ...' + for match in re.finditer(r'Rates:\s*([\d, ]+)', content): + for x in re.findall(r'\d+', match.group(1)): + all_rates.add(int(x)) + except Exception: + pass + + # Quelle 2: codec#0 (HDA-Codec: ALC1220, HDMI) + codec_file = os.path.join(proc_dir, card_dir, 'codec#0') + if os.path.isfile(codec_file): + try: + with open(codec_file) as f: + content = f.read() + # 'rates [0x7f0]: 32000 44100 48000 ...' + for match in re.finditer(r'rates\s*\[0x[0-9a-f]+\]:\s*([\d ]+)', + content): + for x in match.group(1).split(): + if x.isdigit(): + all_rates.add(int(x)) + except Exception: + pass + + if all_rates: + rates_by_card[card_num] = sorted(all_rates) + return rates_by_card + + +def list_devices(): + """Listet alle verfuegbaren Audio-Geraete auf.""" + devices = sd.query_devices() + native_rates = get_native_sample_rates() + print("Verfuegbare Audio-Geraete:") + print(f"{'Idx':>4} {'Typ':>6} {'Kanäle':>10} {'Samplerate':>16} Name") + print("-" * 75) + for i, dev in enumerate(devices): + if dev["max_input_channels"] > 0 or dev["max_output_channels"] > 0: + typ = [] + if dev["max_input_channels"] > 0: + typ.append(f"In:{dev['max_input_channels']}") + if dev["max_output_channels"] > 0: + typ.append(f"Out:{dev['max_output_channels']}") + # Native Hardware-Rate aus /proc/asound suchen + hw_rate = "?" + name_lower = dev["name"].lower() + for card_num, rates in native_rates.items(): + if f"hw:{card_num}" in name_lower or f"card{card_num}" in name_lower: + hw_rate = ",".join(str(r) for r in rates) + break + # Fuer PipeWire/Pulse-Geraete: versuche Name zu matchen + if hw_rate == "?" and "respeaker" in name_lower: + hw_rate = "48000" + print(f" [{i:>2}] {', '.join(typ):>10} {hw_rate:>14} {dev['name']}") + # Echten System-Default anzeigen (PipeWire, nicht PortAudio) + real_default_in = get_system_default_device("input") + real_default_out = get_system_default_device("output") + if real_default_in is not None: + print(f"\nDefault Input: [{real_default_in}] {devices[real_default_in]['name']}") + else: + di = sd.default.device[0] + print(f"\nDefault Input: [{di}] " + f"{devices[di]['name'] if di is not None else 'keiner'} " + f"(PortAudio-Fallback)") + if real_default_out is not None: + print(f"Default Output: [{real_default_out}] {devices[real_default_out]['name']}") + else: + do = sd.default.device[1] + print(f"Default Output: [{do}] " + f"{devices[do]['name'] if do is not None else 'keiner'} " + f"(PortAudio-Fallback)") + + +def find_device(keyword, kind="input"): + """Findet ein Geraet per Keyword-Suche (case-insensitive). + + kind: 'input' oder 'output'. + Gibt den Device-Index zurueck oder None. + """ + devices = sd.query_devices() + channel_key = "max_input_channels" if kind == "input" else "max_output_channels" + for i, dev in enumerate(devices): + name_lower = dev["name"].lower() + if keyword.lower() in name_lower and dev[channel_key] > 0: + return i + return None + + +def resolve_device(spec, kind="input"): + """Loest eine Device-Spezifikation auf. + + Spezifikationen: + - None oder 'default': System-Default + - Zahl (z.B. '5'): Device-Index + - String (z.B. 'respeaker'): Keyword-Suche + """ + if spec is None or spec == "default": + # Zuerst PipeWire/PulseAudio-Default abfragen (portaudio ist unzuverlaessig) + idx = get_system_default_device(kind) + if idx is not None: + return idx + # Fallback: PortAudio-Default + return sd.default.device[0] if kind == "input" else sd.default.device[1] + + # Numerisch? + try: + idx = int(spec) + devices = sd.query_devices() + if 0 <= idx < len(devices): + channel_key = ("max_input_channels" if kind == "input" + else "max_output_channels") + if devices[idx][channel_key] > 0: + return idx + else: + print(f"FEHLER: Device [{idx}] hat keine " + f"{'Input' if kind == 'input' else 'Output'}-Kanaele.") + sys.exit(1) + else: + print(f"FEHLER: Device-Index {idx} existiert nicht " + f"(0-{len(devices)-1}).") + sys.exit(1) + except ValueError: + pass + + # Keyword-Suche + idx = find_device(spec, kind) + if idx is not None: + return idx + + print(f"FEHLER: Kein {kind}-Geraet mit Namen '{spec}' gefunden.") + print("Verfuegbare Geraete (siehe --list-devices):") + devices = sd.query_devices() + channel_key = "max_input_channels" if kind == "input" else "max_output_channels" + for i, dev in enumerate(devices): + if dev[channel_key] > 0: + print(f" [{i}] {dev['name']}") + sys.exit(1) + + +def get_device_info(device_idx, kind="input"): + """Fragt Samplerate und Kanalzahl eines Geraets ab.""" + if device_idx is None: + device_idx = (sd.default.device[0] if kind == "input" + else sd.default.device[1]) + devices = sd.query_devices() + if device_idx is not None and 0 <= device_idx < len(devices): + dev = devices[device_idx] + channel_key = ("max_input_channels" if kind == "input" + else "max_output_channels") + max_ch = dev[channel_key] + sr = int(dev["default_samplerate"]) + name = dev["name"] + return {"index": device_idx, "name": name, + "channels": min(max_ch, 2), "samplerate": sr} + return {"index": None, "name": "default", + "channels": 2, "samplerate": 48000} + + +def is_respeaker(device_name): + """Prueft ob das Geraet ein ReSpeaker XVF3800 ist.""" + name_lower = device_name.lower() + return any(k in name_lower for k in ["xvf3800", "respeaker", "array"]) + + +def get_system_default_device(kind="input"): + """Fragt das echte System-Default-Geraet von PipeWire/PulseAudio ab. + + PortAudio/sounddevice liefert oft das falsche Default-Geraet (den ersten + ALSA-Hardware-Treiber statt den PipeWire-Default). Diese Funktion nutzt + 'pactl' um den echten System-Default zu finden und auf den passenden + sounddevice-Index abzubilden. + + kind: 'input' oder 'output' + Gibt den sounddevice-Index zurueck oder None. + """ + import re as _re + if kind == "input": + cmd_default = ["pactl", "get-default-source"] + cmd_list = ["pactl", "list", "sources"] + else: + cmd_default = ["pactl", "get-default-sink"] + cmd_list = ["pactl", "list", "sinks"] + + # 1. Default-Source/Sink-Name abfragen + try: + result = subprocess.run(cmd_default, capture_output=True, + text=True, timeout=5) + if result.returncode != 0 or not result.stdout.strip(): + return None + default_name = result.stdout.strip() + except Exception: + return None + + # 2. alsa.card fuer dieses Geraet finden + try: + result = subprocess.run(cmd_list, capture_output=True, + text=True, timeout=5) + output = result.stdout + except Exception: + return None + + # Sektionen durchsuchen + sections = _re.split(r'\n(?=Quelle #|Ziel #|Source #|Sink #)', output) + for section in sections: + if f"Name: {default_name}" in section: + card_match = _re.search(r'alsa\.card\s*=\s*"(\d+)"', section) + if card_match: + card_num = card_match.group(1) + # Auf sounddevice-Index mappen + devices = sd.query_devices() + channel_key = ("max_input_channels" if kind == "input" + else "max_output_channels") + for i, dev in enumerate(devices): + if (f"hw:{card_num}" in dev["name"] + and dev[channel_key] > 0): + return i + break + + # 3. Fallback: Keyword aus PA-Namen extrahieren + name_lower = default_name.lower() + if "respeaker" in name_lower or "xvf3800" in name_lower: + return find_device("respeaker", kind) + elif "motu" in name_lower: + return find_device("motu", kind) + + return None + + +# === Audio Processing ======================================================== + +def stereo_to_mono(audio, device_name): + """Extrahiert Mono-Audio aus einem Mehrkanal-Stream. + + Fuer ReSpeaker XVF3800: rechter Kanal (ASR-Output, fuer STT optimiert). + Fuer andere Geraete: Durchschnitt aller Kanaele. + """ + if audio.ndim == 1: + return audio + if is_respeaker(device_name) and audio.shape[1] >= 2: + # Rechter Kanal = ASR-Output (Index 1) + return audio[:, 1] + # Durchschnitt aller Kanaele + return audio.mean(axis=1) + + +def resample(audio, src_sr, target_sr=WHISPER_SR): + """Resampelt Audio auf die Ziel-Samplerate.""" + if src_sr == target_sr: + return audio + from math import gcd + g = gcd(src_sr, target_sr) + up = target_sr // g + down = src_sr // g + return resample_poly(audio, up, down) + + +def compute_rms(audio_float): + if len(audio_float) == 0: + return 0.0 + return float(np.sqrt(np.mean(audio_float ** 2))) + + +def normalize_audio(audio, target_peak=0.7): + """Normalisiert Audio auf Ziel-Peak-Level (-3 dBFS).""" + peak = float(np.max(np.abs(audio))) + if peak > 0.001: + gain = target_peak / peak + audio = audio * gain + audio = np.clip(audio, -1.0, 1.0) + print(f" Normalisierung: Gain x{gain:.1f} " + f"(Peak {peak:.3f} -> {target_peak:.3f})") + else: + print(" Warnung: Sehr leises Signal, keine Normalisierung.") + return audio + + +def save_wav(audio_data, filename, sr=WHISPER_SR): + audio_int16 = (audio_data * 32767).astype(np.int16) + with wave.open(filename, "wb") as wf: + wf.setnchannels(1) + wf.setsampwidth(2) + wf.setframerate(sr) + wf.writeframes(audio_int16.tobytes()) + + +# === TTS Announcement ======================================================== + +def play_announcement(text, output_device_idx, lang="de", volume=3.0): + """Erzeugt TTS mit espeak-ng und spielt es ueber das Ausgabegeraet ab.""" + print(f"Ansage: '{text}'") + + tmp_wav = tempfile.NamedTemporaryFile( + suffix=".wav", delete=False, prefix="tts_" + ) + tmp_wav.close() + + espeak_lang = {"de": "de", "en": "en", "fr": "fr", "es": "es"}.get(lang, "de") + subprocess.run( + ["espeak-ng", "-v", espeak_lang, "-s", "150", + "-w", tmp_wav.name, text], + check=True, capture_output=True + ) + + # Verstaerken und konvertieren + tmp_loud = tempfile.NamedTemporaryFile( + suffix=".wav", delete=False, prefix="tts_loud_" + ) + tmp_loud.close() + + out_info = get_device_info(output_device_idx, "output") + out_sr = out_info["samplerate"] + out_ch = out_info["channels"] + + subprocess.run( + ["ffmpeg", "-y", "-i", tmp_wav.name, + "-filter:a", f"volume={volume}", + "-ar", str(out_sr), "-ac", str(out_ch), + tmp_loud.name], + check=True, capture_output=True + ) + + # Wiedergabe ueber sounddevice (funktioniert mit jedem Device) + try: + with wave.open(tmp_loud.name, "rb") as wf: + frames = wf.readframes(wf.getnframes()) + audio = np.frombuffer(frames, dtype=np.int16) + sr = wf.getframerate() + ch = wf.getnchannels() + if ch == 2: + audio_stereo = audio.reshape(-1, 2) + else: + audio_stereo = audio.reshape(-1, 1) + sd.play(audio_stereo, samplerate=sr, device=output_device_idx) + sd.wait() + except Exception as e: + # Fallback: paplay + print(f" sounddevice-Wiedergabe fehlgeschlagen ({e}), versuche paplay...") + sink = None + if output_device_idx is not None: + devices = sd.query_devices() + if output_device_idx < len(devices): + name = devices[output_device_idx]["name"] + # Versuche den PulseAudio-Sink-Namen zu erraten + if is_respeaker(name): + sink = ("alsa_output.usb-Seeed_Studio_reSpeaker_" + "XVF3800_4-Mic_Array_114993700261100055-00" + ".analog-stereo") + cmd = ["paplay", "--volume", "1.5"] + if sink: + cmd += ["--device", sink] + cmd += [tmp_loud.name] + try: + subprocess.run(cmd, check=True, capture_output=True) + except subprocess.CalledProcessError: + print(" Warnung: Wiedergabe nicht moeglich — Ansage uebersprungen.") + + os.unlink(tmp_wav.name) + os.unlink(tmp_loud.name) + print(" (Ansage beendet)\n") + + +# === Whisper STT ============================================================= + +def load_whisper_model(model_size, device, compute_type, gpu_index): + """Laedt ein Whisper-Modell.""" + print(f"Lade Whisper-Modell '{model_size}' " + f"(device={device}, compute={compute_type})...") + if device == "cpu": + model = WhisperModel(model_size, device=device, compute_type=compute_type) + else: + model = WhisperModel(model_size, device=device, + compute_type=compute_type, + device_index=gpu_index) + print(" Modell bereit.") + return model + + +def transcribe(model, audio_data, language, initial_prompt=None, + vad_filter=False, beam_size=5): + """Transkribiert Audio mit faster-whisper.""" + print("Transkribiere...\n") + + kwargs = dict( + language=language, + beam_size=beam_size, + vad_filter=vad_filter, + ) + if initial_prompt: + kwargs["initial_prompt"] = initial_prompt + + segments, info = model.transcribe(audio_data, **kwargs) + + print(f" Sprache: {info.language} ({info.language_probability:.1%})") + print(f" Dauer: {info.duration:.1f}s\n") + + full_text = "" + print("--- Whisper-Transkription (roh) ----------------------") + for seg in segments: + ts = f"[{seg.start:5.1f}s - {seg.end:5.1f}s]" + print(f" {ts} {seg.text}") + full_text += seg.text + " " + print("------------------------------------------------------\n") + + return full_text.strip() + + +def check_stopword(audio_16k, model, language, stopword_list, vad_threshold): + """Transkribiert einen Chunk und prueft auf Stoppwort.""" + rms = compute_rms(audio_16k) + if rms < vad_threshold: + return False, "", rms + + segments, _ = model.transcribe( + audio_16k, + language=language, + beam_size=1, + vad_filter=False, # VAD wuerde kurze Stopwoerter filtern + temperature=0.0, + ) + text = " ".join(seg.text.strip().lower() for seg in segments) + + # Wortgrenzen-Matching: nur ganze Woerter, keine Teilwort-Treffer + # ('over' in 'wochenende' soll NICHT matchen) + import re + for word in stopword_list: + if re.search(r'\b' + re.escape(word) + r'\b', text): + return True, text, rms + return False, text, rms + + +def record_until_stopword(input_device_idx, input_info, model, language, + stopword_list, vad_threshold, max_duration): + """Nimmt kontinuierlich auf bis Stoppwort erkannt wird. + Verwendet callback-basierte Aufnahme mit Queue, damit Audio auch waehrend + der Whisper-Transkription kontinuierlich aufgezeichnet wird (keine Buffer + Overflows).""" + device_sr = input_info["samplerate"] + device_channels = input_info["channels"] + device_name = input_info["name"] + + print(f"Aufnahme laeuft — sage '{stopword_list[0]}' zum Stoppen.") + print(f" Geraet: [{input_device_idx}] {device_name} " + f"({device_channels} Kanäle, {device_sr} Hz)") + print(f" Max. {max_duration}s\n") + + chunk_samples = int(CHUNK_SEC * device_sr) + all_audio = [] + total_duration = 0.0 + + # Callback-basierte Aufnahme: Audio laeuft in Queue, Hauptthread verarbeitet + audio_queue = queue_module.Queue(maxsize=64) + + def audio_callback(indata, frames, time_info, status): + if status: + print(f" Stream-Status: {status}") + try: + audio_queue.put_nowait(indata.copy()) + except queue_module.Full: + pass # Queue voll — alter Daten verwerfen + + stream = sd.InputStream( + samplerate=device_sr, + channels=device_channels, + dtype="float32", + device=input_device_idx, + blocksize=chunk_samples, + callback=audio_callback, + ) + stream.start() + + try: + while total_duration < max_duration: + try: + chunk = audio_queue.get(timeout=max_duration) + except queue_module.Empty: + break + + chunk_mono = stereo_to_mono(chunk, device_name) + chunk_16k = resample(chunk_mono, device_sr, WHISPER_SR) + all_audio.append(chunk_mono) + + total_duration += CHUNK_SEC + rms = compute_rms(chunk_16k) + + if rms >= vad_threshold: + is_stop, text, rms = check_stopword( + chunk_16k, model, language, stopword_list, vad_threshold + ) + bar = "#" * min(int(rms * 200), 30) + status = f" [{total_duration:5.1f}s] RMS:{rms:.3f} {bar}" + if text: + status += f" '{text}'" + print(status) + + if is_stop: + print(f"\n>>> Stoppwort erkannt: '{text}'") + break + else: + print(f" [{total_duration:5.1f}s] RMS:{rms:.3f} (Stille)") + finally: + stream.stop() + stream.close() + + # Restliche Queue leeren + while not audio_queue.empty(): + try: + chunk = audio_queue.get_nowait() + chunk_mono = stereo_to_mono(chunk, device_name) + all_audio.append(chunk_mono) + except queue_module.Empty: + break + + full_audio = np.concatenate(all_audio) + full_audio_16k = resample(full_audio, device_sr, WHISPER_SR) + full_audio_16k = normalize_audio(full_audio_16k) + + print(f"\nAufnahme beendet. Gesamtdauer: " + f"{len(full_audio_16k)/WHISPER_SR:.1f}s\n") + return full_audio_16k + + +# === LLM Correction ========================================================== + +def llm_correct(text, language, llm_model, llm_url): + """Korrigiert den transkribierten Text mit einem lokalen LLM (Ollama).""" + if not text: + return text + + system_prompt = LLM_SYSTEM_PROMPT.get(language, LLM_SYSTEM_PROMPT["de"]) + print(f"LLM-Korrektur mit '{llm_model}' (Ollama)...\n") + + try: + response = requests.post( + llm_url, + json={ + "model": llm_model, + "messages": [ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": text}, + ], + "stream": False, + "options": { + "temperature": 0.0, + "num_predict": 500, + }, + }, + timeout=180, + ) + response.raise_for_status() + result = response.json().get("message", {}).get("content", "").strip() + + if result: + print("--- LLM-korrigierter Text ----------------------------") + print(f" {result}") + print("------------------------------------------------------\n") + return result + else: + print("LLM lieferte leere Antwort — verwende Whisper-Text.\n") + return text + + except requests.exceptions.RequestException as e: + print(f"Fehler bei LLM-Korrektur: {e}") + print("Verwende unkorrigierten Whisper-Text.\n") + return text + + +def list_ollama_models(): + """Listet verfuegbare Ollama-Modelle auf.""" + try: + result = subprocess.run( + ["ollama", "list"], + capture_output=True, text=True + ) + print(result.stdout) + except FileNotFoundError: + print("ollama nicht gefunden oder nicht im PATH.") + + +# === File Mode =============================================================== + +def transcribe_file(filepath, language, model, initial_prompt, vad_filter, + beam_size, use_llm, llm_model, llm_url): + """Transkribiert eine vorhandene WAV-Datei.""" + print(f"Lade Datei: {filepath}\n") + with wave.open(filepath, "rb") as wf: + frames = wf.readframes(wf.getnframes()) + audio_int16 = np.frombuffer(frames, dtype=np.int16) + audio_float = audio_int16.astype(np.float32) / 32767.0 + file_sr = wf.getframerate() + + print(f" Dauer: {len(audio_float) / file_sr:.1f}s") + print(f" Samplerate: {file_sr} Hz\n") + + if file_sr != WHISPER_SR: + print(f" Resample: {file_sr} -> {WHISPER_SR} Hz") + audio_float = resample(audio_float, file_sr, WHISPER_SR) + + whisper_text = transcribe(model, audio_float, language, initial_prompt, + vad_filter, beam_size) + + if whisper_text and use_llm: + return llm_correct(whisper_text, language, llm_model, llm_url) + return whisper_text + + +# === Main ==================================================================== + +def main(): + parser = argparse.ArgumentParser( + description="STT-Tool: Audio -> Whisper -> LLM-Korrektur", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=( + "Beispiele:\n" + " python3 stt.py --list-devices\n" + " python3 stt.py --input-device respeaker\n" + " python3 stt.py --stt-model large-v3 --input-device respeaker\n" + " python3 stt.py --file audio.wav --no-llm\n" + " python3 stt.py --stt-model medium --device cpu\n" + ), + ) + + # Allgemeine Optionen + parser.add_argument( + "--list-devices", action="store_true", + help="Verfuegbare Audio-Geraete auflisten und beenden", + ) + parser.add_argument( + "--list-models", action="store_true", + help="Verfuegbare Ollama-Modelle auflisten und beenden", + ) + parser.add_argument( + "language", type=str, nargs="?", default=DEFAULT_LANG, + help="Sprachcode: de, en, fr, es, ... (Standard: de)", + ) + + # Eingabe (Mikrofon) + parser.add_argument( + "--input-device", "-i", type=str, default="default", + help="Eingabegeraet: Index (z.B. 5), Name (z.B. respeaker), " + "oder 'default' (Standard: default = System-Mic)", + ) + + # Ausgabe (Lautsprecher) + parser.add_argument( + "--output-device", "-o", type=str, default="default", + help="Ausgabegeraet fuer TTS-Ansage: Index, Name, oder 'default' " + "(Standard: default = System-Speaker)", + ) + + # STT-Modell + parser.add_argument( + "--stt-model", "-m", type=str, default=DEFAULT_STT_MODEL, + choices=["tiny", "base", "small", "medium", "large-v3"], + help=f"Whisper-Modell fuer STT (Standard: {DEFAULT_STT_MODEL})", + ) + + # LLM-Modell + parser.add_argument( + "--llm-model", type=str, default=DEFAULT_LLM_MODEL, + help=f"Ollama-Modell fuer LLM-Korrektur (Standard: {DEFAULT_LLM_MODEL})", + ) + parser.add_argument( + "--no-llm", action="store_true", + help="LLM-Korrektur ueberspringen (nur Whisper-Text)", + ) + + # Compute-Device + parser.add_argument( + "--device", type=str, default="auto", + choices=["auto", "cpu", "cuda"], + help="Compute-Device fuer Whisper: auto, cpu, cuda (Standard: auto)", + ) + parser.add_argument( + "--gpu-index", type=int, default=None, + help="GPU-Index bei Multi-GPU-Systemen (Standard: automatisch)", + ) + + # Datei-Modus + parser.add_argument( + "--file", "-f", type=str, default=None, + help="WAV-Datei transkribieren statt live aufzunehmen", + ) + + # Feintuning + parser.add_argument( + "--initial-prompt", type=str, default=INITIAL_PROMPT, + help="Kontext-Hinweis fuer Whisper (Eigennamen, Fachbegriffe)", + ) + parser.add_argument( + "--max-duration", type=int, default=DEFAULT_MAX_DURATION, + help=f"Maximale Aufnahmedauer in Sekunden (Standard: {DEFAULT_MAX_DURATION})", + ) + parser.add_argument( + "--vad-threshold", type=float, default=DEFAULT_VAD_THRESHOLD, + help=f"RMS-Schwelle fuer Stilleerkennung (Standard: {DEFAULT_VAD_THRESHOLD})", + ) + parser.add_argument( + "--vad-filter", action="store_true", + help="Whisper VAD-Filter aktivieren (entfernt Stille aus Transkription)", + ) + parser.add_argument( + "--beam-size", type=int, default=5, + help="Beam-Size fuer Whisper (Standard: 5, hoeher = genauer aber langsamer)", + ) + parser.add_argument( + "--no-announcement", action="store_true", + help="TTS-Ansage 'Aufnahme laeuft' unterdruecken", + ) + parser.add_argument( + "--announcement-volume", type=float, default=DEFAULT_ANNOUNCEMENT_VOLUME, + help=f"Lautstaerke der Ansage (Standard: {DEFAULT_ANNOUNCEMENT_VOLUME})", + ) + parser.add_argument( + "--delete-wav", action="store_true", + help="Temporaere WAV-Datei nach Transkription loeschen", + ) + + args = parser.parse_args() + + # --list-devices + if args.list_devices: + list_devices() + return + + # --list-models + if args.list_models: + list_ollama_models() + return + + print("=" * 60) + print(" STT-Tool: Audio -> Whisper -> LLM-Korrektur") + print("=" * 60) + + # Compute-Device bestimmen + if args.device == "auto": + device, compute_type, gpu_index = detect_compute() + elif args.device == "cpu": + device, compute_type, gpu_index = "cpu", "int8", None + else: # cuda + compute_type = "float16" + gpu_index = args.gpu_index if args.gpu_index is not None else 0 + device = "cuda" + + print(f"Compute: {device} ({compute_type})" + + (f", GPU {gpu_index}" if gpu_index is not None else "")) + print(f"STT-Modell: {args.stt_model}") + if not args.no_llm: + print(f"LLM-Modell: {args.llm_model}") + print() + + # Whisper-Modell laden (einmal, fuer beide Zwecke genutzt) + model = load_whisper_model(args.stt_model, device, compute_type, gpu_index) + + llm_url = "http://localhost:11434/api/chat" + + # === Datei-Modus === + if args.file: + if not os.path.isfile(args.file): + print(f"Datei nicht gefunden: {args.file}") + sys.exit(1) + text = transcribe_file( + args.file, args.language, model, args.initial_prompt, + args.vad_filter, args.beam_size, + not args.no_llm, args.llm_model, llm_url + ) + if text: + print(f"\n{'='*60}") + print(f"FINALES ERGEBNIS") + print(f"{'='*60}") + print(text) + print() + else: + print("Keine Sprache erkannt.") + return + + # === Live-Aufnahme-Modus === + input_device_idx = resolve_device(args.input_device, "input") + input_info = get_device_info(input_device_idx, "input") + print(f"Eingabe: [{input_device_idx}] {input_info['name']} " + f"({input_info['channels']} Kanäle, {input_info['samplerate']} Hz)") + + output_device_idx = resolve_device(args.output_device, "output") + output_info = get_device_info(output_device_idx, "output") + print(f"Ausgabe: [{output_device_idx}] {output_info['name']} " + f"({output_info['channels']} Kanäle, {output_info['samplerate']} Hz)") + print() + + stopword_list = STOPWORDS.get(args.language, STOPWORDS["de"]) + + # TTS-Ansage + if not args.no_announcement: + play_announcement("Aufnahme läuft.", output_device_idx, + args.language, args.announcement_volume) + + # Aufnehmen bis Stoppwort + audio = record_until_stopword( + input_device_idx, input_info, model, args.language, + stopword_list, args.vad_threshold, args.max_duration + ) + + # WAV zwischenspeichern + tmp_wav = tempfile.NamedTemporaryFile( + suffix=".wav", delete=False, prefix="stt_" + ) + save_wav(audio, tmp_wav.name) + print(f"Aufnahme gespeichert: {tmp_wav.name}\n") + + # Whisper-Transkription + whisper_text = transcribe( + model, audio, args.language, args.initial_prompt, + args.vad_filter, args.beam_size + ) + + # LLM-Korrektur + if whisper_text and not args.no_llm: + final_text = llm_correct(whisper_text, args.language, + args.llm_model, llm_url) + else: + final_text = whisper_text + + # Aufraeumen + if args.delete_wav: + try: + os.unlink(tmp_wav.name) + print(f"Temp-Datei geloescht: {tmp_wav.name}") + except OSError: + pass + else: + print(f"Aufnahme behalten: {tmp_wav.name}") + + # Ergebnis + if final_text: + print(f"\n{'='*60}") + print(f"FINALES ERGEBNIS") + print(f"{'='*60}") + print(final_text) + print() + print("Fertig!") + else: + print("Keine Sprache erkannt — versuche es mit lauterer Stimme.") + + +if __name__ == "__main__": + main() diff --git a/stt_respeaker.py b/stt_respeaker.py new file mode 100644 index 0000000..e56d63d --- /dev/null +++ b/stt_respeaker.py @@ -0,0 +1,476 @@ +#!/usr/bin/env python3 +""" +STT-Skript: Nimmt Audio vom ReSpeaker XVF3800 auf und transkribiert es +mit faster-whisper (lokal, CPU-only). + +Features: + - Ansage "Aufnahme laeuft" ueber den ReSpeaker-Lautsprecher (espeak-ng TTS) + - Kontinuierliche Aufnahme bis das Stoppwort "Ende" erkannt wird + - Echtzeit-Stoppworterkennung mit Whisper tiny-Modell + - Finale Transkription mit Whisper medium-Modell (hoehere Genauigkeit) + - LLM-Korrektur mit gemma3:latest (Ollama) fuer grammatikalisch korrekten Text + +Verwendung: + python3 stt_respeaker.py # Aufnehmen bis "Ende" gesagt wird + python3 stt_respeaker.py en # Sprache Englisch, Stoppwort "end" + python3 stt_respeaker.py --file audio.wav # Vorhandene Datei transkribieren + +Voraussetzungen: + pip install faster-whisper sounddevice numpy scipy requests + espeak-ng installiert + ollama + gemma3:latest installiert + +Beim ersten Aufruf werden die Whisper-Modelle automatisch heruntergeladen. +""" + +import sys +import os +import wave +import time +import tempfile +import argparse +import subprocess +import numpy as np +import sounddevice as sd +from scipy.signal import resample_poly +from faster_whisper import WhisperModel +import requests + +# Output-Buffering deaktivieren +sys.stdout.reconfigure(line_buffering=True) + +# === Konfiguration =========================================================== +DEVICE_SR = 48000 +WHISPER_SR = 16000 +CHANNELS = 2 +DTYPE = "float32" +MODEL_SIZE = "large-v3" # Auf GPU: beste Genauigkeit, ~0.1s pro 30s Audio +STOPWORD_MODEL = "large-v3" # Gleiche Modell auf GPU, keine zusaetzliche Ladezeit +DEVICE_COMPUTE = "cuda" # RTX 3090 +COMPUTE_TYPE = "float16" # Beste GPU-Praezision +GPU_DEVICE_INDEX = 1 # Zweite RTX 3090 (erste ist durch Ollama belegt) +DEFAULT_LANG = "de" +DEVICE_KEYWORD = "XVF3800" +INITIAL_PROMPT = "Nachrichten, Politik, Friedrich Merz, CDU, CSU, SPD, Olaf Scholz, Markus Söder, Kanzler, Merz-Akte" + +# LLM-Korrektur (Ollama) +LLM_MODEL = "mistral-small3.2:24b" +LLM_URL = "http://localhost:11434/api/chat" +LLM_SYSTEM_PROMPT = { + "de": ( + "Du bist ein Korrektor fuer deutsche Sprachtranskripte. " + "Korrigiere Grammatik, Rechtschreibung, korrigiere falsch verstandene Woerter und ergaenze fehlende Woerter " + "zu einem sinnvollen Text. " + "Achte auf korrekte Namen: Friedrich Merz (CDU), Markus Soeder (CSU), " + "Olaf Scholz (SPD). " + "Entferne das Stoppwort 'Ende' am Schluss falls vorhanden. " + "Antworte NUR mit dem korrigierten Text, ohne Einleitung oder Kommentar." + ), + "en": ( + "You are a proofreader for English speech transcripts. " + "Correct grammar, spelling, correct wrong unerstood words and add missing words to make coherent text. " + "Remove the stopword 'end' at the end. " + "Reply ONLY with the corrected text, no preamble." + ), +} + +STOPWORDS = { + "de": ["ende", "end", "stop"], + "en": ["end", "stop", "done", "finish"], + "fr": ["fin", "stop", "termine"], + "es": ["fin", "stop", "terminado"], +} + +CHUNK_SEC = 2.0 +MAX_DURATION = 60 +VAD_THRESHOLD = 0.008 # Angepasst fuer ASR-Kanal (rechter Kanal) + + +def find_respeaker_device(): + """Sucht den ReSpeaker XVF3800 in der Liste der Audiogeraete.""" + devices = sd.query_devices() + for i, dev in enumerate(devices): + name_lower = dev["name"].lower() + if (("xvf3800" in name_lower or "respeaker" in name_lower or "array" in name_lower) + and dev["max_input_channels"] > 0): + print(f"ReSpeaker gefunden: [{i}] {dev['name']}" + f" ({dev['max_input_channels']} In, " + f"{dev['max_output_channels']} Out)") + return i + print(f"Kein Geraet mit '{DEVICE_KEYWORD}' gefunden.") + print("Verfuegbare Eingabegeraete:") + for i, dev in enumerate(devices): + if dev["max_input_channels"] > 0: + print(f" [{i}] {dev['name']}") + return None + + +def play_announcement(text, device_idx, lang="de"): + """Erzeugt TTS mit espeak-ng und spielt es ueber den ReSpeaker ab.""" + print(f"Ansage: '{text}'") + + tmp_wav = tempfile.NamedTemporaryFile( + suffix=".wav", delete=False, prefix="tts_" + ) + tmp_wav.close() + + espeak_lang = {"de": "de", "en": "en", "fr": "fr", "es": "es"}.get(lang, "de") + + subprocess.run( + ["espeak-ng", "-v", espeak_lang, "-s", "150", + "-w", tmp_wav.name, text], + check=True, capture_output=True + ) + + tmp_loud = tempfile.NamedTemporaryFile( + suffix=".wav", delete=False, prefix="tts_loud_" + ) + tmp_loud.close() + + subprocess.run( + ["ffmpeg", "-y", "-i", tmp_wav.name, + "-filter:a", "volume=3.0", + "-ar", str(DEVICE_SR), "-ac", str(CHANNELS), + tmp_loud.name], + check=True, capture_output=True + ) + + sink = ("alsa_output.usb-Seeed_Studio_reSpeaker_" + "XVF3800_4-Mic_Array_114993700261100055-00.analog-stereo") + subprocess.run( + ["paplay", "--device", sink, + "--volume", "1.5", tmp_loud.name], + check=True, capture_output=True + ) + + os.unlink(tmp_wav.name) + os.unlink(tmp_loud.name) + print(" (Ansage beendet)\n") + + +def resample_to_16k(audio_48k_mono): + return resample_poly(audio_48k_mono, 1, 3) + + +def stereo_to_mono(audio): + """Extrahiert den rechten Kanal (ASR-Output des XVF3800). + Der rechte Kanal ist der fuer Spracherkennung optimierte ASR-Output + (AEC residual / auto-select beam). Der linke Kanal ist der verarbeitete + Beamforming-Output. Beide zu mischen verschlechtert die Qualitaet.""" + if audio.ndim > 1: + # Rechter Kanal = Index 1 (ASR-Output, optimiert fuer Spracherkennung) + return audio[:, 1] + return audio + + +def compute_rms(audio_float): + return float(np.sqrt(np.mean(audio_float ** 2))) + + +def check_stopword(audio_16k, model, language, stopword_list): + rms = compute_rms(audio_16k) + if rms < VAD_THRESHOLD: + return False, "", rms + + segments, _ = model.transcribe( + audio_16k, + language=language, + beam_size=1, + vad_filter=True, + temperature=0.0, + ) + text = " ".join(seg.text.strip().lower() for seg in segments) + + for word in stopword_list: + if word in text: + return True, text, rms + return False, text, rms + + +def record_until_stopword(device_idx, language, stopword_list): + """Nimmt kontinuierlich auf bis Stoppwort erkannt wird.""" + print(f"Lade Whisper-Modell '{STOPWORD_MODEL}' (GPU)...") + sw_model = WhisperModel( + STOPWORD_MODEL, device=DEVICE_COMPUTE, compute_type=COMPUTE_TYPE, + device_index=GPU_DEVICE_INDEX, + ) + print(f"Aufnahme laeuft — sage '{stopword_list[0]}' zum Stoppen.") + print(f" (Notbremse nach {MAX_DURATION}s)\n") + + chunk_samples = int(CHUNK_SEC * DEVICE_SR) + all_audio_48k = [] + total_duration = 0.0 + + stream = sd.InputStream( + samplerate=DEVICE_SR, + channels=CHANNELS, + dtype=DTYPE, + device=device_idx, + blocksize=chunk_samples, + ) + stream.start() + + try: + while total_duration < MAX_DURATION: + chunk, overflowed = stream.read(chunk_samples) + if overflowed: + print(" Buffer Overflow!") + + chunk_mono = stereo_to_mono(chunk) + chunk_16k = resample_to_16k(chunk_mono) + all_audio_48k.append(chunk_mono) + + total_duration += CHUNK_SEC + rms = compute_rms(chunk_16k) + + if rms >= VAD_THRESHOLD: + is_stop, text, rms = check_stopword( + chunk_16k, sw_model, language, stopword_list + ) + bar = "#" * min(int(rms * 200), 30) + status = f" [{total_duration:5.1f}s] RMS:{rms:.3f} {bar}" + if text: + status += f" '{text}'" + print(status) + + if is_stop: + print(f"\n>>> Stoppwort erkannt: '{text}'") + break + else: + print(f" [{total_duration:5.1f}s] RMS:{rms:.3f} (Stille)") + finally: + stream.stop() + stream.close() + + full_audio_48k = np.concatenate(all_audio_48k) + full_audio_16k = resample_to_16k(full_audio_48k) + + # Software-Verstaerkung: Audio normalisieren auf -3 dBFS Peak + # Der XVF3800 liefert sehr leises Signal, daher nachverstaerken + peak = float(np.max(np.abs(full_audio_16k))) + if peak > 0.001: + target_peak = 0.7 # -3 dBFS + gain = target_peak / peak + full_audio_16k = full_audio_16k * gain + full_audio_16k = np.clip(full_audio_16k, -1.0, 1.0) + print(f" Lautstaerke-Anpassung: Gain x{gain:.1f} " + f"(Peak {peak:.3f} -> {target_peak:.3f})") + else: + print(" Warnung: Sehr leises Signal, keine Verstaerkung.") + + print(f"\nAufnahme beendet. Gesamtdauer: " + f"{len(full_audio_16k)/WHISPER_SR:.1f}s\n") + return full_audio_16k + + +def save_wav(audio_data, filename): + audio_int16 = (audio_data * 32767).astype(np.int16) + with wave.open(filename, "wb") as wf: + wf.setnchannels(1) + wf.setsampwidth(2) + wf.setframerate(WHISPER_SR) + wf.writeframes(audio_int16.tobytes()) + + +def transcribe(audio_data, language="de", model_size=MODEL_SIZE): + """Transkribiert Audio mit faster-whisper.""" + print(f"Lade Whisper-Modell '{model_size}'" + f" (device={DEVICE_COMPUTE}, compute={COMPUTE_TYPE})...") + print(" (Beim ersten Aufruf wird das Modell heruntergeladen.)\n") + + model = WhisperModel( + model_size, + device=DEVICE_COMPUTE, + compute_type=COMPUTE_TYPE, + device_index=GPU_DEVICE_INDEX, + ) + + print("Transkribiere...\n") + transcribe_kwargs = dict( + language=language, + beam_size=5, + vad_filter=True, + best_of=5, + temperature=[0.0, 0.2, 0.4, 0.6], + ) + if INITIAL_PROMPT: + transcribe_kwargs["initial_prompt"] = INITIAL_PROMPT + + segments, info = model.transcribe(audio_data, **transcribe_kwargs) + + print(f" Sprache erkannt: {info.language}" + f" ({info.language_probability:.1%})") + print(f" Dauer: {info.duration:.1f}s\n") + + full_text = "" + print("--- Whisper-Transkription (roh) ----------------------") + for seg in segments: + timestamp = f"[{seg.start:5.1f}s - {seg.end:5.1f}s]" + print(f" {timestamp} {seg.text}") + full_text += seg.text + " " + print("------------------------------------------------------\n") + + return full_text.strip() + + +def llm_correct(text, language="de"): + """Korrigiert den transkribierten Text mit einem lokalen LLM (Ollama Chat API).""" + if not text: + return text + + system_prompt = LLM_SYSTEM_PROMPT.get(language, LLM_SYSTEM_PROMPT["de"]) + + print(f"LLM-Korrektur mit '{LLM_MODEL}' (Ollama)...\n") + + try: + response = requests.post( + LLM_URL, + json={ + "model": LLM_MODEL, + "messages": [ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": text}, + ], + "stream": False, + "options": { + "temperature": 0.0, + "num_predict": 500, + }, + }, + timeout=180, + ) + response.raise_for_status() + result = response.json().get("message", {}).get("content", "").strip() + + if result: + print("--- LLM-korrigierter Text ----------------------------") + print(f" {result}") + print("------------------------------------------------------\n") + return result + else: + print("LLM lieferte leere Antwort — verwende Whisper-Text.\n") + return text + + except requests.exceptions.RequestException as e: + print(f"Fehler bei LLM-Korrektur: {e}") + print("Verwende unkorrigierten Whisper-Text.\n") + return text + + +def transcribe_file(filepath, language, model_size=MODEL_SIZE): + """Transkribiert eine vorhandene WAV-Datei.""" + print(f"Lade Datei: {filepath}\n") + with wave.open(filepath, "rb") as wf: + frames = wf.readframes(wf.getnframes()) + audio_int16 = np.frombuffer(frames, dtype=np.int16) + audio_float = audio_int16.astype(np.float32) / 32767.0 + file_sr = wf.getframerate() + + print(f" Dauer: {len(audio_float) / file_sr:.1f}s") + print(f" Samplerate: {file_sr} Hz\n") + + if file_sr != WHISPER_SR: + print(f" Resample: {file_sr} -> {WHISPER_SR} Hz") + from math import gcd + g = gcd(file_sr, WHISPER_SR) + up = WHISPER_SR // g + down = file_sr // g + audio_float = resample_poly(audio_float, up, down) + + whisper_text = transcribe(audio_float, language, model_size) + + if whisper_text: + corrected = llm_correct(whisper_text, language) + return corrected + return whisper_text + + +def main(): + parser = argparse.ArgumentParser( + description="STT mit ReSpeaker XVF3800 + faster-whisper + LLM-Korrektur" + ) + parser.add_argument( + "language", + type=str, + nargs="?", + default=DEFAULT_LANG, + help="Sprachcode: de, en, fr, es, ... (Standard: de)", + ) + parser.add_argument( + "--file", + type=str, + default=None, + help="Vorhandene WAV-Datei transkribieren statt live aufzunehmen", + ) + parser.add_argument( + "--model", + type=str, + default=MODEL_SIZE, + help=f"Whisper-Modell: tiny|base|small|medium|large-v3" + f" (Standard: {MODEL_SIZE})", + ) + parser.add_argument( + "--no-llm", + action="store_true", + help="LLM-Korrektur ueberspringen (nur Whisper-Text ausgeben)", + ) + args = parser.parse_args() + + print("=" * 55) + print(" ReSpeaker XVF3800 -> Whisper + LLM-Korrektur") + print("=" * 55) + + # Datei-Modus + if args.file: + if not os.path.isfile(args.file): + print(f"Datei nicht gefunden: {args.file}") + sys.exit(1) + text = transcribe_file(args.file, args.language, args.model) + if text: + print(f"\n=== FINALES ERGEBNIS ===\n{text}\n") + else: + print("Keine Sprache erkannt.") + return + + # Live-Aufnahme-Modus + device_idx = find_respeaker_device() + if device_idx is None: + print("Kein ReSpeaker XVF3800 gefunden.") + sys.exit(1) + + stopword_list = STOPWORDS.get(args.language, STOPWORDS["de"]) + + # Ansage + play_announcement("Aufnahme laeuft.", device_idx, args.language) + + # Aufnehmen bis Stoppwort + audio = record_until_stopword(device_idx, args.language, stopword_list) + + # WAV zwischenspeichern + tmp_wav = tempfile.NamedTemporaryFile( + suffix=".wav", delete=False, prefix="respeaker_" + ) + save_wav(audio, tmp_wav.name) + print(f"Aufnahme gespeichert: {tmp_wav.name}\n") + + # Whisper-Transkription + whisper_text = transcribe(audio, args.language, args.model) + + # LLM-Korrektur + if whisper_text and not args.no_llm: + final_text = llm_correct(whisper_text, args.language) + else: + final_text = whisper_text + + # Aufnahme NICHT loeschen — fuer spaetere Wiederverwendung + print(f"Aufnahme behalten: {tmp_wav.name}\n") + + if final_text: + print(f"\n=== FINALES ERGEBNIS ===\n{final_text}\n") + print("Fertig!") + else: + print("Keine Sprache erkannt — versuche es mit lauterer Stimme.") + + +if __name__ == "__main__": + main()