render: derive caption offsets from the rendered segments, not the EDL

helpers/render.py

@@ -85,6 +85,16 @@ def resolve_grade_filter(grade_field: str | None) -> str:
     return grade_field
 
 
+def probe_duration(path: Path) -> float:
+    """Container duration of a rendered file, in seconds."""
+    out = subprocess.run(
+        ["ffprobe", "-v", "error", "-show_entries", "format=duration",
+         "-of", "default=noprint_wrappers=1:nokey=1", str(path)],
+        capture_output=True, text=True, check=True,
+    )
+    return float(out.stdout.strip())
+
+
 def resolve_path(maybe_path: str, base: Path) -> Path:
     """Resolve a path that may be absolute or relative to `base`."""
     p = Path(maybe_path)
@@ -416,24 +426,49 @@ def _words_in_range(transcript: dict, t_start: float, t_end: float) -> list[dict
     return out
 
 
-def build_master_srt(edl: dict, edit_dir: Path, out_path: Path) -> None:
+def build_master_srt(edl: dict, edit_dir: Path, out_path: Path,
+                     segment_paths: list[Path] | None = None) -> None:
     """Build an output-timeline SRT from per-source transcripts.
 
     - 2-word chunks (break on any punctuation in between)
     - UPPERCASE text
     - Output times computed as word.start - segment_start + segment_offset
+
+    `segment_paths`, when given, are the extracted clips in concat order. Their
+    measured durations are used for the per-segment offset instead of the EDL's
+    `end - start`; see the comment below for why that matters.
     """
     transcripts_dir = edit_dir / "transcripts"
     sources = edl["sources"]
 
+    # The offset has to be where the segment actually STARTS in the concatenated
+    # output. An extract is quantised to whole frames, so it is a fraction of a
+    # frame longer than `end - start`, and summing the EDL's floats accumulates
+    # that error: on a 30-segment, 3m34s edit the captions ran 0.598s early by the
+    # final cue. Measure the rendered clips when they are available.
+    measured: list[float] | None = None
+    if segment_paths:
+        if len(segment_paths) != len(edl["ranges"]):
+            print(f"  warning: {len(segment_paths)} clips for {len(edl['ranges'])} ranges;"
+                  f" falling back to EDL durations for caption offsets")
+        else:
+            try:
+                measured = [probe_duration(p) for p in segment_paths]
+                drift = sum(measured) - sum(float(r["end"]) - float(r["start"])
+                                            for r in edl["ranges"])
+                print(f"  caption offsets from measured segments (drift vs EDL: {drift:+.3f}s)")
+            except (subprocess.CalledProcessError, ValueError, OSError) as exc:
+                print(f"  warning: could not measure segments ({exc}); using EDL durations")
+                measured = None
+
     entries: list[tuple[float, float, str]] = []
     seg_offset = 0.0
 
-    for r in edl["ranges"]:
+    for seg_i, r in enumerate(edl["ranges"]):
         src_name = r["source"]
         seg_start = float(r["start"])
         seg_end = float(r["end"])
-        seg_duration = seg_end - seg_start
+        seg_duration = measured[seg_i] if measured else (seg_end - seg_start)
 
         tr_path = transcripts_dir / f"{src_name}.json"
         if not tr_path.exists():
@@ -743,7 +778,7 @@ def main() -> None:
     if not args.no_subtitles:
         if args.build_subtitles:
             subs_path = edit_dir / "master.srt"
-            build_master_srt(edl, edit_dir, subs_path)
+            build_master_srt(edl, edit_dir, subs_path, segment_paths)
         elif edl.get("subtitles"):
             subs_path = resolve_path(edl["subtitles"], edit_dir)
             if not subs_path.exists():

tests/test_render_caption_offsets.py

@@ -0,0 +1,88 @@
+"""Caption offsets must come from the rendered segments, not the EDL's floats.
+
+An extract is quantised to whole frames, so a segment is a fraction of a frame
+longer than `end - start`. Summing the EDL's floats accumulates that error and
+puts every cue progressively early.
+"""
+import importlib.util
+import json
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+
+MODULE_PATH = Path(__file__).parents[1] / "helpers" / "render.py"
+SPEC = importlib.util.spec_from_file_location("video_use_render", MODULE_PATH)
+assert SPEC and SPEC.loader
+render = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(render)
+
+
+def _word(text, start, end):
+    return {"type": "word", "text": text, "start": start, "end": end}
+
+
+class CaptionOffsetTests(unittest.TestCase):
+    """Three 2.0s segments whose extracts each come out 0.1s long."""
+
+    EDL = {
+        "sources": {"A": "/tmp/A.mov"},
+        "ranges": [
+            {"source": "A", "start": 0.0, "end": 2.0},
+            {"source": "A", "start": 10.0, "end": 12.0},
+            {"source": "A", "start": 20.0, "end": 22.0},
+        ],
+    }
+    MEASURED = [2.1, 2.1, 2.1]
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.edit = Path(self.tmp.name)
+        (self.edit / "transcripts").mkdir()
+        (self.edit / "transcripts" / "A.json").write_text(json.dumps({"words": [
+            _word("one.", 0.5, 1.0),
+            _word("two.", 10.5, 11.0),
+            _word("three.", 20.5, 21.0),
+        ]}))
+        self.addCleanup(self.tmp.cleanup)
+
+    def _cues(self, segment_paths=None):
+        out = self.edit / "master.srt"
+        with patch.object(render, "probe_duration", side_effect=self.MEASURED):
+            render.build_master_srt(self.EDL, self.edit, out, segment_paths)
+        starts = []
+        for line in out.read_text().splitlines():
+            if " --> " in line:
+                h, m, rest = line.split(" --> ")[0].split(":")
+                starts.append(int(h) * 3600 + int(m) * 60 + float(rest.replace(",", ".")))
+        return starts
+
+    def test_without_segment_paths_offsets_come_from_the_edl(self):
+        # 0.5, then 2.0 + 0.5, then 4.0 + 0.5 — the pre-existing behaviour
+        self.assertEqual(self._cues(), [0.5, 2.5, 4.5])
+
+    def test_with_segment_paths_offsets_come_from_the_measured_clips(self):
+        # each clip is 0.1s longer than the EDL claims, and the error accumulates
+        paths = [Path(f"/tmp/seg{i}.mp4") for i in range(3)]
+        self.assertEqual(self._cues(paths), [0.5, 2.6, 4.7])
+
+    def test_a_wrong_number_of_clips_falls_back_to_the_edl(self):
+        # a mismatch must not silently pair clips with the wrong ranges
+        self.assertEqual(self._cues([Path("/tmp/seg0.mp4")]), [0.5, 2.5, 4.5])
+
+    def test_a_failed_probe_falls_back_to_the_edl(self):
+        out = self.edit / "master.srt"
+        paths = [Path(f"/tmp/seg{i}.mp4") for i in range(3)]
+        with patch.object(render, "probe_duration", side_effect=OSError("no ffprobe")):
+            render.build_master_srt(self.EDL, self.edit, out, paths)
+        starts = [
+            int(l.split(" --> ")[0].split(":")[1]) * 60
+            + float(l.split(" --> ")[0].split(":")[2].replace(",", "."))
+            for l in out.read_text().splitlines() if " --> " in l
+        ]
+        self.assertEqual(starts, [0.5, 2.5, 4.5])
+
+
+if __name__ == "__main__":
+    unittest.main()