"""Upload a video once, then turn it into the two things the app actually reads. WHAT CHANGED AND WHY. This used to decode one PNG per source frame and store every one of them. A 7.6-second 1440x1920 take is 112MB that way, and the 900-frame limit is 1.1GB — for pixels whose only consumer was a canvas that MediaPipe then read once. The page now detects from the video itself (see `frontend/src/arthur/flow/ ingest.cljs`), so this produces: THE PROXY. One browser-safe H.264/yuv420p MP4, CFR, `+faststart`. The same take is 6MB. This is the analysis source, and it is re-encoded RATHER THAN KEPT AS UPLOADED even when the upload is already H.264, for two reasons that are both about not guessing: an iPhone's HEVC is not decodable in every browser, and the footage's identity is the digest of this file — one produced by one ffmpeg invocation, not one that depends on which branch the source happened to take. THE TRACING STILLS. One JPEG per frame, long edge capped, for the tracing editor to draw over. Reference images; nothing measures them. They are not in the footage digest — see `models.Footage`. The proxy is probed after it is written rather than before. `width`, `height` and `frames` are properties of the file the browser will decode, and taking them from the source instead is how a scaler or a dropped frame becomes a silent one-frame offset between the landmarks and the audio. """ import hashlib import json import subprocess import tempfile import threading import time from fractions import Fraction from pathlib import Path from django.db import close_old_connections, transaction from . import blobs from .models import Blob, Extraction, Footage, FootageFrame _active = set() _lock = threading.Lock() TIMEOUT = 3600 # Visually lossless enough that landmarks do not move: measured against the same # frames as PNGs, IMAGE-mode landmarks shifted at most 0.0033 of frame width. PROXY_CRF = "18" # The long edge of a tracing still. The proxy keeps full resolution because the # detector reads it; a still only has to be good enough to draw a cel over. TRACING_EDGE = 1280 TRACING_QUALITY = "4" def _command(args): result = subprocess.run(args, capture_output=True, text=True, timeout=TIMEOUT) if result.returncode: raise ValueError((result.stderr or result.stdout or "media tool failed")[-1200:]) return result.stdout def _run_with_progress(job, args, root, name, total, span): """Run one ffmpeg and publish its live frame count as `span` of the job. ffmpeg's `-progress` file is the only honest source for this: parsing its stderr means parsing a format that is explicitly not an interface, and a spinner that is not attached to frames is a spinner that lies on a long take. """ progress_path = root / f"{name}.progress" log_path = root / f"{name}.log" first, last = span args = ["ffmpeg", "-hide_banner", "-loglevel", "error", "-y", "-stats_period", "0.25", "-progress", str(progress_path)] + args with open(log_path, "wb") as log: proc = subprocess.Popen(args, stdout=log, stderr=subprocess.STDOUT) deadline = time.monotonic() + TIMEOUT try: while proc.poll() is None: if time.monotonic() >= deadline: raise TimeoutError(f"{name} timed out") if progress_path.exists(): lines = progress_path.read_text(errors="replace").splitlines() count = next((int(line[6:].strip()) for line in reversed(lines) if line.startswith("frame=") and line[6:].strip().isdigit()), 0) if count and total: reached = first + int((last - first) * min(1.0, count / total)) if reached > job.progress: job.progress = reached job.save(update_fields=["progress", "updated"]) time.sleep(0.2) finally: if proc.poll() is None: proc.kill() proc.wait() if proc.returncode: raise ValueError(log_path.read_text(errors="replace")[-1200:] or f"{name} failed") def _encode_proxy(job, source_path, proxy_path, facts, root): """The uploaded video -> one H.264 file every browser can decode and seek.""" total = facts.get("reported_frames") or round(facts["duration"] * facts["fps"]) _run_with_progress( job, ["-i", str(source_path), "-an", # Constant frame rate at the source's own rate. `probe` has already # refused VFR, so this asserts that rather than resampling. "-fps_mode", "cfr", "-r", str(facts["fps"]), "-c:v", "libx264", "-preset", "veryfast", "-crf", PROXY_CRF, # yuv420p and an even frame size are what makes this playable everywhere # rather than only in the browser that happened to be tested. "-pix_fmt", "yuv420p", "-vf", "scale=trunc(iw/2)*2:trunc(ih/2)*2", "-movflags", "+faststart", str(proxy_path)], root, "proxy", total, (0, 55)) def _extract_stills(job, proxy_path, frames_dir, frames, root): """The proxy -> one tracing JPEG per frame, long edge capped.""" _run_with_progress( job, ["-i", str(proxy_path), "-fps_mode", "passthrough", "-vf", f"scale='if(gt(iw,ih),min({TRACING_EDGE},iw),-2)':" f"'if(gt(iw,ih),-2,min({TRACING_EDGE},ih))'", "-q:v", TRACING_QUALITY, str(frames_dir / "%04d.jpg")], root, "stills", frames, (55, 85)) def probe(path): data = json.loads(_command(["ffprobe", "-v", "error", "-show_streams", "-show_format", "-of", "json", str(path)])) video = next((s for s in data.get("streams", []) if s.get("codec_type") == "video"), None) if not video: raise ValueError("the uploaded file has no video stream") nominal = Fraction(video.get("r_frame_rate") or "0") average = Fraction(video.get("avg_frame_rate") or "0") if nominal <= 0 or average <= 0: raise ValueError("the video's frame rate is unknown") vfr = abs(float(nominal / average) - 1) > 0.001 if vfr: raise ValueError("variable-frame-rate video needs timestamp-aware playback") frames = video.get("nb_frames") duration = float(data.get("format", {}).get("duration") or 0) if ((frames and frames.isdigit() and int(frames) > 900) or (duration > 0 and duration * float(average) > 901)): raise ValueError("video is longer than the 900-frame footage limit") return {"fps": float(average), "nominal_fps": float(nominal), "width": int(video["width"]), "height": int(video["height"]), "duration": duration, "reported_frames": int(frames) if frames and frames.isdigit() else None, "has_audio": any(s.get("codec_type") == "audio" for s in data.get("streams", [])), "vfr": False} def count_frames(path): """How many frames a file really holds, counted rather than reported. `nb_frames` is a container's claim. This is the decoder's answer, and it is what the page will get when it walks the proxy — so a disagreement between the two has to be settled before the count reaches a manifest, not after it has become a one-frame audio offset nobody can find. """ text = _command(["ffprobe", "-v", "error", "-select_streams", "v:0", "-count_frames", "-show_entries", "stream=nb_read_frames", "-of", "default=nokey=1:noprint_wrappers=1", str(path)]) counted = text.strip() if not counted.isdigit(): raise ValueError("could not count the proxy's frames") return int(counted) def extraction_key(source, settings): text = json.dumps({"scheme": 2, "source": source.blob_id, "settings": settings}, sort_keys=True, separators=(",", ":")) return "sha256:" + hashlib.sha256(text.encode()).hexdigest() def _register(job, proxy_path, stills, audio_path, facts): proxy_digest, proxy_size = blobs.adopt(proxy_path) audio_digest, audio_size = blobs.adopt(audio_path) still_blobs = [(index, *blobs.adopt(path)) for index, path in enumerate(stills)] width, height, fps, frames = facts["width"], facts["height"], facts["fps"], facts["frames"] # The footage's own identity: the bytes the page will measure, the audio it # will clock against, and the rate that ties them together. Scheme 2 — scheme # 1 hashed a PNG per frame, and those footages name pixels this no longer has. h = hashlib.sha256() h.update(f"arthur-footage-2/{fps}/{frames}/{width}x{height}\n".encode()) h.update(proxy_digest.encode()) h.update(audio_digest.encode()) with transaction.atomic(): proxy_blob, _ = Blob.objects.get_or_create( digest=proxy_digest, defaults={"size": proxy_size, "media_type": "video/mp4"}) audio_blob, _ = Blob.objects.get_or_create( digest=audio_digest, defaults={"size": audio_size, "media_type": "audio/wav"}) footage, created = Footage.objects.get_or_create( digest=h.hexdigest(), defaults={"label": job.source.filename[:200], "source": job.source.filename[:200], "fps": fps, "frames": frames, "width": width, "height": height, "audio": audio_blob, "video": proxy_blob}) if created: rows = [] for index, digest, size in still_blobs: blob, _ = Blob.objects.get_or_create( digest=digest, defaults={"size": size, "media_type": "image/jpeg"}) rows.append(FootageFrame(footage=footage, index=index, blob=blob)) FootageFrame.objects.bulk_create(rows) return footage def run(key): close_old_connections() try: job = Extraction.objects.select_related("source", "source__blob").get(key=key) job.state, job.progress, job.error = "running", 0, "" job.save(update_fields=["state", "progress", "error", "updated"]) facts = job.source.probe source_path = blobs.path_for(job.source.blob_id) with tempfile.TemporaryDirectory(prefix="arthur-extract-") as directory: root = Path(directory) proxy_path = root / "proxy.mp4" _encode_proxy(job, source_path, proxy_path, facts, root) # Everything downstream describes the PROXY, not the upload. proxy_facts = probe(proxy_path) frames = count_frames(proxy_path) if not 1 <= frames <= 900: raise ValueError(f"the proxy holds {frames} frames; the limit is 1–900") expected = facts.get("reported_frames") if expected and frames != expected: raise ValueError( f"the proxy holds {frames} frames and the upload reports {expected}; " "refusing footage whose picture and audio would drift") proxy_facts["frames"] = frames frames_dir = root / "stills" frames_dir.mkdir() _extract_stills(job, proxy_path, frames_dir, frames, root) stills = sorted(frames_dir.glob("*.jpg")) if len(stills) != frames: raise ValueError(f"wrote {len(stills)} tracing stills for {frames} frames") job.progress = 85 job.save(update_fields=["progress", "updated"]) audio_path = root / "audio.wav" if facts["has_audio"]: _command(["ffmpeg", "-hide_banner", "-loglevel", "error", "-y", "-i", str(source_path), "-vn", "-ac", "1", "-ar", "44100", str(audio_path)]) else: _command(["ffmpeg", "-hide_banner", "-loglevel", "error", "-y", "-f", "lavfi", "-i", "anullsrc=r=44100:cl=mono", "-t", str(frames / proxy_facts["fps"]), "-c:a", "pcm_s16le", str(audio_path)]) footage = _register(job, proxy_path, stills, audio_path, proxy_facts) job.footage, job.state, job.progress = footage, "done", 100 job.save(update_fields=["footage", "state", "progress", "updated"]) except Exception as exc: Extraction.objects.filter(key=key).update(state="failed", error=str(exc)[:2000]) finally: with _lock: _active.discard(key) close_old_connections() def enqueue(key): with _lock: if key in _active: return _active.add(key) threading.Thread(target=run, args=(key,), daemon=True, name=f"arthur-extract-{key[7:15]}").start()