#!/usr/bin/env python3 """Verify burst-video masters against their render manifest and fully decode them.""" import argparse import csv import json import subprocess from pathlib import Path from typing import Optional def parser() -> argparse.ArgumentParser: value = argparse.ArgumentParser() value.add_argument("batch", type=Path) value.add_argument("--manifest", type=Path) value.add_argument("--codec", choices=("prores", "h264"), required=True) value.add_argument("--profile") value.add_argument("--width", type=int) value.add_argument("--height", type=int) value.add_argument("--pix-fmt") value.add_argument("--minimum-duration", type=float, default=3.0) value.add_argument("--report", type=Path) return value def resolve_tool(name: str) -> str: candidates = ( Path("/opt/homebrew/bin") / name, Path("/usr/local/bin") / name, Path("/usr/bin") / name, ) for candidate in candidates: if not candidate.exists(): continue try: check = subprocess.run( (str(candidate), "-version"), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) except OSError: continue if check.returncode == 0: return str(candidate) raise SystemExit(f"Could not find an executable {name}") def run(*args: str) -> subprocess.CompletedProcess[str]: return subprocess.run(args, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) def rate_value(rate: str) -> float: numerator, denominator = rate.split("/", 1) return float(numerator) / float(denominator) def choose_manifest(batch: Path, explicit: Optional[Path]) -> Path: if explicit: return explicit.resolve() candidates = sorted(batch.glob("*render_manifest.csv")) if len(candidates) != 1: raise SystemExit("Specify --manifest when the batch has zero or multiple render manifests") return candidates[0] def main() -> int: args = parser().parse_args() batch = args.batch.resolve() manifest_path = choose_manifest(batch, args.manifest) report_path = (args.report or batch / "verification_report.md").resolve() json_path = report_path.with_suffix(".json") ffprobe = resolve_tool("ffprobe") ffmpeg = resolve_tool("ffmpeg") with manifest_path.open(newline="", encoding="utf-8") as handle: manifest = list(csv.DictReader(handle)) expected_rows = [ row for row in manifest if row.get("output_file") and row.get("status", "rendered") == "rendered" ] expected = {row["output_file"]: row for row in expected_rows} files = sorted(batch / name for name in expected) results = [] for index, path in enumerate(files, 1): errors = [] stream = {} row = expected[path.name] if not path.is_file(): errors.append("missing output file") else: probe = run( ffprobe, "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=codec_name,profile,width,height,pix_fmt,avg_frame_rate,nb_frames,duration", "-of", "json", str(path), ) if probe.returncode: errors.append(f"ffprobe failed: {probe.stderr.strip()}") else: streams = json.loads(probe.stdout).get("streams", []) if streams: stream = streams[0] else: errors.append("no video stream") if stream: if stream.get("codec_name") != args.codec: errors.append(f"codec={stream.get('codec_name')!r}, expected {args.codec!r}") checks = { "profile": args.profile, "width": args.width, "height": args.height, "pix_fmt": args.pix_fmt, } for key, wanted in checks.items(): if wanted is not None and stream.get(key) != wanted: errors.append(f"{key}={stream.get(key)!r}, expected {wanted!r}") try: fps = rate_value(stream["avg_frame_rate"]) manifest_fps = float(row.get("playback_fps") or row["fps"]) if abs(fps - manifest_fps) > 0.01: errors.append(f"fps={fps:g}, manifest={manifest_fps:g}") except Exception as exc: errors.append(f"invalid frame rate: {exc}") try: duration = float(stream["duration"]) manifest_duration = float(row["duration_seconds"]) if duration < args.minimum_duration: errors.append( f"duration={duration:.3f}s below {args.minimum_duration:g}s minimum" ) if abs(duration - manifest_duration) > 0.02: errors.append( f"duration={duration:.3f}, manifest={manifest_duration:.3f}" ) except Exception as exc: errors.append(f"invalid duration: {exc}") if stream.get("nb_frames") != row.get("output_frames"): errors.append( f"nb_frames={stream.get('nb_frames')}, manifest={row.get('output_frames')}" ) decode_returncode = None if path.is_file(): decode = run( ffmpeg, "-v", "error", "-xerror", "-i", str(path), "-map", "0:v:0", "-f", "null", "-", ) decode_returncode = decode.returncode if decode.returncode: errors.append(f"full decode failed: {decode.stderr.strip()}") results.append({ "file": path.name, "bytes": path.stat().st_size if path.is_file() else 0, "stream": stream, "decode_returncode": decode_returncode, "errors": errors, }) print(f"[{index:02d}/{len(files):02d}] {'PASS' if not errors else 'FAIL'} {path.name}", flush=True) failures = [item for item in results if item["errors"]] overall = bool(expected_rows) and not failures payload = { "overall": "PASS" if overall else "FAIL", "manifest": str(manifest_path), "expected_files": len(expected_rows), "failed_files": len(failures), "results": results, } json_path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") total_bytes = sum(item["bytes"] for item in results) durations = [float(item["stream"]["duration"]) for item in results if item["stream"]] lines = [ "# Burst-video verification", "", f"**Overall: {payload['overall']}**", "", f"- Manifest outputs: {len(expected_rows)}", f"- Fully decoded without error: {len(results) - len(failures)} of {len(results)}", f"- Total size: {total_bytes / (1024 ** 3):.2f} GiB", ] if durations: lines.append(f"- Duration range: {min(durations):.3f} to {max(durations):.3f} seconds") if failures: lines.extend(("", "## Failures", "")) for item in failures: lines.append(f"- {item['file']}: {'; '.join(item['errors'])}") report_path.write_text("\n".join(lines) + "\n", encoding="utf-8") print(f"OVERALL {payload['overall']}") print(f"REPORT {report_path}") return 0 if overall else 1 if __name__ == "__main__": raise SystemExit(main())