Batch & CLI

Batch-process a whole directory of voice clips with an OOV/confidence QA report and incremental re-runs, and the openfacefx command-line entry point that wires every stage and exporter together.

openfacefx.batch

Batch directory processing: a tree of voice lines in, a tree of tracks out.

For every .wav under --dir, look for a same-stem .TextGrid (MFA -- accurate path, preferred) or .txt transcript (naive path), generate a track, and write it to the mirrored path under --out. A manifest makes --modified-only re-runs incremental; a summary (printed table + JSON) reports per-file status, counts, OOV words that fell through to the G2P rule fallback, and worst-first aligner confidence when the aligner supplies it. Per-file failures do not stop the batch; the exit code reports them at the end.

Three opt-in additions (issue #35) layer on top without touching the default output -- with their flags absent the printed table and batch_summary.json are byte-identical to before:

• --machine-readable streams an NDJSON event log to stderr (start/progress/warning/failure/done, one JSON object per line) so a supervising process can follow a large run live. Events are emitted in processing order (the summary table stays worst-first sorted).
• --ledger FILE appends one NDJSON record per run to a file (args snapshot, per-input size/mtime, outcome counts) for reproducibility/audit; it survives --modified-only and carries a deterministic, wall-clock-free run id.
• --cue-warnings folds qa.cue_flags() counts into each summary row and the worst-first ranking, so too-short/too-long phoneme cues rank alongside OOV words and low confidence.

MANIFEST_NAME = '.openfacefx-manifest.json' module-attribute

SUMMARY_NAME = 'batch_summary.json' module-attribute

LEDGER_FORMAT = 'openfacefx.batch.ledger' module-attribute

LEDGER_VERSION = 1 module-attribute

find_jobs(in_dir: str, out_dir: str, recurse: bool, ext: str) -> List[dict]

Source code in src/openfacefx/batch.py

def find_jobs(in_dir: str, out_dir: str, recurse: bool, ext: str) -> List[dict]:
    jobs = []
    for root, dirs, files in os.walk(in_dir):
        if not recurse:
            dirs.clear()
        for f in sorted(files):
            if not f.lower().endswith(".wav"):
                continue
            stem = os.path.splitext(f)[0]
            wav = os.path.join(root, f)
            rel = os.path.relpath(wav, in_dir)
            tg = os.path.join(root, stem + ".TextGrid")
            txt = os.path.join(root, stem + ".txt")
            out_rel = os.path.splitext(rel)[0] + "." + ext
            jobs.append(dict(
                rel=rel, wav=wav,
                textgrid=tg if os.path.exists(tg) else None,
                txt=txt if os.path.exists(txt) else None,
                out=os.path.join(out_dir, out_rel),
                out_rel=out_rel,
            ))
    return jobs

run_batch(in_dir: str, out_dir: str, recurse: bool = False, modified_only: bool = False, jobs: int = 1, mapping: Optional[str] = None, cmudict: Optional[str] = None, fps: float = 60.0, ext: str = 'json', machine_readable: bool = False, quiet: bool = False, ledger: Optional[str] = None, cue_warnings: bool = False, min_cue: Optional[float] = None, max_cue: Optional[float] = None, manifest_file: Optional[str] = None, captions: Optional[str] = None) -> int

Returns a process exit code (0 = all ok, 1 = at least one failure).

Every parameter after ext is additive and opt-in (issue #35); with the defaults the printed table and batch_summary.json are byte-identical to before. machine_readable streams an NDJSON event log to stderr, quiet suppresses the human table (the summary JSON and any NDJSON/ledger are still written), ledger appends one NDJSON run record to a file, and cue_warnings folds qa.cue_flags() counts into each row and the worst-first sort (min_cue/max_cue default to qa's thresholds).

manifest_file (issue #40) selects the loc-table driver: instead of walking in_dir, read a CSV/TSV manifest and emit one track per row through the same pipeline + writers + summary/NDJSON/ledger. With it absent the directory-walk path is byte-identical. captions ("srt" / "vtt", issue #41) writes a caption sidecar next to each naive-mode track from the same word alignment; None (the default) writes none.

Source code in src/openfacefx/batch.py

def run_batch(in_dir: str, out_dir: str, recurse: bool = False,
              modified_only: bool = False, jobs: int = 1,
              mapping: Optional[str] = None, cmudict: Optional[str] = None,
              fps: float = 60.0, ext: str = "json",
              machine_readable: bool = False, quiet: bool = False,
              ledger: Optional[str] = None, cue_warnings: bool = False,
              min_cue: Optional[float] = None,
              max_cue: Optional[float] = None,
              manifest_file: Optional[str] = None,
              captions: Optional[str] = None) -> int:
    """Returns a process exit code (0 = all ok, 1 = at least one failure).

    Every parameter after ``ext`` is additive and opt-in (issue #35); with the
    defaults the printed table and ``batch_summary.json`` are byte-identical to
    before. ``machine_readable`` streams an NDJSON event log to stderr, ``quiet``
    suppresses the human table (the summary JSON and any NDJSON/ledger are still
    written), ``ledger`` appends one NDJSON run record to a file, and
    ``cue_warnings`` folds ``qa.cue_flags()`` counts into each row and the
    worst-first sort (``min_cue``/``max_cue`` default to qa's thresholds).

    ``manifest_file`` (issue #40) selects the loc-table driver: instead of
    walking ``in_dir``, read a CSV/TSV manifest and emit one track per row
    through the same pipeline + writers + summary/NDJSON/ledger. With it absent
    the directory-walk path is byte-identical. ``captions`` (``"srt"`` / ``"vtt"``,
    issue #41) writes a caption sidecar next to each naive-mode track from the
    same word alignment; ``None`` (the default) writes none."""
    lo = MIN_CUE if min_cue is None else min_cue
    hi = MAX_CUE if max_cue is None else max_cue
    cue = (lo, hi) if cue_warnings else None

    def emit(obj: dict) -> None:
        if machine_readable:
            sys.stderr.write(json.dumps(obj) + "\n")
            sys.stderr.flush()

    def emit_progress(index: int, row: dict) -> None:
        warns = _row_warnings(row)
        emit({"event": "progress", "index": index, "file": row["file"],
              "out": row["out"], "status": row["status"], "mode": row["mode"],
              "channels": row["channels"], "keyframes": row["keyframes"],
              "oov": row["oov"], "cue_warnings": row.get("cue_warnings", 0),
              "min_confidence": row["min_confidence"], "warnings": warns})
        for w in warns:
            emit({"event": "warning", "index": index, "file": row["file"],
                  "message": w})
        if row["status"] != "ok":
            emit({"event": "failure", "index": index, "file": row["file"],
                  "error": row["error"]})

    def args_snapshot() -> dict:
        snap = {"dir": in_dir, "out": out_dir, "recurse": recurse,
                "modified_only": modified_only, "jobs": jobs, "ext": ext,
                "mapping": mapping, "cmudict": cmudict, "fps": fps,
                "cue_warnings": cue_warnings, "min_cue": lo, "max_cue": hi}
        if manifest_file is not None:            # only in manifest mode -> the
            snap["manifest"] = manifest_file     # directory-mode snapshot is
        if captions is not None:                 # byte-identical without either
            snap["captions"] = captions
        return snap

    def input_fingerprints(work: List[dict]) -> List[dict]:
        fps_out = []
        for job in work:
            st = _stamp(job["wav"]) or [None, None]
            kind = ("mfa" if job["textgrid"] else
                    "naive" if (job["txt"] or job.get("text")) else "none")
            fps_out.append({"file": job["rel"], "mtime": st[0], "size": st[1],
                            "transcript": kind})
        fps_out.sort(key=lambda r: r["file"])
        return fps_out

    if manifest_file is not None:
        from .batch_manifest import manifest_jobs, read_manifest
        base = os.path.dirname(os.path.abspath(manifest_file))
        try:
            work = manifest_jobs(read_manifest(manifest_file), out_dir, ext, base)
        except (OSError, ValueError) as exc:
            if not quiet:
                print(f"cannot read manifest {manifest_file}: {exc}")
            return 1
    else:
        work = find_jobs(in_dir, out_dir, recurse, ext)
    if not work:
        emit({"event": "start", "total": 0, "todo": 0, "skipped": 0,
              "jobs": jobs, "ext": ext, "recurse": recurse})
        emit({"event": "done", "processed": 0, "failed": 0, "skipped": 0,
              "exit": 1})
        if ledger:
            _append_ledger(ledger, _ledger_record(
                args_snapshot=args_snapshot(), inputs=[],
                outcome={"processed": 0, "failed": 0, "skipped": 0, "exit": 1},
                ext=ext))
        if not quiet:
            print(f"no rows found in manifest {manifest_file}" if manifest_file
                  else f"no .wav files found under {in_dir}")
        return 1

    manifest_path = os.path.join(out_dir, MANIFEST_NAME)
    manifest = {}
    if os.path.exists(manifest_path):
        with open(manifest_path, encoding="utf-8") as fh:
            manifest = json.load(fh)

    def fingerprint(job):
        fp = {
            "wav": _stamp(job["wav"]),
            "transcript": _stamp(job["textgrid"] or job["txt"] or ""),
            "mapping": _stamp(mapping) if mapping else None,
            "out": job["out"],
        }
        # manifest jobs carry the transcript inline and may name a per-row
        # mapping; fold both into the modified-only key. Directory jobs have
        # neither, so their manifest fingerprint stays byte-identical.
        if job.get("text") is not None:
            fp["text"] = hashlib.sha256(
                job["text"].encode("utf-8")).hexdigest()[:16]
        if job.get("mapping"):
            fp["mapping"] = _stamp(job["mapping"])
        return fp

    todo, skipped = [], 0
    for job in work:
        if (modified_only and manifest.get(job["rel"]) == fingerprint(job)
                and os.path.exists(job["out"])):
            skipped += 1
            continue
        todo.append(job)

    emit({"event": "start", "total": len(work), "todo": len(todo),
          "skipped": skipped, "jobs": jobs, "ext": ext, "recurse": recurse})

    args = [(job, mapping, cmudict, fps, cue, captions) for job in todo]
    if jobs > 1 and len(todo) > 1:
        with Pool(processes=jobs) as pool:
            rows = pool.map(_process_one, args)
        for i, row in enumerate(rows):
            emit_progress(i, row)
    else:
        rows = []
        for i, a in enumerate(args):
            row = _process_one(a)
            rows.append(row)
            emit_progress(i, row)

    for job, row in zip(todo, rows):
        if row["status"] == "ok":
            manifest[job["rel"]] = fingerprint(job)
        else:
            manifest.pop(job["rel"], None)
    os.makedirs(out_dir, exist_ok=True)
    with open(manifest_path, "w", encoding="utf-8") as fh:
        json.dump(manifest, fh, indent=2)

    # worst files first: failures, then lowest confidence, then most OOV. With
    # --cue-warnings the cue count is an extra final tiebreaker -- a strict
    # superset of the old key, so the order (and thus the bytes) are unchanged
    # without the flag.
    def _key(r):
        base = (r["status"] == "ok",
                r["min_confidence"] if r["min_confidence"] is not None else 2.0,
                -len(r["oov"]))
        return base + (-r["cue_warnings"],) if cue_warnings else base
    rows.sort(key=_key)

    summary = dict(processed=len(rows), skipped_unchanged=skipped,
                   failed=sum(1 for r in rows if r["status"] != "ok"),
                   rows=rows)
    with open(os.path.join(out_dir, SUMMARY_NAME), "w", encoding="utf-8") as fh:
        json.dump(summary, fh, indent=2)
    rc = 1 if summary["failed"] else 0

    emit({"event": "done", "processed": len(rows), "failed": summary["failed"],
          "skipped": skipped, "exit": rc})
    if ledger:
        _append_ledger(ledger, _ledger_record(
            args_snapshot=args_snapshot(), inputs=input_fingerprints(work),
            outcome={"processed": len(rows), "failed": summary["failed"],
                     "skipped": skipped, "exit": rc}, ext=ext))

    if not quiet:
        width = max([len(r["file"]) for r in rows] + [4])
        head = (f"{'file':<{width}}  {'status':<7} {'mode':<5} {'dur':>6} "
                f"{'keys':>5}")
        if cue_warnings:
            head += f" {'cue':>4}"
        head += "  oov"
        print(head)
        for r in rows:
            dur = f"{r['duration']:.2f}" if r["duration"] is not None else "-"
            oov = ",".join(r["oov"][:4]) + ("…" if len(r["oov"]) > 4 else "")
            line = (f"{r['file']:<{width}}  {r['status']:<7} "
                    f"{r['mode'] or '-':<5} {dur:>6} {r['keyframes']:>5}")
            if cue_warnings:
                line += f" {r['cue_warnings']:>4}"
            line += f"  {oov}"
            print(line)
            if r["error"]:
                print(f"{'':<{width}}  ! {r['error']}")
        print(f"\n{len(rows)} processed, {skipped} skipped (unchanged), "
              f"{summary['failed']} failed -> "
              f"{os.path.join(out_dir, SUMMARY_NAME)}")
    return rc

Loc-table manifests (--manifest)

The alternative to the directory walk (issue #40): drive the batch from a localization string table — a CSV/TSV keyed by loc-ID, one row per line (audio, text, language, character, optional mapping/style/out) — the way Unity / Godot / Unreal export String Table Collections and FaceFX keys VO to an entrytag. read_manifest parses the table with stdlib csv (header-matched forgivingly against COLUMN_ALIASES), and manifest_jobs turns the rows into the same jobs the directory walk feeds batch._process_one, so every row runs through the unchanged pipeline, writers, summary, NDJSON stream and ledger. A missing / unreadable / malformed row is an isolated per-row failure; with --manifest absent the directory-walk output is byte-identical.

openfacefx.batch_manifest

Loc-table / dialogue-database manifest ingestion for batch (issue #40).

Real game VO is driven by a localization string table, not a directory of same-stem .wav + .txt files: Unity / Godot / Unreal export String Table Collections keyed by a loc-ID, and FaceFX keys VO to an entrytag. This module is the stdlib csv ingestion layer that feeds those tables into the existing :mod:openfacefx.batch pipeline — distinct from the i18n pronunciation model (#8); this is purely the ingestion front-end.

:func:read_manifest parses a CSV/TSV table into normalized rows, and :func:manifest_jobs turns those rows into the same job dicts :func:openfacefx.batch._process_one already consumes, so each row runs through the unchanged pipeline + output writers and lands in the same summary table, NDJSON stream and ledger. A --manifest run and a directory-walk run share everything downstream of job discovery.

The column mapping is header-driven and forgiving — headers are matched case-insensitively with punctuation/spacing ignored, against a table of common names (see :data:COLUMN_ALIASES). The recognized fields:

• id (id / key / loc-id / entrytag / name) — the row key; also the summary label and the derived output stem.
• audio (audio / wav / file / voice / sound) — the voice clip, resolved relative to the manifest's own directory.
• text (text / transcript / line / dialogue / string) — the spoken transcript, inline in the table.
• language (lang / language / locale) and character (character / speaker / actor) — threaded onto the row as metadata.
• mapping (mapping / rig) and style (style / coart) — per-row solve options (a rig JSON, a coarticulation style preset).
• textgrid (textgrid / alignment) — an optional per-row MFA TextGrid (the accurate path); out (out / output / track) — an explicit output path, else <id>.<ext> under the output tree.

CSV/TSV only; PO / XLIFF and the pivoted one-column-per-locale layout are noted future follow-ups (a pivoted table degrades to per-row "no transcript" failures, never a crash). Stdlib only (csv); deterministic.

COLUMN_ALIASES: Dict[str, str] = {'id': 'id', 'key': 'id', 'locid': 'id', 'loc': 'id', 'locionid': 'id', 'entrytag': 'id', 'name': 'id', 'stringid': 'id', 'line_id': 'id', 'lineid': 'id', 'audio': 'audio', 'wav': 'audio', 'file': 'audio', 'audiofile': 'audio', 'audiopath': 'audio', 'voice': 'audio', 'sound': 'audio', 'clip': 'audio', 'voiceclip': 'audio', 'filename': 'audio', 'text': 'text', 'transcript': 'text', 'line': 'text', 'dialogue': 'text', 'string': 'text', 'source': 'text', 'value': 'text', 'utterance': 'text', 'lang': 'language', 'language': 'language', 'locale': 'language', 'character': 'character', 'char': 'character', 'speaker': 'character', 'actor': 'character', 'voiceactor': 'character', 'npc': 'character', 'mapping': 'mapping', 'rig': 'mapping', 'map': 'mapping', 'style': 'style', 'coart': 'style', 'coartstyle': 'style', 'textgrid': 'textgrid', 'alignment': 'textgrid', 'align': 'textgrid', 'out': 'out', 'output': 'out', 'outpath': 'out', 'track': 'out', 'outputpath': 'out', 'outfile': 'out'} module-attribute

read_manifest(path: str) -> List[Dict[str, Optional[str]]]

Parse a CSV/TSV loc-table into normalized rows.

Each row is a dict over :data:the canonical fields <COLUMN_ALIASES> with None for any absent/blank cell. The header is matched forgivingly (case, spacing and punctuation are ignored); the first column mapping to a canonical field wins. A UTF-8 BOM (common in Unity exports) is stripped. Raises ValueError if the file has no header row (a file-level error, distinct from a bad row — which surfaces later as a per-row failure).

Source code in src/openfacefx/batch_manifest.py

def read_manifest(path: str) -> List[Dict[str, Optional[str]]]:
    """Parse a CSV/TSV loc-table into normalized rows.

    Each row is a dict over :data:`the canonical fields <COLUMN_ALIASES>` with
    ``None`` for any absent/blank cell. The header is matched forgivingly (case,
    spacing and punctuation are ignored); the first column mapping to a canonical
    field wins. A UTF-8 BOM (common in Unity exports) is stripped. Raises
    ``ValueError`` if the file has no header row (a file-level error, distinct
    from a bad row — which surfaces later as a per-row failure)."""
    with open(path, encoding="utf-8-sig", newline="") as fh:
        text = fh.read()
    reader = csv.DictReader(io.StringIO(text), delimiter=_delimiter(path, text))
    if not reader.fieldnames:
        raise ValueError(f"manifest {path} has no header row")
    colmap: Dict[str, str] = {}
    for raw in reader.fieldnames:
        canon = COLUMN_ALIASES.get(_norm_header(raw))
        if canon and canon not in colmap.values():   # first column wins
            colmap[raw] = canon
    rows: List[Dict[str, Optional[str]]] = []
    for raw_row in reader:
        row: Dict[str, Optional[str]] = {f: None for f in _FIELDS}
        for raw, canon in colmap.items():
            val = (raw_row.get(raw) or "").strip()
            row[canon] = val or None
        rows.append(row)
    return rows

manifest_jobs(rows: List[Dict[str, Optional[str]]], out_dir: str, ext: str, base_dir: str) -> List[Dict]

Turn normalized manifest rows into batch._process_one job dicts.

Audio / mapping / textgrid paths resolve relative to base_dir (the manifest's directory). The output path is the row's explicit out column (relative to out_dir, or absolute) when given, else <id>.<ext> under out_dir. The loc-ID is the row key, summary label and, when needed, the output stem; a row with neither id nor audio falls back to row<N>.

Source code in src/openfacefx/batch_manifest.py

def manifest_jobs(rows: List[Dict[str, Optional[str]]], out_dir: str, ext: str,
                  base_dir: str) -> List[Dict]:
    """Turn normalized manifest rows into ``batch._process_one`` job dicts.

    Audio / mapping / textgrid paths resolve relative to ``base_dir`` (the
    manifest's directory). The output path is the row's explicit ``out`` column
    (relative to ``out_dir``, or absolute) when given, else ``<id>.<ext>`` under
    ``out_dir``. The loc-ID is the row key, summary label and, when needed, the
    output stem; a row with neither id nor audio falls back to ``row<N>``."""
    jobs: List[Dict] = []
    for i, row in enumerate(rows):
        loc_id = row.get("id") or (
            os.path.splitext(os.path.basename(row["audio"]))[0]
            if row.get("audio") else "row%d" % (i + 1))
        out_col = row.get("out")
        if out_col:
            out_rel = out_col
            out = out_col if os.path.isabs(out_col) else os.path.join(out_dir,
                                                                      out_col)
        else:
            out_rel = _out_stem(loc_id) + "." + ext
            out = os.path.join(out_dir, out_rel)
        jobs.append(dict(
            id=loc_id, rel=loc_id, out=out, out_rel=out_rel,
            wav=_resolve(row.get("audio"), base_dir) or "",
            text=row.get("text"),                 # inline transcript (may be None)
            textgrid=_resolve(row.get("textgrid"), base_dir),
            txt=None,
            mapping=_resolve(row.get("mapping"), base_dir),
            style=row.get("style"),
            language=row.get("language"),
            character=row.get("character"),
        ))
    return jobs

openfacefx.cli

Command-line interface.

Examples

Naive (text + audio duration, no models needed): python -m openfacefx naive --text "hello world" --wav voice.wav -o out.json

From an MFA alignment (accurate): python -m openfacefx mfa --textgrid voice.TextGrid -o out.json

main(argv=None) -> int

Source code in src/openfacefx/cli.py

def main(argv=None) -> int:
    # FaceFXWrapper.exe drop-in shim (issue #33): intercept BEFORE argparse so the
    # raw positional args pass through verbatim. A real consumer command carries
    # values argparse would choke on — a leading-dash token read as an option, or
    # a resampled-WAV/text path — so the whole tail goes straight to the shim,
    # dispatched on the literal 'facefxwrapper' token.
    raw = sys.argv[1:] if argv is None else list(argv)
    if raw and raw[0] == "facefxwrapper":
        return facefxwrapper.run(raw[1:])

    p = argparse.ArgumentParser(prog="openfacefx")
    sub = p.add_subparsers(dest="cmd", required=True)

    n = sub.add_parser("naive", help="text + duration -> curves (no models)")
    n.add_argument("--text", help="transcript (required unless "
                   "--anchors-format srt supplies it from the cue text)")
    g = n.add_mutually_exclusive_group(required=True)
    g.add_argument("--wav", help="WAV file to read duration from")
    g.add_argument("--duration", type=float, help="duration in seconds")
    n.add_argument("--cmudict", help="optional CMUdict file for better G2P")
    n.add_argument("--anchors", help="word/segment timing file that pins the "
                   "aligner at known boundaries (SRT cues or TTS word timings)")
    n.add_argument("--anchors-format", choices=_ANCHOR_FORMATS,
                   help="format of --anchors: srt|words|azure|elevenlabs|kokoro|"
                        "google (only valid together with --anchors)")
    n.add_argument("--fps", type=float, default=60.0)
    n.add_argument("-o", "--out", required=True)
    n.add_argument("--emit-segments", metavar="PATH",
                   help="also write phoneme segments as JSON for the HTML "
                        "previewer's --segments lane (see tools/build_preview.py)")
    n.add_argument("--emit-captions", metavar="PATH",
                   help="also write SRT/WebVTT captions (issue #41) from the same "
                        "word alignment the curves use — .vtt for WebVTT, any "
                        "other extension for SubRip; tune with --cps/--max-line/"
                        "--max-lines/--gap/--karaoke")
    _add_caption_options(n)
    _add_output_options(n)
    _add_coart_options(n)
    _add_smoothing_options(n)
    _add_gesture_options(n)
    _add_event_options(n)
    _add_prosody_options(n)
    _add_edits_options(n)
    _add_qa_options(n)
    n.add_argument("--no-normalize", action="store_true",
                   help="disable the default transcript normalization pass "
                        "(Unicode ellipsis/dashes/curly-quotes/NBSP -> ASCII "
                        "before G2P). Substitutions are reported in --json/"
                        "--report; ASCII transcripts are unaffected either way")
    n.add_argument("--tags", action="store_true",
                   help="parse transcript text tags in --text (issue #7): curve "
                        "tags '[Name type=ct v1=1]word[/Name]' -> channels, event "
                        "tags '[event:NAME ...]'/'[gesture:NAME]' -> the event "
                        "layer, '[emphasis]word[/emphasis]' -> local articulation "
                        "boost, '<T>' chunk markers / '[pause:SEC]' -> timeline "
                        "silence. Auto-enabled when a clear tag is present. Tags "
                        "are stripped before G2P, so the words are still lip-synced; "
                        "a tagless transcript is byte-identical to no --tags")
    n.add_argument("--ssml", action="store_true",
                   help="parse --text as SSML (issue #52): the same W3C markup you "
                        "feed Azure/Google/Polly TTS drives lip-sync via a thin "
                        "front-end over #7 — <break>/<emphasis>/<mark>/<sub>/<p>/"
                        "<s>/<say-as> map onto the text-tag primitives. "
                        "Auto-enabled when --text opens with a <speak> root; a "
                        "<speak> with no constructs is byte-identical to plain naive")

    m = sub.add_parser("mfa", help="MFA TextGrid -> curves (accurate)")
    m.add_argument("--textgrid", required=True)
    m.add_argument("--wav", help="optional 16-bit PCM WAV the audio-driven layers "
                   "read: energy-scaled --gestures and --prosody events (the "
                   "TextGrid alone has no audio). Without it those layers degrade "
                   "to timing-only / are unavailable")
    m.add_argument("--fps", type=float, default=60.0)
    m.add_argument("-o", "--out", required=True)
    m.add_argument("--emit-segments", metavar="PATH",
                   help="also write phoneme segments as JSON for the HTML "
                        "previewer's --segments lane (see tools/build_preview.py)")
    _add_output_options(m)
    _add_coart_options(m)
    _add_smoothing_options(m)
    _add_gesture_options(m)
    _add_event_options(m)
    _add_prosody_options(m)
    _add_edits_options(m)
    _add_qa_options(m)

    t = sub.add_parser("from-timing",
                       help="TTS phoneme/viseme timing -> curves (skip the "
                            "aligner: espeak .pho, Piper, Cartesia, Azure, Polly)")
    t.add_argument("--file", required=True, help="the timing dump to parse")
    t.add_argument("--format", required=True, choices=sorted(_TIMING_PARSERS),
                   help="pho|piper|cartesia (phoneme-unit) or azure|polly "
                        "(viseme-unit, uses the built-in vendor remap preset)")
    t.add_argument("--sample-rate", type=int,
                   help="Piper voice sample rate in Hz (required for "
                        "--format piper; samples -> seconds)")
    t.add_argument("--final-duration", type=float, default=0.08,
                   help="seconds held by the last start-only event "
                        "(azure/polly); default 0.08")
    t.add_argument("--fps", type=float, default=60.0)
    t.add_argument("-o", "--out", required=True)
    _add_output_options(t)
    _add_coart_options(t)
    _add_smoothing_options(t)
    _add_edits_options(t)
    _add_qa_options(t)

    fc = sub.add_parser("from-cues",
                        help="import a stepped mouth-cue file back into a track "
                             "(Rhubarb tsv/xml/json, Moho .dat, Papagayo .pgo) — "
                             "the inverse of the cue exporters (issue #44)")
    fc.add_argument("file", help="the mouth-cue file to import")
    fc.add_argument("--format", choices=["tsv", "xml", "json-cues", "dat", "pgo"],
                    help="override the format (else auto-detected by extension + "
                         "first line); json-cues reads a Rhubarb JSON, not a track")
    fc.add_argument("--fps", type=float,
                    help="frame rate for the rate-less Moho .dat (default 24); "
                         "Papagayo .pgo carries its own and Rhubarb is seconds-"
                         "based (reconstructed at 100 fps), so this is ignored there")
    fc.add_argument("--coarticulate", action="store_true",
                    help="re-solve the stepped cues through the coarticulation "
                         "dominance blend to smooth the hard steps (default: keep "
                         "the stepped [0,1] switches)")
    fc.add_argument("-o", "--out", required=True)
    _add_output_options(fc)
    _add_qa_options(fc)

    cap = sub.add_parser("captions",
                         help="write SRT/WebVTT subtitles from a transcript + "
                              "duration, timed by the SAME alignment the lip "
                              "curves use — captions and lip-sync from one source "
                              "(issue #41)")
    cap.add_argument("--text", required=True, help="the transcript to caption")
    capg = cap.add_mutually_exclusive_group(required=True)
    capg.add_argument("--wav", help="WAV file to read the duration from")
    capg.add_argument("--duration", type=float, help="duration in seconds")
    cap.add_argument("--cmudict", help="optional CMUdict file for better G2P")
    cap.add_argument("-o", "--out", required=True,
                     help="output path: .vtt -> WebVTT, any other extension -> "
                          "SubRip .srt")
    _add_caption_options(cap)

    au = sub.add_parser("audit",
                        help="reconcile a delivered VO folder against a loc-table "
                             "manifest — missing / orphan / duration / empty / "
                             "naming + a language-coverage matrix (read-only QA "
                             "gate, issue #42)")
    au.add_argument("--manifest", required=True, help="the loc-table manifest "
                    "(the #40 CSV/TSV; audio paths resolve under --delivered)")
    au.add_argument("--delivered", required=True,
                    help="the delivered audio folder (read-only)")
    au.add_argument("--duration-tolerance", type=float, default=0.5,
                    dest="duration_tolerance", metavar="FRACTION",
                    help="allowed +/- fraction of the len(text)/CPS length "
                         "estimate before a take is a duration outlier (0.5)")
    au.add_argument("--cps", type=float, default=14.0, metavar="CHARS",
                    help="speaking rate (characters/sec) for the length estimate "
                         "(default 14)")
    au.add_argument("--json", action="store_true",
                    help="emit the full itemized report as JSON instead of the "
                         "human worst-first table")

    cs = sub.add_parser("from-csv",
                        help="import a blendshape-weight CSV into a track: the "
                             "OpenFaceFX long time,channel,value format or a wide "
                             "per-frame ARKit / Live Link Face CSV (issue #45)")
    cs.add_argument("file", help="the blendshape-weight CSV to import")
    cs.add_argument("--fps", type=float, default=60.0,
                    help="frame rate timing the wide per-frame rows "
                         "(frame/timecode -> seconds; default 60). The long "
                         "time,channel,value format carries its own times")
    cs.add_argument("--timecode-col", metavar="NAME",
                    help="column holding a SMPTE 'HH:MM:SS:FF' timecode in a wide "
                         "CSV (else a literal 'Timecode' column, else the row "
                         "index drives the timeline)")
    cs.add_argument("-o", "--out", required=True)
    _add_output_options(cs)
    _add_qa_options(cs)

    cv = sub.add_parser("convert",
                        help="re-export or retarget an existing track.json to any "
                             "format (Unity/Godot/Live2D/cues/.lip/CSV/JSON) "
                             "WITHOUT re-running the solver (issue #46)")
    cv.add_argument("infile", help="the source .track.json to load")
    cv.add_argument("-o", "--out", required=True,
                    help="output; the exporter is chosen by extension, exactly as "
                         "the generate commands' -o (so behaviour cannot drift)")
    cv.add_argument("--fps", type=float,
                    help="re-stamp the track's frame rate before export "
                         "(default: keep the loaded track's fps)")
    _add_output_options(cv)
    _add_edits_options(cv)

    ins = sub.add_parser("inspect",
                         help="read-only stats for a track.json: duration, "
                              "channel/keyframe counts, per-channel coverage, "
                              "event/variant counts (issue #47)")
    ins.add_argument("file", help="the .track.json to inspect")
    ins.add_argument("--json", action="store_true",
                     help="emit the schema-stable JSON stats instead of the "
                          "human table")

    val = sub.add_parser("validate",
                         help="lint a track / *.edits.json / events file against "
                              "the format contract; exits nonzero on a violation "
                              "(a CI gate, issue #47)")
    val.add_argument("file", help="the .track.json, *.edits.json, or events JSON")
    val.add_argument("--strict", action="store_true",
                     help="promote warnings (empty channels, zero-length track) "
                          "to errors")
    val.add_argument("--json", action="store_true",
                     help="emit the deterministic JSON problem list instead of "
                          "the human summary")

    tf = sub.add_parser("transform",
                        help="retime / mirror / trim an existing track.json "
                             "without re-running the solver (issue #48)")
    tf.add_argument("infile", help="the source .track.json to load")
    tf.add_argument("-o", "--out", required=True)
    rt = tf.add_mutually_exclusive_group()
    rt.add_argument("--retime", type=float, metavar="FACTOR",
                    help="scale all keyframe and event times by FACTOR")
    rt.add_argument("--duration", type=float, metavar="D",
                    help="retime so the clip lasts D seconds")
    rt.add_argument("--wav", metavar="WAV",
                    help="retime so the clip matches the WAV's duration")
    tf.add_argument("--anchor", type=float, default=0.0,
                    help="time pinned fixed by --retime/--duration/--wav "
                         "(default 0, i.e. scale from the clip start)")
    tf.add_argument("--mirror", action="store_true",
                    help="swap *Left/*Right channels and negate the signed "
                         "lateral pose channels (headYaw/headRoll/eyeYaw)")
    tf.add_argument("--trim", type=float, nargs=2, metavar=("T0", "T1"),
                    help="keep [T0, T1] seconds and rebase to 0")
    tf.add_argument("--max-channels", type=int, metavar="N",
                    help="keep only the N highest-energy channels (drop the "
                         "low-energy secondary ones); writes a <out>.budget.json "
                         "energy-ranking sidecar (issue #37)")
    _add_output_options(tf)

    sq = sub.add_parser("sequence",
                        help="splice several track.json files end-to-end into one "
                             "timeline, with optional gaps/crossfade (issue #51)")
    sq.add_argument("infiles", nargs="+",
                    help="the .track.json files to concatenate, in order")
    sq.add_argument("--gap", type=float, default=0.0,
                    help="silence in seconds inserted between each segment "
                         "(default 0 = abutting)")
    sq.add_argument("--crossfade", type=float, default=0.0,
                    help="linear crossfade in seconds at each abutting seam "
                         "(default 0 = hard cut)")
    sq.add_argument("-o", "--out", required=True)
    _add_output_options(sq)

    lo = sub.add_parser("lod",
                        help="offline LOD export: K thinned/resampled variants "
                             "of one track, finest first (issue #36)")
    lo.add_argument("infile", help="the source .track.json to derive LODs from")
    lo.add_argument("-o", "--out", required=True, metavar="DIR/PREFIX",
                    help="output path prefix; writes PREFIX_lod0.EXT ... and a "
                         "PREFIX_lod.json metadata sidecar")
    lo.add_argument("--rdp", metavar="E1,E2,..",
                    help="RDP epsilons per tier, finest first "
                         "(default 0.002,0.01,0.04)")
    lo.add_argument("--fps", metavar="F1,F2,..",
                    help="update rate per tier (default 60,30,15); each is capped "
                         "at the source fps (a tier at the source rate is pure RDP)")
    lo.add_argument("--format", choices=["json", "csv"], default="json",
                    help="variant file format (default json); the metadata "
                         "sidecar is always json")
    lo.add_argument("--max-channels", metavar="N1,N2,..",
                    help="per-tier channel budget: keep the N highest-energy "
                         "channels at each LOD (same length as the tiers, higher "
                         "LODs fewer); nested by the source ranking (issue #37)")

    el = sub.add_parser("export-layers",
                        help="write a track.json carrying an additive speech / "
                             "emotion / gesture layer decomposition + blend-weight "
                             "and priority metadata (issue #39)")
    el.add_argument("infile", help="the merged .track.json to decompose")
    el.add_argument("-o", "--out", required=True,
                    help="output .track.json: the flat channels (unchanged) plus a "
                         "top-level 'layers' block")

    df = sub.add_parser("diff",
                        help="read-only A/B drift report between two track.json "
                             "files; exits nonzero on drift over --tolerance "
                             "(a golden-file CI gate, issue #50)")
    df.add_argument("a", help="the first (baseline / golden) .track.json")
    df.add_argument("b", help="the second .track.json to compare against it")
    df.add_argument("--tolerance", type=float, default=0.0,
                    help="max allowed per-delta drift before the exit is nonzero "
                         "(default 0.0 => exact match required)")
    df.add_argument("--json", action="store_true",
                    help="emit the full deterministic JSON report instead of the "
                         "worst-first human table")

    e = sub.add_parser("energy",
                       help="audio loudness -> mouth-open curves (no "
                            "transcript; amplitude fallback, not viseme sync)")
    e.add_argument("--wav", required=True,
                   help="16-bit PCM WAV (mono or stereo; stereo is downmixed). "
                        "Convert other codecs first: ffmpeg -c:a pcm_s16le")
    e.add_argument("--intensity", type=float, default=1.0,
                   help="gain on the mouth opening (1.0 = as-is; >1 opens "
                        "wider on quiet speech, <1 is subtler)")
    e.add_argument("--fps", type=float, default=60.0)
    e.add_argument("-o", "--out", required=True)
    _add_output_options(e)
    _add_smoothing_options(e)
    _add_gesture_options(e)
    _add_event_options(e)
    _add_prosody_options(e)
    _add_edits_options(e)
    _add_qa_options(e)

    lc = sub.add_parser("lip-calibrate",
                        help="EXPERIMENTAL: write one .lip per grid slot "
                             "(slot_NN.lip, single slot swept 0->1->0) to map "
                             "slots to mouth targets in-game (issue #12)")
    lc.add_argument("--out", required=True,
                    help="output directory for slot_NN.lip + README.txt")
    lc.add_argument("--seconds", type=float, default=2.0,
                    help="sweep length per slot (default 2.0)")
    lc.add_argument("--lip-game", default="skyrim", choices=["skyrim"],
                    help="target game (Skyrim only; #12)")

    b = sub.add_parser("batch", help="process a directory tree of voice lines")
    b.add_argument("--dir", help="input tree of .wav files with same-stem "
                   ".TextGrid or .txt transcripts (directory-walk mode; mutually "
                   "exclusive with --manifest)")
    b.add_argument("--manifest", help="a CSV/TSV loc-table driving one track per "
                   "row (issue #40): header-mapped id/audio/text/language/"
                   "character/mapping/style/out columns. Selects manifest mode; "
                   "the directory-walk mode is untouched")
    b.add_argument("--out", required=True, help="mirrored output tree")
    b.add_argument("--recurse", action="store_true")
    b.add_argument("--modified-only", action="store_true",
                   help="skip files unchanged since the last run (manifest)")
    b.add_argument("--jobs", type=int, default=1, help="parallel workers")
    b.add_argument("--ext", choices=["json", "csv"], default="json")
    b.add_argument("--mapping", dest="batch_mapping",
                   help="mapping JSON applied to every file")
    b.add_argument("--cmudict", dest="batch_cmudict",
                   help="CMUdict file for better G2P on naive-path files")
    b.add_argument("--captions", choices=["srt", "vtt"],
                   help="also write an SRT/WebVTT caption sidecar next to each "
                        "naive-mode track, from the same word alignment (issue "
                        "#41). Opt-in: without it the output tree is unchanged")
    b.add_argument("--fps", type=float, default=60.0)
    b.add_argument("--machine-readable", action="store_true",
                   help="stream an NDJSON event log to stderr (one JSON object "
                        "per line: start/progress/warning/failure/done) so a "
                        "supervising process can track a large run live. Events "
                        "are in processing order; stdout and batch_summary.json "
                        "are unchanged")
    b.add_argument("--quiet", action="store_true",
                   help="suppress the human progress table on stdout; the "
                        "batch_summary.json and any --machine-readable/--ledger "
                        "output are still written")
    b.add_argument("--ledger", metavar="FILE",
                   help="append one NDJSON record per run to FILE (args snapshot, "
                        "per-input size/mtime, outcome counts) for reproducibility"
                        "/audit. Survives --modified-only; the run id is a "
                        "deterministic, wall-clock-free hash of the inputs")
    b.add_argument("--cue-warnings", action="store_true",
                   help="add a too-short/too-long phoneme-cue count (qa.cue_flags) "
                        "to each summary row and fold it into the worst-first "
                        "ranking. Opt-in: without it the table and summary JSON "
                        "are byte-identical to before")
    b.add_argument("--min-cue", type=float, default=None, metavar="SECONDS",
                   help="cue shorter than this counts toward --cue-warnings "
                        "(default 0.03; needs --cue-warnings)")
    b.add_argument("--max-cue", type=float, default=None, metavar="SECONDS",
                   help="cue longer than this counts toward --cue-warnings "
                        "(default 0.5; needs --cue-warnings)")

    de = sub.add_parser("diff-edits",
                        help="capture hand-edits: diff a BASE track vs an EDITED "
                             "track into an openfacefx.edits sidecar (issue #9)")
    de.add_argument("base", help="the generated baseline .track.json")
    de.add_argument("edited", help="the hand-edited .track.json")
    de.add_argument("-o", "--out", required=True, help="output .edits.json sidecar")
    de.add_argument("--mode", choices=["offset", "replace"], default="offset",
                    help="offset (default) stores deltas relative to the baseline "
                         "-- survives later intensity/coart/energy changes; "
                         "replace stores absolute edited values (full ownership)")
    de.add_argument("--span", type=float, nargs=2, metavar=("T0", "T1"),
                    help="restrict capture to a time window (a locked region); "
                         "the rest of each channel stays analysis-owned")
    de.add_argument("--source", metavar="WAV",
                    help="WAV whose sha1 keys the sidecar to its audio (source_id)")

    em = sub.add_parser("emotion",
                        help="bake an additive emotion/expression envelope over a "
                             "solved track (reference-pose delta, issue #38)")
    em.add_argument("track", help="the solved .track.json to bake onto")
    em.add_argument("envelope", help="an openfacefx.emotion envelope JSON: direct "
                    "emotion-channel keyframes, or a valence/arousal track")
    em.add_argument("-o", "--out", required=True,
                    help="baked output track (.json/.csv/.anim/.tres/"
                         ".motion3.json/cue formats, like the generate commands)")
    em.add_argument("--intensity", type=float, default=1.0,
                    help="global dial scaling the emotion delta (linear); "
                         "0 => output byte-identical to the input track")
    em.add_argument("--eps", type=float, default=0.015,
                    help="RDP thinning epsilon for the re-thinned baked channels "
                         "(default 0.015, matching the solver)")
    em.add_argument("--clamp", action="append", nargs=3,
                    metavar=("CHANNEL", "LO", "HI"),
                    help="per-channel output clamp 0<=LO<=HI<=1, repeatable; "
                         "overrides the envelope's own clamps")
    _add_output_options(em)

    args = p.parse_args(argv)
    args._warnings = []          # QA summary sink; see _warn / _emit_summary

    if args.cmd == "diff-edits":
        from .io_export import read_json
        from .edits import diff_edits, save_edits, _sha1_source
        try:
            base = read_json(args.base)
            edited = read_json(args.edited)
        except (OSError, ValueError) as ex:
            raise SystemExit(f"diff-edits: {ex}")
        source_id = _sha1_source(args.source) if args.source else None
        doc = diff_edits(base, edited, mode=args.mode,
                         span=tuple(args.span) if args.span else None,
                         source_id=source_id)
        save_edits(doc, args.out)
        print(f"wrote {args.out}: {len(doc.channels)} edited channel(s), "
              f"mode={args.mode}" + (f", span={args.span}" if args.span else ""))
        return 0

    if args.cmd == "lip-calibrate":
        written = lip_calibrate(args.out, game=args.lip_game,
                                seconds=args.seconds)
        print(f"wrote {len(written)} calibration lips to {args.out} — "
              f"EXPERIMENTAL: load each on a voiced line in-game and report "
              f"which mouth part moves on issue #12")
        return 0

    if args.cmd == "batch":
        from .batch import run_batch
        if bool(args.dir) == bool(args.manifest):
            raise SystemExit("batch needs exactly one of --dir or --manifest")
        lo, hi = _cue_thresholds(args) if args.cue_warnings else (None, None)
        return run_batch(args.dir, args.out, recurse=args.recurse,
                         modified_only=args.modified_only, jobs=args.jobs,
                         mapping=args.batch_mapping,
                         cmudict=args.batch_cmudict,
                         fps=args.fps, ext=args.ext,
                         machine_readable=args.machine_readable,
                         quiet=args.quiet, ledger=args.ledger,
                         cue_warnings=args.cue_warnings,
                         min_cue=lo, max_cue=hi,
                         manifest_file=args.manifest,
                         captions=args.captions)

    if args.cmd == "emotion":
        from .io_export import read_json
        from .emotion import bake_emotion, load_envelope
        try:
            track = read_json(args.track)
            env = load_envelope(args.envelope)
        except (OSError, ValueError) as ex:
            raise SystemExit(f"emotion: {ex}")
        try:
            baked = bake_emotion(track, env, intensity=args.intensity,
                                 clamps=_emotion_clamps(args), eps=args.eps)
        except ValueError as ex:
            raise SystemExit(f"emotion: {ex}")
        _write(baked, args.out, args)
        return 0

    if args.cmd == "from-cues":
        from .importers import import_cues
        try:
            track, warnings = import_cues(args.file, fmt=args.format,
                                          fps=args.fps,
                                          coarticulate=args.coarticulate)
        except (OSError, ValueError) as ex:
            raise SystemExit(f"from-cues: {ex}")
        for w in warnings:
            _warn(args, w)
        _write(track, args.out, args)
        _emit_summary(args, track)
        return 0

    if args.cmd == "captions":
        from .export_captions import write_captions
        dur = args.duration if args.duration else wav_duration(args.wav)
        text, _ = normalize_transcript(args.text)   # same fold the naive path uses
        g2p = G2P()
        if args.cmudict:
            g2p.load_cmudict(args.cmudict)
        cues = write_captions(text, dur, args.out, g2p=g2p, karaoke=args.karaoke,
                              cps=args.cps, max_line=args.max_line,
                              max_lines=args.max_lines, gap=args.caption_gap)
        _say(args, f"wrote {args.out}: {len(cues)} caption cue(s)")
        return 0

    if args.cmd == "audit":
        from .vo_audit import audit_delivery, audit_report_text
        try:
            report = audit_delivery(args.manifest, args.delivered,
                                    duration_tolerance=args.duration_tolerance,
                                    cps=args.cps)
        except (OSError, ValueError) as ex:
            raise SystemExit(f"audit: {ex}")
        print(json.dumps(report, indent=2) if args.json
              else audit_report_text(report))
        return 1 if report["counts"]["issues"] else 0    # nonzero = QA gate

    if args.cmd == "from-csv":
        from .importers_csv import read_csv
        try:
            track, warnings = read_csv(args.file, fps=args.fps,
                                       timecode_col=args.timecode_col)
        except (OSError, ValueError) as ex:
            raise SystemExit(f"from-csv: {ex}")
        for w in warnings:
            _warn(args, w)
        _write(track, args.out, args)
        _emit_summary(args, track)
        return 0

    if args.cmd == "convert":
        # Re-serialise an existing track through the SAME edits->_write path the
        # generate commands use (see naive/mfa/from-timing/energy above), so the
        # output is byte-identical to generating that track — by construction, no
        # solver, audio or RNG. Events on the loaded track are preserved as-is.
        from .io_export import read_json
        try:
            track = read_json(args.infile)
        except (OSError, ValueError) as ex:
            raise SystemExit(f"convert: {ex}")
        if args.fps is not None:
            track.fps = args.fps
        track = _apply_edits_layer(track, args)
        _write(track, args.out, args)
        return 0

    if args.cmd == "inspect":
        from .io_export import read_json
        from .inspect import inspect_track, render_inspect
        try:
            track = read_json(args.file)
        except (OSError, ValueError) as ex:
            raise SystemExit(f"inspect: {ex}")
        doc = inspect_track(track)
        print(json.dumps(doc) if args.json else render_inspect(doc))
        return 0

    if args.cmd == "validate":
        from .inspect import validate_file, render_problems
        kind, problems = validate_file(args.file, strict=args.strict)
        ok = not any(p["severity"] == "error" for p in problems)
        if args.json:
            print(json.dumps({"format": "openfacefx.validate", "version": 1,
                              "kind": kind, "ok": ok, "problems": problems}))
        else:
            print(render_problems(kind, problems))
        return 0 if ok else 1

    if args.cmd == "transform":
        from .io_export import read_json
        from . import transforms as _tf
        try:
            track = read_json(args.infile)
        except (OSError, ValueError) as ex:
            raise SystemExit(f"transform: {ex}")
        applied = False
        try:
            if args.retime is not None:
                track = _tf.retime(track, args.retime, anchor=args.anchor)
                applied = True
            elif args.duration is not None:
                track = _tf.retime_to_duration(track, args.duration,
                                               anchor=args.anchor)
                applied = True
            elif args.wav is not None:
                track = _tf.retime_to_duration(track, wav_duration(args.wav),
                                               anchor=args.anchor)
                applied = True
            if args.mirror:
                track = _tf.mirror(track)
                applied = True
            if args.trim is not None:
                track = _tf.trim(track, args.trim[0], args.trim[1])
                applied = True
            ranking = None
            if args.max_channels is not None:      # energy-ranked cap (issue #37)
                from .budget import budget_channels
                track, ranking = budget_channels(track, args.max_channels)
                applied = True
        except (ValueError, OSError) as ex:
            raise SystemExit(f"transform: {ex}")
        if not applied:
            raise SystemExit("transform: choose at least one of "
                             "--retime/--duration/--wav, --mirror, --trim, "
                             "--max-channels")
        _write(track, args.out, args)
        if ranking is not None:
            _write_budget_sidecar(args.out, ranking, args.max_channels, args)
        return 0

    if args.cmd == "sequence":
        from .io_export import read_json
        from .transforms import concat
        try:
            tracks = [read_json(f) for f in args.infiles]
            gaps = [args.gap] * (len(tracks) - 1) if args.gap else None
            out = concat(tracks, gaps=gaps, crossfade=args.crossfade)
        except (OSError, ValueError) as ex:
            raise SystemExit(f"sequence: {ex}")
        _write(out, args.out, args)
        return 0

    if args.cmd == "lod":
        import os
        from .io_export import read_json, write_csv, write_json
        from .lod import generate_lods, lod_metadata
        source_ranking = None
        try:
            track = read_json(args.infile)
            variants, levels = generate_lods(
                track, rdp=_parse_floats(args.rdp, "--rdp"),
                fps=_parse_floats(args.fps, "--fps"))
            budgets = _parse_ints(args.max_channels, "--max-channels")
            if budgets is not None:
                if len(budgets) != len(variants):
                    raise ValueError(
                        f"--max-channels must list one budget per tier "
                        f"({len(variants)}), got {len(budgets)}")
                # rank the SOURCE once so the kept channel sets nest across LODs
                # (pose channels pass through every tier; issue #37)
                from .budget import keep_top_weight, rank_channels
                source_ranking = rank_channels(track)
                variants = [keep_top_weight(v, source_ranking, n)
                            for v, n in zip(variants, budgets)]
        except (OSError, ValueError) as ex:
            raise SystemExit(f"lod: {ex}")
        prefix, ext = args.out, args.format
        outdir = os.path.dirname(prefix)
        if outdir:
            os.makedirs(outdir, exist_ok=True)
        writer = write_csv if ext == "csv" else write_json
        files = []
        for i, v in enumerate(variants):
            path = f"{prefix}_lod{i}.{ext}"
            writer(v, path)
            files.append(os.path.basename(path))
        meta = lod_metadata(track, levels, variants, files)
        if source_ranking is not None:                # issue #37: per-tier budget
            meta["ranking"] = source_ranking
            for i, entry in enumerate(meta["levels"]):
                entry["max_channels"] = budgets[i]
        meta_path = f"{prefix}_lod.json"
        with open(meta_path, "w", encoding="utf-8") as fh:
            json.dump(meta, fh, indent=2)
            fh.write("\n")
        _say(args, f"wrote {len(variants)} LOD variants + {meta_path}: " +
             ", ".join(f"lod{m['index']}={m['keyframes']}kf@{m['fps']}fps"
                       for m in meta["levels"]))
        return 0

    if args.cmd == "export-layers":
        from .io_export import read_json, to_dict
        from .layers import build_layers
        try:
            track = read_json(args.infile)
        except (OSError, ValueError) as ex:
            raise SystemExit(f"export-layers: {ex}")
        layers = build_layers(track)
        with open(args.out, "w", encoding="utf-8") as fh:
            json.dump(to_dict(track, layers=layers), fh, indent=2)
        _say(args, f"wrote {args.out}: flat track + {len(layers)} layer(s) "
             f"({', '.join(l.name for l in layers)})")
        return 0

    if args.cmd == "diff":
        from .io_export import read_json
        from .trackdiff import diff_tracks, render_diff
        try:
            a = read_json(args.a)
            b = read_json(args.b)
        except (OSError, ValueError) as ex:
            raise SystemExit(f"diff: {ex}")
        report = diff_tracks(a, b, tolerance=args.tolerance)
        print(json.dumps(report) if args.json else render_diff(report))
        return 0 if report["ok"] else 1

    mapping = Mapping.from_json(args.mapping) if args.mapping else None
    params = (_coart_params(args)
              if args.cmd in ("naive", "mfa", "from-timing") else None)

    if args.cmd == "naive":
        if bool(args.anchors) != bool(args.anchors_format):
            raise SystemExit(
                "--anchors and --anchors-format are only valid together")
        dur = args.duration if args.duration else wav_duration(args.wav)
        subs = []
        if args.text and not args.no_normalize:
            args.text, subs = normalize_transcript(args.text)
        g2p = G2P()
        if args.cmudict:
            added = g2p.load_cmudict(args.cmudict)
            _say(args, f"loaded {added} CMUdict entries")
        from .texttags import has_tags
        from .ssml import looks_like_ssml, parse_ssml
        # SSML (issue #52) is a thin front-end over the #7 tag path: parse it to
        # the same (clean_text, tags) pair and route it through _naive_tagged.
        # Explicit --ssml or an auto-detected <speak> root; empty tags still land
        # on the byte-identical plain path.
        ssml_clean = ssml_tags = None
        if getattr(args, "ssml", False) or (bool(args.text)
                                            and looks_like_ssml(args.text)):
            ssml_clean, ssml_tags = parse_ssml(args.text or "")
        use_tags = ssml_tags is not None or getattr(args, "tags", False) or (
            bool(args.text) and has_tags(args.text))
        if use_tags:
            if args.anchors:
                raise SystemExit("--tags cannot be combined with --anchors")
            if args.out.endswith(".lip"):
                raise SystemExit(
                    "--tags is not supported for -o .lip output: tags drive "
                    "curves/events, which the phoneme .lip format cannot carry")
            track, segs, clean = _naive_tagged(args, dur, g2p, mapping, params,
                                               clean=ssml_clean, tags=ssml_tags)
            oov = g2p.oov_words(clean) if (clean and _want_summary(args)) else []
            _emit_segments(segs, args)
            _emit_captions(args, dur, g2p, clean)
            track = _apply_edits_layer(track, args)
            _event_layer(track, args, segments=segs)
            _write(track, args.out, args)
            _emit_summary(args, track, segments=segs, oov_words=oov,
                          substitutions=subs)
            return 0
        segs = _naive_input_segments(args, dur, g2p)
        oov = g2p.oov_words(args.text) if (args.text and _want_summary(args)) else []
        _emit_segments(segs, args)
        _emit_captions(args, dur, g2p, args.text)
        if args.out.endswith(".lip"):
            _write_lip(segs, dur, args)
            _emit_summary(args, None, segments=segs, oov_words=oov,
                          substitutions=subs)
        else:
            track = generate_from_alignment(segs, fps=args.fps, mapping=mapping,
                                            params=params,
                                            gestures=_gesture_params(args),
                                            wav=args.wav)
            track = _apply_edits_layer(track, args)
            _event_layer(track, args, segments=segs)
            _write(track, args.out, args)
            _emit_summary(args, track, segments=segs, oov_words=oov,
                          substitutions=subs)
    elif args.cmd == "mfa":
        segs = load_mfa_textgrid(args.textgrid)
        _emit_segments(segs, args)
        if args.out.endswith(".lip"):
            _write_lip(segs, segs[-1].end if segs else 0.0, args)
            _emit_summary(args, None, segments=segs)
        else:
            track = generate_from_alignment(segs, fps=args.fps, mapping=mapping,
                                            params=params,
                                            gestures=_gesture_params(args),
                                            wav=getattr(args, "wav", None))
            track = _apply_edits_layer(track, args)
            _event_layer(track, args, segments=segs)
            _write(track, args.out, args)
            _emit_summary(args, track, segments=segs)
    elif args.cmd == "from-timing":
        parser_fn, is_viseme = _TIMING_PARSERS[args.format]
        with open(args.file, encoding="utf-8") as fh:
            text = fh.read()
        if args.format == "piper":
            if not args.sample_rate:
                raise SystemExit("--sample-rate (Hz) is required for --format piper")
            events = parser_fn(text, args.sample_rate)
        else:
            events = parser_fn(text)
        events = resolve_ends(events, final_duration=args.final_duration)
        if is_viseme:
            if mapping is not None:
                raise SystemExit(
                    f"--mapping does not apply to --format {args.format}: viseme "
                    "formats use the built-in vendor remap preset")
            table = _VISEME_TABLES[args.format]
            segs, warnings = viseme_events_to_segments(events, table)
            for w in warnings:
                _warn(args, w)
            track = generate_from_alignment(segs, fps=args.fps,
                                            mapping=build_vendor_mapping(table),
                                            params=params)
        else:
            segs = to_segments(events)
            active = mapping
            if active is None:
                # pho (MBROLA SAMPA) and piper/cartesia (IPA) don't speak
                # ARPABET, so default them to the built-in IPA preset; an
                # explicit --mapping still wins. Unknown symbols route to
                # silence with a QA warning, once per distinct symbol.
                active = IPA_MAPPING
                for w in ipa_unknown_symbols(e.symbol for e in events):
                    _warn(args, w)
            track = generate_from_alignment(segs, fps=args.fps, mapping=active,
                                            params=params)
        track = _apply_edits_layer(track, args)
        _write(track, args.out, args)
        _emit_summary(args, track, segments=segs)
    elif args.cmd == "energy":
        from .energy import generate_from_energy
        track = generate_from_energy(args.wav, fps=args.fps,
                                     intensity=args.intensity, mapping=mapping,
                                     gestures=_gesture_params(args),
                                     smooth=_smooth_seconds(args),
                                     lag=_lag_seconds(args))
        track = _apply_edits_layer(track, args)
        et = ev = None
        if getattr(args, "events", False):
            from .energy import energy_envelope
            et, ev = energy_envelope(args.wav, fps=args.fps)
        _event_layer(track, args, env_times=et, env=ev)
        _write(track, args.out, args)
        _emit_summary(args, track)
    return 0