ElebobaandClaude Opus 5.5
d700608825
CI / ruff · pytest (pull_request) Successful in 2m23s
Live check on synthetic speech (espeak) via OpenRouter gemini-2.5-flash: 3–5 s per clip, silence → no speech. Refs #116 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
108 lines
3.4 KiB
Python
108 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
|
|
assert body["reasoning"] == {"enabled": False}
|
|
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
|