feat(gamebook): Kyutai neural TTS for phone pack
Add a local Kyutai TTS 1.6B (en_fr, MLX) backend to the PLIP phone pack generator (--tts kyutai): POST to the local :8302 server, resample 24 -> 16 kHz mono via afconvert, canonicalise. Much nicer French voice than say. Add --only to build/deploy stories incrementally (the 960 MB PLIP SD fits ~3-4); menu + library cover exactly the built books; filename cache makes long runs resumable. Bumps fw (MAX_BODY 3 MB).
This commit was merged in pull request #187.
This commit is contained in:
+1
-1
Submodule ESP32_ZACUS updated: a8af29068b...a4efce4c20
@@ -28,12 +28,15 @@ import json
|
||||
import struct
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unicodedata
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
import yaml
|
||||
|
||||
KYUTAI_URL = "http://127.0.0.1:8302/tts" # local Kyutai TTS 1.6B en_fr (MLX)
|
||||
|
||||
REPO = Path(__file__).resolve().parents[2]
|
||||
BOOKS_DIR = REPO / "game" / "gamebooks"
|
||||
OUT_DIR = REPO / "tools" / "gamebook" / "build_phone"
|
||||
@@ -64,11 +67,33 @@ def clean_wav(raw: bytes) -> bytes:
|
||||
+ b"data" + struct.pack("<I", len(pcm)) + pcm)
|
||||
|
||||
|
||||
def say_wav(text: str, out: Path, force: bool) -> bool:
|
||||
def _kyutai_tts(text: str, out: Path) -> bool:
|
||||
"""Local Kyutai TTS 1.6B (en_fr) → 24 kHz WAV, resampled to 16 kHz mono via
|
||||
afconvert (native macOS), then canonicalised. High-quality French voice."""
|
||||
req = urllib.request.Request(
|
||||
KYUTAI_URL, data=json.dumps({"text": text}).encode(),
|
||||
headers={"Content-Type": "application/json"}, method="POST")
|
||||
raw = urllib.request.urlopen(req, timeout=300).read() # synth is slow
|
||||
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp:
|
||||
tmp.write(raw)
|
||||
tmp_path = tmp.name
|
||||
# 24 kHz → 16 kHz mono 16-bit PCM for the phone firmware.
|
||||
subprocess.run(["afconvert", "-f", "WAVE", "-d", "LEI16@16000", "-c", "1",
|
||||
tmp_path, str(out)], check=True, capture_output=True, timeout=60)
|
||||
Path(tmp_path).unlink(missing_ok=True)
|
||||
out.write_bytes(clean_wav(out.read_bytes()))
|
||||
return out.stat().st_size > 44
|
||||
|
||||
|
||||
def synth_wav(text: str, out: Path, force: bool, backend: str) -> bool:
|
||||
"""Synthesise `text` to a 16 kHz mono WAV at `out`. Cached by filename
|
||||
(skipped if present) so a long run is resumable; pass --force to rebuild."""
|
||||
if out.exists() and out.stat().st_size > 44 and not force:
|
||||
return True
|
||||
out.parent.mkdir(parents=True, exist_ok=True)
|
||||
try:
|
||||
if backend == "kyutai":
|
||||
return _kyutai_tts(text, out)
|
||||
subprocess.run(
|
||||
["say", "-v", DEFAULT_VOICE, "--file-format=WAVE",
|
||||
"--data-format=LEI16@16000", "-o", str(out), text],
|
||||
@@ -77,20 +102,25 @@ def say_wav(text: str, out: Path, force: bool) -> bool:
|
||||
out.write_bytes(clean_wav(out.read_bytes()))
|
||||
return out.stat().st_size > 44
|
||||
except Exception as exc: # noqa: BLE001
|
||||
print(f" ! say failed: {exc}", file=sys.stderr)
|
||||
print(f" ! {backend} synth failed: {str(exc)[:140]}", file=sys.stderr)
|
||||
return False
|
||||
|
||||
|
||||
def push(ip: str, name: str, data: bytes) -> tuple[bool, str]:
|
||||
def push(ip: str, name: str, data: bytes, tries: int = 4) -> tuple[bool, str]:
|
||||
"""POST one file, retrying a few times — the phone can be briefly busy
|
||||
writing the SD over ~800 files, so transient timeouts are expected."""
|
||||
url = f"http://{ip}/game/file?path=sd/gamebook/{name}"
|
||||
last = ""
|
||||
for _ in range(tries):
|
||||
req = urllib.request.Request(url, data=data,
|
||||
headers={"Content-Type": "application/octet-stream"},
|
||||
method="POST")
|
||||
try:
|
||||
return True, urllib.request.urlopen(req, timeout=25).read().decode()
|
||||
return True, urllib.request.urlopen(req, timeout=30).read().decode()
|
||||
except Exception as exc: # noqa: BLE001
|
||||
body = getattr(exc, "read", lambda: b"")()
|
||||
return False, f"{exc} {body.decode('utf-8','replace') if body else ''}".strip()
|
||||
last = f"{exc} {body.decode('utf-8','replace') if body else ''}".strip()
|
||||
return False, last
|
||||
|
||||
|
||||
def passage_narration(text: str, choices: list[dict]) -> str:
|
||||
@@ -118,6 +148,9 @@ def main() -> int:
|
||||
ap = argparse.ArgumentParser(description=__doc__,
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||
ap.add_argument("--ip", default="192.168.0.138", help="PLIP IP (default .138)")
|
||||
ap.add_argument("--tts", choices=["say", "kyutai"], default="say",
|
||||
help="TTS backend (kyutai = local neural FR, much nicer/slower)")
|
||||
ap.add_argument("--only", default="", help="comma-separated book ids to build")
|
||||
ap.add_argument("--force", action="store_true", help="re-synthesise cached WAVs")
|
||||
ap.add_argument("--dry-run", action="store_true", help="synthesise only, no push")
|
||||
args = ap.parse_args()
|
||||
@@ -127,22 +160,28 @@ def main() -> int:
|
||||
print(f"note: {len(paths)} books, phone menu keeps first {MAX_BOOKS}")
|
||||
paths = paths[:MAX_BOOKS]
|
||||
|
||||
# The phone menu + library.json cover exactly the books built this run
|
||||
# (so dialing a number always maps to a story that's actually on the SD).
|
||||
# Add stories incrementally by re-running with the cumulative --only list;
|
||||
# cached WAVs are skipped, only new books synthesise.
|
||||
only = {s.strip() for s in args.only.split(",") if s.strip()}
|
||||
build_paths = [p for p in paths if not only or p.stem in only]
|
||||
books_meta = [{"id": p.stem, "title": yaml.safe_load(p.read_text()).get("title", p.stem)}
|
||||
for p in paths]
|
||||
print(f"phone library: {len(paths)} books -> "
|
||||
f"{'(dry-run)' if args.dry_run else f'{args.ip}:/sdcard/gamebook/'}")
|
||||
for p in build_paths]
|
||||
print(f"phone library: {len(build_paths)} books "
|
||||
f"via {args.tts} -> {'(dry-run)' if args.dry_run else f'{args.ip}:/sdcard/gamebook/'}")
|
||||
|
||||
fail = pushed = 0
|
||||
|
||||
# Menu WAV (the spoken story picker).
|
||||
menu = OUT_DIR / "menu.wav"
|
||||
if say_wav(menu_narration(books_meta), menu, args.force):
|
||||
if synth_wav(menu_narration(books_meta), menu, args.force, args.tts):
|
||||
if not args.dry_run:
|
||||
ok, _ = push(args.ip, "menu.wav", menu.read_bytes()); pushed += ok; fail += not ok
|
||||
else:
|
||||
fail += 1
|
||||
|
||||
for path in paths:
|
||||
for path in build_paths:
|
||||
bid = path.stem
|
||||
book = yaml.safe_load(path.read_text())
|
||||
passages = book["passages"]
|
||||
@@ -152,7 +191,7 @@ def main() -> int:
|
||||
choices = p.get("choices") or []
|
||||
wav = f"{bid}_{pid}.wav"
|
||||
out = OUT_DIR / wav
|
||||
if not say_wav(passage_narration(p["text"], choices), out, args.force):
|
||||
if not synth_wav(passage_narration(p["text"], choices), out, args.force, args.tts):
|
||||
fail += 1
|
||||
continue
|
||||
n_ok += 1
|
||||
|
||||
Reference in New Issue
Block a user