- 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 <noreply@anthropic.com>
62 lines
1.9 KiB
Python
62 lines
1.9 KiB
Python
"""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)
|