Skip to content

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.

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

SUMMARY_NAME = 'batch_summary.json' 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') -> int

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

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") -> int:
    """Returns a process exit code (0 = all ok, 1 = at least one failure)."""
    work = find_jobs(in_dir, out_dir, recurse, ext)
    if not work:
        print(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):
        return {
            "wav": _stamp(job["wav"]),
            "transcript": _stamp(job["textgrid"] or job["txt"] or ""),
            "mapping": _stamp(mapping) if mapping else None,
            "out": job["out"],
        }

    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)

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

    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
    rows.sort(key=lambda r: (r["status"] == "ok",
                             r["min_confidence"] if r["min_confidence"]
                             is not None else 2.0,
                             -len(r["oov"])))
    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)

    width = max([len(r["file"]) for r in rows] + [4])
    print(f"{'file':<{width}}  {'status':<7} {'mode':<5} {'dur':>6} "
          f"{'keys':>5}  oov")
    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 "")
        print(f"{r['file']:<{width}}  {r['status']:<7} {r['mode'] or '-':<5} "
              f"{dur:>6} {r['keyframes']:>5}  {oov}")
        if r["error"]:
            print(f"{'':<{width}}  ! {r['error']}")
    print(f"\n{len(rows)} processed, {skipped} skipped (unchanged), "
          f"{summary['failed']} failed -> {os.path.join(out_dir, SUMMARY_NAME)}")
    return 1 if summary["failed"] else 0

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)")
    _add_output_options(n)
    _add_coart_options(n)
    _add_gesture_options(n)
    _add_event_options(n)
    _add_prosody_options(n)

    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_gesture_options(m)
    _add_event_options(m)
    _add_prosody_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)

    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_gesture_options(e)
    _add_event_options(e)
    _add_prosody_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", required=True, help="input tree of .wav files "
                   "with same-stem .TextGrid or .txt transcripts")
    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("--fps", type=float, default=60.0)

    args = p.parse_args(argv)

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

    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)
        g2p = G2P()
        if args.cmudict:
            added = g2p.load_cmudict(args.cmudict)
            print(f"loaded {added} CMUdict entries")
        segs = _naive_input_segments(args, dur, g2p)
        _emit_segments(segs, args)
        if args.out.endswith(".lip"):
            _write_lip(segs, dur, args)
        else:
            track = generate_from_alignment(segs, fps=args.fps, mapping=mapping,
                                            params=params,
                                            gestures=_gesture_params(args),
                                            wav=args.wav)
            _event_layer(track, args, segments=segs)
            _write(track, args.out, args)
    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)
        else:
            track = generate_from_alignment(segs, fps=args.fps, mapping=mapping,
                                            params=params,
                                            gestures=_gesture_params(args),
                                            wav=getattr(args, "wav", None))
            _event_layer(track, args, segments=segs)
            _write(track, args.out, args)
    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:
                print(f"warning: {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):
                    print(f"warning: {w}")
            track = generate_from_alignment(segs, fps=args.fps, mapping=active,
                                            params=params)
        _write(track, args.out, args)
    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))
        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)
    return 0