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>
62 lines
2.3 KiB
Python
62 lines
2.3 KiB
Python
import asyncio
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
pytest.importorskip("wyoming")
|
|
|
|
from wyoming.asr import Transcribe, Transcript # noqa: E402
|
|
from wyoming.audio import AudioChunk, AudioStart, AudioStop # noqa: E402
|
|
from wyoming.client import AsyncTcpClient # noqa: E402
|
|
from wyoming.info import Describe, Info # noqa: E402
|
|
|
|
import stt_service as service # noqa: E402
|
|
import stt_wyoming # noqa: E402
|
|
from conftest import FakeEngine, tone # noqa: E402
|
|
|
|
|
|
async def _with_server(scenario):
|
|
engine = FakeEngine()
|
|
worker = service.TranscriptionWorker(engine)
|
|
server = await stt_wyoming.start_server(worker, "127.0.0.1", 0)
|
|
port = server._server.sockets[0].getsockname()[1]
|
|
try:
|
|
return await scenario(port), engine
|
|
finally:
|
|
await server.stop()
|
|
worker.shutdown()
|
|
|
|
|
|
def test_describe_returns_asr_info():
|
|
async def scenario(port):
|
|
async with AsyncTcpClient("127.0.0.1", port) as client:
|
|
await client.write_event(Describe().event())
|
|
return Info.from_event(await client.read_event())
|
|
|
|
info, _ = asyncio.run(_with_server(scenario))
|
|
assert info.asr[0].models[0].name == "fake"
|
|
assert "de" in info.asr[0].models[0].languages
|
|
|
|
|
|
def test_transcribe_converts_audio_format():
|
|
"""HA schickt 16 kHz mono; hier absichtlich 22,05 kHz stereo → Konvertierung."""
|
|
rate = 22050
|
|
left = tone(1.0, rate=rate)
|
|
stereo = np.stack([left, left], axis=1).reshape(-1)
|
|
pcm = (stereo * 32767).astype("<i2").tobytes()
|
|
|
|
async def scenario(port):
|
|
async with AsyncTcpClient("127.0.0.1", port) as client:
|
|
await client.write_event(Transcribe(language="de-DE", transcript_names=["Wohnzimmer"]).event())
|
|
await client.write_event(AudioStart(rate=rate, width=2, channels=2).event())
|
|
step = 4410 * 2
|
|
for i in range(0, len(pcm), step):
|
|
await client.write_event(AudioChunk(rate=rate, width=2, channels=2, audio=pcm[i:i + step]).event())
|
|
await client.write_event(AudioStop().event())
|
|
return Transcript.from_event(await client.read_event())
|
|
|
|
transcript, engine = asyncio.run(_with_server(scenario))
|
|
assert transcript.text == "audio 1.0s" and transcript.language == "de"
|
|
call = engine.calls[0]
|
|
assert call["hotwords"] == "Wohnzimmer" and call["language"] == "de-DE"
|
|
assert abs(call["samples"] - 16000) < 50
|