Lane A: capture protocol + ingest pipeline
docs/CAPTURE.md — one-page filming protocol (landscape, feet in frame, T-pose bookends, 5-30s takes, AirDrop to capture/inbox/). ingest.py — scans capture/inbox/, probes each clip (ffprobe, display-orientation aware so phone rotation tags are respected), rejects portrait/malformed to capture/rejected/ with a reason, normalizes the rest to 30fps h264 (audio stripped), renames YYYYMMDD_<name>.mp4, rsyncs to the m3ultra pose queue, and logs every disposition to capture/LEDGER.tsv. Failed ships keep the normalized file in capture/staged/ so a re-run retries without re-encoding. --selfcheck exercises probe+validate on synthesized landscape/portrait clips. Verified end-to-end: inbox drop -> one command -> clip in m3ultra queue, portrait rejected with clear error, ledger rows written. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
type-two
parent
619d35cc79
commit
468f35be7d
36
docs/CAPTURE.md
Normal file
36
docs/CAPTURE.md
Normal file
@ -0,0 +1,36 @@
|
||||
# Filming protocol — read once, keep on the tripod
|
||||
|
||||
One page. Follow it and the pose engine has an easy time; break it and Lane C fights soup.
|
||||
|
||||
## The shot
|
||||
- **Phone landscape**, on a tripod or leaned so it can't move. Never handheld, never portrait.
|
||||
- **1080p, 60fps.** 4K is wasted — pose models downscale anyway. 60fps → we resample to 30.
|
||||
- **Whole body in frame the entire take — feet included.** If a foot leaves the frame, the
|
||||
ground contact is lost and the clip drifts. Frame with headroom top and bottom.
|
||||
- **2–3 m from the phone.** Close enough to fill the frame, far enough that a step doesn't
|
||||
clip you out.
|
||||
- **Even light, plain-ish background.** No harsh single-side shadow, no busy patterns behind you.
|
||||
- **Fitted clothes.** Baggy hoodies and swishy pants = joint error. Snug beats stylish here.
|
||||
|
||||
## The take
|
||||
1. **Start in a T-pose, hold ~1 s.** Arms straight out, legs together, facing the phone.
|
||||
2. Do **one move** — a walk, a DJ scratch, a pick-up. One move per take.
|
||||
3. **End in a T-pose, hold ~1 s.**
|
||||
4. Keep takes **5–30 s.** Short and clean beats long and mushy. Re-film rather than salvage.
|
||||
|
||||
The T-pose bookends are not optional: they're the calibration reference and the anchor Lane C
|
||||
uses to auto-trim the clip. No bookends → manual trimming.
|
||||
|
||||
## Getting it to the pipeline
|
||||
- **AirDrop** the clip from the phone to **ultra**, save into
|
||||
`~/Documents/MOCAPGOD/capture/inbox/`.
|
||||
- Then run **`python3 ingest.py`** (or let the Lane D `mocap` CLI do it). Ingest validates,
|
||||
normalizes to 30fps, and ships it to the m3ultra queue. Rejects (portrait, no feet, garbage)
|
||||
land in `capture/rejected/` with a reason in `capture/LEDGER.tsv` — check there if a clip vanishes.
|
||||
|
||||
## Common rejections (and the fix)
|
||||
| Symptom | Fix |
|
||||
|---|---|
|
||||
| `portrait` | Rotate the phone. Landscape only. |
|
||||
| `too short / malformed` | Re-film; hold the T-poses, aim for 5–30 s. |
|
||||
| drifty / foot-skating result | Feet left the frame, or baggy clothes — re-film closer to protocol. |
|
||||
200
ingest.py
Normal file
200
ingest.py
Normal file
@ -0,0 +1,200 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Lane A ingest — capture/inbox/*.{mp4,mov} → validate → 30fps h264 → rename → rsync to
|
||||
the m3ultra pose queue → capture/LEDGER.tsv.
|
||||
|
||||
python3 ingest.py # scan capture/inbox/ and ship everything
|
||||
python3 ingest.py clip.mov --name dj_scratch_01 # one file, explicit clip name
|
||||
python3 ingest.py --selfcheck # run built-in probe/validate self-test, no network
|
||||
|
||||
Rejects (portrait / malformed) move to capture/rejected/ with a reason in the ledger.
|
||||
Successful originals move to capture/done/; on rsync failure the normalized file is kept
|
||||
in capture/staged/ so a re-run retries the ship without re-encoding.
|
||||
Remote defaults to m3ultra; override with MOCAP_REMOTE=user@host:/path.
|
||||
"""
|
||||
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"
|
||||
REMOTE = 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 ship(staged):
|
||||
host, _, rpath = REMOTE.partition(":")
|
||||
run(["ssh", host, f"mkdir -p {rpath}"]) # 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}")
|
||||
|
||||
|
||||
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):
|
||||
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 ship
|
||||
normalize(src, staged)
|
||||
ship(staged)
|
||||
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} → {REMOTE}/{out_name} ({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 m3ultra 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("--selfcheck", action="store_true", help="run the built-in self-test and exit")
|
||||
args = ap.parse_args()
|
||||
|
||||
if args.selfcheck:
|
||||
return selfcheck()
|
||||
if args.video:
|
||||
return 0 if process_one(args.video, args.name) 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} → {REMOTE}")
|
||||
n_ok = sum(process_one(p) 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())
|
||||
Loading…
Reference in New Issue
Block a user