cca423dac3
PiperTTSProvider ruft das piper-Binary (--output-raw) async auf, liest die native Sample-Rate aus der .onnx.json und resampelt per ffmpeg auf 24000 Hz (Gateway-Norm). Nicht passende Stimmen (z. B. Cloud-Stimme 'Zephyr' aus der Route) fallen auf die konfigurierte Default-Stimme zurueck. Damit ist eine voll-lokale Konstellation (faster-whisper + Ollama + piper) moeglich -> keine API-Kosten, max. Datenschutz. - config: PIPER_BIN/PIPER_VOICES_DIR/PIPER_VOICE/TTS_SAMPLE_RATE (+ .env.example) - dependencies: piper-Factory mit Settings verdrahtet - tests: tests/test_piper_tts.py (offline, Fake-Binary; Resample-Test skippt ohne ffmpeg); e2e/auth-Tests nutzen jetzt einen Stub-TTS statt 'piper' als Pseudo-Stub - docs: README, BEDIENUNGSANLEITUNG (voll-lokal-Beispiel), Architektur-Roadmap Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
76 lines
2.8 KiB
Python
76 lines
2.8 KiB
Python
"""Tests fuer den lokalen Piper-TTS-Provider (offline, mit Fake-Binary)."""
|
|
|
|
import asyncio
|
|
import json
|
|
import shutil
|
|
import stat
|
|
|
|
import pytest
|
|
|
|
from app.providers.tts.piper import PiperTTSProvider
|
|
|
|
# PCM-Nutzlast des Fake-Binaries: gross genug, dass der ffmpeg-Resampler Samples ausgibt
|
|
# (bei sehr kurzen Eingaben puffert er und liefert nichts). 4000 Bytes = 2000 s16le-Samples.
|
|
FAKE_PCM = b"\x01\x02" * 2000
|
|
|
|
|
|
def _run(coro):
|
|
return asyncio.run(coro)
|
|
|
|
|
|
def _make_fake_voice(tmp_path, sample_rate):
|
|
"""Legt ein Fake-piper-Binary + Stimmmodell (.onnx/.onnx.json) an."""
|
|
voices = tmp_path / "voices"
|
|
voices.mkdir()
|
|
(voices / "de_DE-test.onnx").write_bytes(b"fake-model")
|
|
(voices / "de_DE-test.onnx.json").write_text(
|
|
json.dumps({"audio": {"sample_rate": sample_rate}})
|
|
)
|
|
|
|
# Fake-Binary: ignoriert Argumente, schreibt feste Roh-PCM-Bytes nach stdout.
|
|
binp = tmp_path / "piper"
|
|
binp.write_text(
|
|
"#!/usr/bin/env python3\n"
|
|
"import sys\n"
|
|
f"sys.stdout.buffer.write({FAKE_PCM!r})\n"
|
|
)
|
|
binp.chmod(binp.stat().st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH)
|
|
return str(binp), str(voices)
|
|
|
|
|
|
def test_piper_returns_pcm_without_resampling(tmp_path):
|
|
# Native Rate == Ziel-Rate -> kein ffmpeg noetig, Bytes unveraendert.
|
|
binp, voices = _make_fake_voice(tmp_path, sample_rate=24000)
|
|
provider = PiperTTSProvider(binp, voices, "de_DE-test", target_rate=24000)
|
|
pcm = _run(provider.synthesize("Hallo Welt"))
|
|
assert pcm == FAKE_PCM
|
|
|
|
|
|
def test_piper_wraps_wav_on_request(tmp_path):
|
|
binp, voices = _make_fake_voice(tmp_path, sample_rate=24000)
|
|
provider = PiperTTSProvider(binp, voices, "de_DE-test", target_rate=24000)
|
|
wav = _run(provider.synthesize("Hallo", audio_format="wav"))
|
|
assert wav.startswith(b"RIFF") and b"WAVE" in wav[:16]
|
|
|
|
|
|
def test_piper_rejects_empty_text(tmp_path):
|
|
binp, voices = _make_fake_voice(tmp_path, sample_rate=24000)
|
|
provider = PiperTTSProvider(binp, voices, "de_DE-test")
|
|
with pytest.raises(ValueError):
|
|
_run(provider.synthesize(" "))
|
|
|
|
|
|
def test_piper_unknown_voice_raises(tmp_path):
|
|
binp, voices = _make_fake_voice(tmp_path, sample_rate=24000)
|
|
provider = PiperTTSProvider(binp, voices, "gibt-es-nicht")
|
|
with pytest.raises(RuntimeError):
|
|
_run(provider.synthesize("Hallo"))
|
|
|
|
|
|
@pytest.mark.skipif(not shutil.which("ffmpeg"), reason="ffmpeg nicht installiert")
|
|
def test_piper_resamples_when_rates_differ(tmp_path):
|
|
# Native 16000 -> Ziel 24000: ffmpeg laeuft, Ergebnis ist gueltiges (nicht leeres) PCM.
|
|
binp, voices = _make_fake_voice(tmp_path, sample_rate=16000)
|
|
provider = PiperTTSProvider(binp, voices, "de_DE-test", target_rate=24000)
|
|
pcm = _run(provider.synthesize("Hallo"))
|
|
assert isinstance(pcm, bytes) and len(pcm) > 0
|