#!/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())