feat(tts): Chatterbox-Provider (hohe Qualitaet + Voice-Cloning) anbinden
Loest #7. Der chatterbox-Stub wird durch eine echte Anbindung an den lokalen Chatterbox-HTTP-Dienst ersetzt (POST /speak -> /status pollen -> GET /audio, WAV). no_playback=true -> der Dienst spielt nicht lokal ab, liefert nur Bytes. WAV->PCM (24 kHz, Resampling bei Bedarf). Waehlbar via tts_provider=chatterbox; piper bleibt der schnelle Default (chatterbox ist ~echtzeit-langsam, dafuer klonbare Stimme). - config: CHATTERBOX_BASE_URL/_VOICE/_LANG/_SPEED/_TIMEOUT; Registry verdrahtet - Tests: gemockter httpx (Synthese, WAV/Resample, Job-Fehler, Stimmenwahl) - Doku: README, Architektur (#7), deploy/README (GPU-Pinning per UUID, no_playback) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
parent
77026bff1b
commit
5f01607b79
7 changed files with 223 additions and 7 deletions
82
tests/test_chatterbox_tts.py
Normal file
82
tests/test_chatterbox_tts.py
Normal file
|
|
@ -0,0 +1,82 @@
|
|||
import io
|
||||
import wave
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
import app.providers.tts.chatterbox as cb
|
||||
from app.providers.tts.chatterbox import ChatterboxTTSProvider
|
||||
|
||||
|
||||
def _wav(rate=24000, payload=b"\x01\x02" * 2400) -> bytes:
|
||||
buf = io.BytesIO()
|
||||
with wave.open(buf, "wb") as w:
|
||||
w.setnchannels(1)
|
||||
w.setsampwidth(2)
|
||||
w.setframerate(rate)
|
||||
w.writeframes(payload)
|
||||
return buf.getvalue()
|
||||
|
||||
|
||||
def _install(monkeypatch, handler):
|
||||
real = httpx.AsyncClient # echte Klasse sichern (vor dem Patch)
|
||||
|
||||
def make_client(*args, **kwargs):
|
||||
kwargs.pop("timeout", None)
|
||||
return real(transport=httpx.MockTransport(handler))
|
||||
monkeypatch.setattr(cb.httpx, "AsyncClient", make_client)
|
||||
|
||||
|
||||
def _ok_handler(status="done", wav_rate=24000):
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
p = request.url.path
|
||||
if request.method == "POST" and p == "/speak":
|
||||
return httpx.Response(200, json={"job_id": "j1", "status": "pending"})
|
||||
if p == "/status":
|
||||
return httpx.Response(200, json={"recent_jobs": [{"id": "j1", "status": status,
|
||||
"error": "boom" if status == "error" else None}]})
|
||||
if p == "/audio/j1":
|
||||
return httpx.Response(200, content=_wav(wav_rate), headers={"content-type": "audio/wav"})
|
||||
return httpx.Response(404)
|
||||
return handler
|
||||
|
||||
|
||||
def test_synthesize_returns_pcm(monkeypatch):
|
||||
_install(monkeypatch, _ok_handler())
|
||||
p = ChatterboxTTSProvider("http://x:9999", voice="/ref.wav")
|
||||
pcm = __import__("asyncio").run(p.synthesize("Hallo Welt"))
|
||||
assert pcm == b"\x01\x02" * 2400 # 24000 Hz -> kein Resampling, Frames unveraendert
|
||||
|
||||
|
||||
def test_synthesize_wav_format(monkeypatch):
|
||||
_install(monkeypatch, _ok_handler())
|
||||
p = ChatterboxTTSProvider("http://x:9999")
|
||||
wav = __import__("asyncio").run(p.synthesize("Hallo", audio_format="wav"))
|
||||
assert wav.startswith(b"RIFF") and b"WAVE" in wav[:16]
|
||||
|
||||
|
||||
def test_resample_when_rate_differs(monkeypatch):
|
||||
_install(monkeypatch, _ok_handler(wav_rate=22050))
|
||||
p = ChatterboxTTSProvider("http://x:9999", target_rate=24000)
|
||||
pcm = __import__("asyncio").run(p.synthesize("Hallo"))
|
||||
assert isinstance(pcm, bytes) and len(pcm) > 0
|
||||
|
||||
|
||||
def test_job_error_raises(monkeypatch):
|
||||
_install(monkeypatch, _ok_handler(status="error"))
|
||||
p = ChatterboxTTSProvider("http://x:9999")
|
||||
with pytest.raises(RuntimeError):
|
||||
__import__("asyncio").run(p.synthesize("Hallo"))
|
||||
|
||||
|
||||
def test_empty_text_raises(monkeypatch):
|
||||
p = ChatterboxTTSProvider("http://x:9999")
|
||||
with pytest.raises(ValueError):
|
||||
__import__("asyncio").run(p.synthesize(" "))
|
||||
|
||||
|
||||
def test_ref_voice_selection():
|
||||
p = ChatterboxTTSProvider("http://x:9999", voice="/default.wav")
|
||||
assert p._ref_voice(None) == "/default.wav"
|
||||
assert p._ref_voice("Zephyr") == "/default.wav" # Cloud-Name -> Default
|
||||
assert p._ref_voice("/custom.wav") == "/custom.wav" # Pfad -> uebernommen
|
||||
Loading…
Add table
Add a link
Reference in a new issue