"""QuickVeed encode study (October 2026): H.264 vs H.265 vs AV1 (vs VP9) on our own short clips. Reproducible and offline (no network). Run from the repo root: ~/.pyenv/versions/ai_tools_env/bin/python scripts/studies/video_encode_study.py [--ffmpeg PATH] [--runs N] Every *.mp4 in is a source. The decoded source is the reference (our clips are already H.264, so scores measure loss against what we had, not against a camera master). Each source is re-encoded video-only (-an) at every setting in SETTINGS; we record output bytes, kbps, MB per minute, wall-clock encode time (median of N runs), VMAF (libvmaf, default model vmaf_v0.6.1, mean over frames) and SSIM (ffmpeg ssim filter, "All"). One extra run muxes AAC 128 kb/s audio onto the H.264 CRF 23 encode of the clip that has an audio track, to show the audio share of the file. Writes blog/quickveed_content/data/encode_study_2026_10.json. """ import argparse import glob import json import math import os import platform import re import shutil import statistics import subprocess import sys import tempfile import time from datetime import date ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) OUT = os.path.join(ROOT, "blog", "quickveed_content", "data", "encode_study_2026_10.json") # (codec label, encoder, CRF, extra ffmpeg args) SETTINGS = [ ("H.264", "libx264", 18, ["-preset", "medium"]), ("H.264", "libx264", 23, ["-preset", "medium"]), ("H.264", "libx264", 28, ["-preset", "medium"]), ("H.265", "libx265", 20, ["-preset", "medium", "-tag:v", "hvc1", "-x265-params", "log-level=error"]), ("H.265", "libx265", 26, ["-preset", "medium", "-tag:v", "hvc1", "-x265-params", "log-level=error"]), ("H.265", "libx265", 30, ["-preset", "medium", "-tag:v", "hvc1", "-x265-params", "log-level=error"]), ("AV1", "libsvtav1", 30, ["-preset", "8"]), ("AV1", "libsvtav1", 38, ["-preset", "8"]), ("AV1", "libsvtav1", 45, ["-preset", "8"]), ("VP9", "libvpx-vp9", 31, ["-b:v", "0"]), ("VP9", "libvpx-vp9", 37, ["-b:v", "0"]), ] TARGET_VMAF = 94 # preferred matched-quality point (middle of the 93-95 band); see main() def run(cmd): return subprocess.run(cmd, capture_output=True, text=True, check=True) def probe(ffprobe, path): out = run([ffprobe, "-v", "error", "-show_entries", "stream=index,codec_type,codec_name,profile,width,height,r_frame_rate,bit_rate,nb_frames,pix_fmt" ":format=duration,size,bit_rate", "-of", "json", path]).stdout return json.loads(out) def encode_cmd(ffmpeg, src, dst, encoder, crf, extra, audio=False): cmd = [ffmpeg, "-y", "-hide_banner", "-loglevel", "error", "-i", src, "-map", "0:v:0", "-c:v", encoder, "-crf", str(crf), *extra, "-pix_fmt", "yuv420p"] if audio: cmd += ["-map", "0:a:0", "-c:a", "aac", "-b:a", "128k"] else: cmd += ["-an"] return cmd + ["-movflags", "+faststart", dst] def vmaf(ffmpeg, dist, ref, workdir): log = os.path.join(workdir, "vmaf.json") graph = ("[0:v]setpts=PTS-STARTPTS[d];[1:v]setpts=PTS-STARTPTS[r];" f"[d][r]libvmaf=log_fmt=json:log_path={log}:n_threads=8") run([ffmpeg, "-hide_banner", "-loglevel", "error", "-i", dist, "-i", ref, "-lavfi", graph, "-f", "null", "-"]) with open(log) as fh: data = json.load(fh) pooled = data["pooled_metrics"]["vmaf"] return round(pooled["mean"], 2), round(pooled["min"], 2) def ssim(ffmpeg, dist, ref): graph = "[0:v]setpts=PTS-STARTPTS[d];[1:v]setpts=PTS-STARTPTS[r];[d][r]ssim" res = subprocess.run([ffmpeg, "-hide_banner", "-i", dist, "-i", ref, "-lavfi", graph, "-f", "null", "-"], capture_output=True, text=True, check=True) m = re.search(r"All:([0-9.]+)", res.stderr) return round(float(m.group(1)), 4) def interp_at_vmaf(points, target): """points: [(vmaf, bytes)] for one codec. Log-linear interpolation of bytes at target VMAF. Returns (bytes, method) where method is "interpolated" when target lies between two measured points, else None (we do not extrapolate).""" pts = sorted(points) for (v1, b1), (v2, b2) in zip(pts, pts[1:]): if v1 <= target <= v2 and v2 > v1: t = (target - v1) / (v2 - v1) return math.exp(math.log(b1) + t * (math.log(b2) - math.log(b1))), "interpolated" return None, None def lib_version(name): link = f"/opt/homebrew/opt/{name}" if os.path.islink(link): return os.path.basename(os.path.realpath(link)) return None def main(): ap = argparse.ArgumentParser() ap.add_argument("source_dir") ap.add_argument("--ffmpeg", default=shutil.which("ffmpeg") or "/opt/homebrew/bin/ffmpeg") ap.add_argument("--runs", type=int, default=3, help="encode repeats for the timing median") args = ap.parse_args() ffmpeg = args.ffmpeg ffprobe = os.path.join(os.path.dirname(ffmpeg), "ffprobe") sources = sorted(glob.glob(os.path.join(args.source_dir, "*.mp4"))) if not sources: sys.exit("no .mp4 files in " + args.source_dir) ff_version = run([ffmpeg, "-hide_banner", "-version"]).stdout.splitlines()[0] try: cpu = run(["sysctl", "-n", "machdep.cpu.brand_string"]).stdout.strip() mem = int(run(["sysctl", "-n", "hw.memsize"]).stdout.strip()) // (1024 ** 3) except Exception: cpu, mem = platform.processor(), None result = { "study": "QuickVeed encode study: H.264 vs H.265 vs AV1 vs VP9 on our own short clips", "date": date.today().isoformat(), "machine": {"cpu": cpu, "memory_gb": mem, "logical_cpus": os.cpu_count(), "os": f"macOS {platform.mac_ver()[0]}" if platform.mac_ver()[0] else platform.platform()}, "ffmpeg_version": ff_version, "library_versions": {k: lib_version(k) for k in ("x264", "x265", "svt-av1", "libvpx", "libvmaf")}, "metrics": {"vmaf": "libvmaf default model (vmaf_v0.6.1), mean over all frames, n_threads=8", "ssim": "ffmpeg ssim filter, All (Y/U/V weighted)", "reference": "the decoded source file (already H.264), not a camera master", "timing": f"wall clock, median of {args.runs} runs, ffmpeg default threading"}, "preferred_target_vmaf": TARGET_VMAF, "settings": [{"codec": c, "encoder": e, "crf": q, "extra_args": x} for c, e, q, x in SETTINGS], "sources": [], "encodes": [], "matched_quality": [], "audio_run": None, } with tempfile.TemporaryDirectory() as work: for src in sources: name = os.path.basename(src) info = probe(ffprobe, src) v = next(s for s in info["streams"] if s["codec_type"] == "video") has_audio = any(s["codec_type"] == "audio" for s in info["streams"]) duration = float(info["format"]["duration"]) num, den = v["r_frame_rate"].split("/") result["sources"].append({ "file": name, "width": v["width"], "height": v["height"], "fps": round(int(num) / int(den), 3), "frames": int(v.get("nb_frames", 0)), "duration_s": round(duration, 3), "codec": v["codec_name"], "profile": v.get("profile"), "file_bytes": int(info["format"]["size"]), "video_kbps": round(int(v["bit_rate"]) / 1000, 1) if v.get("bit_rate") else None, "has_audio": has_audio, }) # Video-only reference stream size, so ratios compare like with like. ref_v = os.path.join(work, "ref_video.mp4") run([ffmpeg, "-y", "-hide_banner", "-loglevel", "error", "-i", src, "-map", "0:v:0", "-c", "copy", ref_v]) ref_v_bytes = os.path.getsize(ref_v) result["sources"][-1]["video_only_bytes"] = ref_v_bytes for codec, encoder, crf, extra in SETTINGS: dst = os.path.join(work, f"out_{encoder}_{crf}.mp4") cmd = encode_cmd(ffmpeg, src, dst, encoder, crf, extra) times = [] for _ in range(args.runs): t0 = time.perf_counter() run(cmd) times.append(time.perf_counter() - t0) size = os.path.getsize(dst) vm_mean, vm_min = vmaf(ffmpeg, dst, src, work) ss = ssim(ffmpeg, dst, src) enc_s = statistics.median(times) frames = result["sources"][-1]["frames"] row = { "source": name, "codec": codec, "encoder": encoder, "crf": crf, "command": " ".join(["ffmpeg"] + [name if c == src else c for c in cmd[1:-1]] + ["out.mp4"]), "bytes": size, "kbps": round(size * 8 / duration / 1000, 1), "mb_per_min": round(size / duration * 60 / 1e6, 2), "vs_source_video_pct": round(100 * size / ref_v_bytes, 1), "encode_s": round(enc_s, 3), "encode_fps": round(frames / enc_s, 1) if frames else None, "vmaf": vm_mean, "vmaf_min": vm_min, "ssim": ss, } result["encodes"].append(row) print(f"{name:22} {codec:6} crf {crf:2} {size:9d} B {row['kbps']:7.1f} kbps " f"{row['encode_s']:6.2f} s VMAF {vm_mean:6.2f} SSIM {ss}", flush=True) # Matched quality: bytes per codec at one VMAF, log-linear between measured CRFs (no # extrapolation). The target is the whole VMAF nearest TARGET_VMAF that every codec's # tested CRF range reaches on this clip. codecs = ("H.264", "H.265", "AV1", "VP9") ranges = [] for codec in codecs: vs = [r["vmaf"] for r in result["encodes"] if r["source"] == name and r["codec"] == codec] ranges.append((min(vs), max(vs))) lo, hi = max(r[0] for r in ranges), min(r[1] for r in ranges) inside = [t for t in range(60, 100) if lo <= t <= hi] target = min(inside, key=lambda t: abs(t - TARGET_VMAF)) if inside else None result["sources"][-1]["matched_vmaf_target"] = target result["sources"][-1]["vmaf_overlap"] = [lo, hi] for codec in codecs: pts = [(r["vmaf"], r["bytes"]) for r in result["encodes"] if r["source"] == name and r["codec"] == codec] b, method = interp_at_vmaf(pts, target) if target is not None else (None, None) result["matched_quality"].append({ "source": name, "codec": codec, "target_vmaf": target, "method": method, "bytes": round(b) if b else None, "kbps": round(b * 8 / duration / 1000, 1) if b else None, "mb_per_min": round(b / duration * 60 / 1e6, 2) if b else None, }) if has_audio and result["audio_run"] is None: dst = os.path.join(work, "out_audio.mp4") cmd = encode_cmd(ffmpeg, src, dst, "libx264", 23, ["-preset", "medium"], audio=True) run(cmd) info2 = probe(ffprobe, dst) a = next(s for s in info2["streams"] if s["codec_type"] == "audio") vv = next(s for s in info2["streams"] if s["codec_type"] == "video") total = int(info2["format"]["size"]) a_bytes = int(a["bit_rate"]) * float(info2["format"]["duration"]) / 8 result["audio_run"] = { "source": name, "video": "libx264 CRF 23 preset medium", "audio": "AAC 128 kb/s", "command": " ".join(["ffmpeg"] + [name if c == src else c for c in cmd[1:-1]] + ["out.mp4"]), "total_bytes": total, "mb_per_min": round(total / duration * 60 / 1e6, 2), "audio_kbps": round(int(a["bit_rate"]) / 1000, 1), "video_kbps": round(int(vv["bit_rate"]) / 1000, 1), "audio_bytes_est": round(a_bytes), "audio_share_pct": round(100 * a_bytes / total, 1), } # Averages across sources at matched quality, relative to H.264. summary = [] for codec in ("H.265", "AV1", "VP9"): ratios = [] for s in result["sources"]: base = next(m for m in result["matched_quality"] if m["source"] == s["file"] and m["codec"] == "H.264") mine = next(m for m in result["matched_quality"] if m["source"] == s["file"] and m["codec"] == codec) if base["bytes"] and mine["bytes"]: ratios.append(mine["bytes"] / base["bytes"]) summary.append({"codec": codec, "clips_compared": len(ratios), "min_size_vs_h264_pct": round(100 * min(ratios), 1) if ratios else None, "max_size_vs_h264_pct": round(100 * max(ratios), 1) if ratios else None, "mean_size_vs_h264_pct": round(100 * statistics.mean(ratios), 1) if ratios else None, "per_clip_pct": [round(100 * r, 1) for r in ratios]}) # Encode-time ratio vs H.264 at the middle CRF of each codec. mid = {"H.264": 23, "H.265": 26, "AV1": 38, "VP9": 31} times = [] for codec, crf in mid.items(): ratios = [] for s in result["sources"]: base = next(r for r in result["encodes"] if r["source"] == s["file"] and r["codec"] == "H.264" and r["crf"] == 23) mine = next(r for r in result["encodes"] if r["source"] == s["file"] and r["codec"] == codec and r["crf"] == crf) ratios.append(mine["encode_s"] / base["encode_s"]) times.append({"codec": codec, "crf": crf, "mean_time_vs_h264_x": round(statistics.mean(ratios), 2), "per_clip_x": [round(r, 2) for r in ratios]}) result["summary"] = {"matched_quality_vs_h264": summary, "encode_time_vs_h264": times} os.makedirs(os.path.dirname(OUT), exist_ok=True) with open(OUT, "w") as fh: json.dump(result, fh, indent=1) fh.write("\n") print("wrote", OUT) print(json.dumps(result["summary"], indent=1)) if __name__ == "__main__": main()