wakeforge / src /backends /piper.py
eoinedge's picture
Upload src/backends/piper.py with huggingface_hub
ab75466 verified
Raw
History Blame Contribute Delete
3.7 kB
"""Free, local Piper TTS backend.
Voice models are pulled from the ``rhasspy/piper-voices`` repo on the
Hugging Face Hub and synthesized with the ``piper`` Python package. No
network access is needed at synthesis time and no paid service is used.
"""
from __future__ import annotations
import tempfile
import wave
from pathlib import Path
from typing import List, Optional
from ..audio import read_wav_file, resample
from ..config import PIPER_VOICES
from .base import SynthesisResult, TTSBackend, Voice
class PiperTTSBackend(TTSBackend):
source = "piper_tts"
def __init__(
self,
max_voices: int,
sample_rate_hz: int,
cache_dir: Optional[str] = None,
) -> None:
self.max_voices = max_voices
self.sample_rate_hz = sample_rate_hz
self.cache_dir = Path(cache_dir) if cache_dir else Path(tempfile.gettempdir()) / "piper_voices"
self._voices: List[Voice] = []
self._models: dict[str, "object"] = {} # voice name -> PiperVoice instance
# ------------------------------------------------------------------ #
def prepare(self) -> None:
from huggingface_hub import hf_hub_download # local import: heavy dep
from piper import PiperVoice as PiperModel # local import: heavy dep
self.cache_dir.mkdir(parents=True, exist_ok=True)
selected = PIPER_VOICES[: self.max_voices]
for spec in selected:
try:
onnx_path = hf_hub_download(
repo_id="rhasspy/piper-voices",
filename=f"{spec.repo_path}.onnx",
cache_dir=str(self.cache_dir),
)
config_path = hf_hub_download(
repo_id="rhasspy/piper-voices",
filename=f"{spec.repo_path}.onnx.json",
cache_dir=str(self.cache_dir),
)
model = PiperModel.load(onnx_path, config_path=config_path)
self._models[spec.voice_id] = model
self._voices.append(
Voice(
name=spec.voice_id,
language_code=spec.locale,
description=spec.description,
)
)
except Exception as exc: # noqa: BLE001 - skip individual bad voices
print(f"[piper] Skipping voice {spec.voice_id}: {exc}")
if not self._voices:
raise RuntimeError("No Piper voices could be downloaded or loaded.")
def voices(self) -> List[Voice]:
return list(self._voices)
def synthesize(self, text: str, voice: Voice) -> SynthesisResult:
model = self._models.get(voice.name)
if model is None:
raise RuntimeError(f"Piper voice not loaded: {voice.name}")
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp:
tmp_path = Path(tmp.name)
try:
with wave.open(str(tmp_path), "wb") as wav_file:
# piper-tts >= 1.3 renames the WAV writer to ``synthesize_wav``;
# the old ``synthesize(text, wav_file)`` no longer sets the WAV
# header (raising "# channels not specified"). Support both.
if hasattr(model, "synthesize_wav"):
model.synthesize_wav(text, wav_file)
else:
model.synthesize(text, wav_file)
audio, src_rate = read_wav_file(tmp_path)
finally:
tmp_path.unlink(missing_ok=True)
audio = resample(audio, src_rate, self.sample_rate_hz)
return SynthesisResult(audio=audio, sample_rate_hz=self.sample_rate_hz)