mocapgod/mocap

#!/usr/bin/env python3
"""MOCAPGOD Lane D — `mocap`: one command, phone video → game-ready mocap clip.

    ./mocap capture/inbox/vid.mp4 --name dj_scratch_01
    ./mocap spin.mov --name spin --character male --travel

Chains the three proven stages, all on ultra:
  Lane A  ingest.py       video → 30fps h264 → queue/<date>_<name>.mp4      (MOVES the source)
  Lane B  pose_engine.py  video → HSMR/SKEL → out/<stem>/motion.bvh + sidecar  [HSMR venv, MPS]
  Lane C  retarget.py     BVH → mixamorig → clips_out/<name>.fbx + <name>_qc.mp4 [headless Blender]

Result: clips_out/<name>.fbx (anim-only, Mixamo "Without Skin") + <name>_qc.mp4 turntable.
House rule: eyeball the QC mp4 before the clip enters a bank.

Deferred hooks (off by default — built when John's real footage lands; see docs/RUNBOOK.md):
  clip-bank GLB merge, --to-unreal, --publish-3god. merge_anims.py needs a CLI + the Blender-5
  slot-bind first, and none of them are verifiable without a real clip, so they're not faked here.
"""
import argparse, datetime, re, subprocess, sys
from pathlib import Path

REPO    = Path(__file__).resolve().parent
QUEUE   = REPO / "queue"
OUT     = REPO / "out"
CLIPS   = REPO / "clips_out"
VENV_PY = REPO / ".engine/HSMR/.venv/bin/python"          # Lane B lives in its own env
BLENDER = Path("/Applications/Blender.app/Contents/MacOS/Blender")
KIT     = Path.home() / "Documents/character_kit"
RIGS = {                                                  # --character → (pose gender, retarget rig)
    "female": ("female", KIT / "female/female_game.glb"),  # v1 proven path
    "male":   ("male",   KIT / "hum_character.glb"),        # ponytail: male rig unverified until a male clip exists
}


def slug(s):                                              # mirrors ingest.slug (kept standalone on purpose)
    return re.sub(r"[^A-Za-z0-9_-]+", "_", s).strip("_") or "clip"


def paths_for(name):
    """The naming contract that threads a clip through all three stages. The one bit of real
    logic here, so it gets the self-check below."""
    stem = f"{datetime.date.today():%Y%m%d}_{name}"       # ingest's queue naming
    return stem, QUEUE / f"{stem}.mp4", OUT / stem / "motion.bvh", CLIPS / f"{name}.fbx"


def stage(title, cmd):
    print(f"\n▶ {title}\n  $ {' '.join(str(c) for c in cmd)}", flush=True)
    if subprocess.run([str(c) for c in cmd]).returncode != 0:
        sys.exit(f"✗ {title} failed — chain stopped.")


def selfcheck():
    """Path-threading only: no engine, no network, no Blender."""
    stem, q, bvh, fbx = paths_for("dj_scratch_01")
    assert stem.endswith("_dj_scratch_01"), stem
    assert q == QUEUE / f"{stem}.mp4", q
    assert bvh == OUT / stem / "motion.bvh", bvh
    assert fbx == CLIPS / "dj_scratch_01.fbx", fbx        # final name stays clean (retarget --name)
    assert slug("DJ Scratch #1!") == "DJ_Scratch_1", slug("DJ Scratch #1!")
    for c in RIGS.values():
        assert c[0] in ("male", "female")
    print("selfcheck OK — name threads inbox→queue→out→clips, final clip name stays clean")
    return 0


def main():
    ap = argparse.ArgumentParser(description="Lane D: one command, video → game-ready mocap clip.")
    ap.add_argument("video", nargs="?", help="source clip (e.g. capture/inbox/move.mp4) — ingest MOVES it")
    ap.add_argument("--name", help="clip name → clips_out/<name>.fbx")
    ap.add_argument("--character", choices=list(RIGS), default="female")
    ap.add_argument("--gender", choices=["male", "female"], help="override pose gender (default: from --character)")
    ap.add_argument("--travel", action="store_true", help="keep world root translation (default: in-place)")
    ap.add_argument("--no-qc", action="store_true", help="skip the QC turntable render")
    ap.add_argument("--selfcheck", action="store_true", help="verify the path-threading logic and exit")
    a = ap.parse_args()

    if a.selfcheck:
        return selfcheck()
    if not a.video or not a.name:
        ap.error("video and --name are required (or use --selfcheck)")
    if not VENV_PY.exists():
        sys.exit(f"✗ HSMR venv missing: {VENV_PY} — run .engine/setup_hsmr.sh (Lane B).")
    if not BLENDER.exists():
        sys.exit(f"✗ Blender not found: {BLENDER} (Lane C needs it).")

    name = slug(a.name)
    gender, rig = RIGS[a.character]
    gender = a.gender or gender
    if not rig.exists():
        sys.exit(f"✗ rig for --character {a.character} not found: {rig}")
    stem, queued, bvh, fbx = paths_for(name)

    stage("Lane A · ingest", ["python3", REPO / "ingest.py", a.video, "--name", name])
    if not queued.exists():
        sys.exit(f"✗ no {queued} after ingest — clip rejected? check capture/LEDGER.tsv")

    stage("Lane B · pose_engine (HSMR/SKEL, MPS — slow, ~1.6× realtime)",
          [VENV_PY, REPO / "pose_engine.py", queued, "--gender", gender])
    if not bvh.exists():
        sys.exit(f"✗ no {bvh} after pose_engine.")

    rt = [BLENDER, "--background", "--python", REPO / "retarget.py", "--",
          "--bvh", bvh, "--rig", rig, "--name", name]
    if a.travel: rt.append("--travel")
    if a.no_qc:  rt.append("--no-qc")
    stage("Lane C · retarget + QC (Blender)", rt)
    if not fbx.exists():
        sys.exit(f"✗ no {fbx} after retarget.")

    qc = CLIPS / f"{name}_qc.mp4"
    print(f"\n✅ clip ready: {fbx}")
    if qc.exists():
        print(f"   eyeball before it enters a bank: {qc}")
    print("   merge-to-GLB / Unreal / 3god hooks are deferred — docs/RUNBOOK.md.")
    return 0


if __name__ == "__main__":
    sys.exit(main())