diff --git a/skills/taste-application/scripts/forge.py b/skills/taste-application/scripts/forge.py index ffb554987..09cd98f63 100644 --- a/skills/taste-application/scripts/forge.py +++ b/skills/taste-application/scripts/forge.py @@ -22,7 +22,6 @@ from __future__ import annotations import argparse import math import tempfile -import sys from datetime import datetime, timezone from pathlib import Path @@ -223,13 +222,12 @@ def forge( rot = 0.0 if wash else float(rng.uniform(-0.6, 0.6)) opa = overlay_opacity * (0.75 if wash else 1.25) dst = work_dir / f"ov_{i:03d}.mp4" - try: - stamped.append(asm.overlay( - clip, plate, dst, opacity=min(0.95, opa), scale=sc, - position=pos, rotate=rot, width=W, height=H)) - continue - except RuntimeError as exc: - print(f" !! overlay failed on shot {i}: {str(exc)[:80]}", file=sys.stderr) + # Requested overlays are part of the output contract. A failed + # composite must not produce a successful, unstamped handoff. + stamped.append(asm.overlay( + clip, plate, dst, opacity=min(0.95, opa), scale=sc, + position=pos, rotate=rot, width=W, height=H)) + continue stamped.append(clip) order = stamped diff --git a/skills/taste-application/scripts/taste/assemble.py b/skills/taste-application/scripts/taste/assemble.py index 519f3b8c3..c4ec815ac 100644 --- a/skills/taste-application/scripts/taste/assemble.py +++ b/skills/taste-application/scripts/taste/assemble.py @@ -162,7 +162,9 @@ def overlay( ) _run([ "ffmpeg", "-nostdin", "-loglevel", "error", "-y", - "-i", str(clip), "-i", str(image), "-filter_complex", fc, + # Keep the still alive until the video ends; shortest=1 otherwise + # terminates every shot after the image's single decoded frame. + "-i", str(clip), "-loop", "1", "-i", str(image), "-filter_complex", fc, "-c:v", "libx264", "-crf", "14", "-preset", "veryfast", "-pix_fmt", "yuv420p", "-an", str(dst), ]) diff --git a/skills/taste-distillation/scripts/taste/assemble.py b/skills/taste-distillation/scripts/taste/assemble.py index 519f3b8c3..c4ec815ac 100644 --- a/skills/taste-distillation/scripts/taste/assemble.py +++ b/skills/taste-distillation/scripts/taste/assemble.py @@ -162,7 +162,9 @@ def overlay( ) _run([ "ffmpeg", "-nostdin", "-loglevel", "error", "-y", - "-i", str(clip), "-i", str(image), "-filter_complex", fc, + # Keep the still alive until the video ends; shortest=1 otherwise + # terminates every shot after the image's single decoded frame. + "-i", str(clip), "-loop", "1", "-i", str(image), "-filter_complex", fc, "-c:v", "libx264", "-crf", "14", "-preset", "veryfast", "-pix_fmt", "yuv420p", "-an", str(dst), ]) diff --git a/tests/test_taste_overlays.py b/tests/test_taste_overlays.py new file mode 100644 index 000000000..b728baad6 --- /dev/null +++ b/tests/test_taste_overlays.py @@ -0,0 +1,132 @@ +"""Requested image overlays must fail closed if compositing fails.""" + +import importlib.util +import shutil +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch + +if any( + importlib.util.find_spec(name) is None for name in ("numpy", "cv2", "scenedetect") +): + raise unittest.SkipTest("Install taste-application requirements for overlay tests") + +SCRIPTS = Path(__file__).resolve().parents[1] / "skills/taste-application/scripts" +sys.path.insert(0, str(SCRIPTS)) +spec = importlib.util.spec_from_file_location("overlay_forge", SCRIPTS / "forge.py") +forge = importlib.util.module_from_spec(spec) +spec.loader.exec_module(forge) + + +class OverlayFailureTests(unittest.TestCase): + @unittest.skipUnless( + shutil.which("ffmpeg") and shutil.which("ffprobe"), "FFmpeg required" + ) + def test_still_overlay_preserves_all_video_frames(self): + import cv2 + + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + take, plate, out = ( + root / name for name in ("take.mp4", "plate.png", "out.mp4") + ) + subprocess.run( + [ + "ffmpeg", + "-nostdin", + "-v", + "error", + "-f", + "lavfi", + "-i", + "color=c=black:s=64x64:r=30:d=0.5", + "-c:v", + "libx264", + str(take), + ], + check=True, + timeout=20, + ) + image = forge.np.full((16, 16, 4), 255, dtype=forge.np.uint8) + self.assertTrue(cv2.imwrite(str(plate), image)) + forge.asm.overlay(take, plate, out, width=64, height=64) + cap = cv2.VideoCapture(str(out)) + frames = [] + while True: + ok, frame = cap.read() + if not ok: + break + frames.append(frame) + cap.release() + self.assertEqual(len(frames), 15) + self.assertTrue(all(frame.max() > 100 for frame in frames)) + + def test_failed_requested_overlay_prevents_final_video_and_manifest(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + take, plate, out = ( + root / name for name in ("take.mp4", "plate.png", "out.mp4") + ) + take.write_bytes(b"original video") + plate.write_bytes(b"original image") + with ( + patch.object( + forge.pack_mod, + "load", + return_value=SimpleNamespace( + grade_path="grade", cadence_path="cadence" + ), + ), + patch.object(forge.grade_mod, "load_stats"), + patch.object( + forge.cad_mod, + "load", + return_value=SimpleNamespace( + mean_shot=1, + cuts_per_min=60, + rhythm_variance=0, + plan_shots=lambda _: [1], + ), + ), + patch.object( + forge.frame_mod, + "probe", + return_value=SimpleNamespace( + width=320, height=180, fps=30, duration=1 + ), + ), + patch.object(forge.asm, "normalize", return_value=take), + patch.object(forge.grade_mod, "grade_clip_direct"), + patch.object(forge.asm, "cut_take", return_value=[take]), + patch.object(forge.plate_mod, "tighten", return_value=plate), + patch.object(forge.plate_mod, "plate_coverage", return_value=0.3), + patch.object( + forge.asm, "overlay", side_effect=RuntimeError("compositor failed") + ), + patch.object(forge.asm, "concat") as concat, + patch.object(forge.tl_mod, "write_timeline") as timeline, + patch.object(forge.asm, "write_manifest") as manifest, + ): + with self.assertRaisesRegex(RuntimeError, "compositor failed"): + forge.forge( + "look", + [str(take)], + str(out), + overlays=[str(plate)], + work=str(root / "work"), + fps=30, + ) + concat.assert_not_called() + timeline.assert_not_called() + manifest.assert_not_called() + self.assertFalse(out.exists()) + self.assertEqual(take.read_bytes(), b"original video") + self.assertEqual(plate.read_bytes(), b"original image") + + +if __name__ == "__main__": + unittest.main()