Praxis CI
85 lines
2.7 KiB
Python
85 lines
2.7 KiB
Python
"""Piper self-hosted TTS adapter behind the TTSProvider interface (D-014).
|
|
|
|
Piper is the R4 mitigation (ARCHITECTURE.md): self-hosted, ~80ms first-audio on
|
|
CPU, open-weights, $0 marginal cost. Selected via PRAXIS_TTS=piper. A voice
|
|
model must be downloaded separately (see docs/latency-report.md §Piper
|
|
pre-staging). The adapter degrades gracefully if the voice model is absent.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import time
|
|
from pathlib import Path
|
|
from typing import AsyncIterator
|
|
|
|
from server.services.base import TTSProvider, TTSResult
|
|
|
|
|
|
class PiperTTS(TTSProvider):
|
|
"""Piper self-hosted TTS adapter (D-014 fallback / R4 mitigation)."""
|
|
|
|
name = "piper"
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
voice_model: str | None = None,
|
|
voice_id: str | None = None,
|
|
sample_rate: int = 22050,
|
|
) -> None:
|
|
self._voice_model = (
|
|
voice_model or os.environ.get("PIPER_VOICE_MODEL", "")
|
|
).strip()
|
|
self._voice_id = (voice_id or "piper-en_CA-medium").strip()
|
|
self._sample_rate = sample_rate
|
|
self._voice = None # loaded lazily
|
|
|
|
@property
|
|
def voice_id(self) -> str:
|
|
return self._voice_id
|
|
|
|
def _model_available(self) -> bool:
|
|
return bool(self._voice_model) and Path(self._voice_model).exists()
|
|
|
|
def _load_voice(self):
|
|
if self._voice is not None:
|
|
return self._voice
|
|
if not self._model_available():
|
|
return None
|
|
try:
|
|
from piper import PiperVoice # type: ignore
|
|
except ImportError:
|
|
return None
|
|
self._voice = PiperVoice.load(self._voice_model)
|
|
return self._voice
|
|
|
|
async def synthesize(self, text: str) -> AsyncIterator[bytes]:
|
|
"""Stream PCM s16le audio chunks from Piper."""
|
|
voice = self._load_voice()
|
|
if voice is None:
|
|
# Graceful no-op when the voice model isn't provisioned.
|
|
return
|
|
import io
|
|
|
|
for chunk in voice.synthesize(text):
|
|
# Piper yields AudioChunk with .audio_int16_bytes (PCM s16le).
|
|
yield chunk.audio_int16_bytes
|
|
|
|
async def synthesize_all(self, text: str) -> tuple[bytes, TTSResult]:
|
|
t0 = time.perf_counter()
|
|
chunks = bytearray()
|
|
first_audio_ms: float | None = None
|
|
async for chunk in self.synthesize(text):
|
|
if first_audio_ms is None:
|
|
first_audio_ms = (time.perf_counter() - t0) * 1000.0
|
|
chunks += chunk
|
|
return bytes(chunks), TTSResult(
|
|
first_audio_ms=first_audio_ms,
|
|
chars=len(text),
|
|
voice_id=self._voice_id,
|
|
sample_rate=self._sample_rate,
|
|
)
|
|
|
|
|
|
__all__ = ["PiperTTS"] |