le-mystere-professeur-zacus/tools/gamebook/build_phone_gamebook.py

#!/usr/bin/env python3
"""Build the AUDIO gamebook pack for the PLIP phone and stage it on its SD.

Unlike the master/screen pack, the PLIP has NO screen: the choices must be
spoken. So each passage's narration WAV = the passage text + the choices read
out with their dial digit ("Pour examiner la machine, faites le 1. Pour
fouiller le bureau, faites le 2. Composez votre choix."). A menu WAV lets the
player pick a story by dialing 1..N at pickup.

Outputs, staged onto the PLIP SD at /sdcard/gamebook/ via POST /game/file:
  * menu.wav              — "Pour le dragon, faites le 1. ..."
  * <id>.json             — {start, passages:{pid:{wav, choices:[{goto}]}}}
  * <id>_<passage>.wav    — narration + spoken numbered choices
  * library.json          — [{id,title,book}]

Plays fully offline on the phone: pick up, dial to choose. No model, no network
at play time.

Usage:
    python3 tools/gamebook/build_phone_gamebook.py            # all -> .138
    python3 tools/gamebook/build_phone_gamebook.py --dry-run
    python3 tools/gamebook/build_phone_gamebook.py --ip 192.168.0.138 --force
"""
from __future__ import annotations

import argparse
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"
DEFAULT_VOICE = "Thomas"
EXCLUDE = {"zacus_demo"}
MAX_BOOKS = 9            # the dial menu uses single digits 1..9


def clean_wav(raw: bytes) -> bytes:
    """Strip Apple's FLLR filler → canonical 44-byte-header PCM WAV."""
    if raw[:4] != b"RIFF" or raw[8:12] != b"WAVE":
        return raw
    i, fmt, pcm = 12, None, None
    while i + 8 <= len(raw):
        cid = raw[i:i + 4]
        sz = int.from_bytes(raw[i + 4:i + 8], "little")
        body = raw[i + 8:i + 8 + sz]
        if cid == b"fmt ":
            fmt = body[:16]
        elif cid == b"data":
            pcm = body
            break
        i += 8 + sz + (sz & 1)
    if fmt is None or pcm is None:
        return raw
    return (b"RIFF" + struct.pack("<I", 4 + 8 + len(fmt) + 8 + len(pcm)) + b"WAVE"
            + b"fmt " + struct.pack("<I", len(fmt)) + fmt
            + b"data" + struct.pack("<I", len(pcm)) + pcm)


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],
            check=True, capture_output=True, timeout=120,
        )
        out.write_bytes(clean_wav(out.read_bytes()))
        return out.stat().st_size > 44
    except Exception as exc:  # noqa: BLE001
        print(f"  ! {backend} synth failed: {str(exc)[:140]}", file=sys.stderr)
        return False


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=30).read().decode()
        except Exception as exc:  # noqa: BLE001
            body = getattr(exc, "read", lambda: b"")()
            last = f"{exc} {body.decode('utf-8','replace') if body else ''}".strip()
    return False, last


def passage_narration(text: str, choices: list[dict]) -> str:
    """Passage text + spoken numbered choices (or the ending prompt)."""
    parts = [" ".join(text.split())]
    if choices:
        for i, c in enumerate(choices, start=1):
            parts.append(f"Pour {c['label'].rstrip('.').lower()}, faites le {i}.")
        parts.append("Composez votre choix.")
    else:
        parts.append("Fin de l'histoire. Faites le 0 pour en choisir une autre, "
                     "ou raccrochez.")
    return " ".join(parts)


def menu_narration(books: list[dict]) -> str:
    parts = ["Bienvenue dans les histoires dont vous êtes le héros."]
    for i, b in enumerate(books, start=1):
        parts.append(f"Pour {b['title'].rstrip('.').lower()}, faites le {i}.")
    parts.append("Composez le numéro de votre histoire.")
    return " ".join(parts)


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

    paths = sorted(p for p in BOOKS_DIR.glob("*.yaml") if p.stem not in EXCLUDE)
    if len(paths) > MAX_BOOKS:
        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 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). ALWAYS regenerated: its content
    # depends on the (cumulative) book list, so a cached menu.wav would still
    # announce an old, shorter set of stories.
    menu = OUT_DIR / "menu.wav"
    menu.unlink(missing_ok=True)
    if synth_wav(menu_narration(books_meta), menu, True, 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 build_paths:
        bid = path.stem
        book = yaml.safe_load(path.read_text())
        passages = book["passages"]
        manifest = {"start": book["start"], "passages": {}}
        n_ok = 0
        for pid, p in passages.items():
            choices = p.get("choices") or []
            wav = f"{bid}_{pid}.wav"
            out = OUT_DIR / wav
            if not synth_wav(passage_narration(p["text"], choices), out, args.force, args.tts):
                fail += 1
                continue
            n_ok += 1
            manifest["passages"][pid] = {
                "wav": wav,
                "choices": [{"goto": c["goto"]} for c in choices],
            }
            if not args.dry_run:
                ok, _ = push(args.ip, wav, out.read_bytes()); pushed += ok; fail += not ok
        bookfile = f"{bid}.json"
        (OUT_DIR / bookfile).write_text(json.dumps(manifest, ensure_ascii=False))
        if not args.dry_run:
            ok, resp = push(args.ip, bookfile, (OUT_DIR / bookfile).read_bytes())
            fail += not ok
        print(f"  {bid:18s} {len(passages):3d} passages  synth={n_ok}")

    library = {"library": [{"id": b["id"], "title": b["title"], "book": f"{b['id']}.json"}
                           for b in books_meta]}
    (OUT_DIR / "library.json").write_text(json.dumps(library, ensure_ascii=False, indent=1))
    if not args.dry_run:
        ok, resp = push(args.ip, "library.json", (OUT_DIR / "library.json").read_bytes())
        print(f"  {'OK' if ok else 'XX'} library.json -> {resp[:70]}")
        fail += not ok

    print(f"\nfail={fail}" + ("" if args.dry_run else f" pushed={pushed}"))
    print(f"local: {OUT_DIR}")
    return 1 if fail else 0


if __name__ == "__main__":
    raise SystemExit(main())