feat(tracks): probe dv tracks for hdr10+ and list them as DV+HDR10P

A DV track that carries HDR10+ keeps its DV range and gets Video.hdr10plus, sent in the REST track JSON and to remote clients.

docs/dev/rest-api/endpoints.md

@@ -332,7 +332,7 @@ When you give both `season` and `episode`, the server combines them into a `"{se
           "bitrate": 8000, "bitrate_bps": 8000000, "width": 3840, "height": 2160,
           "resolution": "2160p", "fps": "24", "range": "HDR10",
           "range_display": "HDR10", "scan_type": null, "closed_captions": [],
-          "dv_compatible_bitstream": false, "language": "en", "is_original_lang": true,
+          "dv_compatible_bitstream": false, "hdr10plus": false, "language": "en", "is_original_lang": true,
           "name": null, "needs_repack": false, "edition": [], "data": null,
           "drm": [], "descriptor": "DASH"
         }

docs/reference/configuration/download.md

@@ -379,6 +379,10 @@ reads the first fragment of HEVC HDR10 tracks from DASH, HLS fMP4 and one-file s
 whose first frame carries HDR10+ (SMPTE ST 2094-40) metadata becomes `HDR10P`. Then `-r HDR10P`,
 `--list` and [`--generate-dv`](../../guide/downloading.md#dolby-vision-from-hdr10) use it.
 
+unshackle also probes HEVC Dolby Vision tracks. A DV track that carries HDR10+ keeps its `DV`
+range, so `-r DV` and `-r HYBRID` select it as before, and `--list` shows it as `DV+HDR10P`. The
+HDR10 tracks and the DV tracks of a title are sampled as two separate sets.
+
 When two HDR10 tracks have the same resolution, they can be an HDR10 encode and an HDR10+ encode
 side by side, so unshackle probes every track. When each track has its own resolution, unshackle
 probes the two highest and the two lowest bitrates first:

tests/core/test_hdr10plus_probe.py

@@ -45,10 +45,11 @@ def test_encrypted_or_truncated_bytes_prove_nothing() -> None:
     assert fragment_has_hdr10plus(b"\x47" * 188) is None
 
 
-def hevc(track_id: str, height: int, bitrate: int = 1) -> Any:
+def hevc(track_id: str, height: int, bitrate: int = 1, range_: Video.Range = Video.Range.HDR10) -> Any:
     return SimpleNamespace(
         id=track_id,
-        range=Video.Range.HDR10,
+        range=range_,
+        hdr10plus=False,
         codec=Video.Codec.HEVC,
         width=height * 16 // 9,
         height=height,
@@ -61,6 +62,13 @@ def ladder(
     monkeypatch: pytest.MonkeyPatch, pattern: str, heights: list[int] | None = None
 ) -> tuple[list[Any], list[str]]:
     """HEVC HDR10 tracks, highest bitrate first; '+' marks the ones whose fragment carries HDR10+."""
+    heights = heights or [2160 - 100 * i for i in range(len(pattern))]
+    videos = [hevc(f"v{i}{mark}", height, 100 - i) for i, (mark, height) in enumerate(zip(pattern, heights))]
+    return videos, probe(monkeypatch, videos)
+
+
+def probe(monkeypatch: pytest.MonkeyPatch, videos: list[Any]) -> list[str]:
+    """Run the probe with a fake fragment: a track id that ends in '+' carries HDR10+. Returns the probed ids."""
     probed: list[str] = []
 
     def fake_fragment(video: Any, session: Any) -> bytes:
@@ -69,10 +77,8 @@ def ladder(
 
     monkeypatch.setattr(hdr10plus, "first_fragment", fake_fragment)
     monkeypatch.setattr(hdr10plus, "fragment_has_hdr10plus", lambda data: data == b"+")
-    heights = heights or [2160 - 100 * i for i in range(len(pattern))]
-    videos = [hevc(f"v{i}{mark}", height, 100 - i) for i, (mark, height) in enumerate(zip(pattern, heights))]
     hdr10plus.probe_hdr10plus(videos, None, log=logging.getLogger("test"), workers=4)  # type: ignore[arg-type]
-    return videos, probed
+    return probed
 
 
 def ranges(videos: list[Any]) -> str:
@@ -103,6 +109,24 @@ def test_two_encodes_at_one_resolution_are_all_probed(monkeypatch: pytest.Monkey
     assert len(probed) == 5
 
 
+def test_a_dv_ladder_keeps_its_range_and_is_sampled_apart_from_hdr10(monkeypatch: pytest.MonkeyPatch) -> None:
+    hdr = [hevc(f"h{i}-", 2160 - 100 * i, 100 - i) for i in range(6)]
+    dv = [hevc(f"d{i}+", 2160 - 100 * i, 100 - i, Video.Range.DV) for i in range(6)]
+    assert len(probe(monkeypatch, hdr + dv)) == 8
+    assert ranges(hdr) == "------"
+    assert all(v.range == Video.Range.DV and v.hdr10plus for v in dv)
+
+
+def test_a_dv_track_with_hdr10plus_lists_as_dv_plus_hdr10p() -> None:
+    video = Video(
+        url="https://x/v.mp4", language="en", codec=Video.Codec.HEVC, range_=Video.Range.DV, width=3840, height=2160
+    )
+    assert "[H.265, DV]" in str(video)
+    video.hdr10plus = True
+    assert "[H.265, DV+HDR10P]" in str(video)
+    assert Video.from_dict(video.to_dict()).hdr10plus
+
+
 def test_on_start_runs_only_when_there_is_a_track_to_probe(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setattr(hdr10plus, "first_fragment", lambda video, session: None)
     started: list[int] = []

unshackle/core/api/handlers.py

@@ -1744,10 +1744,11 @@ def serialize_video_track(track: Video, include_url: bool = False) -> Dict[str,
         "resolution": f"{track.width}x{track.height}" if track.width and track.height else None,
         "fps": track.fps if track.fps else None,
         "range": range_name,
-        "range_display": DYNAMIC_RANGE_MAP.get(range_name, range_name),
+        "range_display": DYNAMIC_RANGE_MAP.get(range_name, range_name) + ("+HDR10P" if track.hdr10plus else ""),
         "scan_type": enum_name(track.scan_type) if track.scan_type else None,
         "closed_captions": list(getattr(track, "closed_captions", None) or []),
         "dv_compatible_bitstream": bool(getattr(track, "dv_compatible_bitstream", False)),
+        "hdr10plus": bool(getattr(track, "hdr10plus", False)),
     }
     result.update(serialize_track_common(track))
     if include_url and hasattr(track, "url") and track.url:

unshackle/core/remote_service.py

@@ -281,6 +281,7 @@ def deserialize_video(data: Dict[str, Any]) -> Video:
         scan_type=enum_get(Video.ScanType, data.get("scan_type")),
         closed_captions=data.get("closed_captions") or None,
         dv_compatible_bitstream=bool(data.get("dv_compatible_bitstream", False)),
+        hdr10plus=bool(data.get("hdr10plus", False)),
     )
 
 

unshackle/core/tracks/video.py

@@ -233,6 +233,7 @@ class Video(Track):
         scan_type: Optional[Video.ScanType] = None,
         closed_captions: Optional[list[dict[str, Any]]] = None,
         dv_compatible_bitstream: bool = False,
+        hdr10plus: bool = False,
         **kwargs: Any,
     ) -> None:
         """
@@ -251,6 +252,7 @@ class Video(Track):
             fps: A number, float, or string representing the frames/s of the video.
                 Strings may represent numbers, floats, or a fraction (num/den).
                 All strings will be cast to either a number or float.
+            hdr10plus: The DV track also carries HDR10+ metadata. The HDR10+ probe sets it.
 
         Note: If codec, bitrate, width, height, or fps is not specified, unshackle can
         skip some checks or assume a value. Give as much information as possible.
@@ -299,6 +301,7 @@ class Video(Track):
         self.closed_captions: list[dict[str, Any]] = closed_captions or []
         self.needs_duration_fix = False
         self.dv_compatible_bitstream = dv_compatible_bitstream
+        self.hdr10plus = hdr10plus
         self.hybrid_base_only = False
 
     def to_dict(self) -> dict[str, Any]:
@@ -313,6 +316,7 @@ class Video(Track):
                 "fps": str(self.fps) if self.fps else None,
                 "scan_type": self.scan_type.name if self.scan_type else None,
                 "dv_compatible_bitstream": self.dv_compatible_bitstream,
+                "hdr10plus": self.hdr10plus,
             }
         )
         return data
@@ -330,15 +334,21 @@ class Video(Track):
             fps=data.get("fps"),
             scan_type=Video.ScanType[data["scan_type"]] if data.get("scan_type") else None,
             dv_compatible_bitstream=data.get("dv_compatible_bitstream", False),
+            hdr10plus=data.get("hdr10plus", False),
         )
 
+    @property
+    def range_label(self) -> str:
+        """The range name, with ``+HDR10P`` on a DV track that also carries HDR10+."""
+        return self.range.name + ("+HDR10P" if self.hdr10plus else "")
+
     def __str__(self) -> str:
         return " | ".join(
             filter(
                 bool,
                 [
                     "VID",
-                    "[" + (", ".join(filter(bool, [self.codec.value if self.codec else None, self.range.name]))) + "]",
+                    "[" + (", ".join(filter(bool, [self.codec.value if self.codec else None, self.range_label]))) + "]",
                     str(self.language),
                     ", ".join(
                         filter(

unshackle/core/utils/hdr10plus.py

@@ -146,26 +146,27 @@ def probe_hdr10plus(
     workers: int,
     on_start: Optional[Callable[[], None]] = None,
 ) -> None:
-    """Relabel HEVC HDR10 tracks whose first sample carries HDR10+ SEI as HDR10P.
+    """Relabel HEVC HDR10 tracks whose first sample carries HDR10+ SEI as HDR10P, and flag such DV tracks.
 
-    A ladder with one track per resolution is sampled at both ends first. Two tracks at one resolution
-    can be an HDR10 and an HDR10+ encode side by side, so such a ladder is probed whole. The rules are
-    in the ``hdr10plus_probe`` section of docs/reference/configuration/download.md. A failed probe never
-    changes a label. ``workers`` caps the probes in flight, and ``on_start`` runs once before the first.
+    A DV track keeps its DV range and gets ``hdr10plus`` set, so selection and hybrid pairing are unchanged.
+    HDR10 and DV are separate ladders. A ladder with one track per resolution is sampled at both ends
+    first. Two tracks at one resolution can be an HDR10 and an HDR10+ encode side by side, so such a
+    ladder is probed whole. The rules are in the ``hdr10plus_probe`` section of
+    docs/reference/configuration/download.md. A failed probe never changes a label. ``workers`` caps the
+    probes in flight, and ``on_start`` runs once before the first.
     """
     from unshackle.core.tracks import Video
 
-    candidates = sorted(
-        (v for v in videos if v.range == Video.Range.HDR10 and v.codec == Video.Codec.HEVC),
-        key=lambda v: v.bitrate or 0,
-        reverse=True,
-    )
-    if candidates and on_start:
+    ladders = [
+        sorted(
+            (v for v in videos if v.range == range_ and v.codec == Video.Codec.HEVC),
+            key=lambda v: v.bitrate or 0,
+            reverse=True,
+        )
+        for range_ in (Video.Range.HDR10, Video.Range.DV)
+    ]
+    if any(ladders) and on_start:
         on_start()
-    if len(candidates) <= 4 or len({(v.width, v.height) for v in candidates}) < len(candidates):
-        samples, rest = candidates, []
-    else:
-        samples, rest = candidates[:2] + candidates[-2:], candidates[2:-2]
 
     def probe(video: "Video") -> Optional[bool]:
         try:
@@ -177,18 +178,38 @@ def probe_hdr10plus(
         log.debug(f"{video.id}: HDR10+ probe {found}")
         return found
 
-    def probe_all(tracks: list["Video"]) -> list[Optional[bool]]:
+    def probe_all(tracks: list["Video"]) -> dict[int, Optional[bool]]:
         if not tracks:
-            return []
+            return {}
         with ThreadPoolExecutor(max_workers=min(workers, len(tracks))) as executor:
-            return list(executor.map(probe, tracks))
+            return dict(zip(map(id, tracks), executor.map(probe, tracks)))
 
-    results = probe_all(samples)
-    if any(results) and not all(results):
-        samples, results = samples + rest, results + probe_all(rest)
-    hits = candidates if all(results) else [v for v, found in zip(samples, results) if found]
-    for video in hits:
-        video.range = Video.Range.HDR10P
+    plans = []
+    for candidates in ladders:
+        if len(candidates) <= 4 or len({(v.width, v.height) for v in candidates}) < len(candidates):
+            plans.append((candidates, candidates, []))
+        else:
+            plans.append((candidates, candidates[:2] + candidates[-2:], candidates[2:-2]))
+
+    # one round for the samples of every ladder, then one for the rest of each mixed ladder
+    found = probe_all([v for _, samples, _ in plans for v in samples])
+    found.update(
+        probe_all(
+            [
+                v
+                for _, samples, rest in plans
+                if any(found[id(s)] for s in samples) and not all(found[id(s)] for s in samples)
+                for v in rest
+            ]
+        )
+    )
+    for candidates, samples, _ in plans:
+        hits = candidates if all(found[id(s)] for s in samples) else [v for v in candidates if found.get(id(v))]
+        for video in hits:
+            if video.range == Video.Range.DV:
+                video.hdr10plus = True
+            else:
+                video.range = Video.Range.HDR10P
 
 
 def probed_tracks(