speak.py

"""
TTS speaker: reads text from stdin, synthesizes and plays audio.

Protocol (stdin/stdout):
  stdin:  text to speak (one line = one utterance)
  stdin:  STOP — interrupt current utterance
  stdout: "DONE\n" after each utterance finishes (or is stopped)

Usage:
  echo "Hello world" | python speak.py
  python speak.py  # interactive, type lines
"""

import argparse
import os
import select
import signal
import subprocess
import sys
import tempfile
import wave
from pathlib import Path

from piper import PiperVoice

running = True
_sigcount = 0


def _handle_signal(*_):
    global running, _sigcount
    _sigcount += 1
    running = False
    if _sigcount >= 2:
        os._exit(1)


signal.signal(signal.SIGTERM, _handle_signal)
signal.signal(signal.SIGINT, _handle_signal)

_DIR = Path(__file__).resolve().parent


def read_stdin():
    """Blocking read of one non-empty line from stdin. Returns None on EOF."""
    while running:
        ready, _, _ = select.select([sys.stdin], [], [], 0.5)
        if not ready:
            continue
        line = sys.stdin.readline()
        if not line:
            return None
        text = line.strip()
        if text:
            return text
    return None


def drain_stop():
    """Non-blocking check if STOP is waiting on stdin."""
    while True:
        ready, _, _ = select.select([sys.stdin], [], [], 0)
        if not ready:
            return False
        line = sys.stdin.readline()
        if not line:
            return False
        if line.strip() == "STOP":
            return True


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--model", default=str(_DIR / "models" / "en_GB-cori-high.onnx"))
    args = parser.parse_args()

    print(f"TTS: loading {Path(args.model).stem}...", file=sys.stderr, flush=True)
    voice = PiperVoice.load(args.model)
    print("TTS: ready", file=sys.stderr, flush=True)

    while running:
        text = read_stdin()
        if text is None:
            break
        if text == "STOP":
            continue

        # Synthesize (blocking)
        tmp = tempfile.mktemp(suffix=".wav")
        with wave.open(tmp, "wb") as wav_file:
            voice.synthesize_wav(text, wav_file)

        # Check if STOP arrived during inference
        if drain_stop():
            Path(tmp).unlink(missing_ok=True)
            print("DONE", flush=True)
            continue

        # Play
        proc = subprocess.Popen(["afplay", tmp])
        while proc.poll() is None:
            if not running:
                proc.terminate()
                proc.wait()
                break
            # Check for STOP during playback
            ready, _, _ = select.select([sys.stdin], [], [], 0.1)
            if ready:
                line = sys.stdin.readline()
                if line.strip() == "STOP":
                    proc.terminate()
                    proc.wait()
                    break

        Path(tmp).unlink(missing_ok=True)
        print("DONE", flush=True)


if __name__ == "__main__":
    main()