fix(movie): make frame and concat inputs portable

skills/proving-it-works-with-a-movie/scripts/assemble

@@ -30,10 +30,12 @@ import os
 import shutil
 import subprocess
 import sys
+import tempfile
 import urllib.parse
 from pathlib import Path
 
 import yaml
+from media_paths import ffconcat_entry, stage_frames
 
 BROWSERS = [
     "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome",
@@ -94,7 +96,7 @@ def make_card(scene, png, w, h, browser):
         sub=scene.get("subtitle_size", max(16, h // 32)),
         TITLE=scene.get("title", ""), SUB=scene.get("subtitle", ""))
     tmp = png.with_suffix(".html")
-    tmp.write_text(html)
+    tmp.write_text(html, encoding="utf-8")
     run([browser, "--headless=new", "--disable-gpu", "--hide-scrollbars",
          f"--screenshot={png}", f"--window-size={w},{h}",
          "--force-device-scale-factor=1", "file://" + urllib.parse.quote(str(tmp))])
@@ -113,7 +115,7 @@ def main():
         if not shutil.which(tool):
             die(f"{tool} not on PATH")
 
-    doc = yaml.safe_load(args.scenes.read_text())
+    doc = yaml.safe_load(args.scenes.read_text(encoding="utf-8"))
     base = args.scenes.parent
     res = doc.get("resolution", {}) or {}
     W, H = int(res.get("width", 1920)), int(res.get("height", 1080))
@@ -146,16 +148,28 @@ def main():
                  "-c:v", "libx264", "-preset", "medium", "-pix_fmt", "yuv420p",
                  "-c:a", "aac", "-ar", "44100", "-ac", "2", str(seg)])
         else:
+            frame_staging = None
             if kind == "frames":
                 src = base / sc["src"]
-                n = len(list(Path(src).glob("*.png")))
-                if not n:
-                    die(f"scene {sid}: no PNGs in {src}")
                 rate = float(sc.get("rate", FPS))
+                frame_staging = tempfile.TemporaryDirectory(
+                    prefix=f"frames-{sid}-", dir=work
+                )
+                try:
+                    staged = stage_frames(
+                        src, Path(frame_staging.name) / "sequence"
+                    )
+                except ValueError as error:
+                    frame_staging.cleanup()
+                    die(f"scene {sid}: {error}")
+                except BaseException:
+                    frame_staging.cleanup()
+                    raise
+                n = len(staged)
                 vis = n / rate
                 target = max(nard, vis)
-                vin = ["-framerate", str(rate), "-pattern_type", "glob",
-                       "-i", str(Path(src) / "*.png")]
+                vin = ["-framerate", str(rate), "-start_number", "0",
+                       "-i", str(staged[0].parent / "frame-%08d.png")]
                 # freeze the last frame when narration outlasts the action
                 vf = fit + f",tpad=stop_mode=clone:stop_duration={max(0.0, target - vis):.3f}"
             else:
@@ -174,11 +188,15 @@ def main():
 
             ain = (["-i", str(nar)] if nar.exists()
                    else ["-f", "lavfi", "-i", "anullsrc=r=44100:cl=stereo"])
-            run(["ffmpeg", "-nostdin", "-y", "-v", "error", *vin, *ain,
-                 "-vf", vf, "-af", "apad", "-r", str(FPS), "-t", f"{target:.3f}",
-                 "-map", "0:v:0", "-map", "1:a:0",
-                 "-c:v", "libx264", "-preset", "medium", "-pix_fmt", "yuv420p",
-                 "-c:a", "aac", "-ar", "44100", "-ac", "2", str(seg)])
+            try:
+                run(["ffmpeg", "-nostdin", "-y", "-v", "error", *vin, *ain,
+                     "-vf", vf, "-af", "apad", "-r", str(FPS), "-t", f"{target:.3f}",
+                     "-map", "0:v:0", "-map", "1:a:0",
+                     "-c:v", "libx264", "-preset", "medium", "-pix_fmt", "yuv420p",
+                     "-c:a", "aac", "-ar", "44100", "-ac", "2", str(seg)])
+            finally:
+                if frame_staging is not None:
+                    frame_staging.cleanup()
 
         actual = dur(seg)
         # only scenes that speak get a subtitle offset; a movie played as
@@ -186,14 +204,16 @@ def main():
         if nar.exists() and kind != "movie":
             offsets[sid] = round(clock, 3)
         clock += actual
-        concat_lines.append(f"file '{seg.resolve()}'")
+        concat_lines.append(ffconcat_entry(seg))
         print(f"{sid}: {actual:.1f}s{' (own audio)' if kind == 'movie' else ''}")
 
     listing = work / "concat.txt"
-    listing.write_text("\n".join(concat_lines) + "\n")
+    listing.write_text("".join(concat_lines), encoding="utf-8")
     run(["ffmpeg", "-nostdin", "-y", "-v", "error", "-f", "concat", "-safe", "0",
          "-i", str(listing), "-c", "copy", str(args.out)])
-    (work / "offsets.json").write_text(json.dumps(offsets, indent=2))
+    (work / "offsets.json").write_text(
+        json.dumps(offsets, indent=2), encoding="utf-8"
+    )
     print(f"\nassembled {args.out} ({dur(args.out):.1f}s)")
     print(f"scene offsets -> {work / 'offsets.json'} "
           f"(feed to make-subtitles --offsets-json)")

skills/proving-it-works-with-a-movie/scripts/media_paths.py

@@ -0,0 +1,24 @@
+"""Portable media path preparation for the movie assembly tools."""
+
+import shutil
+from pathlib import Path
+
+
+def stage_frames(source: Path, destination: Path) -> list[Path]:
+    files = sorted(source.glob("*.png"))
+    if not files:
+        raise ValueError(f"no PNG frames in {source}")
+    destination.mkdir(parents=True, exist_ok=False)
+    result = []
+    for index, frame in enumerate(files):
+        target = destination / f"frame-{index:08d}.png"
+        shutil.copyfile(frame, target)
+        result.append(target)
+    return result
+
+
+def ffconcat_entry(path: Path) -> str:
+    value = path.resolve().as_posix()
+    if "\n" in value or "\r" in value:
+        raise ValueError("FFconcat paths cannot contain line breaks")
+    return "file '" + value.replace("'", "'\\''") + "'\n"

tests/proving-it-works-with-a-movie/run-tests.py

@@ -15,9 +15,9 @@ IMPLEMENTED_SUITES = {
     "assembly": "test_assembly.py",
     "checker": "test_checker.py",
     "narration": "test_narration.py",
+    "paths": "test_paths.py",
 }
 RESERVED_SUITES = {
-    "paths",
     "browser",
     "subtitles",
     "processes",

tests/proving-it-works-with-a-movie/test_assembly.py

@@ -1,9 +1,149 @@
 import json
+import os
+import shutil
+import subprocess
+import sys
 import tempfile
 import unittest
 from pathlib import Path
 
 import fixtures
+from PIL import Image
+
+
+def run_ffmpeg(args: list[str], *, cwd: Path) -> subprocess.CompletedProcess[bytes]:
+    ffmpeg = shutil.which("ffmpeg")
+    if ffmpeg is None:
+        raise RuntimeError("ffmpeg is required for assembly fixtures")
+    result = subprocess.run(
+        [ffmpeg, "-nostdin", "-y", "-v", "error", *args],
+        cwd=cwd,
+        capture_output=True,
+        timeout=fixtures.TIMEOUT_SECONDS,
+    )
+    if result.returncode != 0:
+        raise RuntimeError(
+            f"ffmpeg fixture generation failed: {fixtures.output_text(result)}"
+        )
+    return result
+
+
+def make_tone(path: Path, duration: float, frequency: int, *, cwd: Path) -> None:
+    run_ffmpeg(
+        [
+            "-f",
+            "lavfi",
+            "-i",
+            f"sine=frequency={frequency}:sample_rate=8000:duration={duration}",
+            "-c:a",
+            "pcm_s16le",
+            str(path),
+        ],
+        cwd=cwd,
+    )
+
+
+def make_movie(path: Path, duration: float, frequency: int, *, cwd: Path) -> None:
+    run_ffmpeg(
+        [
+            "-f",
+            "lavfi",
+            "-i",
+            f"color=c=green:size=160x90:rate=10:duration={duration}",
+            "-f",
+            "lavfi",
+            "-i",
+            f"sine=frequency={frequency}:sample_rate=8000:duration={duration}",
+            "-c:v",
+            "libx264",
+            "-pix_fmt",
+            "yuv420p",
+            "-c:a",
+            "aac",
+            "-shortest",
+            str(path),
+        ],
+        cwd=cwd,
+    )
+
+
+def decoded_pixel(path: Path, timestamp: float, *, cwd: Path) -> tuple[int, int, int]:
+    result = run_ffmpeg(
+        [
+            "-ss",
+            str(timestamp),
+            "-i",
+            str(path),
+            "-frames:v",
+            "1",
+            "-vf",
+            "scale=1:1",
+            "-f",
+            "rawvideo",
+            "-pix_fmt",
+            "rgb24",
+            "-",
+        ],
+        cwd=cwd,
+    )
+    if len(result.stdout) < 3:
+        raise AssertionError(f"no decoded pixel from {path}")
+    return tuple(result.stdout[:3])
+
+
+def decoded_frequency(path: Path, *, cwd: Path) -> float:
+    sample_rate = 8000
+    result = run_ffmpeg(
+        [
+            "-ss",
+            "0.1",
+            "-t",
+            "0.5",
+            "-i",
+            str(path),
+            "-map",
+            "0:a:0",
+            "-f",
+            "s16le",
+            "-acodec",
+            "pcm_s16le",
+            "-ac",
+            "1",
+            "-ar",
+            str(sample_rate),
+            "-",
+        ],
+        cwd=cwd,
+    )
+    samples = [
+        int.from_bytes(result.stdout[index:index + 2], "little", signed=True)
+        for index in range(0, len(result.stdout) - 1, 2)
+    ]
+    nonzero = [sample for sample in samples if sample]
+    crossings = sum(
+        (left < 0 <= right) or (left > 0 >= right)
+        for left, right in zip(nonzero, nonzero[1:])
+    )
+    seconds = len(samples) / sample_rate
+    return crossings / (2 * seconds)
+
+
+def available_browser() -> str | None:
+    candidates = [
+        "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome",
+        "/Applications/Chromium.app/Contents/MacOS/Chromium",
+        "chromium",
+        "chromium-browser",
+        "google-chrome",
+        "google-chrome-stable",
+    ]
+    for candidate in candidates:
+        if Path(candidate).is_file():
+            return candidate
+        found = shutil.which(candidate)
+        if found:
+            return found
+    return None
 
 
 class AssemblyRegression(unittest.TestCase):
@@ -57,6 +197,221 @@ class AssemblyRegression(unittest.TestCase):
                 delta=0.001,
             )
 
+    def test_image_frames_and_movie_paths_timing_order_and_cleanup(self):
+        missing = fixtures.missing_executables("uv", "ffmpeg", "ffprobe")
+        if missing:
+            self.skipTest(f"required executable(s) not on PATH: {', '.join(missing)}")
+
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            launch = root / "launch"
+            launch.mkdir()
+            project = root / "movie O'Brien λ & [take]"
+            assets = project / "assets"
+            frames = assets / "frames"
+            subtitles = assets / "subtitles" / "nested"
+            narration = project / "narration"
+            frames.mkdir(parents=True)
+            subtitles.mkdir(parents=True)
+            narration.mkdir()
+
+            still = assets / "still image.png"
+            Image.new("RGB", (160, 90), (230, 230, 230)).save(still)
+            Image.new("RGB", (160, 90), (255, 0, 0)).save(frames / "a.png")
+            Image.new("RGB", (160, 90), (0, 0, 255)).save(frames / "b.png")
+            source_pngs = {
+                path: path.read_bytes() for path in [still, *sorted(frames.glob("*.png"))]
+            }
+
+            source_movie = assets / "source movie.mp4"
+            make_movie(source_movie, 0.8, 440, cwd=launch)
+            make_tone(narration / "image.wav", 0.8, 660, cwd=launch)
+            make_tone(narration / "frames.wav", 1.4, 770, cwd=launch)
+            make_tone(narration / "movie.wav", 1.6, 880, cwd=launch)
+
+            nested_srt = subtitles / "captions.srt"
+            nested_srt.write_bytes(
+                b"1\r\n00:00:00,000 --> 00:00:00,500\r\nPortable paths\r\n"
+            )
+            scenes = project / "scenes.yaml"
+            yaml_text = f"""resolution: {{ width: 160, height: 90 }}
+fps: 10
+scenes:
+  - id: image
+    kind: image
+    src: assets/still image.png
+    duration: 0.3
+  - id: frames
+    kind: frames
+    src: {frames.resolve().as_posix()}
+    rate: 2.0
+  - id: movie
+    kind: movie
+    src: assets/source movie.mp4
+"""
+            scenes.write_bytes(yaml_text.replace("\n", "\r\n").encode("utf-8"))
+
+            work = root / "generated outside launch"
+            output = project / "assembled output.mp4"
+            result = fixtures.run_tool(
+                "assemble",
+                [
+                    os.path.relpath(scenes, launch),
+                    os.path.relpath(output, launch),
+                    "--narration",
+                    str(narration.resolve()),
+                    "--work",
+                    str(work.resolve()),
+                ],
+                cwd=launch,
+            )
+            self.assertEqual(result.returncode, 0, fixtures.output_text(result))
+
+            image_duration = fixtures.duration(work / "image.mp4")
+            frames_duration = fixtures.duration(work / "frames.mp4")
+            movie_duration = fixtures.duration(work / "movie.mp4")
+            self.assertAlmostEqual(
+                image_duration,
+                max(fixtures.duration(narration / "image.wav"), 0.3),
+                delta=0.25,
+            )
+            self.assertAlmostEqual(
+                frames_duration,
+                max(fixtures.duration(narration / "frames.wav"), 1.0),
+                delta=0.25,
+            )
+            self.assertAlmostEqual(
+                movie_duration,
+                fixtures.duration(source_movie),
+                delta=0.25,
+            )
+            self.assertLess(movie_duration, fixtures.duration(narration / "movie.wav") - 0.4)
+            self.assertAlmostEqual(
+                decoded_frequency(work / "movie.mp4", cwd=launch),
+                440,
+                delta=15,
+            )
+
+            offsets = json.loads(
+                (work / "offsets.json").read_text(encoding="utf-8")
+            )
+            self.assertAlmostEqual(offsets["image"], 0.0, delta=0.001)
+            self.assertAlmostEqual(offsets["frames"], image_duration, delta=0.001)
+            self.assertNotIn("movie", offsets)
+
+            first = decoded_pixel(work / "frames.mp4", 0.2, cwd=launch)
+            second = decoded_pixel(work / "frames.mp4", 0.7, cwd=launch)
+            self.assertGreater(first[0], first[2] + 100)
+            self.assertGreater(second[2], second[0] + 100)
+            self.assertEqual(
+                {path: path.read_bytes() for path in source_pngs},
+                source_pngs,
+            )
+            self.assertEqual(list(work.glob("frames-frames-*")), [])
+
+            subtitled = project / "subtitled output.mp4"
+            subtitle_result = fixtures.run_tool(
+                "burn-subtitles",
+                [
+                    str(output.resolve()),
+                    os.path.relpath(nested_srt, launch),
+                    str(subtitled.resolve()),
+                    "--soft",
+                ],
+                cwd=launch,
+            )
+            self.assertEqual(
+                subtitle_result.returncode,
+                0,
+                fixtures.output_text(subtitle_result),
+            )
+            self.assertTrue(subtitled.is_file())
+
+    def test_card_uses_longer_narration_duration(self):
+        if sys.platform == "win32":
+            self.skipTest(
+                "native Windows card completion is pending Task5 browser discovery/file-URI work"
+            )
+        missing = fixtures.missing_executables("uv", "ffmpeg", "ffprobe")
+        if missing:
+            self.skipTest(f"required executable(s) not on PATH: {', '.join(missing)}")
+        browser = available_browser()
+        if browser is None:
+            self.skipTest("required browser unavailable for card assembly")
+
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            narration = root / "narration"
+            narration.mkdir()
+            make_tone(narration / "card.wav", 0.8, 550, cwd=root)
+            scenes = root / "scenes.yaml"
+            scenes.write_bytes(
+                b"resolution: { width: 160, height: 90 }\r\n"
+                b"fps: 10\r\n"
+                b"scenes:\r\n"
+                b"  - id: card\r\n"
+                b"    kind: card\r\n"
+                b"    title: Portable\r\n"
+                b"    subtitle: paths\r\n"
+                b"    duration: 0.3\r\n"
+            )
+            work = root / "work"
+            result = fixtures.run_tool(
+                "assemble",
+                [
+                    str(scenes),
+                    str(root / "out.mp4"),
+                    "--work",
+                    str(work),
+                    "--browser",
+                    browser,
+                ],
+                cwd=root,
+            )
+            self.assertEqual(result.returncode, 0, fixtures.output_text(result))
+            self.assertAlmostEqual(
+                fixtures.duration(work / "card.mp4"),
+                max(fixtures.duration(narration / "card.wav"), 0.3),
+                delta=0.25,
+            )
+            offsets = json.loads(
+                (work / "offsets.json").read_text(encoding="utf-8")
+            )
+            self.assertAlmostEqual(offsets["card"], 0.0, delta=0.001)
+
+    def test_frame_sources_survive_failed_assembly_cleanup(self):
+        missing = fixtures.missing_executables("uv", "ffmpeg", "ffprobe")
+        if missing:
+            self.skipTest(f"required executable(s) not on PATH: {', '.join(missing)}")
+
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            source = root / "movie O'Brien λ & [take]" / "frames"
+            source.mkdir(parents=True)
+            bad_frame = source / "a.png"
+            bad_frame.write_bytes(b"not a PNG")
+            scenes = root / "scenes.yaml"
+            scenes.write_text(
+                f"""resolution: {{ width: 160, height: 90 }}
+fps: 10
+scenes:
+  - id: broken
+    kind: frames
+    src: {source.resolve().as_posix()}
+    rate: 1.0
+""",
+                encoding="utf-8",
+            )
+            work = root / "work"
+            result = fixtures.run_tool(
+                "assemble",
+                [str(scenes), str(root / "out.mp4"), "--work", str(work)],
+                cwd=root,
+            )
+            self.assertNotEqual(result.returncode, 0)
+            self.assertEqual(bad_frame.read_bytes(), b"not a PNG")
+            self.assertEqual(list(work.glob("frames-broken-*")), [])
+
 
 if __name__ == "__main__":
     unittest.main()

tests/proving-it-works-with-a-movie/test_paths.py

@@ -0,0 +1,104 @@
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+import wave
+from pathlib import Path
+
+
+SCRIPTS = (
+    Path(__file__).resolve().parents[2]
+    / "skills/proving-it-works-with-a-movie/scripts"
+)
+sys.path.insert(0, str(SCRIPTS))
+
+import media_paths
+
+
+class MediaPathRegression(unittest.TestCase):
+    def test_frame_staging_uses_ordinary_ordered_files(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            source = root / "movie O'Brien λ & [take]"
+            source.mkdir()
+            (source / "b.png").write_bytes(b"second")
+            (source / "a.png").write_bytes(b"first")
+
+            staged = media_paths.stage_frames(source, root / "staged")
+
+            self.assertEqual(
+                [path.read_bytes() for path in staged],
+                [b"first", b"second"],
+            )
+            self.assertTrue(all(not path.is_symlink() for path in staged))
+            self.assertEqual(len(list(source.iterdir())), 2)
+
+    def test_frame_staging_rejects_an_empty_source(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            source = root / "empty"
+            source.mkdir()
+
+            with self.assertRaisesRegex(ValueError, "no PNG frames"):
+                media_paths.stage_frames(source, root / "staged")
+
+            self.assertFalse((root / "staged").exists())
+
+    def test_ffconcat_entry_is_accepted_by_ffmpeg(self):
+        ffmpeg = shutil.which("ffmpeg")
+        if ffmpeg is None:
+            self.skipTest("required executable not on PATH: ffmpeg")
+
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory) / "movie O'Brien λ & [take]"
+            nested = root / "subtitles" / "nested"
+            nested.mkdir(parents=True)
+            audio = nested / "tone.wav"
+            with wave.open(str(audio), "wb") as output:
+                output.setnchannels(1)
+                output.setsampwidth(2)
+                output.setframerate(8000)
+                output.writeframes(b"\x00\x00" * 800)
+
+            if sys.platform == "win32":
+                self.assertRegex(str(audio.resolve()), r"^[A-Za-z]:\\")
+
+            listing = root / "concat.txt"
+            listing.write_text(
+                "ffconcat version 1.0\n" + media_paths.ffconcat_entry(audio),
+                encoding="utf-8",
+            )
+            result = subprocess.run(
+                [
+                    ffmpeg,
+                    "-nostdin",
+                    "-v",
+                    "error",
+                    "-f",
+                    "concat",
+                    "-safe",
+                    "0",
+                    "-i",
+                    str(listing),
+                    "-f",
+                    "null",
+                    "-",
+                ],
+                cwd=Path(directory),
+                capture_output=True,
+                timeout=30,
+            )
+            self.assertEqual(
+                result.returncode,
+                0,
+                (result.stdout + result.stderr).decode("utf-8", errors="replace"),
+            )
+
+    def test_ffconcat_entry_rejects_line_breaks(self):
+        with self.assertRaisesRegex(ValueError, "cannot contain line breaks"):
+            media_paths.ffconcat_entry(Path("bad\nname.wav"))
+
+
+if __name__ == "__main__":
+    unittest.main()