dschlueter
a986cbe42a
- 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>
92 lines
2.8 KiB
Python
92 lines
2.8 KiB
Python
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("<i2").tobytes())
|
|
elif width == 4:
|
|
w.writeframes((np.clip(audio, -1, 1) * 2147483647).astype("<i4").tobytes())
|
|
return buf.getvalue()
|
|
|
|
|
|
def tone(seconds: float, freq: float = 220.0, amp: float = 0.5, rate: int = SR) -> 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()
|