diff --git a/scripts/export_public_demo_evidence.py b/scripts/export_public_demo_evidence.py index 584e001f..1094be88 100644 --- a/scripts/export_public_demo_evidence.py +++ b/scripts/export_public_demo_evidence.py @@ -355,6 +355,55 @@ def _probe_frame_presentation_times_us( return presentation_times_us +def _probe_video_sample_end_us( + ffprobe: str, + media_path: Path, +) -> int: + """Return the exact end of the final encoded video sample. + + A frame timestamp identifies the start of a sample, not the media duration. + The public viewer must retain the complete terminal hold, so bind the + timeline to ``max(PTS + duration)`` across the encoded video packets. + """ + + result = subprocess.run( + [ + ffprobe, + "-v", + "error", + "-select_streams", + "v:0", + "-show_entries", + "packet=pts_time,duration_time", + "-of", + "json", + str(media_path), + ], + check=True, + capture_output=True, + text=True, + ) + payload = json.loads(result.stdout) + raw_packets = payload.get("packets") if isinstance(payload, dict) else None + if not isinstance(raw_packets, list) or not raw_packets: + raise EvidencePackError(f"ffprobe found no video packets in {media_path}") + sample_end_us = 0 + for item in raw_packets: + pts = item.get("pts_time") if isinstance(item, dict) else None + duration = item.get("duration_time") if isinstance(item, dict) else None + if not isinstance(pts, str) or not isinstance(duration, str): + raise EvidencePackError(f"ffprobe omitted packet timing in {media_path}") + end_us = (Decimal(pts) + Decimal(duration)) * Decimal(1_000_000) + if end_us != end_us.to_integral_value(): + raise EvidencePackError( + f"ffprobe returned a sub-microsecond sample end in {media_path}" + ) + sample_end_us = max(sample_end_us, int(end_us)) + if sample_end_us <= 0: + raise EvidencePackError(f"ffprobe returned an empty duration for {media_path}") + return sample_end_us + + def _write_presentation_clip( *, capture: _PresentationCapture, @@ -397,7 +446,7 @@ def _write_presentation_clip( ffmpeg, ffprobe = _presentation_tools() presentation_dir.mkdir(parents=True, exist_ok=True) - media_path = presentation_dir / f"{clip_id}.webm" + media_path = presentation_dir / f"{clip_id}.mp4" timeline_path = presentation_dir / f"{clip_id}.control-overlay.v2.json" pts_path = presentation_dir / f"{clip_id}.frame-pts-us.json" with tempfile.TemporaryDirectory( @@ -453,17 +502,23 @@ def _write_presentation_clip( str(concat_path), "-an", "-fps_mode", - "vfr", + "passthrough", + "-enc_time_base", + "demux", "-c:v", - "libvpx-vp9", - "-lossless", - "1", - "-deadline", - "good", - "-cpu-used", - "2", + "libx264", + "-preset", + "slow", + "-crf", + "18", + "-bf", + "0", "-pix_fmt", "yuv420p", + "-video_track_timescale", + "1000000", + "-movflags", + "+faststart", str(media_path), ], check=True, @@ -493,7 +548,16 @@ def _write_presentation_clip( ) media_sha256 = _sha256(media_path) - duration_ms = presentation_times_us[-1] // 1_000 + sample_end_us = _probe_video_sample_end_us(ffprobe, media_path) + if sample_end_us <= presentation_times_us[-1]: + raise EvidencePackError( + f"{clip_id} final decoded sample has no retained duration" + ) + if sample_end_us % 1_000: + raise EvidencePackError( + f"{clip_id} final decoded sample does not end on a media millisecond" + ) + duration_ms = sample_end_us // 1_000 timeline = build_runtime_control_overlay_timeline_v2( capture.frames, data_classification=ControlOverlayDataClassification.SYNTHETIC, @@ -1247,6 +1311,7 @@ def _media_type(path: Path) -> str: ".html": "text/html", ".py": "text/x-python", ".png": "image/png", + ".mp4": "video/mp4", ".webm": "video/webm", }.get(path.suffix.lower(), "application/octet-stream") @@ -1258,7 +1323,7 @@ def _role(path: str) -> str: return "compiled_bundle" if "/qualification/" in path: return "qualification" - if path.endswith(".webm"): + if path.endswith((".mp4", ".webm")): return "media" if "/cases/" in path: return "case_evidence" @@ -1345,7 +1410,7 @@ def _assemble_manifest( def presentation_media(clip_id: str) -> dict[str, Any]: return { - "media": _ref_for(root, presentation / f"{clip_id}.webm"), + "media": _ref_for(root, presentation / f"{clip_id}.mp4"), "timeline": _ref_for( root, presentation / f"{clip_id}.control-overlay.v2.json" ), @@ -1754,7 +1819,7 @@ def _validate_presentation_artifacts( pts_relative = str(clip["frame_pts"]["path"]) expected_prefix = f"artifacts/presentation/{clip_id}" if ( - media_relative != f"{expected_prefix}.webm" + media_relative not in {f"{expected_prefix}.mp4", f"{expected_prefix}.webm"} or timeline_relative != f"{expected_prefix}.control-overlay.v2.json" or pts_relative != f"{expected_prefix}.frame-pts-us.json" ): @@ -1806,13 +1871,22 @@ def _validate_presentation_artifacts( raise EvidencePackError( f"exact-PTS sidecar does not bind decoded media: {pts_relative}" ) + expected_duration_ms = presentation_times_us[-1] // 1_000 + if media_path.suffix.lower() == ".mp4": + _ffmpeg, ffprobe = _presentation_tools() + sample_end_us = _probe_video_sample_end_us(ffprobe, media_path) + if sample_end_us % 1_000: + raise EvidencePackError( + f"presentation media does not end on a millisecond: {media_relative}" + ) + expected_duration_ms = sample_end_us // 1_000 if ( timeline.get("schema_version") != "openadapt.control-overlay-timeline/v2" or timeline.get("data_classification") != "synthetic" or timeline.get("evidence_pack_id") != pack_id or timeline.get("media_sha256") != media_sha256 or timeline.get("media_frame_count") != len(presentation_times_us) - or timeline.get("duration_ms") != presentation_times_us[-1] // 1_000 + or timeline.get("duration_ms") != expected_duration_ms ): raise EvidencePackError( f"control-overlay timeline does not bind media: {timeline_relative}" diff --git a/tests/test_public_demo_presentation_media.py b/tests/test_public_demo_presentation_media.py new file mode 100644 index 00000000..67970cb8 --- /dev/null +++ b/tests/test_public_demo_presentation_media.py @@ -0,0 +1,96 @@ +"""Characterization of exact presentation-media timing.""" + +from __future__ import annotations + +import io +import json +import shutil +import subprocess +from pathlib import Path +from types import SimpleNamespace + +import pytest +from PIL import Image + +from scripts.export_public_demo_evidence import ( + _PresentationCapture, + _probe_video_sample_end_us, + _write_presentation_clip, +) + + +def _png(color: tuple[int, int, int]) -> bytes: + output = io.BytesIO() + Image.new("RGB", (64, 64), color).save(output, format="PNG") + return output.getvalue() + + +def test_sample_end_includes_the_final_packet_duration( + monkeypatch, tmp_path: Path +) -> None: + payload = { + "packets": [ + {"pts_time": "0.000000", "duration_time": "0.250000"}, + {"pts_time": "0.250000", "duration_time": "1.750000"}, + {"pts_time": "2.000000", "duration_time": "0.001000"}, + ] + } + monkeypatch.setattr( + subprocess, + "run", + lambda *_args, **_kwargs: SimpleNamespace(stdout=json.dumps(payload)), + ) + + assert _probe_video_sample_end_us("ffprobe", tmp_path / "clip.mp4") == 2_001_000 + + +@pytest.mark.skipif( + shutil.which("ffmpeg") is None or shutil.which("ffprobe") is None, + reason="requires separately provisioned ffmpeg and ffprobe", +) +def test_exported_mp4_preserves_exact_pts_and_full_terminal_sample( + tmp_path: Path, +) -> None: + capture = _PresentationCapture(mode="replay") + capture.emitter.begin(profile="standard") + capture.emitter.emit_phase("observing", observation_png=_png((0, 0, 0))) + capture.emitter.emit_terminal("HALTED", observation_png=_png((255, 255, 255))) + capture.frames = [ + frame.model_copy(update={"observed_at_monotonic_ms": 1_000.0 + offset}) + for frame, offset in zip(capture.frames, (0, 250), strict=True) + ] + + _write_presentation_clip( + capture=capture, + pack_id="exact-timing-fixture", + clip_id="halted", + presentation_dir=tmp_path, + ) + + pts = json.loads((tmp_path / "halted.frame-pts-us.json").read_text()) + timeline = json.loads((tmp_path / "halted.control-overlay.v2.json").read_text()) + media = tmp_path / "halted.mp4" + assert pts["presentation_times_us"] == [0, 250_000, 2_000_000] + assert _probe_video_sample_end_us(str(shutil.which("ffprobe")), media) == 2_001_000 + assert timeline["duration_ms"] == 2_001 + + probe = subprocess.run( + [ + str(shutil.which("ffprobe")), + "-v", + "error", + "-select_streams", + "v:0", + "-show_entries", + "stream=time_base:packet=pts,dts", + "-of", + "json", + str(media), + ], + check=True, + capture_output=True, + text=True, + ) + timing = json.loads(probe.stdout) + assert timing["streams"] == [{"time_base": "1/1000000"}] + assert all(packet["pts"] == packet["dts"] for packet in timing["packets"])