63fcfb7ba2
Step 3 (귀): add WhisperSTT + whisper_worker, a warm out-of-venv worker mirroring the MeloTTS shape (whisper312 venv, small/int8 on CPU). transcribe() closes the voice round trip (MeloTTS wav -> whisper text); utterances() turns an injected audio_source into Utterances (Discord voice feed pending). Wired into factory as WSAI_STT=whisper. Also address the arbiter's TTS follow-ups: - melo worker error handling: capture stderr (drained in a bounded background task so the pipe can't fill), surface the real failure cause, and defend against an empty/invalid ready line instead of dying on JSONDecodeError. - pipeline pre-warm: load slow backends (warmup()) at startup so the first utterance is answered warm; a warmup failure is logged, not fatal. Verified: real TTS->STT round trip recovers the sentence near-perfectly; warm transcribe ~1.2s (CPU). 12 tests pass. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
61 lines
1.8 KiB
Python
61 lines
1.8 KiB
Python
"""Unit tests for the WhisperSTT source plumbing.
|
|
|
|
These do NOT load a model or spawn the worker (that needs the whisper312 venv and
|
|
is exercised by the manual TTS->STT round trip). They pin the SpeechToText
|
|
contract: how `utterances()` turns an audio source into Utterances, and how it
|
|
behaves with no source wired yet.
|
|
"""
|
|
|
|
import asyncio
|
|
from typing import AsyncIterator
|
|
|
|
from wsai.backends.whisper import WhisperSTT
|
|
from wsai.interfaces import SpeechToText, Utterance
|
|
|
|
|
|
def _collect(stt: WhisperSTT) -> list[Utterance]:
|
|
async def run():
|
|
return [u async for u in stt.utterances()]
|
|
|
|
return asyncio.run(asyncio.wait_for(run(), timeout=5))
|
|
|
|
|
|
async def _paths(items) -> AsyncIterator[str]:
|
|
for it in items:
|
|
yield it
|
|
|
|
|
|
def test_is_speech_to_text():
|
|
assert isinstance(WhisperSTT(), SpeechToText)
|
|
|
|
|
|
def test_no_audio_source_yields_nothing():
|
|
# Discord voice receiver not wired yet -> the loop idles and returns.
|
|
assert _collect(WhisperSTT(audio_source=None)) == []
|
|
|
|
|
|
def test_utterances_transcribes_each_chunk(monkeypatch):
|
|
stt = WhisperSTT(audio_source=_paths(["a.wav", "b.wav"]))
|
|
|
|
async def fake_transcribe(wav_path, *, language=None):
|
|
return f"text::{wav_path}"
|
|
|
|
monkeypatch.setattr(stt, "transcribe", fake_transcribe)
|
|
|
|
utts = _collect(stt)
|
|
assert [u.text for u in utts] == ["text::a.wav", "text::b.wav"]
|
|
assert all(u.source == "voice" for u in utts)
|
|
|
|
|
|
def test_empty_transcript_is_skipped(monkeypatch):
|
|
# Silence / VAD-filtered audio transcribes to "" and must not become a turn.
|
|
stt = WhisperSTT(audio_source=_paths(["silent.wav", "real.wav"]))
|
|
|
|
async def fake_transcribe(wav_path, *, language=None):
|
|
return "" if wav_path == "silent.wav" else "안녕"
|
|
|
|
monkeypatch.setattr(stt, "transcribe", fake_transcribe)
|
|
|
|
utts = _collect(stt)
|
|
assert [u.text for u in utts] == ["안녕"]
|