ElebobaandClaude Opus 5.5
7443b61974
CI / ruff · pytest (pull_request) Successful in 2m57s
POST /api/v1/me/{token}/stt takes a 16 kHz mono WAV and only transcribes:
the patient checks the text and sends it as an ordinary message, so safety,
scrub() and the LLM gateway still apply. Audio is neither stored nor logged.
OpenRouter STT is synthetic-data only (ADR-014); SaluteSpeech for the pilot is #121.
Refs #116
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
107 lines
3.4 KiB
Python
107 lines
3.4 KiB
Python
"""Speech-to-text providers (#116): OpenRouter request shape, retries, silence, WAV parsing, provider choice."""
|
|
|
|
import base64
|
|
import io
|
|
import json
|
|
import wave
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.config import Settings
|
|
from app.llm.stt import (
|
|
InvalidAudio,
|
|
MockSTT,
|
|
OpenRouterSTT,
|
|
SttError,
|
|
SttNoSpeech,
|
|
get_stt,
|
|
load_texts,
|
|
wav_seconds,
|
|
)
|
|
|
|
|
|
def make_wav(seconds: float, rate: int = 16_000) -> bytes:
|
|
buf = io.BytesIO()
|
|
with wave.open(buf, "wb") as w:
|
|
w.setnchannels(1)
|
|
w.setsampwidth(2)
|
|
w.setframerate(rate)
|
|
w.writeframes(b"\x00\x00" * int(rate * seconds))
|
|
return buf.getvalue()
|
|
|
|
|
|
def _settings(**kw) -> Settings:
|
|
base = {"openrouter_api_key": "test", "stt_provider": "openrouter", "stt_timeout_s": 5.0}
|
|
return Settings(_env_file=None, **{**base, **kw})
|
|
|
|
|
|
def _fake(responses: list[httpx.Response]) -> tuple[OpenRouterSTT, list[dict]]:
|
|
sent: list[dict] = []
|
|
replies = iter(responses)
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
sent.append(json.loads(request.content))
|
|
return next(replies)
|
|
|
|
stt = OpenRouterSTT(_settings(), client=httpx.AsyncClient(transport=httpx.MockTransport(handler)))
|
|
return stt, sent
|
|
|
|
|
|
def _ok(text: str) -> httpx.Response:
|
|
return httpx.Response(200, json={"choices": [{"message": {"content": text}}]})
|
|
|
|
|
|
def test_wav_seconds() -> None:
|
|
assert wav_seconds(make_wav(2.5)) == pytest.approx(2.5)
|
|
with pytest.raises(InvalidAudio):
|
|
wav_seconds(b"OggS" + b"\x00" * 100)
|
|
with pytest.raises(InvalidAudio):
|
|
wav_seconds(b"")
|
|
|
|
|
|
async def test_openrouter_sends_audio_and_returns_transcript() -> None:
|
|
audio = make_wav(1)
|
|
stt, sent = _fake([_ok("«Сегодня плохо спал.»")])
|
|
assert await stt.transcribe(audio) == "Сегодня плохо спал."
|
|
body = sent[0]
|
|
assert body["model"] == "google/gemini-2.5-flash" and body["temperature"] == 0
|
|
part = body["messages"][1]["content"][1]
|
|
assert part["type"] == "input_audio" and part["input_audio"]["format"] == "wav"
|
|
assert base64.b64decode(part["input_audio"]["data"]) == audio
|
|
|
|
|
|
async def test_silence_marker_is_no_speech() -> None:
|
|
stt, _ = _fake([_ok(load_texts().silence_marker.upper())])
|
|
with pytest.raises(SttNoSpeech):
|
|
await stt.transcribe(make_wav(1))
|
|
|
|
|
|
async def test_retries_then_succeeds(monkeypatch) -> None:
|
|
monkeypatch.setattr("app.llm.stt.asyncio.sleep", _no_sleep)
|
|
stt, sent = _fake([httpx.Response(503), _ok("текст")])
|
|
assert await stt.transcribe(make_wav(1)) == "текст"
|
|
assert len(sent) == 2
|
|
|
|
|
|
async def test_http_error_raises(monkeypatch) -> None:
|
|
monkeypatch.setattr("app.llm.stt.asyncio.sleep", _no_sleep)
|
|
stt, _ = _fake([httpx.Response(401, json={"error": {"message": "bad key"}})])
|
|
with pytest.raises(SttError):
|
|
await stt.transcribe(make_wav(1))
|
|
|
|
|
|
async def _no_sleep(_: float) -> None:
|
|
return None
|
|
|
|
|
|
async def test_mock_is_offline() -> None:
|
|
assert await MockSTT().transcribe(b"") == load_texts().mock_transcript
|
|
|
|
|
|
def test_provider_choice() -> None:
|
|
assert isinstance(get_stt(_settings(stt_provider="auto")), OpenRouterSTT)
|
|
assert isinstance(get_stt(_settings(stt_provider="auto", openrouter_api_key="")), MockSTT)
|
|
assert get_stt(_settings(stt_provider="openrouter", openrouter_api_key="")) is None
|
|
assert get_stt(_settings(stt_provider="off")) is None
|