fix(hybrid): read the container file for a base that had the DV fixup

The fixup replaces a track with a raw HEVC file, which has no duration and cannot be seeked, so Level 5 crop detection raised KeyError.

docs/dev/service-architecture.md

@@ -429,6 +429,11 @@ extracts the RPU with `dovi_tool`, and injects it onto the base layer, to make a
 track that carries both HDR10 metadata and the DV RPU. Both are automatic once the
 pipeline selects the right tracks. A service only needs to give accurate ranges.
 
+The fixup replaces the track file with a raw HEVC file. A raw HEVC file has no duration and
+FFmpeg cannot seek in it. When a hybrid base also gets the fixup, the pipeline keeps the
+container file of that track until its hybrid step is complete, and the hybrid step reads
+the container file.
+
 ## Request and HTTP session objects
 
 Everything a service does over the network flows through objects the base constructor

tests/core/test_hybrid_dv_fixup.py

@@ -0,0 +1,84 @@
+"""Unit tests for the path from the DV fixup to the hybrid step, which must read the
+container file and not the raw stream."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any
+
+import pytest
+
+from unshackle.commands.dl import hybrid_source_tracks, select_hybrid_bases
+from unshackle.core.tracks import Video
+from unshackle.core.tracks import dv_fixup as dv_fixup_mod
+from unshackle.core.tracks.dv_fixup import apply_dv_fixup
+
+pytestmark = pytest.mark.unit
+
+
+def fake_video(tmp_path: Path, track_id: str, range_: Video.Range, height: int = 2160) -> Any:
+    path = tmp_path / f"{track_id}.mp4"
+    path.write_bytes(b"x")
+    return SimpleNamespace(id=track_id, path=path, height=height, range=range_, dv_compatible_bitstream=True)
+
+
+@pytest.fixture
+def fake_demux(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(dv_fixup_mod, "FFMPEG", "ffmpeg")
+    monkeypatch.setattr(dv_fixup_mod, "log_event", lambda *a, **k: None)
+    monkeypatch.setattr(dv_fixup_mod, "run_step", lambda args, **k: Path(args[-1]).write_bytes(b"y"))
+
+
+@pytest.mark.parametrize("keep_source", [False, True])
+def test_dv_fixup_keeps_the_source_on_request(fake_demux: None, tmp_path: Path, keep_source: bool) -> None:
+    video = fake_video(tmp_path, "abc", Video.Range.HDR10P)
+    source = video.path
+
+    kept = apply_dv_fixup(video, keep_source=keep_source)
+
+    assert video.path == tmp_path / "abc.dv.hevc"
+    assert source.exists() is keep_source
+    assert kept == (source if keep_source else None)
+
+
+def test_dv_fixup_that_falls_back_keeps_nothing(
+    fake_demux: None, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    """A failed fixup leaves the track on its source, so there is no second file to clean up."""
+
+    def failed_demux(args: list[Any], **kwargs: Any) -> None:
+        raise RuntimeError("ffmpeg failed")
+
+    monkeypatch.setattr(dv_fixup_mod, "run_step", failed_demux)
+    video = fake_video(tmp_path, "abc", Video.Range.HDR10P)
+    source = video.path
+
+    assert apply_dv_fixup(video, keep_source=True) is None
+    assert video.path == source and source.exists()
+
+
+def test_hybrid_step_reads_the_container_after_the_fixup(fake_demux: None, tmp_path: Path) -> None:
+    base = fake_video(tmp_path, "base", Video.Range.HDR10P)
+    dv = fake_video(tmp_path, "dv", Video.Range.DV)
+    container = apply_dv_fixup(base, keep_source=True)
+    assert container is not None
+
+    hybrid_base, hybrid_dv = hybrid_source_tracks(base, dv, container)
+
+    assert base.path == tmp_path / "base.dv.hevc"
+    assert hybrid_base.path == tmp_path / "base.mp4"
+    assert hybrid_dv.path == dv.path
+    assert hybrid_base.needs_duration_fix and hybrid_dv.needs_duration_fix
+
+
+def test_one_hybrid_base_for_each_height(tmp_path: Path) -> None:
+    videos = [
+        fake_video(tmp_path, "sdr", Video.Range.SDR, 1080),
+        fake_video(tmp_path, "low", Video.Range.HDR10, 1080),
+        fake_video(tmp_path, "high", Video.Range.HDR10P, 2160),
+        fake_video(tmp_path, "high2", Video.Range.HDR10, 2160),
+        fake_video(tmp_path, "dv", Video.Range.DV, 2160),
+    ]
+
+    assert [v.id for v in select_hybrid_bases(videos)] == ["high", "low"]

tests/core/test_hybrid_l6.py

@@ -1,6 +1,7 @@
 """Unit tests for the Hybrid Level 6 path: the FFprobe HDR10 metadata probe and the
-rule that decides when the RPU L6 block gets overwritten. Nothing runs FFprobe or
+rule that decides when the RPU L6 block gets overwritten. One test covers the Level 5
-dovi_tool; the tests mock subprocess.run and the dovi wrappers."""
+duration probe. Nothing runs FFprobe or dovi_tool; the tests mock subprocess.run and
+the dovi wrappers."""
 
 from __future__ import annotations
 
@@ -65,6 +66,17 @@ def patch_probe_run(monkeypatch: pytest.MonkeyPatch, result: FakeCompleted) -> l
     return calls
 
 
+def test_level_5_skips_a_stream_with_no_duration(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    """FFprobe gives a raw stream an empty format object with exit code 0."""
+    calls = patch_probe_run(monkeypatch, FakeCompleted(0, '{"format": {}}'))
+    hybrid = make_hybrid()
+
+    hybrid.level_5(tmp_path / "base.dv.hevc")
+
+    assert hybrid.rpu_file == "RPU.bin"
+    assert len(calls) == 1
+
+
 def probe_output(*side_data: dict) -> str:
     return json.dumps({"frames": [{"side_data_list": list(side_data)}]})
 

unshackle/commands/dl.py

@@ -210,6 +210,31 @@ def group_videos_by_variant(videos: list[Video], *, merge: bool) -> list[list[Vi
     return list(groups.values())
 
 
+def select_hybrid_bases(videos: Sequence[Video]) -> list[Video]:
+    """The HDR10 or HDR10+ track that is the hybrid base for each height, tallest first."""
+    bases: dict[int, Video] = {}
+    for video in sorted(
+        (v for v in videos if v.range in (Video.Range.HDR10P, Video.Range.HDR10)),
+        key=lambda v: v.height or 0,
+        reverse=True,
+    ):
+        bases.setdefault(video.height or 0, video)
+    return list(bases.values())
+
+
+def hybrid_source_tracks(base: Video, dv: Video, base_container: Optional[Path] = None) -> list[Video]:
+    """Copies of the base and DV tracks for the hybrid step, with the base on its container file.
+
+    The hybrid step reads the duration and seeks, and the raw stream from the DV fixup
+    permits neither.
+    """
+    tracks = [deepcopy(base), deepcopy(dv)]
+    tracks[0].path = base_container or base.path
+    for track in tracks:
+        track.needs_duration_fix = True
+    return tracks
+
+
 def title_wanted(candidate: Any, wanted: Collection[str]) -> bool:
     """Whether ``-w`` selects this title. A title type without selection keys is always kept.
 
@@ -3278,6 +3303,8 @@ class dl:
                     # Track hybrid-processing outputs explicitly so we can always clean them up,
                     # even if muxing fails early (e.g. SystemExit) before the normal delete loop.
                     hybrid_temp_paths: list[Path] = []
+                    hybrid_base_ids: set[str] = set()
+                    container_paths: dict[str, Path] = {}
 
                     def clone_tracks_for_audio(base_tracks: Tracks, audio_tracks: list[Audio]) -> Tracks:
                         task_tracks = Tracks()
@@ -3356,7 +3383,10 @@ class dl:
                     def mux_video_group(video_tracks: list[Optional[Video]]) -> None:
                         for video_track in video_tracks:
                             if video_track and video_track.dv_compatible_bitstream:
-                                apply_dv_fixup(video_track)
+                                container = apply_dv_fixup(video_track, keep_source=video_track.id in hybrid_base_ids)
+                                if container:
+                                    container_paths[video_track.id] = container
+                                    hybrid_temp_paths.append(container)
 
                         task_description = "Multiplexing"
                         # All tracks in a merged group share height/range/codec, so describe from the first.
@@ -3403,55 +3433,45 @@ class dl:
                                     standalone_groups: list[list[Optional[Video]]] = [
                                         list(g) for g in group_videos_by_variant(standalone_videos, merge=merge_video)
                                     ]
+                                    dv_tracks = [v for v in title.tracks.videos if v.range == Video.Range.DV]
+                                    hybrid_bases = select_hybrid_bases(title.tracks.videos) if dv_tracks else []
+                                    hybrid_base_ids.update(v.id for v in hybrid_bases)
+
                                     for group in sorted(standalone_groups, key=group_file_size, reverse=True):
                                         mux_video_group(group)
 
-                                    resolutions_processed = set()
+                                    for hdr10_track in hybrid_bases:
-                                    base_tracks_list = sorted(
+                                        resolution = hdr10_track.height or 0
-                                        (
+                                        matching_dv = min(dv_tracks, key=lambda v: v.height)
-                                            v
+                                        container = container_paths.get(hdr10_track.id)
-                                            for v in title.tracks.videos
+                                        resolution_tracks = hybrid_source_tracks(hdr10_track, matching_dv, container)
-                                            if v.range in (Video.Range.HDR10P, Video.Range.HDR10)
-                                        ),
-                                        key=lambda v: v.height,
-                                        reverse=True,
-                                    )
-                                    dv_tracks = [v for v in title.tracks.videos if v.range == Video.Range.DV]
 
-                                    for hdr10_track in base_tracks_list:
+                                        hybrid_filename = f"HDR10-DV-{resolution}p.hevc"
-                                        resolution = hdr10_track.height
+                                        hybrid_output_path = config.directories.temp / hybrid_filename
-                                        if resolution in resolutions_processed:
+                                        hybrid_temp_paths.append(hybrid_output_path)
-                                            continue
 
-                                        matching_dv = min(dv_tracks, key=lambda v: v.height) if dv_tracks else None
+                                        Hybrid(resolution_tracks, self.service, output_name=hybrid_filename)
 
-                                        if matching_dv:
+                                        if container:
-                                            resolutions_processed.add(resolution)
+                                            try:
+                                                container.unlink(missing_ok=True)
+                                            except PermissionError:
+                                                pass  # still in hybrid_temp_paths, so the final cleanup tries again
 
-                                            resolution_tracks = [deepcopy(hdr10_track), deepcopy(matching_dv)]
+                                        task_description = f"Multiplexing Hybrid HDR10+DV {resolution}p"
-                                            for track in resolution_tracks:
+                                        task_tracks = (
-                                                track.needs_duration_fix = True
+                                            Tracks(title.tracks) + title.tracks.chapters + title.tracks.attachments
+                                        )
 
-                                            hybrid_filename = f"HDR10-DV-{resolution}p.hevc"
+                                        hybrid_track = deepcopy(hdr10_track)
-                                            hybrid_output_path = config.directories.temp / hybrid_filename
+                                        hybrid_track.id = f"hybrid_{hdr10_track.id}_{resolution}"
-                                            hybrid_temp_paths.append(hybrid_output_path)
+                                        hybrid_track.path = hybrid_output_path
+                                        hybrid_track.range = Video.Range.DV
+                                        hybrid_track.needs_duration_fix = True
+                                        title.tracks.add(hybrid_track)
+                                        task_tracks.videos = [hybrid_track]
 
-                                            Hybrid(resolution_tracks, self.service, output_name=hybrid_filename)
+                                        enqueue_mux_tasks(task_description, task_tracks, hybrid=True)
-
-                                            task_description = f"Multiplexing Hybrid HDR10+DV {resolution}p"
-                                            task_tracks = (
-                                                Tracks(title.tracks) + title.tracks.chapters + title.tracks.attachments
-                                            )
-
-                                            hybrid_track = deepcopy(hdr10_track)
-                                            hybrid_track.id = f"hybrid_{hdr10_track.id}_{resolution}"
-                                            hybrid_track.path = hybrid_output_path
-                                            hybrid_track.range = Video.Range.DV
-                                            hybrid_track.needs_duration_fix = True
-                                            title.tracks.add(hybrid_track)
-                                            task_tracks.videos = [hybrid_track]
-
-                                            enqueue_mux_tasks(task_description, task_tracks, hybrid=True)
 
                                     console.print()
                                 else:

unshackle/core/tracks/dv_fixup.py

@@ -15,7 +15,7 @@ from __future__ import annotations
 
 import logging
 from pathlib import Path
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Optional
 
 from rich.padding import Padding
 from rich.rule import Rule
@@ -83,18 +83,24 @@ class DVFixup:
         return fixed_hevc
 
 
-def apply_dv_fixup(video: "Video") -> None:
+def apply_dv_fixup(video: "Video", keep_source: bool = False) -> Optional[Path]:
-    """Do the DV fixup on `video` if flagged as DV-composite. Updates `video.path` in place
+    """Do the DV fixup on `video` if flagged as DV-composite and set `video.path` to the result.
-    and deletes the original source file so the standard mux cleanup handles the new path."""
+
+    Deletes the source file. With `keep_source`, returns its path for the caller to delete.
+    """
     if not getattr(video, "dv_compatible_bitstream", False):
-        return
+        return None
     if not video.path or not Path(video.path).exists():
-        return
+        return None
     original = Path(video.path)
     fixed = DVFixup(video).run()
-    if fixed != original:
+    if fixed == original:
-        video.path = fixed
+        return None
-        original.unlink(missing_ok=True)
+    video.path = fixed
+    if keep_source:
+        return original
+    original.unlink(missing_ok=True)
+    return None
 
 
 __all__ = ("DVFixup", "apply_dv_fixup")

unshackle/core/tracks/hybrid.py

@@ -232,8 +232,20 @@ class Hybrid:
                 self.log.warning("Could not probe video duration, skipping L5 crop detection")
                 return
 
-            duration_info = json.loads(result_duration.stdout)
+            duration_value = json.loads(result_duration.stdout).get("format", {}).get("duration")
-            duration = float(duration_info["format"]["duration"])
+            if duration_value is None:
+                log_event(
+                    "hybrid_level5",
+                    level="WARNING",
+                    message="Video has no container duration; RPU keeps the active area of the DV track",
+                    context={"input": str(input_video)},
+                )
+                self.log.warning(
+                    "Video has no container duration, skipping L5 crop detection. "
+                    "The RPU keeps the active area of the DV track."
+                )
+                return
+            duration = float(duration_value)
 
             result_streams = subprocess.run(
                 [