feat(tts): echtes lokales TTS via piper (kein Stub mehr)

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>

tests/test_auth.py

@@ -1,5 +1,6 @@
 from fastapi.testclient import TestClient
 
+import app.dependencies as deps
 from app.main import app
 from app.config import settings
 
@@ -67,17 +68,23 @@ def test_tenant_isolation_returns_403(monkeypatch):
 
 def test_user_prefs_applied_to_route(monkeypatch):
     _enable_auth(monkeypatch)
+
+    class StubTTS:
+        async def synthesize(self, text, voice=None, audio_format="pcm"):
+            return b""
+    monkeypatch.setitem(deps.TTS_REGISTRY, "stub-tts", lambda s: StubTTS())
+
     token = _create_user("Pref")
     auth = {"Authorization": f"Bearer {token}"}
 
     client.put(
         "/api/me/prefs",
         headers=auth,
-        json={"tts_provider": "piper", "output_endpoint": "loopback"},
+        json={"tts_provider": "stub-tts", "output_endpoint": "loopback"},
     )
     resp = client.post("/api/speak", headers=auth, json={"text": "hallo"})
     assert resp.status_code == 200
-    assert resp.headers["X-TTS-Provider"] == "piper"
+    assert resp.headers["X-TTS-Provider"] == "stub-tts"
     assert resp.headers["X-Output-Endpoint"] == "loopback"
 
 

tests/test_endpoints_e2e.py

@@ -9,15 +9,25 @@ from tests.conftest import loopback_output
 client = TestClient(app)
 
 
-def test_speak_loopback_collects_chunks():
-    # piper-Stub liefert b"" -> kein Netzcall; Loopback sammelt den Chunk.
+def _stub_tts(monkeypatch, name="stub-tts", audio=b""):
+    """Registriert einen deterministischen TTS-Stub (kein Netz, kein lokales Binary)."""
+    class StubTTS:
+        async def synthesize(self, text, voice=None, audio_format="pcm"):
+            return audio
+    monkeypatch.setitem(deps.TTS_REGISTRY, name, lambda s: StubTTS())
+    return name
+
+
+def test_speak_loopback_collects_chunks(monkeypatch):
+    # Stub-TTS liefert b"" -> kein Netzcall; Loopback sammelt den Chunk.
+    tts = _stub_tts(monkeypatch)
     resp = client.post(
         "/api/speak",
-        json={"text": "Hallo Welt", "tts_provider": "piper", "output_endpoint": "loopback"},
+        json={"text": "Hallo Welt", "tts_provider": tts, "output_endpoint": "loopback"},
     )
     assert resp.status_code == 200
     assert resp.headers["X-Output-Endpoint"] == "loopback"
-    assert resp.headers["X-TTS-Provider"] == "piper"
+    assert resp.headers["X-TTS-Provider"] == tts
     assert len(loopback_output().chunks) == 1
 
 
@@ -35,15 +45,16 @@ def test_unknown_provider_returns_422():
     assert resp.status_code == 422
 
 
-def test_session_route_applies():
+def test_session_route_applies(monkeypatch):
+    tts = _stub_tts(monkeypatch)
     client.post(
         "/api/sessions/s1/route",
-        json={"tts_provider": "piper", "output_endpoint": "loopback"},
+        json={"tts_provider": tts, "output_endpoint": "loopback"},
     )
     resp = client.post("/api/speak?session_id=s1", json={"text": "hallo"})
     assert resp.status_code == 200
     assert resp.headers["X-Output-Endpoint"] == "loopback"
-    assert resp.headers["X-TTS-Provider"] == "piper"
+    assert resp.headers["X-TTS-Provider"] == tts
 
 
 def test_chat_per_request_override_and_loopback(monkeypatch):

tests/test_piper_tts.py

@@ -0,0 +1,76 @@
+"""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