src/backends/piper.py

"""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)