104 lines
3.3 KiB
Python
104 lines
3.3 KiB
Python
"""stt.transcribe() — the "guaranteed" one-shot fallback.
|
|
|
|
There is no separate REST endpoint server-side any more: /desk/stt is a
|
|
websocket relay only, so this connects the exact same way stt_stream.py's
|
|
opportunistic streaming path does (via stt_stream.connect(), unconditionally
|
|
— not gated by STT_STREAMING, since there's nothing left to fall back to).
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from bolt_pet.audio import stt, stt_stream
|
|
|
|
|
|
class FakeSocket:
|
|
def __init__(self, messages=(), fail_to_connect=False):
|
|
self.sent = []
|
|
self.closed = False
|
|
self._fail_to_connect = fail_to_connect
|
|
self._messages = list(messages)
|
|
|
|
def send_binary(self, data):
|
|
self.sent.append(data)
|
|
|
|
def send(self, text):
|
|
self.sent.append(text)
|
|
|
|
def recv(self):
|
|
if self._messages:
|
|
return self._messages.pop(0)
|
|
time.sleep(0.01)
|
|
raise ConnectionError("closed")
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
|
|
def _results(transcript, is_final=True):
|
|
return json.dumps({
|
|
"type": "Results", "is_final": is_final,
|
|
"channel": {"alternatives": [{"transcript": transcript}]},
|
|
})
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _configured(monkeypatch):
|
|
monkeypatch.setattr(stt.config, "SERVER_URL", "http://test-server:5002")
|
|
monkeypatch.setattr(stt.config, "API_KEY", "test-key")
|
|
|
|
|
|
def test_transcribe_raises_when_not_configured(monkeypatch):
|
|
monkeypatch.setattr(stt.config, "API_KEY", "")
|
|
with pytest.raises(stt.SttError, match="BOLT_SERVER_URL"):
|
|
stt.transcribe(np.zeros(320, dtype=np.int16))
|
|
|
|
|
|
def test_transcribe_connects_and_feeds_the_whole_utterance(monkeypatch):
|
|
socket = FakeSocket([_results("turn on the lights")])
|
|
monkeypatch.setattr(stt_stream, "connect", lambda: socket)
|
|
|
|
pcm = np.full(3200, 500, dtype=np.int16)
|
|
text = stt.transcribe(pcm)
|
|
|
|
assert text == "turn on the lights"
|
|
# one binary frame (the whole utterance) plus the close message
|
|
assert len(socket.sent) == 2
|
|
assert socket.closed
|
|
|
|
|
|
def test_transcribe_raises_when_the_server_is_unreachable(monkeypatch):
|
|
def refuse():
|
|
raise OSError("no route to host")
|
|
|
|
monkeypatch.setattr(stt_stream, "connect", refuse)
|
|
|
|
with pytest.raises(stt.SttError, match="couldn't reach"):
|
|
stt.transcribe(np.zeros(320, dtype=np.int16))
|
|
|
|
|
|
def test_transcribe_returns_empty_string_for_silence_not_an_error(monkeypatch):
|
|
"""No speech recognized is a legitimate outcome, not a failure — callers
|
|
(controller.py) treat "" as "say nothing" rather than logging an error."""
|
|
socket = FakeSocket([]) # never says anything back
|
|
monkeypatch.setattr(stt_stream, "connect", lambda: socket)
|
|
|
|
assert stt.transcribe(np.zeros(320, dtype=np.int16)) == ""
|
|
|
|
|
|
def test_transcribe_is_unaffected_by_stt_streaming_being_off(monkeypatch):
|
|
"""The opportunistic accelerator and the guaranteed fallback share a
|
|
connector, but STT_STREAMING must only gate the former."""
|
|
monkeypatch.setattr(stt.config, "STT_STREAMING", False)
|
|
socket = FakeSocket([_results("still works")])
|
|
monkeypatch.setattr(stt_stream, "connect", lambda: socket)
|
|
|
|
assert stt.transcribe(np.zeros(320, dtype=np.int16)) == "still works"
|