narrate.py

"""
Lulluna — Narration Engine (piper-tts)

Two usage modes:
  1. Direct import  — called from app.py on HF Spaces (unified env)
  2. Subprocess     — called by app.py on local dev (separate .venv_tts)
     python narrate.py  <reads text from stdin, writes base64 WAV JSON to stdout>

Voice: en_US-hfc_female-medium from rhasspy/piper-voices (HF Hub).
Downloaded once to the HF Hub cache; reused on every subsequent call.
The PiperVoice is cached after first load so repeated calls within the same
process (import mode) do not reload the ONNX model each time.
"""
import sys
import os
import io
import json
import base64
import wave

_VOICE_HF_REPO   = "rhasspy/piper-voices"
_VOICE_ONNX_PATH = "en/en_US/hfc_female/medium/en_US-hfc_female-medium.onnx"
_VOICE_JSON_PATH = "en/en_US/hfc_female/medium/en_US-hfc_female-medium.onnx.json"

# Slightly slower than normal — calming bedtime pace.
# piper length_scale = 1/speed  (>1 = slower)
_LENGTH_SCALE = 1.0 / float(os.getenv("NARRATION_SPEED", "0.85"))

# Module-level cache — populated on first call.
_voice = None


def _get_voice():
    global _voice
    if _voice is None:
        from piper.voice import PiperVoice
        from huggingface_hub import hf_hub_download
        onnx = hf_hub_download(repo_id=_VOICE_HF_REPO, filename=_VOICE_ONNX_PATH)
        cfg  = hf_hub_download(repo_id=_VOICE_HF_REPO, filename=_VOICE_JSON_PATH)
        _voice = PiperVoice.load(onnx, config_path=cfg)
    return _voice


def narrate(text: str) -> dict:
    """Synthesise speech for *text*. Returns {audio: base64 wav, mime: audio/wav}."""
    voice = _get_voice()
    buf = io.BytesIO()
    with wave.open(buf, "wb") as wav_file:
        wav_file.setnchannels(1)
        wav_file.setsampwidth(2)          # 16-bit PCM
        wav_file.setframerate(voice.config.sample_rate)
        voice.synthesize(text, wav_file, length_scale=_LENGTH_SCALE)
    buf.seek(0)
    b64 = base64.b64encode(buf.read()).decode()
    return {"audio": b64, "mime": "audio/wav"}


if __name__ == "__main__":
    # Subprocess mode: read text from stdin, write JSON to stdout.
    text = sys.stdin.read().strip()
    if not text:
        print("ERROR: empty input", file=sys.stderr)
        sys.exit(1)
    try:
        result = narrate(text)
        print(json.dumps(result), end="")
    except Exception as e:
        print(f"ERROR: {e}", file=sys.stderr)
        sys.exit(1)