#!/usr/bin/env python3 """Recreate two local media-packaging experiments; Python stdlib + FFmpeg only. Run in a fresh directory: python3 reproduce.py --output ./ffmpeg-reproduction Requires ffmpeg with libx264 + AAC and ffprobe on PATH. No network access. """ import argparse import json from pathlib import Path import platform import re import shlex import struct import subprocess import sys from datetime import datetime, timezone parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--output", type=Path, default=Path("ffmpeg-reproduction")) args = parser.parse_args() root = args.output.resolve() root.mkdir(parents=True, exist_ok=True) commands = [] def run(program, *arguments): command = [program, *map(str, arguments)] result = subprocess.run(command, cwd=root, capture_output=True, text=True, check=True) commands.append({"command": shlex.join(command), "exitCode": result.returncode, "stderr": result.stderr}) return result.stdout def save(name, value): (root / name).write_text(json.dumps(value, indent=2) + "\n") def ffmpeg(*arguments): return run("ffmpeg", "-hide_banner", "-loglevel", "error", "-nostdin", "-y", *arguments) def probe(name, *arguments): return json.loads(run("ffprobe", "-v", "error", *arguments, "-of", "json", name)) def boxes(name): """Walk ISO BMFF top-level boxes using declared sizes, not string search.""" data = (root / name).read_bytes() result = [] position = 0 while position < len(data): if position + 8 > len(data): raise ValueError("Incomplete box header") size, kind = struct.unpack_from(">I4s", data, position) header = 8 if size == 1: size = struct.unpack_from(">Q", data, position + 8)[0] header = 16 elif size == 0: size = len(data) - position if size < header or position + size > len(data): raise ValueError("Invalid box size") result.append({"type": kind.decode("ascii"), "offset": position, "size": size}) position += size return result def encode(gop, filename): # Same procedural picture/audio recipe for both GOP variants. They are # independent encodes, not a quality comparison or a second-generation copy. ffmpeg("-f", "lavfi", "-i", "testsrc2=size=320x180:rate=30:duration=8", "-f", "lavfi", "-i", "sine=frequency=660:sample_rate=48000:duration=8", "-map", "0:v:0", "-map", "1:a:0", "-c:v", "libx264", "-preset", "medium", "-crf", "23", "-pix_fmt", "yuv420p", "-g", gop, "-keyint_min", gop, "-sc_threshold", "0", "-bf", "0", "-flags", "+cgop", "-threads:v", "1", "-c:a", "aac", "-b:a", "64k", "-map_metadata", "-1", "-t", "8", filename) version = run("ffmpeg", "-version") (root / "ffmpeg-version.txt").write_text(version) (root / "ffprobe-version.txt").write_text(run("ffprobe", "-version")) encode(90, "ordinary-gop90.mp4") ffmpeg("-i", "ordinary-gop90.mp4", "-map", "0:v:0", "-map", "0:a:0", "-c", "copy", "-movflags", "+faststart", "faststart-gop90.mp4") encode(60, "aligned-gop60.mp4") metadata = {} packets = {} for name in ["ordinary-gop90.mp4", "faststart-gop90.mp4", "aligned-gop60.mp4"]: metadata[name] = probe(name, "-show_streams", "-show_format") packets[name] = probe(name, "-show_packets", "-show_data_hash", "sha256", "-show_entries", "packet=stream_index,pts_time,dts_time,duration_time,flags,size,data_hash") save(name + ".probe.json", metadata[name]) save(name + ".packets.json", packets[name]) ordinary = packets["ordinary-gop90.mp4"]["packets"] faststart = packets["faststart-gop90.mp4"]["packets"] stream_comparison = [] for index, kind in [(0, "video"), (1, "audio")]: original = [p for p in ordinary if p["stream_index"] == index] remuxed = [p for p in faststart if p["stream_index"] == index] stream_comparison.append({ "stream": kind, "inputPacketCount": len(original), "outputPacketCount": len(remuxed), "orderedPacketPayloadHashesEqual": [p["data_hash"] for p in original] == [p["data_hash"] for p in remuxed], "packetTimingAndFlagsEqual": original == remuxed, }) mp4_boxes = {name: boxes(name) for name in ["ordinary-gop90.mp4", "faststart-gop90.mp4"]} hls_results = [] for name, folder in [("ordinary-gop90.mp4", "hls-gop90"), ("aligned-gop60.mp4", "hls-gop60")]: (root / folder).mkdir(exist_ok=True) ffmpeg("-i", name, "-map", "0:v:0", "-map", "0:a:0", "-c", "copy", "-f", "hls", "-hls_time", "2", "-hls_list_size", "0", "-hls_playlist_type", "vod", "-hls_segment_type", "fmp4", "-hls_flags", "independent_segments", "-hls_segment_filename", folder + "/segment-%02d.m4s", folder + "/index.m3u8") playlist = (root / folder / "index.m3u8").read_text() durations = [float(v) for v in re.findall(r"#EXTINF:([0-9.]+)", playlist)] uris = [line for line in playlist.splitlines() if line and not line.startswith("#")] keyframes = [float(p["pts_time"]) for p in packets[name]["packets"] if p["stream_index"] == 0 and "K" in p["flags"]] # Probe and decode the whole local playlist, which includes init.mp4. # A standalone .m4s segment is not a complete MP4. playlist_probe = probe(folder + "/index.m3u8", "-show_streams", "-show_format") save(folder + "/playlist.probe.json", playlist_probe) ffmpeg("-xerror", "-i", folder + "/index.m3u8", "-map", "0:v:0", "-map", "0:a:0", "-f", "null", "-") hls_results.append({"source": name, "playlist": folder + "/index.m3u8", "targetSeconds": 2, "sourceVideoKeyframePtsSeconds": keyframes, "extinfSeconds": durations, "segments": uris, "declaredTargetDuration": int(re.search(r"#EXT-X-TARGETDURATION:(\d+)", playlist)[1]), "allReferencedAssetsPresent": all((root / folder / uri).exists() for uri in ["init.mp4", *uris]), "localPlaylistDecodeExitCode": 0}) checks = { "ordinaryMoovAfterMdat": next(b["offset"] for b in mp4_boxes["ordinary-gop90.mp4"] if b["type"] == "moov") > next(b["offset"] for b in mp4_boxes["ordinary-gop90.mp4"] if b["type"] == "mdat"), "faststartMoovBeforeMdat": next(b["offset"] for b in mp4_boxes["faststart-gop90.mp4"] if b["type"] == "moov") < next(b["offset"] for b in mp4_boxes["faststart-gop90.mp4"] if b["type"] == "mdat"), "payloadsAndTimingPreserved": all(s["orderedPacketPayloadHashesEqual"] and s["packetTimingAndFlagsEqual"] for s in stream_comparison), "gop90SegmentsAre3_3_2": hls_results[0]["extinfSeconds"] == [3.0, 3.0, 2.0], "gop60SegmentsAre2_2_2_2": hls_results[1]["extinfSeconds"] == [2.0, 2.0, 2.0, 2.0], "allPlaylistAssetsPresent": all(h["allReferencedAssetsPresent"] for h in hls_results), } report = { "experiment": "FFMPEG-PACKAGING-2026-09-07", "recordedAt": datetime.now(timezone.utc).isoformat(), "execution": "Native local FFmpeg CLI; no browser, network, CDN, WASM, seek latency or adaptive switching measurements.", "environment": {"system": platform.system(), "release": platform.release(), "machine": platform.machine(), "python": platform.python_version(), "ffmpeg": version.splitlines()[0]}, "source": {"origin": "Procedurally generated by FFmpeg lavfi testsrc2 + sine; no external footage or audio", "durationSeconds": 8, "width": 320, "height": 180, "fps": 30, "audioHz": 660, "sampleRate": 48000, "channels": 1}, "mp4": {"topLevelBoxes": mp4_boxes, "streamComparison": stream_comparison, "fileBytes": {name: (root / name).stat().st_size for name in mp4_boxes}}, "hls": hls_results, "checks": checks, "reproductionNote": "Exact hashes identify this run. Different FFmpeg/libx264 builds may produce different bytes; inspect structural invariants, and retain your version logs. The GOP variants are independent encodes of the same synthetic recipe.", } save("run.json", report) save("commands.json", commands) (root / "commands.sh").write_text("#!/bin/sh\nset -eu\n# Run in a fresh output directory with hls-gop90 and hls-gop60 subdirectories.\nmkdir -p hls-gop90 hls-gop60\n" + "\n".join(c["command"] for c in commands if c["command"].startswith("ffmpeg -hide_banner")) + "\n") print(json.dumps({"output": str(root), "checks": checks, "hls": hls_results}, indent=2)) if not all(checks.values()): sys.exit("Observed results differ from the published invariants; inspect run.json before making a claim.")