From a986cbe42a0ebad9e67807a5480eacb2f700be37 Mon Sep 17 00:00:00 2001 From: dschlueter Date: Sun, 13 Sep 2026 21:34:03 +0200 Subject: [PATCH] STT-Dienst: Kern, HTTP/WebSocket/Wyoming-Service, CLI, Tests, systemd-Unit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - stt_core.py: Whisper-Engine (einmal laden, CPU-Fallback), Audio-Dekodierung jedes Formats, Halluzinationsfilter, Äußerungserkennung per Silero-VAD - stt_service.py: /health, /v1/audio/transcriptions (OpenAI), /transcribe (Rohdaten), WS /ws/stream; Inferenz im Worker-Thread, begrenzte Warteschlange (503) - stt_wyoming.py: Home-Assistant-Anbindung (TCP 10300) - stt_cli.py: Datei/Mikrofon über den Dienst oder lokal - Tests (Unit + Integration), README, requirements.txt, deploy/stt-service.service Co-Authored-By: Claude Opus 5 --- CLAUDE.md | 72 +++--- README.md | 127 +++++++++ deploy/stt-service.service | 21 ++ pytest.ini | 6 + requirements.txt | 26 ++ stt_cli.py | 379 +++++++++++++++++++++++++++ stt_core.py | 509 +++++++++++++++++++++++++++++++++++++ stt_service.py | 439 ++++++++++++++++++++++++++++++++ stt_wyoming.py | 121 +++++++++ tests/conftest.py | 92 +++++++ tests/test_core.py | 165 ++++++++++++ tests/test_integration.py | 62 +++++ tests/test_service.py | 153 +++++++++++ tests/test_wyoming.py | 62 +++++ 14 files changed, 2197 insertions(+), 37 deletions(-) create mode 100644 README.md create mode 100644 deploy/stt-service.service create mode 100644 pytest.ini create mode 100644 requirements.txt create mode 100644 stt_cli.py create mode 100644 stt_core.py create mode 100644 stt_service.py create mode 100644 stt_wyoming.py create mode 100644 tests/conftest.py create mode 100644 tests/test_core.py create mode 100644 tests/test_integration.py create mode 100644 tests/test_service.py create mode 100644 tests/test_wyoming.py diff --git a/CLAUDE.md b/CLAUDE.md index a1b8802..134c700 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -2,55 +2,53 @@ 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). +> Sprache: Code-Kommentare, Doku und Commits auf Deutsch. ## 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. +Lokale, schnelle, zuverlässige Spracherkennung, nicht blockierend. Baustein für Smarthome +(Home Assistant via Wyoming, ESP32/Arduino als Audio-Clients) und **STT-Pendant zu `chatterbox-tts-cli`** +(`~/chatterbox-tts-cli`: CLI-Kern + FastAPI-Service + MCP-Adapter + systemd-User-Service, Port 9999). +Bewährtes STT-Vorbild: `~/my_voice_assistant_v3` (`app/providers/stt/faster_whisper.py`, `app/api/ws.py`). +Git-Remote auf kitux.de (Forgejo) geplant, noch nicht eingerichtet. ## Befehle ```bash -pip install faster-whisper sounddevice numpy scipy requests # plus Systemtools: espeak-ng, ffmpeg, pactl/paplay, ollama +.venv/bin/python -m pytest # Unit-Tests, ohne Modell (~9 s) +.venv/bin/python -m pytest -m integration # echtes Whisper small/CPU + Silero, espeak-ng-Sprache (~20 s) +.venv/bin/python -m pytest tests/test_service.py::test_ws_stream_with_vad_emits_transcripts_per_utterance -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 +CUDA_VISIBLE_DEVICES=GPU-83ba6d1f-ebbe-f8c8-1433-153e0ebc3866 .venv/bin/python stt_service.py # Port 9998, Wyoming 10300 +.venv/bin/python stt_cli.py datei.wav | --once | --local --model small datei.wav ``` -Rauchtest ohne Mikrofon/GPU: `python3 stt.py --file --no-llm -m tiny --device cpu`. +`.venv` ist mit `--system-site-packages` auf miniforge-Base angelegt (faster-whisper, fastapi, onnxruntime kommen +von dort); nur `wyoming` ist in der venv installiert. Tests immer mit `.venv/bin/python`, sonst fehlt `wyoming`. -## GPU-Umgebung +## Architektur -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. +- `stt_core.py` — reine Logik, kein print/sys.exit. `SttEngine` (Modell einmal laden, Lade-Versuche cuda→cpu mit + `cpu_model`, `_run_lock` serialisiert Inferenz, `condition_on_previous_text=False`, Halluzinationsfilter + `is_hallucination`), Audio-Hilfen (`decode_audio_bytes` via PyAV für jedes Format, `pcm16_to_float` mit + Kanalwahl, `StreamResampler` via soxr), `UtteranceSegmenter` (Streaming-Endpointing über 512-Sample-Fenster; + Wahrscheinlichkeitsfunktion injizierbar, Standard `SileroProb`). +- `stt_service.py` — `create_app(engine, settings, vad_config, prob_fn_factory)` als Fabrik (Tests injizieren + `FakeEngine`/`EnergyProb` aus `tests/conftest.py`). `TranscriptionWorker` = ThreadPoolExecutor(1) + Zähler + `pending` → `Overloaded`/503. Preload läuft als erster Worker-Job, ohne den Start zu blockieren. + `_StreamSession` = WebSocket-Protokoll (Empfangsschleife legt Äußerungen in eine Queue, `_consume` transkribiert + der Reihe nach; `send_lock`, weil zwei Tasks senden). +- `stt_wyoming.py` — Wyoming-Handler im selben Prozess/Worker; nimmt HA-Audio komplett an (externe VAD), + `transcript_names/terms` → Whisper-`hotwords`; Verbindung wird nach dem Transkript geschlossen (HA-Konvention). +- `stt_cli.py` — Client: Dateien per `POST /transcribe`, Mikrofon per `/ws/stream`; `--local` nutzt den Kern direkt. + stdout nur Transkripte, stderr Status. Default-Mikrofon über das ALSA-Gerät `pulse` (PortAudios Default ist hier + die falsche Onboard-Karte); ReSpeaker XVF3800 → automatisch Kanal 1 (ASR-Ausgang). +- `stt.py`, `stt_respeaker.py` — **historische** Einzelskripte, nicht weiterentwickeln. -## Architektur `stt.py` (Ist-Zustand) +Sprache: `language=None` → Config-Default, `"auto"` → Erkennung, `"de-DE"` → `"de"` (`SttEngine.resolve_language`). -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`). +## Umgebung -- `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. +GPUs: T600 + 2× RTX 3090. `nvidia-smi` zählt in PCI-Reihenfolge, CUDA „fastest first“ → GPU immer per +`CUDA_VISIBLE_DEVICES=GPU-` pinnen. GPU 1 belegt llama.cpp/ollama fast komplett; der Dienst läuft auf GPU 2 +(`large-v3-turbo` ≈ 2,4 GB) neben chatterbox-tts und open-notebook-stt (:8902). HF-Modelle unter `$HF_HOME`. diff --git a/README.md b/README.md new file mode 100644 index 0000000..7b55b6f --- /dev/null +++ b/README.md @@ -0,0 +1,127 @@ +# SpeechToText + +Lokaler, schneller Spracherkennungs-Dienst (Sprache → Text) auf Basis von +[faster-whisper](https://github.com/SYSTRAN/faster-whisper) — das STT-Gegenstück zu +[chatterbox-tts-cli](https://kitux.de/forgejo/dschlueter/chatterbox-tts-cli). + +Gedacht als Baustein für Sprachassistenten und Smarthome: Home Assistant (Wyoming), +Mikrocontroller wie ESP32 (HTTP/WebSocket), Skripte (CLI) und OpenAI-kompatible Tools. + +- **Schnell:** Modell bleibt geladen. Gemessen auf RTX 3090 mit `large-v3-turbo`: + 5,8 s Audio → 0,2 s, 35 s Audio → 0,7 s Rechenzeit; ~2,4 GB VRAM. +- **Nicht blockierend:** Inferenz in einem Worker-Thread, Warteschlange begrenzt (503 statt Stau). +- **Äußerungsende per VAD** (Silero) statt Stoppwort; Transkript ca. 1 s nach Sprechende. +- **Robust:** GPU-Fehler → automatischer CPU-Fallback; beliebige Audioformate; Halluzinationsfilter + („Untertitel im Auftrag des ZDF“ …). + +## Installation + +```bash +python3 -m venv --system-site-packages .venv # nutzt vorhandenes faster-whisper/torch-Umfeld +.venv/bin/pip install -r requirements.txt +``` + +## Dienst starten + +```bash +# Manuell (GPU per UUID pinnen: nvidia-smi -L) +CUDA_VISIBLE_DEVICES=GPU- .venv/bin/uvicorn stt_service:app --host 0.0.0.0 --port 9998 + +# Als systemd-User-Service (Autostart) +cp deploy/stt-service.service ~/.config/systemd/user/ +systemctl --user daemon-reload && systemctl --user enable --now stt-service +journalctl --user -u stt-service -f + +curl http://127.0.0.1:9998/health +``` + +Beim ersten Start wird das Modell heruntergeladen (`large-v3-turbo` ≈ 1,6 GB). Der Dienst ist sofort +erreichbar; `/health` meldet `loading`, bis das Modell bereit ist. + +### Konfiguration (Umgebungsvariablen) + +| Variable | Standard | Bedeutung | +|---|---|---| +| `STT_MODEL` | `large-v3-turbo` | Whisper-Modell (`tiny` … `large-v3`, `large-v3-turbo`, Pfad) | +| `STT_DEVICE` | `auto` | `auto` / `cuda` / `cpu` | +| `STT_COMPUTE_TYPE` | `auto` | `float16` (GPU) / `int8` (CPU) automatisch | +| `STT_CPU_MODEL` | `small` | Modell beim Ausweichen auf CPU | +| `STT_LANGUAGE` | `de` | Default-Sprache, `auto` = Erkennung | +| `STT_BEAM_SIZE` | `5` | Beam-Search-Breite (1 = schneller) | +| `STT_INITIAL_PROMPT` | – | Kontext/Eigennamen für Whisper | +| `STT_MAX_QUEUE` | `8` | max. wartende Aufträge, darüber HTTP 503 | +| `STT_MAX_UPLOAD_MB` | `50` | max. Uploadgröße | +| `STT_PRELOAD` | `1` | Modell beim Start laden + aufwärmen | +| `STT_VAD_SILENCE_MS` | `700` | Stille bis Äußerungsende (Streaming) | +| `STT_VAD_THRESHOLD` | `0.5` | Silero-Sprachschwelle | +| `STT_VAD_MAX_UTTERANCE_S` | `30` | max. Äußerungslänge | +| `STT_WYOMING_HOST` / `STT_WYOMING_PORT` | `0.0.0.0` / `10300` | Wyoming-Server, Port leer = aus | +| `STT_LOG_LEVEL` | `INFO` | Log-Level | + +## Schnittstellen + +### HTTP + +```bash +# OpenAI-kompatibel (multipart) — response_format: json | text | verbose_json +curl -F file=@aufnahme.mp3 -F language=de http://127.0.0.1:9998/v1/audio/transcriptions + +# Rohdaten im Body — ideal für Mikrocontroller (kein multipart nötig) +curl --data-binary @aufnahme.wav http://127.0.0.1:9998/transcribe +curl --data-binary @roh.pcm "http://127.0.0.1:9998/transcribe?format=pcm&sample_rate=16000&channels=1" +``` + +`/transcribe` liefert `text`, `language`, `duration`, `processing_s`, `segments`. + +### WebSocket `/ws/stream` (Live-Streaming) + +Client schickt s16le-PCM in beliebig großen Stücken; der Dienst erkennt Äußerungen selbst. + +```text +→ {"type":"start","sample_rate":16000,"channels":1,"channel":null,"language":"de","vad":true,"silence_ms":700} +→ … +← {"type":"speech_start","t":1.31} +← {"type":"transcript","text":"Schalte das Licht ein.","language":"de","duration":2.4,"processing_s":0.21,"reason":"silence","t":3.9} +→ {"type":"end"} # laufende Äußerung abschließen → danach {"type":"done"} +→ {"type":"close"} # ausstehende Transkripte senden, dann schließen +``` + +Ohne VAD (`"vad":false`) wird alles bis `end` gesammelt und als Ganzes transkribiert (Push-to-Talk). +Für ESP32: I²S-Mikrofon mit 16 kHz mono lesen und die Puffer direkt als Binär-Frames senden. + +### Home Assistant (Wyoming) + +Einstellungen → Geräte & Dienste → Integration hinzufügen → **Wyoming Protocol** → +Host = dieser Rechner, Port `10300`. Danach in der Sprachassistenten-Pipeline als +Spracherkennung auswählen. ESPHome-Satelliten nutzen den Dienst dann automatisch. + +## CLI + +Nur Transkripte auf stdout, Status auf stderr → skriptfähig. + +```bash +.venv/bin/python stt_cli.py aufnahme.wav # Datei über den Dienst +.venv/bin/python stt_cli.py *.mp3 --json # JSON-Zeilen +.venv/bin/python stt_cli.py # Mikrofon (System-Default), fortlaufend +text=$(.venv/bin/python stt_cli.py --once) # eine Äußerung → Variable +.venv/bin/python stt_cli.py -d respeaker -v # bestimmtes Gerät (ReSpeaker: ASR-Kanal automatisch) +.venv/bin/python stt_cli.py --local --model small aufnahme.wav # ohne Dienst +.venv/bin/python stt_cli.py --list-devices +``` + +Exit-Codes: `0` ok, `1` Fehler, `2` keine Sprache erkannt. + +## Tests + +```bash +.venv/bin/python -m pytest # schnell, ohne Modell (Fake-Engine) +.venv/bin/python -m pytest -m integration # echtes Whisper (small, CPU) + Silero mit espeak-ng-Sprache +.venv/bin/python -m pytest tests/test_core.py::test_segmenter_detects_two_utterances +``` + +## Bekannte Grenzen + +- Wörtlich wiederholte Sätze innerhalb eines 30-s-Fensters fasst Whisper teils zusammen + (bei natürlicher Sprache selten). +- Keine wortweisen Zwischenergebnisse während des Sprechens — Transkript kommt pro Äußerung. +- `stt.py` und `stt_respeaker.py` sind die alten Einzelskripte (historisch). diff --git a/deploy/stt-service.service b/deploy/stt-service.service new file mode 100644 index 0000000..9c0c765 --- /dev/null +++ b/deploy/stt-service.service @@ -0,0 +1,21 @@ +[Unit] +Description=SpeechToText Service (faster-whisper, HTTP :9998, Wyoming :10300) +After=network.target + +[Service] +Type=simple +WorkingDirectory=%h/Python_Programs/SpeechToText +ExecStart=%h/Python_Programs/SpeechToText/.venv/bin/uvicorn stt_service:app --host 0.0.0.0 --port 9998 +Restart=on-failure +RestartSec=5 +StandardOutput=journal +StandardError=journal +# GPU eindeutig per UUID (nvidia-smi -L), NICHT per Index: CUDA zählt "fastest first", +# nvidia-smi in PCI-Reihenfolge. GPU 2 = RTX 3090, teilt sich mit chatterbox-tts (~2,4 GB VRAM). +Environment="CUDA_VISIBLE_DEVICES=GPU-83ba6d1f-ebbe-f8c8-1433-153e0ebc3866" +Environment="STT_MODEL=large-v3-turbo" +Environment="STT_LANGUAGE=de" +Environment="STT_WYOMING_PORT=10300" + +[Install] +WantedBy=default.target diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..ea3a827 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,6 @@ +[pytest] +testpaths = tests +markers = + integration: echtes Whisper-Modell (langsamer, braucht gecachtes Modell) +addopts = -m "not integration" +asyncio_default_fixture_loop_scope = function diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..2f83a3e --- /dev/null +++ b/requirements.txt @@ -0,0 +1,26 @@ +# SpeechToText — Abhängigkeiten +# Getestet mit Python 3.12, CUDA 12.x (ctranslate2 4.6), Ubuntu +# +# GPU: ctranslate2 braucht cuBLAS 12 + cuDNN 9 (z. B. über pip-Pakete +# nvidia-cublas-cu12 / nvidia-cudnn-cu12 oder eine vorhandene CUDA-Installation). + +# --- STT-Kern (stt_core.py) --- +faster-whisper>=1.2.0 # bringt ctranslate2, PyAV (Dekodierung) und Silero-VAD (onnxruntime) mit +numpy>=1.26 +scipy>=1.11 # Resampling ganzer Signale +soxr>=0.3 # Streaming-Resampling (optional, sonst Fallback auf scipy) + +# --- Dienst (stt_service.py) --- +fastapi>=0.115,<1.0 +uvicorn[standard]>=0.32,<1.0 +python-multipart>=0.0.9 # /v1/audio/transcriptions (multipart) +wyoming>=1.6,<2.0 # Home Assistant (stt_wyoming.py) + +# --- CLI (stt_cli.py) --- +sounddevice>=0.4 +requests>=2.31 +websockets>=13 + +# --- Tests --- +pytest>=8 +httpx>=0.27 diff --git a/stt_cli.py b/stt_cli.py new file mode 100644 index 0000000..c4eb40e --- /dev/null +++ b/stt_cli.py @@ -0,0 +1,379 @@ +#!/usr/bin/env python3 +"""STT-CLI – Client für stt_service.py (oder lokal ohne Dienst). + +Nur Transkripte gehen auf stdout (eine Zeile pro Äußerung bzw. Datei); alle +Statusmeldungen auf stderr. Dadurch direkt in Skripten nutzbar: + text=$(python stt_cli.py --once) + +Beispiele: + python stt_cli.py aufnahme.wav # Datei an den Dienst schicken + python stt_cli.py *.mp3 --json # mehrere Dateien, JSON-Zeilen + python stt_cli.py # Mikrofon → fortlaufend transkribieren + python stt_cli.py --once # nach der ersten Äußerung beenden + python stt_cli.py -d respeaker # ReSpeaker (rechter ASR-Kanal automatisch) + python stt_cli.py --local aufnahme.wav # ohne Dienst (lädt Modell selbst) + python stt_cli.py --list-devices + +Exit-Codes: 0 ok, 1 Fehler, 2 keine Sprache erkannt (--once/--timeout, Dateien) +""" + +from __future__ import annotations + +import argparse +import json +import os +import queue +import sys +import threading +import time +from pathlib import Path + +DEFAULT_URL = os.environ.get("STT_URL", "http://127.0.0.1:9998") +BLOCK_SEC = 0.1 + + +def status(msg: str) -> None: + print(msg, file=sys.stderr, flush=True) + + +def emit(result: dict, as_json: bool, source: str | None = None) -> None: + if as_json: + if source: + result = {"source": source, **result} + print(json.dumps(result, ensure_ascii=False), flush=True) + else: + print(result.get("text", ""), flush=True) + + +# === Geräte =================================================================== + +def list_devices() -> None: + import sounddevice as sd + + default_in = sd.default.device[0] + for i, dev in enumerate(sd.query_devices()): + if dev["max_input_channels"] > 0: + mark = "*" if i == default_in else " " + print(f"{mark}[{i:>2}] {dev['name']} ({dev['max_input_channels']} Kanäle, " + f"{int(dev['default_samplerate'])} Hz)") + + +SYSTEM_DEVICES = ("pipewire", "pulse") # ALSA-Plugins, die dem System-Default folgen + + +def pulse_default_source() -> str: + """Name der System-Standardquelle (PipeWire/PulseAudio), '' wenn unbekannt.""" + import subprocess + + try: + return subprocess.run(["pactl", "get-default-source"], capture_output=True, + text=True, timeout=3).stdout.strip() + except Exception: + return "" + + +def resolve_input(spec: str | None) -> tuple[int | None, dict]: + """None → System-Default über PipeWire/Pulse; Zahl → Index; Text → Namenssuche. + + PortAudios eigener Default ist auf PipeWire-Systemen oft die erste ALSA-Karte + (nicht das eingestellte Mikrofon) — daher bevorzugt das 'pulse'-Gerät. + """ + import sounddevice as sd + + devices = sd.query_devices() + if spec is None: + for name in SYSTEM_DEVICES: + for i, dev in enumerate(devices): + if dev["name"] == name and dev["max_input_channels"] > 0: + return i, dev + idx = sd.default.device[0] + return (None, sd.query_devices(kind="input")) if idx is None or idx < 0 else (idx, devices[idx]) + if spec.isdigit(): + idx = int(spec) + if idx < len(devices) and devices[idx]["max_input_channels"] > 0: + return idx, devices[idx] + raise SystemExit(f"Gerät [{idx}] existiert nicht oder hat keine Eingänge (--list-devices)") + for i, dev in enumerate(devices): + if spec.lower() in dev["name"].lower() and dev["max_input_channels"] > 0: + return i, dev + raise SystemExit(f"Kein Eingabegerät mit '{spec}' gefunden (--list-devices)") + + +def is_respeaker(name: str) -> bool: + return any(k in name.lower() for k in ("xvf3800", "respeaker")) + + +class Microphone: + """Nimmt im Callback auf und legt s16le-Mono-Blöcke in eine Queue — der + Audio-Thread wird nie durch Netzwerk oder Inferenz aufgehalten.""" + + def __init__(self, device: str | None, channel: int | None): + import sounddevice as sd + + self.index, info = resolve_input(device) + self.name = info["name"] + self.channels = min(int(info["max_input_channels"]), 2) + if self.name in SYSTEM_DEVICES: + # PipeWire resampelt selbst → direkt in Whisper-Rate aufnehmen + self.rate = 16000 + source = pulse_default_source() + if source: + self.name = f"{self.name} → {source}" + else: + self.rate = int(info["default_samplerate"]) + if channel is None and is_respeaker(self.name) and self.channels >= 2: + channel = 1 # XVF3800: rechter Kanal = ASR-optimierter Ausgang + self.channel = channel + self.blocks: queue.Queue[bytes] = queue.Queue(maxsize=600) # ~60 s Puffer + self.dropped = 0 + self._stream = sd.InputStream( + device=self.index, samplerate=self.rate, channels=self.channels, dtype="int16", + blocksize=int(self.rate * BLOCK_SEC), callback=self._callback, + ) + + def _callback(self, indata, frames, time_info, flags): + import numpy as np + + data = indata[:, self.channel] if self.channel is not None else indata.mean(axis=1) + try: + self.blocks.put_nowait(np.ascontiguousarray(data, dtype=" dict: + import requests + + params = {k: v for k, v in (("language", args.language), ("prompt", args.prompt)) if v} + with open(path, "rb") as fh: + r = requests.post(f"{url}/transcribe", params=params, data=fh, timeout=args.http_timeout) + if r.status_code != 200: + raise RuntimeError(f"HTTP {r.status_code}: {r.text[:300]}") + return r.json() + + +def service_mic(url: str, args) -> int: + from websockets.sync.client import connect + + ws_url = url.replace("http://", "ws://").replace("https://", "wss://").rstrip("/") + "/ws/stream" + got_text = threading.Event() + stop = threading.Event() + result = {"code": 0} + + with Microphone(args.device, args.channel) as mic, connect(ws_url, max_size=None) as ws: + ws.recv() # ready + start = {"type": "start", "sample_rate": mic.rate, "language": args.language, + "prompt": args.prompt} + if args.silence_ms: + start["silence_ms"] = args.silence_ms + ws.send(json.dumps(start)) + ws.recv() # ready (neu konfiguriert) + status("Bereit — sprich. (Strg+C beendet)") + + def receiver(): + try: + for raw in ws: + event = json.loads(raw) + etype = event.get("type") + if etype == "speech_start" and args.verbose: + status(f" … Sprache erkannt ({event['t']:.1f}s)") + elif etype == "transcript": + if not event["text"]: + continue + if args.verbose: + status(f" ({event['duration']:.1f}s Audio, {event['processing_s']:.2f}s Rechenzeit)") + emit(event if args.json else {"text": event["text"]}, args.json) + got_text.set() + if args.once: + stop.set() + return + elif etype == "error": + status(f" Fehler vom Dienst: {event.get('detail')}") + stop.set() # Dienst hat die Verbindung regulär geschlossen + except Exception as exc: # Verbindung weg + if not stop.is_set(): + status(f"Verbindung beendet: {exc}") + result["code"] = 1 + stop.set() + + rx = threading.Thread(target=receiver, daemon=True) + rx.start() + deadline = time.monotonic() + args.timeout if args.timeout else None + try: + while not stop.is_set(): + if deadline and time.monotonic() > deadline: + status("Zeitlimit erreicht.") + break + try: + ws.send(mic.blocks.get(timeout=0.2)) + except queue.Empty: + continue + except KeyboardInterrupt: + status("\nBeendet.") + finally: + if mic.dropped: + status(f"Warnung: {mic.dropped} Audioblöcke verworfen (Netz zu langsam?)") + if not stop.is_set(): + try: + ws.send(json.dumps({"type": "close"})) + rx.join(timeout=60) # ausstehende Transkripte noch abholen + except Exception: + pass + if result["code"]: + return result["code"] + return 0 if got_text.is_set() or not (args.once or args.timeout) else 2 + + +# === Lokaler Modus (ohne Dienst) ============================================== + +def _local_engine(args): + import stt_core as core + + cfg = core.EngineConfig.from_env() + if args.model: + cfg.model = cfg.cpu_model = args.model + if args.compute_device: + cfg.device = args.compute_device + engine = core.SttEngine(cfg) + status("Lade Modell …") + engine.load() + status(f"Modell {engine.active_model} auf {engine.active_device} bereit ({engine.load_seconds:.1f}s)") + return engine + + +def local_files(paths: list[Path], args) -> int: + import stt_core as core + + engine = _local_engine(args) + code = 2 + for path in paths: + audio = core.decode_audio_bytes(path.read_bytes()) + tr = engine.transcribe(audio, language=args.language, initial_prompt=args.prompt) + emit(tr.to_dict() if args.json else {"text": tr.text}, args.json, str(path)) + code = 0 if tr.text else code + return code + + +def local_mic(args) -> int: + from concurrent.futures import ThreadPoolExecutor + + import stt_core as core + + engine = _local_engine(args) + vad = core.VadConfig.from_env() + if args.silence_ms: + vad.silence_ms = args.silence_ms + segmenter = core.UtteranceSegmenter(vad) + executor = ThreadPoolExecutor(max_workers=1) # Inferenz getrennt von der Aufnahme + done = threading.Event() + got_text = threading.Event() + + def on_result(future): + tr = future.result() + if tr.text and not (args.once and got_text.is_set()): + emit(tr.to_dict(with_segments=False) if args.json else {"text": tr.text}, args.json) + got_text.set() + if args.once: + done.set() + + def submit(event): + fut = executor.submit(engine.transcribe, event.audio, language=args.language, + initial_prompt=args.prompt) + fut.add_done_callback(on_result) + + with Microphone(args.device, args.channel) as mic: + resampler = core.StreamResampler(mic.rate) + status("Bereit — sprich. (Strg+C beendet)") + deadline = time.monotonic() + args.timeout if args.timeout else None + try: + while not done.is_set(): + if deadline and time.monotonic() > deadline: + status("Zeitlimit erreicht.") + break + try: + block = mic.blocks.get(timeout=0.2) + except queue.Empty: + continue + audio = resampler.process(core.pcm16_to_float(block)) + for event in segmenter.feed(audio): + if event.kind == "speech_start" and args.verbose: + status(f" … Sprache erkannt ({event.time_s:.1f}s)") + elif event.kind == "utterance": + submit(event) + except KeyboardInterrupt: + status("\nBeendet.") + for event in segmenter.flush(): + submit(event) + executor.shutdown(wait=True) + return 0 if got_text.is_set() or not (args.once or args.timeout) else 2 + + +# === Main ===================================================================== + +def main() -> int: + p = argparse.ArgumentParser( + description="STT-CLI: Datei oder Mikrofon → Text (über stt_service oder lokal)", + formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__.split("Beispiele:")[1], + ) + p.add_argument("files", nargs="*", type=Path, help="Audiodateien (WAV, MP3, OGG, …); ohne = Mikrofon") + p.add_argument("--url", default=DEFAULT_URL, help=f"Dienst-URL (Standard: {DEFAULT_URL}, env STT_URL)") + p.add_argument("--local", action="store_true", help="ohne Dienst: Modell im Prozess laden") + p.add_argument("-l", "--language", default=None, help="Sprachcode (de, en, …) oder 'auto'; Standard: Dienst-Default") + p.add_argument("--prompt", default=None, help="Kontext/Eigennamen für Whisper") + p.add_argument("--json", action="store_true", help="JSON-Zeilen statt reinem Text") + p.add_argument("-d", "--device", default=None, help="Eingabegerät: Index oder Namensteil (Standard: System-Default)") + p.add_argument("--channel", type=int, default=None, help="Kanal wählen (ReSpeaker: automatisch 1)") + p.add_argument("--once", action="store_true", help="nach der ersten erkannten Äußerung beenden") + p.add_argument("--timeout", type=float, default=None, help="Mikrofon-Modus nach N Sekunden beenden") + p.add_argument("--silence-ms", type=int, default=None, help="Stille bis Äußerungsende (Standard: 700)") + p.add_argument("--model", default=None, help="nur --local: Whisper-Modell") + p.add_argument("--compute-device", choices=["auto", "cuda", "cpu"], default=None, help="nur --local") + p.add_argument("--http-timeout", type=float, default=300.0) + p.add_argument("--list-devices", action="store_true", help="Eingabegeräte auflisten") + p.add_argument("-v", "--verbose", action="store_true", help="Statusdetails auf stderr") + args = p.parse_args() + + try: + if args.list_devices: + list_devices() + return 0 + if args.files: + missing = [f for f in args.files if not f.is_file()] + if missing: + status(f"Datei nicht gefunden: {', '.join(map(str, missing))}") + return 1 + if args.local: + return local_files(args.files, args) + code = 2 + for path in args.files: + result = service_file(args.url, path, args) + emit(result if args.json else {"text": result["text"]}, args.json, str(path)) + code = 0 if result["text"] else code + return code + return local_mic(args) if args.local else service_mic(args.url, args) + except KeyboardInterrupt: + return 130 + except SystemExit: + raise + except Exception as exc: + hint = "" + if "Connection refused" in str(exc) or "ConnectionRefused" in type(exc).__name__: + hint = f"\nLäuft der Dienst? (systemctl --user status stt-service, {args.url}/health) — oder --local nutzen." + status(f"Fehler: {exc}{hint}") + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/stt_core.py b/stt_core.py new file mode 100644 index 0000000..471ba4b --- /dev/null +++ b/stt_core.py @@ -0,0 +1,509 @@ +"""STT-Kern: Whisper-Engine, Audio-Hilfen und Äußerungs-Segmentierung (VAD). + +Wird von stt_service.py (HTTP/WebSocket/Wyoming) und stt_cli.py (lokaler Modus) +gemeinsam genutzt. Keine Ausgabe per print, keine sys.exit() — Fehler werden als +Exceptions gemeldet, Ergebnisse als Datenobjekte zurückgegeben. + +Grundsätze (Lehren aus stt.py und my_voice_assistant_v3): +- Modell EINMAL laden und halten; bei GPU-Problemen robust auf CPU ausweichen. +- Äußerungsende per Silero-VAD (Stille nach Sprache) statt Stoppwort. +- Genau eine Transkription pro Äußerung, ohne Kontext-Verkettung + (condition_on_previous_text=False) und mit Halluzinationsfilter. +""" + +from __future__ import annotations + +import io +import logging +import os +import re +import threading +import time +from collections import deque +from dataclasses import asdict, dataclass, field +from math import gcd +from typing import Callable + +import numpy as np + +log = logging.getLogger("stt") + +__version__ = "0.1.0" + +SAMPLE_RATE = 16000 # Whisper arbeitet intern mit 16 kHz mono float32 + + +class AudioDecodeError(ValueError): + """Audio konnte nicht dekodiert werden (defekte Datei, unbekanntes Format).""" + + +# === Audio-Hilfen ============================================================= + +def pcm16_to_float(data: bytes, channels: int = 1, channel: int | None = None) -> np.ndarray: + """Rohes s16le-PCM → float32 mono. + + channels > 1: `channel` wählt einen Kanal (z. B. 1 = rechter ASR-Kanal des + ReSpeaker XVF3800), sonst werden alle Kanäle gemittelt. Unvollständige + Frames am Ende werden verworfen. + """ + frame = 2 * channels + usable = len(data) - (len(data) % frame) + audio = np.frombuffer(data[:usable], dtype=" 1: + audio = audio.reshape(-1, channels) + if channel is not None: + if not 0 <= channel < channels: + raise ValueError(f"channel={channel} ungültig bei {channels} Kanälen") + audio = audio[:, channel] + else: + audio = audio.mean(axis=1) + return np.ascontiguousarray(audio, dtype=np.float32) + + +def float_to_pcm16(audio: np.ndarray) -> bytes: + return (np.clip(audio, -1.0, 1.0) * 32767.0).astype(" np.ndarray: + """Resampelt ein komplettes Signal (nicht für Streaming-Stückchen gedacht).""" + if src_rate == dst_rate or audio.size == 0: + return audio.astype(np.float32, copy=False) + from scipy.signal import resample_poly + + g = gcd(src_rate, dst_rate) + return resample_poly(audio, dst_rate // g, src_rate // g).astype(np.float32) + + +class StreamResampler: + """Resampelt fortlaufende Audio-Stückchen ohne Artefakte an den Stückgrenzen. + + Nutzt soxr.ResampleStream (zustandsbehaftet); ohne soxr Fallback auf + resample_poly pro Stück (kleine Randartefakte, für ASR unkritisch). + """ + + def __init__(self, src_rate: int, dst_rate: int = SAMPLE_RATE): + self.src_rate = src_rate + self.dst_rate = dst_rate + self._stream = None + if src_rate != dst_rate: + try: + import soxr + + self._stream = soxr.ResampleStream(src_rate, dst_rate, 1, dtype="float32") + except ImportError: + self._stream = None + + def process(self, audio: np.ndarray, last: bool = False) -> np.ndarray: + if self.src_rate == self.dst_rate: + return audio + if self._stream is not None: + return self._stream.resample_chunk(audio.astype(np.float32, copy=False), last=last) + return resample(audio, self.src_rate, self.dst_rate) + + +def decode_audio_bytes(data: bytes) -> np.ndarray: + """Dekodiert beliebige Audio-Container (WAV jeder Bittiefe/Kanalzahl, MP3, OGG, + WebM, M4A …) via PyAV nach 16 kHz mono float32.""" + if not data: + raise AudioDecodeError("Leere Audiodaten") + from faster_whisper import decode_audio + + try: + return decode_audio(io.BytesIO(data), sampling_rate=SAMPLE_RATE) + except Exception as exc: # PyAV wirft diverse Typen + raise AudioDecodeError(f"Audio nicht dekodierbar: {exc}") from exc + + +# === Ergebnis-Datenobjekte ==================================================== + +@dataclass +class Segment: + start: float + end: float + text: str + avg_logprob: float = 0.0 + no_speech_prob: float = 0.0 + + +@dataclass +class Transcript: + text: str + language: str | None + language_probability: float + duration: float # Audiodauer in Sekunden + processing_s: float # Rechenzeit in Sekunden + segments: list[Segment] = field(default_factory=list) + dropped_segments: int = 0 # vom Halluzinationsfilter verworfen + + def to_dict(self, with_segments: bool = True) -> dict: + data = asdict(self) + if not with_segments: + data.pop("segments") + return data + + +# === Halluzinationsfilter ===================================================== + +# Whisper erzeugt in Stille/Rauschen typische Floskeln aus seinen Trainingsdaten +# (Untertitel). Gefiltert wird nur, wenn ein Segment KOMPLETT aus so einer Floskel +# besteht — echte Sätze, die solche Wörter enthalten, bleiben erhalten. +_HALLUCINATION_PATTERNS = [ + r"untertitel(ung)? (im auftrag )?(des|der|von|by) .*", + r"untertitel (der|von) amara\.org( community)?", + r"(copyright|©) ?(wdr|swr|zdf|ard|ndr|br|mdr)( \d{4})?", + r"(swr|wdr|zdf|ndr|mdr) \d{4}", + r"vielen dank (fürs|für's|für das) (zuschauen|zusehen|zuhören)", + r"(thanks|thank you) for watching", + r"subtitles by .*", + r"sous-titr(es|age) (réalisés? )?par .*", +] +_HALLUCINATION_RE = re.compile( + "^(?:" + "|".join(_HALLUCINATION_PATTERNS) + ")$", re.IGNORECASE +) +_PUNCT_RE = re.compile(r"[^\w\s©'.-]+|[.!?…]+$") + + +def is_hallucination(seg: Segment) -> bool: + norm = _PUNCT_RE.sub("", seg.text.strip().lower()).strip(" .") + if not norm: + return True + if _HALLUCINATION_RE.match(norm): + return True + # Whisper selbst unsicher UND Stille wahrscheinlich → verwerfen + return seg.no_speech_prob > 0.6 and seg.avg_logprob < -0.8 + + +# === Engine =================================================================== + +def _env(name: str, default, cast=str): + raw = os.environ.get(name) + if raw is None or raw == "": + return default + if cast is bool: + return raw.strip().lower() in ("1", "true", "yes", "on") + return cast(raw) + + +@dataclass +class EngineConfig: + model: str = "large-v3-turbo" # tiny|base|small|medium|large-v3|large-v3-turbo|Pfad + device: str = "auto" # auto|cuda|cpu + compute_type: str = "auto" # auto → float16 (cuda) / int8 (cpu) + device_index: int = 0 # CUDA-Index; besser GPU per CUDA_VISIBLE_DEVICES=UUID pinnen + cpu_model: str = "small" # Modell, wenn (automatisch) auf CPU ausgewichen wird + cpu_threads: int = 0 # 0 = ctranslate2-Default + language: str | None = "de" # Default-Sprache; None/"auto" = automatische Erkennung + beam_size: int = 5 + initial_prompt: str | None = None + download_root: str | None = None + + @classmethod + def from_env(cls) -> "EngineConfig": + d = cls() + language = _env("STT_LANGUAGE", d.language) + return cls( + model=_env("STT_MODEL", d.model), + device=_env("STT_DEVICE", d.device), + compute_type=_env("STT_COMPUTE_TYPE", d.compute_type), + device_index=_env("STT_DEVICE_INDEX", d.device_index, int), + cpu_model=_env("STT_CPU_MODEL", d.cpu_model), + cpu_threads=_env("STT_CPU_THREADS", d.cpu_threads, int), + language=None if language in (None, "auto") else language, + beam_size=_env("STT_BEAM_SIZE", d.beam_size, int), + initial_prompt=_env("STT_INITIAL_PROMPT", d.initial_prompt), + download_root=_env("STT_DOWNLOAD_ROOT", d.download_root), + ) + + +def cuda_device_count() -> int: + try: + import ctranslate2 + + return ctranslate2.get_cuda_device_count() + except Exception: + return 0 + + +class SttEngine: + """Thread-sichere Whisper-Engine. Modell wird beim ersten Bedarf (oder per + load()) genau einmal geladen; parallele Aufrufe werden serialisiert.""" + + def __init__(self, config: EngineConfig | None = None): + self.config = config or EngineConfig() + self._model = None + self._load_lock = threading.Lock() + self._run_lock = threading.Lock() + self.status = "idle" # idle|loading|ready|error + self.error: str | None = None + self.active_model: str | None = None + self.active_device: str | None = None + self.active_compute_type: str | None = None + self.load_seconds: float | None = None + + # --- Laden ---------------------------------------------------------------- + + def _load_attempts(self) -> list[tuple[str, str, str]]: + cfg = self.config + attempts: list[tuple[str, str, str]] = [] + if cfg.device in ("auto", "cuda"): + if cuda_device_count() > 0: + ct = "float16" if cfg.compute_type == "auto" else cfg.compute_type + attempts.append(("cuda", cfg.model, ct)) + else: + log.warning("Keine CUDA-GPU gefunden — nutze CPU") + cpu_model = cfg.model if cfg.device == "cpu" else cfg.cpu_model + cpu_ct = "int8" if cfg.compute_type == "auto" or cfg.device != "cpu" else cfg.compute_type + attempts.append(("cpu", cpu_model, cpu_ct)) + return attempts + + def load(self): + with self._load_lock: + if self._model is not None: + return self._model + from faster_whisper import WhisperModel + + self.status = "loading" + t0 = time.perf_counter() + last_exc: Exception | None = None + for device, model_name, compute_type in self._load_attempts(): + try: + log.info("Lade Whisper-Modell %s (%s, %s)", model_name, device, compute_type) + kwargs = dict( + device=device, + compute_type=compute_type, + cpu_threads=self.config.cpu_threads, + download_root=self.config.download_root, + ) + if device == "cuda": + kwargs["device_index"] = self.config.device_index + self._model = WhisperModel(model_name, **kwargs) + self.active_model, self.active_device = model_name, device + self.active_compute_type = compute_type + break + except Exception as exc: + last_exc = exc + log.warning("Laden auf %s fehlgeschlagen: %s", device, exc) + if self._model is None: + self.status, self.error = "error", str(last_exc) + raise RuntimeError(f"Whisper-Modell nicht ladbar: {last_exc}") from last_exc + self.load_seconds = time.perf_counter() - t0 + self.status, self.error = "ready", None + log.info("Modell bereit nach %.1fs", self.load_seconds) + return self._model + + def warmup(self) -> None: + """Lädt das Modell und rechnet einmal durch (CUDA-Kernel, Decoder).""" + self.load() + noise = (np.random.default_rng(0).standard_normal(SAMPLE_RATE) * 0.01).astype(np.float32) + self.transcribe(noise, vad_filter=False, beam_size=1) + + # --- Transkription --------------------------------------------------------- + + def resolve_language(self, language: str | None) -> str | None: + """None → Default aus Config; 'auto' → Erkennung; 'de-DE' → 'de'.""" + if language is None: + return self.config.language + language = language.strip().lower() + if language in ("", "auto"): + return None + return language.split("-")[0].split("_")[0] + + def transcribe( + self, + audio: np.ndarray, + *, + language: str | None = None, + initial_prompt: str | None = None, + hotwords: str | None = None, + beam_size: int | None = None, + vad_filter: bool = True, + ) -> Transcript: + """Transkribiert 16-kHz-mono-float32-Audio. Blockiert — in Diensten in einem + Worker-Thread aufrufen.""" + duration = float(audio.size) / SAMPLE_RATE + lang = self.resolve_language(language) + if audio.size < SAMPLE_RATE // 10: # < 100 ms: nichts Sinnvolles + return Transcript("", lang, 0.0, duration, 0.0) + + model = self.load() + t0 = time.perf_counter() + with self._run_lock: + segments, info = model.transcribe( + audio.astype(np.float32, copy=False), + language=lang, + beam_size=beam_size or self.config.beam_size, + vad_filter=vad_filter, + vad_parameters={"min_silence_duration_ms": 500, "speech_pad_ms": 200}, + condition_on_previous_text=False, + initial_prompt=initial_prompt or self.config.initial_prompt, + hotwords=hotwords, + ) + raw = list(segments) # Generator: die eigentliche Arbeit passiert hier + + kept: list[Segment] = [] + dropped = 0 + for s in raw: + seg = Segment(round(s.start, 2), round(s.end, 2), s.text.strip(), + round(s.avg_logprob, 3), round(s.no_speech_prob, 3)) + if is_hallucination(seg): + dropped += 1 + log.debug("Halluzination verworfen: %r", seg.text) + else: + kept.append(seg) + text = " ".join(s.text for s in kept).strip() + return Transcript( + text=text, + language=info.language, + language_probability=round(float(info.language_probability), 3), + duration=round(duration, 2), + processing_s=round(time.perf_counter() - t0, 3), + segments=kept, + dropped_segments=dropped, + ) + + +# === Äußerungs-Segmentierung (Streaming-VAD) ================================== + +VAD_WINDOW = 512 # Samples pro Silero-Fenster bei 16 kHz (= 32 ms) +_WINDOW_MS = VAD_WINDOW * 1000 / SAMPLE_RATE + + +class SileroProb: + """Sprachwahrscheinlichkeit je 512-Sample-Fenster mit dem in faster-whisper + mitgelieferten Silero-VAD (ONNX, CPU, kein Download). + + Das Modell leitet den Kontext eines Fensters aus dem vorherigen Fenster im + selben Aufruf ab. Damit das über Aufrufgrenzen stimmt, wird das letzte Fenster + des Vorgängeraufrufs vorangestellt und dessen Ergebnis verworfen. + """ + + def __init__(self): + from faster_whisper.vad import get_vad_model + + self._model = get_vad_model() + self._prev = np.zeros(VAD_WINDOW, dtype=np.float32) + + def __call__(self, windows: np.ndarray) -> np.ndarray: + audio = np.concatenate([self._prev, windows]) + self._prev = windows[-VAD_WINDOW:].copy() + return np.asarray(self._model(audio)).reshape(-1)[1:] + + def reset(self) -> None: + self._prev[:] = 0 + + +@dataclass +class VadConfig: + threshold: float = 0.5 # Sprachwahrscheinlichkeit für "Sprache" + min_speech_ms: int = 200 # so lange Sprache nötig, bis eine Äußerung beginnt + silence_ms: int = 700 # so lange Stille beendet eine Äußerung + pre_roll_ms: int = 300 # Audio vor Sprachbeginn mitnehmen (Anlaute) + max_utterance_s: float = 30.0 # harte Obergrenze (Whisper-Fenster) + + @classmethod + def from_env(cls) -> "VadConfig": + d = cls() + return cls( + threshold=_env("STT_VAD_THRESHOLD", d.threshold, float), + min_speech_ms=_env("STT_VAD_MIN_SPEECH_MS", d.min_speech_ms, int), + silence_ms=_env("STT_VAD_SILENCE_MS", d.silence_ms, int), + pre_roll_ms=_env("STT_VAD_PRE_ROLL_MS", d.pre_roll_ms, int), + max_utterance_s=_env("STT_VAD_MAX_UTTERANCE_S", d.max_utterance_s, float), + ) + + +@dataclass +class VadEvent: + kind: str # "speech_start" | "utterance" + time_s: float # Stream-Zeit des Ereignisses + audio: np.ndarray | None = None + reason: str | None = None # utterance: "silence" | "max_length" | "flush" + + +class UtteranceSegmenter: + """Zerlegt einen fortlaufenden 16-kHz-Stream in Äußerungen. + + feed() liefert Ereignisse, sobald Sprache beginnt bzw. eine Äußerung durch + Stille, Maximallänge oder flush() endet. Rein rechnerisch, kein I/O. + """ + + def __init__(self, config: VadConfig | None = None, + prob_fn: Callable[[np.ndarray], np.ndarray] | None = None): + self.config = config or VadConfig() + self._prob_fn = prob_fn if prob_fn is not None else SileroProb() + pre_windows = int(np.ceil((self.config.pre_roll_ms + self.config.min_speech_ms) / _WINDOW_MS)) + self._pre: deque[np.ndarray] = deque(maxlen=max(pre_windows, 1)) + self._pending = np.zeros(0, dtype=np.float32) + self._speech: list[np.ndarray] | None = None + self._voiced = 0 + self._silent = 0 + self._samples = 0 + + @property + def in_speech(self) -> bool: + return self._speech is not None + + def reset(self) -> None: + self._pre.clear() + self._pending = np.zeros(0, dtype=np.float32) + self._speech = None + self._voiced = self._silent = 0 + if hasattr(self._prob_fn, "reset"): + self._prob_fn.reset() + + def _now(self) -> float: + return self._samples / SAMPLE_RATE + + def _finish(self, reason: str) -> VadEvent: + audio = np.concatenate(self._speech) if self._speech else np.zeros(0, np.float32) + self._speech = None + self._voiced = self._silent = 0 + self._pre.clear() + return VadEvent("utterance", self._now(), audio=audio, reason=reason) + + def feed(self, audio: np.ndarray) -> list[VadEvent]: + cfg = self.config + buf = np.concatenate([self._pending, audio.astype(np.float32, copy=False)]) + n = buf.size // VAD_WINDOW + self._pending = buf[n * VAD_WINDOW:] + if n == 0: + return [] + windows = buf[: n * VAD_WINDOW] + probs = self._prob_fn(windows) + + min_speech = max(1, int(cfg.min_speech_ms / _WINDOW_MS)) + end_silence = max(1, int(cfg.silence_ms / _WINDOW_MS)) + max_windows = int(cfg.max_utterance_s * 1000 / _WINDOW_MS) + neg_threshold = max(cfg.threshold - 0.15, 0.01) + + events: list[VadEvent] = [] + for i in range(n): + win = windows[i * VAD_WINDOW:(i + 1) * VAD_WINDOW] + p = float(probs[i]) + self._samples += VAD_WINDOW + if self._speech is None: + self._pre.append(win) + self._voiced = self._voiced + 1 if p >= cfg.threshold else 0 + if self._voiced >= min_speech: + self._speech = list(self._pre) + self._pre.clear() + self._silent = 0 + events.append(VadEvent("speech_start", self._now())) + else: + self._speech.append(win) + self._silent = self._silent + 1 if p < neg_threshold else 0 + if self._silent >= end_silence: + events.append(self._finish("silence")) + elif len(self._speech) >= max_windows: + events.append(self._finish("max_length")) + return events + + def flush(self) -> list[VadEvent]: + """Beendet eine laufende Äußerung (z. B. Stream-Ende).""" + if self._speech is None: + self._pending = np.zeros(0, dtype=np.float32) + return [] + if self._pending.size: + self._speech.append(self._pending) + self._pending = np.zeros(0, dtype=np.float32) + return [self._finish("flush")] diff --git a/stt_service.py b/stt_service.py new file mode 100644 index 0000000..a2b7752 --- /dev/null +++ b/stt_service.py @@ -0,0 +1,439 @@ +#!/usr/bin/env python3 +"""STT-Service – lokaler Spracherkennungs-Dienst (Pendant zu chatterbox-tts-cli). + +Start: + .venv/bin/uvicorn stt_service:app --host 0.0.0.0 --port 9998 + +Endpunkte: + GET /health – Status, Modell, Gerät, Warteschlange + POST /v1/audio/transcriptions – OpenAI-kompatibel (multipart: file, language, prompt, + response_format=json|text|verbose_json) + POST /transcribe – Rohdaten im Body (für Mikrocontroller, ohne multipart): + WAV/MP3/… oder ?format=pcm&sample_rate=16000&channels=1 + WS /ws/stream – Live-Streaming: s16le-PCM rein, Äußerungsende per VAD, + Transkript-Events raus + TCP :10300 (Wyoming) – Home Assistant (STT_WYOMING_PORT, leer = aus) + +Nicht blockierend: Die Inferenz läuft in genau einem Worker-Thread (GPU-schonend), +der Event-Loop bleibt frei. Ist die Warteschlange voll, wird sofort mit 503 +abgelehnt statt Anfragen unbegrenzt zu stauen. + +Konfiguration per Umgebung (siehe README): STT_MODEL, STT_DEVICE, STT_LANGUAGE, +STT_MAX_QUEUE, STT_VAD_*, STT_WYOMING_HOST/PORT, STT_PRELOAD … +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import os +from concurrent.futures import ThreadPoolExecutor +from contextlib import asynccontextmanager +from dataclasses import dataclass +from functools import partial +from typing import Callable + +import numpy as np +from fastapi import FastAPI, File, Form, HTTPException, Query, Request, UploadFile, WebSocket +from fastapi.responses import PlainTextResponse +from starlette.websockets import WebSocketDisconnect, WebSocketState + +import stt_core as core + +log = logging.getLogger("stt.service") + +__version__ = core.__version__ + + +class Overloaded(RuntimeError): + """Warteschlange voll.""" + + +@dataclass +class ServiceSettings: + max_queue: int = 8 + max_upload_mb: int = 50 + preload: bool = True + wyoming_host: str = "0.0.0.0" + wyoming_port: int | None = 10300 + + @classmethod + def from_env(cls) -> "ServiceSettings": + d = cls() + port = os.environ.get("STT_WYOMING_PORT", str(d.wyoming_port)) + return cls( + max_queue=core._env("STT_MAX_QUEUE", d.max_queue, int), + max_upload_mb=core._env("STT_MAX_UPLOAD_MB", d.max_upload_mb, int), + preload=core._env("STT_PRELOAD", d.preload, bool), + wyoming_host=core._env("STT_WYOMING_HOST", d.wyoming_host), + wyoming_port=int(port) if port.strip() else None, + ) + + +class TranscriptionWorker: + """Serialisiert Inferenz in einem Thread; begrenzt die Anzahl wartender Aufträge.""" + + def __init__(self, engine: core.SttEngine, max_queue: int = 8): + self.engine = engine + self.max_queue = max_queue + self.pending = 0 # nur im Event-Loop verändert → kein Lock nötig + self._executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="stt-worker") + + async def run(self, fn: Callable, *args, **kwargs): + loop = asyncio.get_running_loop() + return await loop.run_in_executor(self._executor, partial(fn, *args, **kwargs)) + + async def transcribe(self, audio: np.ndarray, **kwargs) -> core.Transcript: + if self.pending >= self.max_queue: + raise Overloaded(f"Warteschlange voll ({self.pending} Aufträge)") + self.pending += 1 + try: + return await self.run(self.engine.transcribe, audio, **kwargs) + finally: + self.pending -= 1 + + def shutdown(self) -> None: + self._executor.shutdown(wait=False, cancel_futures=True) + + +def create_app( + engine: core.SttEngine | None = None, + settings: ServiceSettings | None = None, + vad_config: core.VadConfig | None = None, + prob_fn_factory: Callable[[], Callable] | None = None, +) -> FastAPI: + """App-Fabrik; Tests injizieren eine Fake-Engine bzw. eine VAD-Funktion.""" + engine = engine or core.SttEngine(core.EngineConfig.from_env()) + settings = settings or ServiceSettings.from_env() + vad_config = vad_config or core.VadConfig.from_env() + prob_fn_factory = prob_fn_factory or core.SileroProb + worker = TranscriptionWorker(engine, settings.max_queue) + + @asynccontextmanager + async def lifespan(app: FastAPI): + preload_future = None + if settings.preload: + # Nicht awaiten: Dienst ist sofort erreichbar (/health = loading); + # erste Anfragen reihen sich im Worker hinter dem Laden ein. + preload_future = asyncio.ensure_future(worker.run(engine.warmup)) + preload_future.add_done_callback(_log_preload_result) + wyoming_server = None + if settings.wyoming_port: + wyoming_server = await _start_wyoming(worker, settings) + app.state.wyoming = wyoming_server + try: + yield + finally: + if wyoming_server is not None: + await wyoming_server.stop() + if preload_future is not None and not preload_future.done(): + preload_future.cancel() + worker.shutdown() + + app = FastAPI(title="STT-Service", version=__version__, lifespan=lifespan) + app.state.engine = engine + app.state.worker = worker + + # --- HTTP ----------------------------------------------------------------- + + @app.get("/health") + async def health(): + return { + "status": "ok" if engine.status == "ready" else engine.status, + "model": engine.active_model or engine.config.model, + "device": engine.active_device, + "compute_type": engine.active_compute_type, + "default_language": engine.config.language or "auto", + "queue": worker.pending, + "max_queue": worker.max_queue, + "load_seconds": engine.load_seconds, + "wyoming_port": settings.wyoming_port, + "error": engine.error, + "version": __version__, + } + + async def _run_transcription(audio: np.ndarray, **kwargs) -> core.Transcript: + try: + return await worker.transcribe(audio, **kwargs) + except Overloaded as exc: + raise HTTPException(status_code=503, detail=str(exc), headers={"Retry-After": "2"}) + except RuntimeError as exc: + log.exception("Transkription fehlgeschlagen") + raise HTTPException(status_code=500, detail=str(exc)) + + def _check_size(size: int) -> None: + if size > settings.max_upload_mb * 1024 * 1024: + raise HTTPException(status_code=413, detail=f"Audio größer als {settings.max_upload_mb} MB") + + async def _decode(data: bytes) -> np.ndarray: + try: + # Dekodieren (PyAV) kostet bei langen Dateien spürbar CPU → nicht im Event-Loop + return await asyncio.to_thread(core.decode_audio_bytes, data) + except core.AudioDecodeError as exc: + raise HTTPException(status_code=400, detail=str(exc)) + + @app.post("/v1/audio/transcriptions") + async def openai_transcriptions( + file: UploadFile = File(...), + model: str | None = Form(default=None), # akzeptiert, aber ignoriert (ein Modell pro Dienst) + language: str | None = Form(default=None), + prompt: str | None = Form(default=None), + response_format: str = Form(default="json"), + temperature: float | None = Form(default=None), # ignoriert + ): + data = await file.read() + _check_size(len(data)) + audio = await _decode(data) + tr = await _run_transcription(audio, language=language, initial_prompt=prompt) + if response_format == "text": + return PlainTextResponse(tr.text) + if response_format == "verbose_json": + return { + "task": "transcribe", + "language": tr.language, + "duration": tr.duration, + "text": tr.text, + "segments": [ + {"id": i, "start": s.start, "end": s.end, "text": s.text, + "avg_logprob": s.avg_logprob, "no_speech_prob": s.no_speech_prob} + for i, s in enumerate(tr.segments) + ], + } + if response_format != "json": + raise HTTPException(status_code=400, detail=f"response_format '{response_format}' nicht unterstützt") + return {"text": tr.text} + + @app.post("/transcribe") + async def transcribe_raw( + request: Request, + format: str = Query(default="auto", pattern="^(auto|pcm)$"), + sample_rate: int = Query(default=16000, ge=4000, le=192000), + channels: int = Query(default=1, ge=1, le=16), + channel: int | None = Query(default=None, ge=0), + language: str | None = Query(default=None), + prompt: str | None = Query(default=None), + ): + data = await request.body() + _check_size(len(data)) + if not data: + raise HTTPException(status_code=400, detail="Leerer Body") + if format == "pcm": + try: + audio = core.resample(core.pcm16_to_float(data, channels, channel), sample_rate) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) + else: + audio = await _decode(data) + tr = await _run_transcription(audio, language=language, initial_prompt=prompt) + return tr.to_dict() + + # --- WebSocket-Streaming -------------------------------------------------- + + @app.websocket("/ws/stream") + async def ws_stream(ws: WebSocket): + await ws.accept() + session = _StreamSession(ws, worker, vad_config, prob_fn_factory) + await session.run() + + return app + + +def _log_preload_result(future: asyncio.Future) -> None: + if future.cancelled(): + return + exc = future.exception() + if exc is not None: + log.error("Vorladen des Modells fehlgeschlagen: %s", exc) + + +class _StreamSession: + """Eine WebSocket-Verbindung. + + Protokoll (Client → Server): + Text {"type":"start", "sample_rate":16000, "channels":1, "channel":null, + "language":"de", "prompt":null, "vad":true, + "silence_ms":700, "threshold":0.5, "max_utterance_s":30} (optional) + Binär s16le-PCM (beliebige Stückgröße) + Text {"type":"end"} – laufende Äußerung abschließen (bzw. ohne VAD: alles + bisher Gesendete transkribieren); danach "done" + Text {"type":"close"} – Verbindung nach ausstehenden Transkripten schließen + + Server → Client: + {"type":"ready", ...} {"type":"speech_start","t":1.23} + {"type":"transcript","text":...,"language":...,"duration":...,"processing_s":..., + "reason":"silence|max_length|flush","t":...} + {"type":"done"} {"type":"error","detail":...} + """ + + def __init__(self, ws: WebSocket, worker: TranscriptionWorker, + vad_config: core.VadConfig, prob_fn_factory: Callable): + self.ws = ws + self.worker = worker + self.base_vad = vad_config + self.prob_fn_factory = prob_fn_factory + self.send_lock = asyncio.Lock() + self.jobs: asyncio.Queue = asyncio.Queue() + self._configure({}) + + def _configure(self, opts: dict) -> None: + self.sample_rate = int(opts.get("sample_rate", 16000)) + self.channels = int(opts.get("channels", 1)) + self.channel = opts.get("channel") + self.language = opts.get("language") + self.prompt = opts.get("prompt") + self.use_vad = bool(opts.get("vad", True)) + vad = core.VadConfig( + threshold=float(opts.get("threshold", self.base_vad.threshold)), + min_speech_ms=int(opts.get("min_speech_ms", self.base_vad.min_speech_ms)), + silence_ms=int(opts.get("silence_ms", self.base_vad.silence_ms)), + pre_roll_ms=int(opts.get("pre_roll_ms", self.base_vad.pre_roll_ms)), + max_utterance_s=float(opts.get("max_utterance_s", self.base_vad.max_utterance_s)), + ) + self.resampler = core.StreamResampler(self.sample_rate) + self.segmenter = core.UtteranceSegmenter(vad, self.prob_fn_factory()) if self.use_vad else None + self.buffer: list[np.ndarray] = [] + self.buffered_samples = 0 + self.max_buffer_samples = int(vad.max_utterance_s * core.SAMPLE_RATE) + + async def send(self, payload: dict) -> None: + if self.ws.application_state != WebSocketState.CONNECTED: + return + async with self.send_lock: + await self.ws.send_json(payload) + + async def run(self) -> None: + consumer = asyncio.create_task(self._consume()) + await self.send({"type": "ready", "sample_rate": self.sample_rate, "vad": self.use_vad}) + try: + while True: + message = await self.ws.receive() + if message["type"] == "websocket.disconnect": + break + if message.get("bytes") is not None: + await self._on_audio(message["bytes"]) + elif message.get("text") is not None: + if not await self._on_control(message["text"]): + break + except WebSocketDisconnect: + pass + finally: + await self.jobs.put(None) + try: + await asyncio.wait_for(consumer, timeout=120) + except (asyncio.TimeoutError, asyncio.CancelledError): + consumer.cancel() + if self.ws.application_state == WebSocketState.CONNECTED: + await self.ws.close() + + async def _on_audio(self, data: bytes) -> None: + try: + audio = core.pcm16_to_float(data, self.channels, self.channel) + except ValueError as exc: + await self.send({"type": "error", "detail": str(exc)}) + return + audio = self.resampler.process(audio) + if self.segmenter is None: + self.buffer.append(audio) + self.buffered_samples += audio.size + if self.buffered_samples >= self.max_buffer_samples: + await self._enqueue_buffer("max_length") + return + for event in self.segmenter.feed(audio): + await self._handle_vad_event(event) + + async def _handle_vad_event(self, event: core.VadEvent) -> None: + if event.kind == "speech_start": + await self.send({"type": "speech_start", "t": round(event.time_s, 2)}) + else: + await self.jobs.put((event.audio, event.reason, event.time_s)) + + async def _enqueue_buffer(self, reason: str) -> None: + if self.buffer: + audio = np.concatenate(self.buffer) + self.buffer, self.buffered_samples = [], 0 + await self.jobs.put((audio, reason, None)) + + async def _on_control(self, raw: str) -> bool: + try: + msg = json.loads(raw) + except ValueError: + await self.send({"type": "error", "detail": "Ungültiger Steuer-Frame (kein JSON)"}) + return True + mtype = msg.get("type") + if mtype == "start": + self._configure(msg) + await self.send({"type": "ready", "sample_rate": self.sample_rate, "vad": self.use_vad}) + elif mtype in ("end", "close"): + tail = self.resampler.process(np.zeros(0, np.float32), last=True) + if self.segmenter is not None: + events = self.segmenter.feed(tail) + self.segmenter.flush() + for event in events: + await self._handle_vad_event(event) + else: + if tail.size: + self.buffer.append(tail) + await self._enqueue_buffer("flush") + self.resampler = core.StreamResampler(self.sample_rate) + await self.jobs.put("done") + return mtype == "end" + else: + await self.send({"type": "error", "detail": f"Unbekannter Frame-Typ: {mtype!r}"}) + return True + + async def _consume(self) -> None: + """Arbeitet Äußerungen der Reihe nach ab; der Empfang läuft derweil weiter.""" + while True: + job = await self.jobs.get() + if job is None: + return + if job == "done": + await self.send({"type": "done"}) + continue + audio, reason, t = job + try: + tr = await self.worker.transcribe(audio, language=self.language, + initial_prompt=self.prompt) + except Overloaded as exc: + await self.send({"type": "error", "detail": str(exc)}) + continue + except Exception as exc: # Verbindung soll Einzelfehler überleben + log.exception("Transkription im Stream fehlgeschlagen") + await self.send({"type": "error", "detail": str(exc)}) + continue + payload = {"type": "transcript", **tr.to_dict(with_segments=False), "reason": reason} + if t is not None: + payload["t"] = round(t, 2) + await self.send(payload) + + +async def _start_wyoming(worker: TranscriptionWorker, settings: ServiceSettings): + try: + import stt_wyoming + except ImportError as exc: + log.warning("Wyoming deaktiviert (Paket 'wyoming' fehlt): %s", exc) + return None + try: + server = await stt_wyoming.start_server(worker, settings.wyoming_host, settings.wyoming_port) + log.info("Wyoming-Server auf tcp://%s:%s", settings.wyoming_host, settings.wyoming_port) + return server + except OSError as exc: + log.error("Wyoming-Server nicht startbar (%s:%s): %s", + settings.wyoming_host, settings.wyoming_port, exc) + return None + + +logging.basicConfig( + level=os.environ.get("STT_LOG_LEVEL", "INFO").upper(), + format="%(asctime)s %(levelname)s %(name)s: %(message)s", +) +for _noisy in ("httpx", "faster_whisper"): # Modell-Downloads / "Processing audio …" pro Anfrage + logging.getLogger(_noisy).setLevel(logging.WARNING) + +app = create_app() + +if __name__ == "__main__": + import uvicorn + + uvicorn.run(app, host=os.environ.get("STT_HOST", "0.0.0.0"), + port=int(os.environ.get("STT_PORT", "9998"))) diff --git a/stt_wyoming.py b/stt_wyoming.py new file mode 100644 index 0000000..b3836bb --- /dev/null +++ b/stt_wyoming.py @@ -0,0 +1,121 @@ +"""Wyoming-Protokoll-Adapter für Home Assistant. + +Läuft im selben Prozess wie stt_service.py und teilt sich Engine + Worker (ein +Modell im Speicher). In Home Assistant: Einstellungen → Geräte & Dienste → +Integration „Wyoming Protocol“ → Host dieses Rechners, Port 10300. + +Ablauf (HA → Server): transcribe → audio-start → audio-chunk* → audio-stop +Antwort: transcript. Das Äußerungsende bestimmt HA bzw. der Satellit (ESPHome) +selbst (requires_external_vad), wir transkribieren das komplette Audio. +""" + +from __future__ import annotations + +import logging +from functools import partial + +from wyoming.asr import Transcribe, Transcript +from wyoming.audio import AudioChunk, AudioChunkConverter, AudioStart, AudioStop +from wyoming.event import Event +from wyoming.info import AsrModel, AsrProgram, Attribution, Describe, Info +from wyoming.server import AsyncEventHandler, AsyncTcpServer + +import stt_core as core + +log = logging.getLogger("stt.wyoming") + +PROJECT_URL = "https://kitux.de/forgejo/dschlueter/SpeechToText" + + +def _whisper_languages() -> list[str]: + try: + from faster_whisper.tokenizer import _LANGUAGE_CODES + + return sorted(_LANGUAGE_CODES) + except ImportError: + return ["de", "en"] + + +def build_info(engine: core.SttEngine) -> Info: + model_name = engine.active_model or engine.config.model + return Info( + asr=[ + AsrProgram( + name="speech-to-text", + description="Lokaler STT-Dienst (faster-whisper)", + attribution=Attribution(name="dschlueter", url=PROJECT_URL), + installed=True, + version=core.__version__, + models=[ + AsrModel( + name=model_name, + description=f"Whisper {model_name}", + attribution=Attribution( + name="OpenAI / SYSTRAN", url="https://github.com/SYSTRAN/faster-whisper" + ), + installed=True, + version=None, + languages=_whisper_languages(), + ) + ], + ) + ] + ) + + +class SttEventHandler(AsyncEventHandler): + def __init__(self, worker, *args, **kwargs): + super().__init__(*args, **kwargs) + self.worker = worker + self._converter = AudioChunkConverter(rate=core.SAMPLE_RATE, width=2, channels=1) + self._reset() + + def _reset(self) -> None: + self._audio = bytearray() + self._language: str | None = None + self._hotwords: str | None = None + + async def handle_event(self, event: Event) -> bool: + if Describe.is_type(event.type): + await self.write_event(build_info(self.worker.engine).event()) + return True + + if Transcribe.is_type(event.type): + req = Transcribe.from_event(event) + self._language = req.language + terms = (req.transcript_names or []) + (req.transcript_terms or []) + self._hotwords = " ".join(terms) or None + return True + + if AudioStart.is_type(event.type): + self._audio = bytearray() + self._converter = AudioChunkConverter(rate=core.SAMPLE_RATE, width=2, channels=1) + return True + + if AudioChunk.is_type(event.type): + chunk = self._converter.convert(AudioChunk.from_event(event)) + self._audio.extend(chunk.audio) + return True + + if AudioStop.is_type(event.type): + audio = core.pcm16_to_float(bytes(self._audio)) + text, language = "", self._language + try: + tr = await self.worker.transcribe(audio, language=self._language, + hotwords=self._hotwords) + text, language = tr.text, tr.language + log.info("Wyoming-Transkript (%.1fs Audio, %.2fs): %s", + tr.duration, tr.processing_s, text) + except Exception as exc: # HA erwartet immer eine Antwort + log.error("Wyoming-Transkription fehlgeschlagen: %s", exc) + await self.write_event(Transcript(text=text, language=language).event()) + self._reset() + return False # HA öffnet pro Anfrage eine neue Verbindung + + return True + + +async def start_server(worker, host: str, port: int) -> AsyncTcpServer: + server = AsyncTcpServer(host, port) + await server.start(partial(SttEventHandler, worker)) + return server diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..6951eff --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,92 @@ +import io +import os +import shutil +import subprocess +import sys +import wave +from pathlib import Path + +import numpy as np +import pytest + +ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT)) +os.environ.setdefault("STT_WYOMING_PORT", "") # Modul-App in Tests ohne TCP-Server + +import stt_core as core # noqa: E402 + +SR = core.SAMPLE_RATE + + +def make_wav(audio: np.ndarray, rate: int = SR, channels: int = 1, width: int = 2) -> bytes: + """float32 (mono oder [n, channels]) → WAV-Bytes.""" + buf = io.BytesIO() + with wave.open(buf, "wb") as w: + w.setnchannels(channels) + w.setsampwidth(width) + w.setframerate(rate) + if width == 2: + w.writeframes((np.clip(audio, -1, 1) * 32767).astype(" np.ndarray: + t = np.arange(int(seconds * rate)) / rate + return (amp * np.sin(2 * np.pi * freq * t)).astype(np.float32) + + +def silence(seconds: float, rate: int = SR) -> np.ndarray: + return np.zeros(int(seconds * rate), dtype=np.float32) + + +class EnergyProb: + """Deterministische Ersatz-VAD für Tests: Fenster mit Energie = Sprache.""" + + def __call__(self, windows: np.ndarray) -> np.ndarray: + frames = windows.reshape(-1, core.VAD_WINDOW) + rms = np.sqrt((frames ** 2).mean(axis=1)) + return (rms > 0.05).astype(np.float32) + + +class FakeEngine(core.SttEngine): + """Engine ohne Modell: 'transkribiert' die Audiodauer.""" + + def __init__(self, delay: float = 0.0): + super().__init__(core.EngineConfig(model="fake")) + self.delay = delay + self.calls: list[dict] = [] + self.status = "ready" + self.active_model, self.active_device = "fake", "cpu" + + def load(self): + return None + + def warmup(self): + return None + + def transcribe(self, audio, **kwargs): + import time + + if self.delay: + time.sleep(self.delay) + self.calls.append({"samples": int(audio.size), **kwargs}) + duration = round(audio.size / SR, 2) + return core.Transcript( + text=f"audio {duration:.1f}s", language=self.resolve_language(kwargs.get("language")) or "de", + language_probability=1.0, duration=duration, processing_s=0.0, + segments=[core.Segment(0.0, duration, f"audio {duration:.1f}s")], + ) + + +def espeak_wav(text: str, voice: str = "de") -> bytes | None: + if not shutil.which("espeak-ng"): + return None + return subprocess.run(["espeak-ng", "-v", voice, "-s", "140", "--stdout", text], + check=True, capture_output=True).stdout + + +@pytest.fixture +def fake_engine(): + return FakeEngine() diff --git a/tests/test_core.py b/tests/test_core.py new file mode 100644 index 0000000..a743859 --- /dev/null +++ b/tests/test_core.py @@ -0,0 +1,165 @@ +import numpy as np +import pytest + +import stt_core as core +from conftest import SR, EnergyProb, make_wav, silence, tone + + +# --- Audio-Hilfen ------------------------------------------------------------- + +def test_pcm16_to_float_mono_and_channel_selection(): + stereo = np.array([[1000, -2000], [3000, -4000]], dtype=" 0.9 * SR + assert not seg.in_speech and seg.flush() == [] + + +def test_silero_prob_streaming_on_silence_and_noise(): + prob = core.SileroProb() + out = prob(np.zeros(core.VAD_WINDOW * 10, np.float32)) + assert out.shape == (10,) and float(out.max()) < 0.2 diff --git a/tests/test_integration.py b/tests/test_integration.py new file mode 100644 index 0000000..a32c7f6 --- /dev/null +++ b/tests/test_integration.py @@ -0,0 +1,62 @@ +"""Echte Transkription mit Whisper (pytest -m integration). + +Nutzt espeak-ng als reproduzierbare Sprachquelle und standardmäßig das kleine, +lokal gecachte Modell 'small' auf CPU (STT_TEST_MODEL / STT_TEST_DEVICE überschreibbar). +""" + +import os + +import numpy as np +import pytest + +import stt_core as core +from conftest import espeak_wav, silence + +pytestmark = pytest.mark.integration + + +@pytest.fixture(scope="module") +def engine(): + cfg = core.EngineConfig(model=os.environ.get("STT_TEST_MODEL", "small"), + device=os.environ.get("STT_TEST_DEVICE", "cpu"), language="de") + eng = core.SttEngine(cfg) + eng.load() + return eng + + +def _speech(text, voice="de"): + wav = espeak_wav(text, voice) + if wav is None: + pytest.skip("espeak-ng fehlt") + return core.decode_audio_bytes(wav) + + +def _words(text): + return set(text.lower().replace(".", " ").replace(",", " ").split()) + + +def test_german_command(engine): + tr = engine.transcribe(_speech("Schalte das Licht im Wohnzimmer ein.")) + assert {"licht", "wohnzimmer"} <= _words(tr.text), tr.text + + +def test_silence_gives_empty_text(engine): + noise = (np.random.default_rng(1).standard_normal(3 * core.SAMPLE_RATE) * 0.003).astype(np.float32) + assert engine.transcribe(noise).text == "" + + +def test_language_detection(engine): + tr = engine.transcribe(_speech("The weather is nice today.", "en"), language="auto") + assert tr.language == "en", tr + + +def test_segmenter_with_silero_on_real_speech(engine): + speech = _speech("Schalte das Licht aus.") + stream = np.concatenate([silence(0.5), speech, silence(1.5), speech, silence(1.5)]) + seg = core.UtteranceSegmenter() + events = [] + for i in range(0, stream.size, 1600): + events += seg.feed(stream[i:i + 1600]) + utterances = [e for e in events if e.kind == "utterance"] + assert len(utterances) == 2, [(e.kind, e.time_s) for e in events] + assert "licht" in _words(engine.transcribe(utterances[0].audio).text) diff --git a/tests/test_service.py b/tests/test_service.py new file mode 100644 index 0000000..37bc9a4 --- /dev/null +++ b/tests/test_service.py @@ -0,0 +1,153 @@ +import asyncio +import json + +import numpy as np +import pytest +from fastapi.testclient import TestClient + +import stt_core as core +import stt_service as service +from conftest import SR, EnergyProb, FakeEngine, make_wav, silence, tone + + +def _client(engine=None, **settings): + settings = service.ServiceSettings(wyoming_port=None, **settings) + app = service.create_app(engine or FakeEngine(), settings, + core.VadConfig(silence_ms=400), prob_fn_factory=EnergyProb) + return TestClient(app) + + +def test_health(): + with _client() as client: + data = client.get("/health").json() + assert data["status"] == "ok" and data["model"] == "fake" and data["queue"] == 0 + + +def test_openai_transcriptions_formats(): + wav = make_wav(tone(1.5)) + with _client() as client: + files = {"file": ("a.wav", wav, "audio/wav")} + assert client.post("/v1/audio/transcriptions", files=files).json() == {"text": "audio 1.5s"} + r = client.post("/v1/audio/transcriptions", files=files, data={"response_format": "text"}) + assert r.text == "audio 1.5s" + r = client.post("/v1/audio/transcriptions", files=files, + data={"response_format": "verbose_json", "language": "en"}) + body = r.json() + assert body["language"] == "en" and body["segments"][0]["text"] == "audio 1.5s" + r = client.post("/v1/audio/transcriptions", files=files, data={"response_format": "srt"}) + assert r.status_code == 400 + + +def test_transcribe_raw_pcm_with_resampling_and_channel(): + engine = FakeEngine() + stereo = np.stack([tone(1.0, rate=48000), silence(1.0, rate=48000)], axis=1) + pcm = (stereo * 32767).astype("