"""Local Teddy audio follower (Chatterbox or Piper). Reads simulated state; never changes game state.""" import argparse import hashlib from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import json import os from pathlib import Path import shutil import subprocess import sys import threading import time import urllib.request import wave ROOT = Path(__file__).resolve().parent DATA = ROOT / ".speech" VOICE = "en_US-ljspeech-medium" MODEL = DATA / "voices" / (VOICE + ".onnx") AUDIO = {} OUTPUT = "local" STATUS = {"status": "starting", "voice": VOICE, "message": "Loading speech follower"} def python_path(): return DATA / "venv" / ("Scripts/python.exe" if os.name == "nt" else "bin/python") def setup(): import venv DATA.mkdir(exist_ok=True) if not python_path().exists(): venv.EnvBuilder(with_pip=True).create(DATA / "venv") subprocess.run([str(python_path()), "-m", "pip", "install", "piper-tts==1.8.0"], check=True) MODEL.parent.mkdir(exist_ok=True) base = "https://huggingface.co/rhasspy/piper-voices/resolve/main/en/en_US/ljspeech/medium/" for name in (MODEL.name, MODEL.name + ".json", "MODEL_CARD"): target = MODEL.parent / name if not target.exists(): temp = target.with_suffix(target.suffix + ".download") print("Downloading " + name, flush=True) urllib.request.urlretrieve(base + name, temp) temp.replace(target) print("Piper ready. Run: py -3 lab.py start-room --speech") def synth(text_file, output): from piper import PiperVoice voice = PiperVoice.load(str(MODEL)) temp = Path(output).with_suffix(".partial") with wave.open(str(temp), "wb") as stream: voice.synthesize_wav(Path(text_file).read_text(encoding="utf-8"), stream) temp.replace(output) class BrowserPlayer: def stop(self): pass def play(self, path): pass class Player: def __init__(self): self.process = None if os.name != "nt": self.command = shutil.which("afplay" if sys.platform == "darwin" else "aplay") if not self.command: raise RuntimeError("Install aplay (alsa-utils) for Linux audio playback") def stop(self): if os.name == "nt": import winsound winsound.PlaySound(None, 0) elif self.process and self.process.poll() is None: self.process.terminate() self.process.wait(timeout=3) self.process = None def play(self, path): self.stop() if os.name == "nt": import winsound winsound.PlaySound(str(path), winsound.SND_FILENAME | winsound.SND_ASYNC | winsound.SND_NODEFAULT) else: self.process = subprocess.Popen([self.command, str(path)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) def cue_for(state): teddy = state.get("teddy", {}) cue = teddy.get("current") if not cue or teddy.get("muted") or (teddy.get("takeover") and not cue.get("key", "").startswith("gm-")) or not teddy.get("online"): return None return (state["run_id"], cue["id"], cue.get("started"), cue["transcript"]) class Follower: def __init__(self, player, engine=None): self.engine = engine self.player = player self.current = None self.job = None self.output = None self.end = None self.failed = False (DATA / "cache").mkdir(parents=True, exist_ok=True) def status(self, value, message): global STATUS STATUS = {"status": value, "voice": VOICE, "message": message} def cancel(self): global AUDIO AUDIO = {} if self.job and self.job.poll() is None: self.job.terminate() self.job.wait(timeout=5) self.job = None self.end = None self.player.stop() def update(self, state): cue = cue_for(state) if state else None if cue != self.current: self.cancel() self.current = cue self.failed = False if cue: digest = hashlib.sha256((VOICE + "\n" + cue[3]).encode()).hexdigest() self.output = DATA / "cache" / (digest + ".wav") if self.output.exists(): self.play() else: text_file = self.output.with_suffix(".txt") text_file.write_text(cue[3], encoding="utf-8") self.job = self.engine.submit(cue[3], self.output) if self.engine else subprocess.Popen([str(python_path()), str(Path(__file__).resolve()), "synth", "--text-file", str(text_file), "--output", str(self.output)], creationflags=subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0) self.status("generating", "Generating Teddy's current cue") if self.job and self.job.poll() is not None: result = self.job.returncode self.job = None if result == 0: self.play() else: self.failed = True self.status("error", "Speech synthesis failed; captions remain available. Replay to retry.") if self.end is not None and time.monotonic() >= self.end: self.end = None self.status("finished", "Cue finished. Advance virtual time to release the next queued cue.") if not cue: self.status("idle" if state else "disconnected", "Waiting for a cue" if state else "Room connection lost; audio stopped") def play(self): try: with wave.open(str(self.output)) as audio: duration = audio.getnframes() / audio.getframerate() self.player.play(self.output) global AUDIO identity = hashlib.sha256(repr(self.current[:3]).encode()).hexdigest() AUDIO = {"path":str(self.output), "audio_url":"/audio/" + identity + ".wav", "cue_id":self.current[1], "run_id":self.current[0]} self.end = None if OUTPUT == "browser" else time.monotonic() + duration self.status("ready" if OUTPUT == "browser" else "playing", "Audio ready for enabled browsers" if OUTPUT == "browser" else "Playing on the lab computer's default audio output") except Exception as error: self.failed = True self.status("error", "Audio playback failed: " + str(error)) class Handler(BaseHTTPRequestHandler): def do_GET(self): origin = self.headers.get("Origin", "") allowed = ("http://127.0.0.1:18777", "http://localhost:18777") if origin and origin not in allowed: self.send_error(403) return audio = AUDIO.copy() status = 200 content_range = None if self.path == "/health": data = json.dumps({**STATUS, "output":OUTPUT, **{k:v for k,v in audio.items() if k != "path"}}).encode() content_type = "application/json" elif audio and self.path == audio.get("audio_url"): data = Path(audio["path"]).read_bytes() content_type = "audio/wav" byte_range = self.headers.get("Range") if byte_range: import re match = re.fullmatch(r"bytes=(\d+)-(\d*)", byte_range) start = int(match[1]) if match else len(data) end = min(int(match[2]) if match and match[2] else len(data)-1,len(data)-1) if start > end: self.send_error(416) return content_range = f"bytes {start}-{end}/{len(data)}" data = data[start:end+1] status = 206 else: self.send_error(404) return self.send_response(status) if origin: self.send_header("Access-Control-Allow-Origin", origin) self.send_header("Content-Type", content_type) self.send_header("Accept-Ranges", "bytes") if content_range: self.send_header("Content-Range", content_range) self.send_header("Cache-Control", "no-store") self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def log_message(self, *_): pass def follow(room_port=18776, status_port=18778, output="local", backend="chatterbox"): global OUTPUT, VOICE OUTPUT = output engine = None if backend == "chatterbox": from chatterbox_client import Engine, voice_id VOICE = voice_id() engine = Engine() follower = Follower(BrowserPlayer() if output == "browser" else Player(), engine) server = ThreadingHTTPServer(("127.0.0.1", status_port), Handler) threading.Thread(target=server.serve_forever, daemon=True).start() follower.status("starting" if engine else "idle", "Loading Teddy voice" if engine else "Waiting for a cue") print(f"TEDDY AUDIO: {backend}, {output} output, status port {status_port}", flush=True) try: while True: if engine and engine.error: follower.cancel() follower.status("error", "Chatterbox unavailable; see .speech/chatterbox.log. Restart speech to retry.") time.sleep(1) continue if engine and not engine.ready.is_set(): time.sleep(.2) continue try: with urllib.request.urlopen(f"http://127.0.0.1:{room_port}/room/state", timeout=1) as response: state = json.load(response)["state"] except Exception: state = None follower.update(state) time.sleep(0.2) finally: follower.cancel() if engine: engine.close() server.shutdown() server.server_close() if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("mode", choices=("setup", "follow", "synth")) parser.add_argument("--backend", choices=("chatterbox", "piper"), default="chatterbox") parser.add_argument("--text-file") parser.add_argument("--output") parser.add_argument("--room-port", type=int, default=18776) parser.add_argument("--status-port", type=int, default=18778) parser.add_argument("--output-device", choices=("local","browser"), default="local") args = parser.parse_args() if args.mode == "setup": setup() elif args.mode == "synth": synth(args.text_file, args.output) else: follow(args.room_port, args.status_port, args.output_device, args.backend)