mocapgod/ingest.py

#!/usr/bin/env python3
"""Lane A ingest — capture/inbox/*.{mp4,mov} → validate → 30fps h264 → rename → pose queue
→ capture/LEDGER.tsv.

  python3 ingest.py                    # scan capture/inbox/ → local queue/ (the v1 default)
  python3 ingest.py clip.mov --name dj_scratch_01   # one file, explicit clip name
  python3 ingest.py --remote-queue     # ship to the m3ultra queue instead (rsync)
  python3 ingest.py --selfcheck        # run built-in probe/validate self-test, no network

The pose engine (HSMR) currently runs on ultra, so ingest defaults to the LOCAL queue/ dir.
--remote-queue [user@host:/path] rsyncs to m3ultra instead (default target: MOCAP_REMOTE).
Rejects (portrait / malformed) move to capture/rejected/ with a reason in the ledger.
Successful originals move to capture/done/; on a failed remote ship the normalized file is
kept in capture/staged/ so a re-run retries without re-encoding.
"""
import argparse, datetime, json, os, re, shutil, subprocess, sys, tempfile
from pathlib import Path

ROOT     = Path(__file__).resolve().parent
INBOX    = ROOT / "capture" / "inbox"
STAGED   = ROOT / "capture" / "staged"
DONE     = ROOT / "capture" / "done"
REJECTED = ROOT / "capture" / "rejected"
LEDGER   = ROOT / "capture" / "LEDGER.tsv"
QUEUE    = ROOT / "queue"        # v1 default: pose engine (HSMR) runs on ultra, next to the repo
# ponytail: v1 inference on ultra; migrate to m3ultra (--remote-queue) when a real throughput need appears
REMOTE_DEFAULT = os.environ.get("MOCAP_REMOTE", "m3ultra@100.89.131.57:~/Documents/MOCAPGOD/queue")
VIDEO_EXT = {".mp4", ".mov", ".m4v"}
HEADERS  = ["utc", "original", "output", "fps_in", "dur_s", "resolution", "audio", "status", "note"]


def run(cmd):
    return subprocess.run(cmd, capture_output=True, text=True)


def _fps(frac):
    n, _, d = str(frac).partition("/")
    try:
        n, d = float(n), float(d or 1)
        return round(n / d, 3) if d else 0.0
    except ValueError:
        return 0.0


def probe(path):
    """Display-oriented video facts, or raise ValueError with a clear reason."""
    r = run(["ffprobe", "-v", "error", "-print_format", "json",
             "-show_format", "-show_streams", str(path)])
    if r.returncode != 0:
        raise ValueError("ffprobe failed: " + (r.stderr.strip() or "unreadable file"))
    data = json.loads(r.stdout or "{}")
    streams = data.get("streams", [])
    v = next((s for s in streams if s.get("codec_type") == "video"), None)
    if not v:
        raise ValueError("no video stream (malformed)")
    w, h = int(v.get("width", 0)), int(v.get("height", 0))
    # phones store portrait as landscape-pixels + a rotation tag — resolve to display orientation
    rot = 0
    for sd in v.get("side_data_list", []):
        if "rotation" in sd:
            rot = int(sd["rotation"]) % 360
    if "rotate" in v.get("tags", {}):
        rot = int(v["tags"]["rotate"]) % 360
    if rot in (90, 270):
        w, h = h, w
    dur = float(data.get("format", {}).get("duration") or v.get("duration") or 0)
    return {
        "w": w, "h": h,
        "fps": _fps(v.get("avg_frame_rate") or v.get("r_frame_rate")),
        "dur": dur,
        "vcodec": v.get("codec_name", "?"),
        "audio": any(s.get("codec_type") == "audio" for s in streams),
    }


def validate(info):
    """None if shippable, else a human-readable rejection reason."""
    if info["w"] == 0 or info["h"] == 0:
        return "zero dimensions (malformed)"
    if info["h"] > info["w"]:
        return f"portrait {info['w']}x{info['h']} — film landscape, feet in frame (docs/CAPTURE.md)"
    if info["dur"] < 1:
        return f"duration {info['dur']:.1f}s — too short / malformed"
    return None  # no-audio is fine: mocap ignores audio and we strip it anyway


def normalize(src, dst):
    dst.parent.mkdir(parents=True, exist_ok=True)
    r = run(["ffmpeg", "-y", "-i", str(src),
             "-r", "30", "-an", "-c:v", "libx264", "-crf", "18",
             "-preset", "veryfast", "-pix_fmt", "yuv420p", str(dst)])
    if r.returncode != 0:
        tail = (r.stderr.strip().splitlines() or ["?"])[-1]
        raise ValueError("ffmpeg normalize failed: " + tail)


def deliver(staged, remote):
    """Put the normalized clip in the pose queue. Local queue/ dir by default; rsync to a
    remote (user@host:/path) when --remote-queue is used. Returns the destination string."""
    if remote is None:
        QUEUE.mkdir(parents=True, exist_ok=True)
        dst = QUEUE / staged.name
        shutil.move(str(staged), dst)                 # same-fs move; local delivery can't half-fail
        return str(dst)
    host, _, rpath = remote.partition(":")
    run(["ssh", host, f"mkdir -p {rpath}"])           # remote queue dir may not exist yet
    r = run(["rsync", "-az", str(staged), remote + "/"])
    if r.returncode != 0:
        tail = (r.stderr.strip().splitlines() or ["?"])[-1]
        raise ConnectionError(f"rsync to {remote} failed: {tail}")
    return f"{remote}/{staged.name}"


def ledger(row):
    LEDGER.parent.mkdir(parents=True, exist_ok=True)
    fresh = not LEDGER.exists()
    with open(LEDGER, "a") as f:
        if fresh:
            f.write("\t".join(HEADERS) + "\n")
        f.write("\t".join(str(c) for c in row) + "\n")


def slug(s):
    return re.sub(r"[^A-Za-z0-9_-]+", "_", s).strip("_") or "clip"


def process_one(src, name=None, remote=None):
    src = Path(src)
    out_name = f"{datetime.date.today():%Y%m%d}_{slug(name or src.stem)}.mp4"
    ts = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")

    def reject(reason, info=None):
        REJECTED.mkdir(parents=True, exist_ok=True)
        shutil.move(str(src), REJECTED / src.name)
        if info:
            ledger([ts, src.name, "", info["fps"], f"{info['dur']:.1f}",
                    f"{info['w']}x{info['h']}", "yes" if info["audio"] else "no", "REJECT", reason])
        else:
            ledger([ts, src.name, "", "", "", "", "", "REJECT", reason])
        print(f"  ✗ {src.name}: {reason}")
        return False

    try:
        info = probe(src)
    except ValueError as e:
        return reject(str(e))
    bad = validate(info)
    if bad:
        return reject(bad, info)

    res, audio = f"{info['w']}x{info['h']}", "yes" if info["audio"] else "no"
    staged = STAGED / out_name
    try:
        if not staged.exists():          # skip re-encode when retrying a failed remote ship
            normalize(src, staged)
        dest = deliver(staged, remote)
    except (ValueError, ConnectionError) as e:
        ledger([ts, src.name, out_name, info["fps"], f"{info['dur']:.1f}", res, audio, "FAIL", str(e)])
        print(f"  ! {src.name}: {e}  (left in inbox; staged kept for retry)")
        return False

    DONE.mkdir(parents=True, exist_ok=True)
    shutil.move(str(src), DONE / src.name)
    staged.unlink(missing_ok=True)
    ledger([ts, src.name, out_name, info["fps"], f"{info['dur']:.1f}", res, audio,
            "OK", "" if info["audio"] else "audio stripped (none present)"])
    print(f"  ✓ {src.name} → {dest}  ({res} {info['fps']}fps {info['dur']:.1f}s)")
    return True


def selfcheck():
    """Exercise probe + validate on synthesized landscape/portrait clips. No network, no queue."""
    d = Path(tempfile.mkdtemp())
    land, port = d / "land.mp4", d / "port.mp4"
    run(["ffmpeg", "-y", "-f", "lavfi", "-i", "testsrc=size=640x480:rate=60:duration=1", str(land)])
    run(["ffmpeg", "-y", "-f", "lavfi", "-i", "testsrc=size=480x640:rate=50:duration=1", str(port)])
    li, pi = probe(land), probe(port)
    assert (li["w"], li["h"]) == (640, 480), li
    assert round(li["fps"]) == 60, li
    assert validate(li) is None, "landscape should pass"
    assert pi["h"] > pi["w"] and validate(pi) is not None, "portrait should be rejected"
    assert li["audio"] is False, "no-audio clip should probe cleanly"
    shutil.rmtree(d, ignore_errors=True)
    print("selfcheck OK — landscape passes, portrait rejected, no-audio handled")
    return 0


def main():
    ap = argparse.ArgumentParser(description="Lane A: ingest phone videos into the pose queue.")
    ap.add_argument("video", nargs="?", help="single file to ingest (default: scan capture/inbox/)")
    ap.add_argument("--name", help="clip name for the output (default: source filename)")
    ap.add_argument("--remote-queue", nargs="?", const=REMOTE_DEFAULT, default=None,
                    metavar="USER@HOST:PATH",
                    help=f"rsync to a remote pose queue instead of the local queue/ dir "
                         f"(default target: {REMOTE_DEFAULT})")
    ap.add_argument("--selfcheck", action="store_true", help="run the built-in self-test and exit")
    args = ap.parse_args()

    if args.selfcheck:
        return selfcheck()
    dest_label = args.remote_queue if args.remote_queue else str(QUEUE)
    if args.video:
        return 0 if process_one(args.video, args.name, args.remote_queue) else 1

    INBOX.mkdir(parents=True, exist_ok=True)
    vids = sorted(p for p in INBOX.iterdir() if p.suffix.lower() in VIDEO_EXT)
    if not vids:
        print(f"inbox empty: {INBOX}")
        return 0
    print(f"ingesting {len(vids)} file(s) from {INBOX} → {dest_label}")
    n_ok = sum(process_one(p, remote=args.remote_queue) for p in vids)
    print(f"done: {n_ok}/{len(vids)} ok")
    return 0 if n_ok == len(vids) else 1


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