mirror of
https://github.com/unshackle-dl/unshackle.git
synced 2026-11-11 19:45:06 +00:00
feat(hybrid): generate dolby vision from hdr10+ with --generate-dv
Builds per-scene L1 in unshackle (PQ mean, letterbox-aware) with a generate_dv_mode config (v29/v40/legacy); a failed hybrid no longer exits 0 with no video.
This commit is contained in:
imSp4rky
@@ -156,8 +156,15 @@ class TrackRequest:
|
||||
codecs: list[Video.Codec] = [] # empty == accept any codec
|
||||
ranges: list[Video.Range] = [Video.Range.SDR]
|
||||
best_available: bool = False
|
||||
generate_dv: bool = False # --generate-dv: make Dolby Vision from HDR10+
|
||||
```
|
||||
|
||||
!!! note "`generate_dv` and a hand-made HYBRID"
|
||||
A service that fetches the HYBRID variants itself (HDR10 and DV) must also fetch
|
||||
HDR10P when `track_request.generate_dv` is set and the title has no DV track.
|
||||
The conversion needs an HEVC HDR10+ track as its base.
|
||||
`get_tracks_for_variants` does this for you.
|
||||
|
||||
!!! note "Read the request, don't second-guess the user"
|
||||
Services may narrow `track_request` for hard technical constraints (e.g.
|
||||
*"this service delivers HDR only as HEVC"*), but must **not** filter
|
||||
|
||||
@@ -471,6 +471,7 @@ Make a download job. It requires `service` and `title_id`. Every other field is
|
||||
| `downloads` | int | `1` | Concurrent tracks. |
|
||||
| `worst` | boolean | `false` | Lowest bitrate within quality. |
|
||||
| `best_available` | boolean | `false` | Fall back to best available. With `all_drm`, a DRM system that does not license is a warning. |
|
||||
| `generate_dv` | boolean | `false` | With `HYBRID` in `range`, make Dolby Vision from the HDR10+ metadata when a title has no DV track. Needs `hdr10plus_tool` on the server. Without `HYBRID` in `range`, the request returns `400`. |
|
||||
| `repack` | boolean | `false` | Add REPACK tag. |
|
||||
| `tag` | string | `null` | Release group tag. |
|
||||
| `tmdb_id` / `imdb_id` / `tvdb_id` / `anilist_id` | - | `null` | External ID overrides. Each resolves its metadata directly instead of by a title search, and is used for tagging. Set `enrich` to also take the title, year and original language. Give at most one of `tmdb_id`, `imdb_id` and `tvdb_id`, since unshackle resolves the others from it. Sending two returns `400`. `anilist_id` still combines with one of them. `tmdb_id` and `tvdb_id` must be positive integers, `imdb_id` must look like `tt1375666`, and `anilist_id` must be a positive integer or a string like `mal:12345`, or the request returns `400`. An ID whose provider is unconfigured, such as `tmdb_id` with no `tmdb_api_key`, fails the job rather than returning `400`. `anilist_id` needs no API key. |
|
||||
|
||||
@@ -130,8 +130,8 @@ hood the client walks a remote session through its lifecycle.
|
||||
The client also downloads playlists and segments through the same proxy,
|
||||
as a local run does. `--no-proxy-download` and `--proxy-download` change
|
||||
that for downloads only.
|
||||
- Track-selection hints (`range_`, `vcodec`, `quality`, `best_available`) so
|
||||
the server fetches the right manifests
|
||||
- Track-selection hints (`range_`, `vcodec`, `quality`, `best_available`,
|
||||
`generate_dv`) so the server fetches the right manifests
|
||||
- Your language and audio codec selection (`-l`, `-vl`, `-al`, `-a`, `-fs`)
|
||||
in a `dl_params` object. Some services fetch one manifest
|
||||
for each audio language, and pick the languages from these values
|
||||
@@ -428,7 +428,7 @@ The server mounts all these routes, even in `--remote-only` mode. Paths use the
|
||||
answers CDM calls), `cdm_system_id` and `cdm_device_type` (Widevine only), `service_config`
|
||||
(the identity config keys the service lists in `client_config`), `client` (a freeform identity
|
||||
object the dashboard shows as sent), and the track-selection hints `range_`, `vcodec`,
|
||||
`quality`, `best_available`, plus arbitrary service CLI options
|
||||
`quality`, `best_available`, `generate_dv`, plus arbitrary service CLI options
|
||||
(`additionalProperties: true`).
|
||||
|
||||
`dl_params` holds the `dl` track selection: `lang`, `v_lang` and `a_lang`
|
||||
|
||||
@@ -133,8 +133,10 @@ See [Proxies and VPN](../guide/proxies-and-vpn.md).
|
||||
!!! note "`TrackRequest`"
|
||||
`self.track_request` is a small dataclass built from the CLI: `codecs`
|
||||
(from `--vcodec`, empty means "accept any"), `ranges` (from `--range`, default
|
||||
`[Video.Range.SDR]`), and `best_available`. Use it in `get_tracks` to decide which
|
||||
variants to fetch.
|
||||
`[Video.Range.SDR]`), `best_available`, and `generate_dv` (from `--generate-dv`).
|
||||
Use it in `get_tracks` to decide which variants to fetch. A service that fetches
|
||||
the HYBRID variants itself must also fetch HDR10P when `generate_dv` is set and
|
||||
the title has no DV track, because the conversion needs an HEVC HDR10+ base.
|
||||
|
||||
## Titles and the `Title` hierarchy
|
||||
|
||||
|
||||
@@ -80,6 +80,7 @@ unshackle dl [OPTIONS] SERVICE [SERVICE ARGS...]
|
||||
| `-vb-range`, `--vbitrate-range` | - | Video bitrate range in kbps, e.g. `6000-7000`; picks highest within. Mutually exclusive with `--vbitrate`. |
|
||||
| `-ab-range`, `--abitrate-range` | - | Audio bitrate range in kbps, e.g. `128-256`. Mutually exclusive with `--abitrate`. |
|
||||
| `-r`, `--range` | `SDR` | Colour range(s): `SDR`, `HLG`, `HDR10`, `HDR10P` (HDR10+), `DV`, `HYBRID`. |
|
||||
| `--generate-dv` | off | With `-r HYBRID`, make Dolby Vision from the HDR10+ metadata when a title has no Dolby Vision track. Needs `hdr10plus_tool`. See [Dolby Vision from HDR10+](downloading.md#dolby-vision-from-hdr10). |
|
||||
| `-c`, `--channels` | - | Audio channels; matches sub-layouts (5.1 ≈ 6.0). |
|
||||
| `-naa`, `--noatmos` | off | Exclude Dolby Atmos audio tracks. |
|
||||
| `--worst` | off | Pick the lowest bitrate within the requested quality. **Requires `-q`.** |
|
||||
|
||||
@@ -149,14 +149,54 @@ unshackle dl -q 2160 -r HDR10,DV EXAMPLE '...'
|
||||
`-r HYBRID` makes a single hybrid track by injecting the Dolby Vision RPU
|
||||
metadata onto an HDR10/HDR10+ base layer with **dovi_tool**.
|
||||
It requires the `dovi_tool` binary, resolved from unshackle's `binaries/` folder,
|
||||
the `binaries` folder of your user data directory, or your `PATH`. The normal case is a DV track plus an HDR10 or HDR10+ base. With HDR10+
|
||||
and no DV, unshackle converts the HDR10+ metadata to DV instead, which also needs
|
||||
`hdr10plus_tool`. With no DV and no HDR10+, the title fails.
|
||||
When HDR10+ is present, unshackle prefers it over HDR10 as the base layer.
|
||||
the `binaries` folder of your user data directory, or your `PATH`. A hybrid needs a DV track plus an HDR10 or
|
||||
HDR10+ base. With no DV track, the title fails, unless you add `--generate-dv` (see below).
|
||||
When HDR10+ is present, unshackle prefers it over HDR10 as the base layer. The base must be
|
||||
HEVC; a title whose HDR10/HDR10+ tracks are all in another codec fails before it downloads.
|
||||
If the hybrid step fails for one resolution, unshackle logs an error, skips that resolution
|
||||
and goes on with the other resolutions of the title. When no hybrid file at all was made,
|
||||
the title fails.
|
||||
|
||||
To keep an HDR10+ deliverable *alongside* the hybrid, request both ranges:
|
||||
`-r HYBRID,HDR10P`. `-r HYBRID` on its own muxes only the merged hybrid.
|
||||
|
||||
#### Dolby Vision from HDR10+
|
||||
|
||||
When a title has HDR10+ but no DV track, `-r HYBRID --generate-dv` makes a Dolby Vision
|
||||
Profile 8.1 track from the HDR10+ metadata. It needs `hdr10plus_tool` as well as `dovi_tool`.
|
||||
When a title has a DV track, unshackle uses that track and ignores the flag. When it has none,
|
||||
the flag also makes unshackle ask the service for its HDR10+ tracks.
|
||||
|
||||
```shell title="Generate Dolby Vision from HDR10+, and keep the HDR10+ file too"
|
||||
unshackle dl -q 2160 -r HYBRID,HDR10P --generate-dv EXAMPLE '...'
|
||||
```
|
||||
|
||||
Each HDR10+ scene becomes one Dolby Vision shot, with its own scene cut and its own Level 1
|
||||
brightness values (peak and average). unshackle calculates the average itself, as a PQ mean of
|
||||
the HDR10+ brightness distribution without the letterbox bars. A scene with no black pixels keeps
|
||||
its whole frame, so a title that changes aspect ratio is also correct. The `dovi_tool generate
|
||||
--hdr10plus-json` average makes most scenes too bright.
|
||||
|
||||
The [`generate_dv_mode`](../reference/configuration/download.md#generate_dv_mode) setting in
|
||||
`unshackle.yaml` changes how the average is set: `v29` (default), `v40` for the 10-nit floor of studio
|
||||
Dolby Vision, or `legacy` for the `dovi_tool` average.
|
||||
|
||||
!!! warning "Generated Dolby Vision is not studio Dolby Vision"
|
||||
- There are no colourist trims (Level 2 and Level 8). The TV does all the tone mapping
|
||||
from the Level 1 values.
|
||||
- Dolby Vision does not let the average go below about 2.4 nits (10 nits with
|
||||
`generate_dv_mode: v40`), so a very dark scene reads a little brighter than its pixels.
|
||||
- The HDR10+ tone-mapping curve (Profile B) has no Dolby Vision equivalent. The HDR10+
|
||||
metadata stays in the file, so an HDR10+ TV still reads it.
|
||||
- The result can be no better than the HDR10+ metadata. Profile A metadata is often less
|
||||
reliable than Profile B, and unshackle shows a warning for Profile A.
|
||||
- A TV or player that supports both formats almost always picks Dolby Vision. On such a
|
||||
TV, the generated Dolby Vision replaces the original HDR10+. To keep the original, also
|
||||
request `HDR10P`, as in the example above.
|
||||
|
||||
An HDR10 track that a service labels HDR10 but that carries HDR10+ metadata does not
|
||||
convert. Only a track with the `HDR10P` range does.
|
||||
|
||||
## Bitrate selection
|
||||
|
||||
By default unshackle keeps the highest-bitrate rendition for each selected
|
||||
|
||||
@@ -52,6 +52,7 @@ Common config keys. This is a useful subset, and every `dl` flag works:
|
||||
| `speed_limit` | str | *(unlimited)* | Total download speed cap across all downloads combined, e.g. `500k`, `5M`, `1.5G` or plain bytes/sec. Values are bytes, not bits (`5M` = 5.0 MB/s). |
|
||||
| `slow` | str | *(unset)* | Inter-title delay, e.g. `"20-40"`. |
|
||||
| `best_available` | bool | `false` | Warn instead of failing when a requested resolution, range, or language is missing, or when one DRM system does not license under `--all-drm`. The CLI flag is `--best-available`, also spelled `--warn-only`. |
|
||||
| `generate_dv` | bool | `false` | With `-r HYBRID`, make Dolby Vision from the HDR10+ metadata when a title has no DV track. Needs `hdr10plus_tool`. See [Dolby Vision from HDR10+](../../guide/downloading.md#dolby-vision-from-hdr10). |
|
||||
| `proxy` | str | *(unset)* | Default proxy URI or 2-letter country. |
|
||||
| `no_folder` | bool | `false` | Do not create a per-title folder. |
|
||||
|
||||
@@ -368,3 +369,24 @@ dual_multi_mode: strict
|
||||
!!! note "Dialects count as one language"
|
||||
Regional variants of the same base language (e.g. `en-US` and `en-GB`) collapse to a single
|
||||
language in both modes, so they never trigger `{dual}` on their own.
|
||||
|
||||
## `generate_dv_mode`
|
||||
|
||||
- **Type:** `str` · **Default:** `v29`
|
||||
|
||||
How [`--generate-dv`](../../guide/downloading.md#dolby-vision-from-hdr10) sets the Level 1 brightness
|
||||
values of the Dolby Vision it makes from HDR10+. All modes make one shot per HDR10+ scene and keep the
|
||||
same peak and letterbox area. They differ in the scene average, which the TV uses to tone-map.
|
||||
|
||||
| Mode | Scene average |
|
||||
|------|---------------|
|
||||
| `v29` (default) | unshackle's average from the HDR10+ brightness distribution, at least about 2.4 nits. Closest to the pixels. |
|
||||
| `v40` | The same average, at least about 10 nits, as in studio Dolby Vision (CM v4.0). Dark scenes read brighter than `v29`. |
|
||||
| `legacy` | The `dovi_tool generate --hdr10plus-json` average. Most scenes read brighter than the pixels. |
|
||||
|
||||
```yaml
|
||||
generate_dv_mode: v29
|
||||
```
|
||||
|
||||
The setting applies on the machine that runs the hybrid step: the client for a remote download, the
|
||||
server for a REST API job.
|
||||
|
||||
@@ -31,7 +31,7 @@ Where the file lives, how unshackle discovers it, and how to edit it:
|
||||
|------|------|
|
||||
| [Directories](directories.md) | `directories`, `filenames`, `services_repo_force` |
|
||||
| [Services & authentication](services.md) | `services`, `credentials`, `firefox_cookies`, `remote_services`, `serve` |
|
||||
| [Download & processing](download.md) | `dl`, `continue_downloads`, `merge_segments`, `subtitle`, `audio`, `muxing`, `language_tags`, `dual_multi_mode` |
|
||||
| [Download & processing](download.md) | `dl`, `continue_downloads`, `merge_segments`, `subtitle`, `audio`, `muxing`, `language_tags`, `dual_multi_mode`, `generate_dv_mode` |
|
||||
| [Output & naming](output.md) | `output_template`, `tag`, `tag_rules`, `tag_group_name`, `tag_imdb_tmdb`, `chapter_fallback_name`, `unicode_filenames` |
|
||||
| [DRM & CDM](drm.md) | `cdm`, `remote_cdm`, `decryption` |
|
||||
| [Network & proxy](network.md) | `network`, `headers`, `proxy_providers` |
|
||||
|
||||
@@ -20,7 +20,9 @@ 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)
|
||||
return SimpleNamespace(
|
||||
id=track_id, path=path, height=height, range=range_, codec=Video.Codec.HEVC, dv_compatible_bitstream=True
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
||||
@@ -0,0 +1,736 @@
|
||||
"""Unit tests for the HDR10+ to Dolby Vision conversion: the L1 built from HDR10+ histograms, the
|
||||
generator config, the active-area share, and the dl and REST API entry points. Nothing runs
|
||||
dovi_tool, hdr10plus_tool or FFmpeg."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import gc
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
import weakref
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from typing import Any, Iterator, Optional
|
||||
|
||||
import pytest
|
||||
|
||||
import unshackle.commands.dl as dl_module
|
||||
from tests.remote.unit.test_remote_service_helpers import _sent_create_data
|
||||
from unshackle.commands.dl import (
|
||||
hybrid_gate_failure,
|
||||
hybrid_plan,
|
||||
hybrid_source_tracks,
|
||||
hybrid_step_error,
|
||||
missing_hybrid_error,
|
||||
select_hybrid_bases,
|
||||
)
|
||||
from unshackle.core.api import download_manager, handlers
|
||||
from unshackle.core.api.handlers import validate_download_parameters
|
||||
from unshackle.core.config import config
|
||||
from unshackle.core.service import Service, TrackRequest
|
||||
from unshackle.core.services import Services
|
||||
from unshackle.core.tracks import Tracks, Video
|
||||
from unshackle.core.tracks import hybrid as hybrid_mod
|
||||
from unshackle.core.tracks.hybrid import Hybrid, hdr10plus_mean_pq, hdr10plus_shots, nits_to_pq
|
||||
|
||||
pytestmark = pytest.mark.unit
|
||||
|
||||
INDEXES = [1, 5, 10, 25, 50, 75, 90, 95, 99]
|
||||
|
||||
|
||||
def histogram(nits: dict[int, float]) -> dict[int, float]:
|
||||
"""A full 9-slot histogram: every distribution slot at 50 nits unless given."""
|
||||
return {index: nits.get(index, 50.0) for index in INDEXES}
|
||||
|
||||
|
||||
def scene_info(values: dict[int, float], average_rgb: float = 30.0) -> dict[str, Any]:
|
||||
return {
|
||||
"LuminanceParameters": {
|
||||
"AverageRGB": round(average_rgb * 10),
|
||||
"LuminanceDistributions": {
|
||||
"DistributionIndex": list(values),
|
||||
"DistributionValues": [round(v * 10) for v in values.values()],
|
||||
},
|
||||
"MaxScl": [0, 0, 0],
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
def metadata(
|
||||
scenes: list[tuple[int, dict[int, float]]], first_frame: int = 0, average_rgb: float = 30.0
|
||||
) -> dict[str, Any]:
|
||||
"""HDR10+ JSON in hdr10plus_tool's layout, one SceneInfo entry per frame."""
|
||||
frames = [scene_info(values, average_rgb) for length, values in scenes for _ in range(length)]
|
||||
firsts, frame = [], first_frame
|
||||
for length, _ in scenes:
|
||||
firsts.append(frame)
|
||||
frame += length
|
||||
return {
|
||||
"JSONInfo": {"HDR10plusProfile": "B", "Version": "1.0"},
|
||||
"SceneInfo": frames,
|
||||
"SceneInfoSummary": {"SceneFirstFrameIndex": firsts, "SceneFrameNumbers": [n for n, _ in scenes]},
|
||||
}
|
||||
|
||||
|
||||
def test_pq_matches_the_st2084_reference_points() -> None:
|
||||
assert round(nits_to_pq(100)) == 2081
|
||||
assert round(nits_to_pq(1000)) == 3079
|
||||
assert round(nits_to_pq(10000)) == 4095
|
||||
assert round(nits_to_pq(0)) == 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize("active", [1.0, 0.75])
|
||||
def test_a_flat_histogram_averages_to_its_own_level(active: float) -> None:
|
||||
values = {index: 120.0 for index in INDEXES}
|
||||
mean = hdr10plus_mean_pq(values, active)
|
||||
assert mean is not None and mean == pytest.approx(nits_to_pq(120.0))
|
||||
|
||||
|
||||
def test_the_slots_outside_the_distribution_do_not_move_the_mean() -> None:
|
||||
"""Index 5 is a luminance-Y value and index 10 a pixel share, so neither is a maxRGB percentile."""
|
||||
base = hdr10plus_mean_pq(histogram({5: 50.0, 10: 50.0}))
|
||||
changed = hdr10plus_mean_pq(histogram({5: 4000.0, 10: 9.7}))
|
||||
assert base == changed
|
||||
|
||||
|
||||
def test_the_mean_is_taken_in_pq_not_in_linear_light() -> None:
|
||||
"""Half the pixels at 1 nit and half at 1000: PQ of the linear mean is far brighter than the PQ mean."""
|
||||
values = histogram({1: 1.0, 25: 1.0, 50: 1.0, 75: 1000.0, 90: 1000.0, 95: 1000.0, 99: 1000.0})
|
||||
mean = hdr10plus_mean_pq(values)
|
||||
assert mean is not None
|
||||
assert mean < nits_to_pq(500.5) - 400
|
||||
|
||||
|
||||
LETTERBOXED = histogram({1: 0.0, 25: 20.0, 50: 40.0, 75: 80.0, 90: 150.0, 95: 200.0, 99: 400.0})
|
||||
|
||||
|
||||
def test_letterbox_bars_are_left_out_of_the_mean() -> None:
|
||||
"""Black up to the bar share and flat above it: the active mean is the plateau itself."""
|
||||
bars_then_plateau = histogram({1: 0.0, 25: 200.0, 50: 200.0, 75: 200.0, 90: 200.0, 95: 200.0, 99: 200.0})
|
||||
full, active = hdr10plus_mean_pq(bars_then_plateau, 1.0), hdr10plus_mean_pq(bars_then_plateau, 0.75)
|
||||
assert full is not None and active is not None
|
||||
assert active == pytest.approx(nits_to_pq(200.0))
|
||||
assert full < active
|
||||
|
||||
|
||||
def test_a_scene_with_no_black_pixels_keeps_its_whole_frame() -> None:
|
||||
"""A title that changes aspect ratio has full-frame scenes, and those have no bars to leave out."""
|
||||
full_frame = {**LETTERBOXED, 1: 0.5}
|
||||
assert hdr10plus_mean_pq(full_frame, 0.75) == hdr10plus_mean_pq(full_frame, 1.0)
|
||||
|
||||
|
||||
def test_a_ten_slot_histogram_uses_the_same_lookup() -> None:
|
||||
nine = histogram({99: 300.0})
|
||||
ten = {**nine, 98: 250.0}
|
||||
assert hdr10plus_mean_pq(ten) == hdr10plus_mean_pq(nine)
|
||||
|
||||
|
||||
def test_a_histogram_without_a_distribution_slot_has_no_mean() -> None:
|
||||
values = histogram({})
|
||||
del values[75]
|
||||
assert hdr10plus_mean_pq(values) is None
|
||||
|
||||
|
||||
def test_each_scene_becomes_one_shot_with_its_own_cut() -> None:
|
||||
meta = metadata([(3, histogram({99: 200.0})), (5, histogram({99: 800.0}))], first_frame=1000)
|
||||
|
||||
shots = hdr10plus_shots(meta)
|
||||
|
||||
assert [(s["start"], s["duration"]) for s in shots] == [(0, 3), (3, 5)]
|
||||
levels = [s["metadata_blocks"][0]["Level1"] for s in shots]
|
||||
assert [lv["max_pq"] for lv in levels] == [round(nits_to_pq(200.0)), round(nits_to_pq(800.0))]
|
||||
assert all(lv["min_pq"] == 0 and lv["avg_pq"] < lv["max_pq"] for lv in levels)
|
||||
|
||||
|
||||
def test_avg_is_the_pq_mean_and_not_the_average_rgb() -> None:
|
||||
"""dovi_tool's PQ(AverageMaxRGB) is far brighter than the PQ mean; the shot must carry the PQ mean."""
|
||||
meta = metadata([(2, LETTERBOXED)], average_rgb=150.0)
|
||||
|
||||
(shot,) = hdr10plus_shots(meta, 0.8)
|
||||
|
||||
mean = hdr10plus_mean_pq(LETTERBOXED, 0.8)
|
||||
assert mean is not None and mean != hdr10plus_mean_pq(LETTERBOXED, 1.0)
|
||||
assert shot["metadata_blocks"][0]["Level1"]["avg_pq"] == round(mean)
|
||||
assert round(mean) < round(nits_to_pq(150.0)) - 100
|
||||
|
||||
|
||||
def test_a_scene_without_a_full_distribution_falls_back_to_the_average_rgb() -> None:
|
||||
values = histogram({})
|
||||
del values[25]
|
||||
meta = metadata([(2, values)])
|
||||
|
||||
(shot,) = hdr10plus_shots(meta)
|
||||
|
||||
assert shot["metadata_blocks"][0]["Level1"]["avg_pq"] == round(nits_to_pq(30.0))
|
||||
|
||||
|
||||
def test_the_peak_skips_the_pixel_share_slot_but_keeps_the_luminance_slot() -> None:
|
||||
"""Index 10 holds a percentage (100 = every pixel under 100 nits), which would read as 10 nits in a
|
||||
near-black scene. Index 5 holds the 99.99% luminance, a real brightness that can top index 99."""
|
||||
dark = {index: 0.2 for index in INDEXES}
|
||||
dark[10] = 10.0
|
||||
(dark_shot,) = hdr10plus_shots(metadata([(2, dark)]))
|
||||
assert dark_shot["metadata_blocks"][0]["Level1"]["max_pq"] == round(nits_to_pq(0.2))
|
||||
|
||||
(bright_shot,) = hdr10plus_shots(metadata([(2, histogram({5: 1400.0, 99: 1200.0}))]))
|
||||
assert bright_shot["metadata_blocks"][0]["Level1"]["max_pq"] == round(nits_to_pq(1400.0))
|
||||
|
||||
|
||||
def test_an_empty_histogram_takes_its_peak_from_maxscl_and_caps_the_avg() -> None:
|
||||
meta = metadata([(2, {})], average_rgb=900.0)
|
||||
meta["SceneInfo"][0]["LuminanceParameters"]["MaxScl"] = [4000, 6000, 5000]
|
||||
|
||||
(shot,) = hdr10plus_shots(meta)
|
||||
|
||||
level1 = shot["metadata_blocks"][0]["Level1"]
|
||||
assert level1["max_pq"] == round(nits_to_pq(600.0))
|
||||
assert level1["avg_pq"] == level1["max_pq"]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def quiet(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||||
monkeypatch.setattr(config.directories, "temp", tmp_path)
|
||||
monkeypatch.setattr(config, "generate_dv_mode", "v29")
|
||||
monkeypatch.setattr(hybrid_mod, "log_event", lambda *a, **k: None)
|
||||
|
||||
@contextlib.contextmanager
|
||||
def no_status(*args: Any, **kwargs: Any) -> Iterator[None]:
|
||||
yield
|
||||
|
||||
monkeypatch.setattr(hybrid_mod.console, "status", no_status)
|
||||
|
||||
|
||||
def make_hybrid() -> Hybrid:
|
||||
"""Build a Hybrid without __init__, which would demux real video tracks."""
|
||||
obj = Hybrid.__new__(Hybrid)
|
||||
obj.log = logging.getLogger("hybrid-test")
|
||||
obj.rpu_file = "RPU.bin"
|
||||
obj.hdr10plus_file = "HDR10Plus.json"
|
||||
return obj
|
||||
|
||||
|
||||
@pytest.mark.parametrize("top, expected", [(277, (2160 - 554) / 2160), (1000, None)])
|
||||
def test_active_area_share(top: int, expected: Optional[float], quiet: None, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""A letterboxed picture gives its crop and share of the frame; an implausibly small one gives None.
|
||||
|
||||
cropdetect settles over its frames, so the last crop line counts.
|
||||
"""
|
||||
|
||||
def fake_run(args: list[Any], **kwargs: Any) -> SimpleNamespace:
|
||||
if "format=duration" in args:
|
||||
return SimpleNamespace(returncode=0, stdout='{"format": {"duration": "100"}}', stderr="")
|
||||
if "stream=width,height" in args:
|
||||
return SimpleNamespace(returncode=0, stdout='{"streams": [{"width": 3840, "height": 2160}]}', stderr="")
|
||||
return SimpleNamespace(
|
||||
returncode=0, stdout="", stderr=f"crop=3840:2160:0:0\ncrop=3840:{2160 - 2 * top}:0:{top}"
|
||||
)
|
||||
|
||||
monkeypatch.setattr(hybrid_mod.subprocess, "run", fake_run)
|
||||
|
||||
area = make_hybrid().detect_active_area(Path("base.mkv"))
|
||||
|
||||
if expected is None:
|
||||
assert area is None
|
||||
else:
|
||||
assert area is not None
|
||||
assert area[0] == (0, top, 0, top)
|
||||
assert area[1] == pytest.approx(expected)
|
||||
|
||||
|
||||
def test_each_crop_sample_evaluates_enough_frames(quiet: None, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""cropdetect ignores its first two frames (skip=2) and only grows the picture over the rest."""
|
||||
cropdetect_calls: list[list[str]] = []
|
||||
|
||||
def fake_run(args: list[Any], **kwargs: Any) -> SimpleNamespace:
|
||||
if "format=duration" in args:
|
||||
return SimpleNamespace(returncode=0, stdout='{"format": {"duration": "100"}}', stderr="")
|
||||
if "stream=width,height" in args:
|
||||
return SimpleNamespace(returncode=0, stdout='{"streams": [{"width": 3840, "height": 2160}]}', stderr="")
|
||||
cropdetect_calls.append([str(a) for a in args])
|
||||
return SimpleNamespace(returncode=0, stdout="", stderr="crop=3840:1600:0:280")
|
||||
|
||||
monkeypatch.setattr(hybrid_mod.subprocess, "run", fake_run)
|
||||
|
||||
make_hybrid().detect_active_area(Path("base.mkv"))
|
||||
|
||||
assert cropdetect_calls
|
||||
for args in cropdetect_calls:
|
||||
filtergraph = args[args.index("-vf") + 1]
|
||||
skip_match = re.search(r"skip=(\d+)", filtergraph)
|
||||
skip = int(skip_match.group(1)) if skip_match else 2
|
||||
assert int(args[args.index("-vframes") + 1]) - skip >= 8
|
||||
|
||||
|
||||
def test_the_parsed_metadata_is_released_before_dovi_tool_runs(
|
||||
quiet: None, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
"""hdr10plus_tool writes one entry per frame, so a film's parsed JSON is hundreds of MB of objects."""
|
||||
|
||||
class Tracked(dict): # type: ignore[type-arg]
|
||||
pass
|
||||
|
||||
(tmp_path / "HDR10Plus.json").write_text(json.dumps(metadata([(2, histogram({}))])))
|
||||
hybrid = make_hybrid()
|
||||
monkeypatch.setattr(hybrid, "probe_hdr_metadata", lambda: (1000, 1, 0, 0))
|
||||
real_load = json.load
|
||||
refs: list[weakref.ReferenceType[Tracked]] = []
|
||||
|
||||
def tracked_load(fp: Any) -> Tracked:
|
||||
loaded = Tracked(real_load(fp))
|
||||
refs.append(weakref.ref(loaded))
|
||||
return loaded
|
||||
|
||||
def fake_generate(config_json: Path, output: Path, **kwargs: Any) -> bytes:
|
||||
gc.collect()
|
||||
assert refs and refs[0]() is None, "the parsed HDR10+ metadata is still alive"
|
||||
output.write_bytes(b"rpu")
|
||||
return b""
|
||||
|
||||
monkeypatch.setattr(hybrid_mod.json, "load", tracked_load)
|
||||
monkeypatch.setattr(hybrid_mod.dovi, "generate", fake_generate)
|
||||
|
||||
hybrid.convert_hdr10plus_to_dv()
|
||||
|
||||
|
||||
def test_conversion_writes_a_cm29_config_with_shots_and_cleans_up(
|
||||
quiet: None, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
(tmp_path / "HDR10Plus.json").write_text(json.dumps(metadata([(4, LETTERBOXED), (2, histogram({99: 900.0}))])))
|
||||
hybrid = make_hybrid()
|
||||
monkeypatch.setattr(hybrid, "probe_hdr_metadata", lambda: (1000, 1, 650, 120))
|
||||
seen: dict[str, Any] = {}
|
||||
|
||||
def fake_generate(config_json: Path, output: Path, **kwargs: Any) -> bytes:
|
||||
seen.update(json.loads(config_json.read_text()))
|
||||
output.write_bytes(b"rpu")
|
||||
return b""
|
||||
|
||||
monkeypatch.setattr(hybrid_mod.dovi, "generate", fake_generate)
|
||||
|
||||
hybrid.convert_hdr10plus_to_dv(0.9)
|
||||
|
||||
assert seen["cm_version"] == "V29"
|
||||
assert seen["level6"]["max_content_light_level"] == 650
|
||||
assert [s["duration"] for s in seen["shots"]] == [4, 2]
|
||||
assert seen["shots"][0]["metadata_blocks"][0]["Level1"]["avg_pq"] == round(hdr10plus_mean_pq(LETTERBOXED, 0.9) or 0)
|
||||
assert "length" not in seen
|
||||
assert (tmp_path / "RPU.bin").exists()
|
||||
assert not (tmp_path / "HDR10Plus.json").exists() and not (tmp_path / "generate.json").exists()
|
||||
|
||||
|
||||
def run_conversion(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, mode: str) -> tuple[dict, dict]:
|
||||
"""Convert a two-scene HDR10+ file in the given generate_dv_mode; return the config and generate's kwargs."""
|
||||
monkeypatch.setattr(config, "generate_dv_mode", mode)
|
||||
(tmp_path / "HDR10Plus.json").write_text(json.dumps(metadata([(4, LETTERBOXED), (2, histogram({99: 900.0}))])))
|
||||
hybrid = make_hybrid()
|
||||
monkeypatch.setattr(hybrid, "probe_hdr_metadata", lambda: (1000, 1, 650, 120))
|
||||
seen: dict[str, Any] = {}
|
||||
called: dict[str, Any] = {}
|
||||
|
||||
def fake_generate(config_json: Path, output: Path, **kwargs: Any) -> bytes:
|
||||
seen.update(json.loads(config_json.read_text()))
|
||||
called.update(
|
||||
kwargs, hdr10plus_existed=bool(kwargs.get("hdr10plus_json") and kwargs["hdr10plus_json"].exists())
|
||||
)
|
||||
output.write_bytes(b"rpu")
|
||||
return b""
|
||||
|
||||
monkeypatch.setattr(hybrid_mod.dovi, "generate", fake_generate)
|
||||
hybrid.convert_hdr10plus_to_dv(0.9)
|
||||
return seen, called
|
||||
|
||||
|
||||
def test_v40_mode_keeps_the_shots_and_asks_for_the_cm40_avg_floor(
|
||||
quiet: None, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
seen, called = run_conversion(monkeypatch, tmp_path, "v40")
|
||||
|
||||
assert seen["cm_version"] == "V29" and seen["l1_avg_pq_cm_version"] == "V40"
|
||||
assert [s["duration"] for s in seen["shots"]] == [4, 2]
|
||||
assert not called.get("hdr10plus_json")
|
||||
|
||||
|
||||
def test_v29_mode_leaves_the_avg_floor_to_the_cm_version(
|
||||
quiet: None, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
seen, called = run_conversion(monkeypatch, tmp_path, "v29")
|
||||
|
||||
assert "l1_avg_pq_cm_version" not in seen and not called.get("hdr10plus_json")
|
||||
|
||||
|
||||
def test_legacy_mode_hands_the_hdr10plus_json_to_dovi_tool(
|
||||
quiet: None, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
seen, called = run_conversion(monkeypatch, tmp_path, "legacy")
|
||||
|
||||
assert "shots" not in seen and seen["length"] == 0 and seen["level6"]["max_content_light_level"] == 650
|
||||
assert called["hdr10plus_existed"]
|
||||
assert not (tmp_path / "HDR10Plus.json").exists() and not (tmp_path / "generate.json").exists()
|
||||
|
||||
|
||||
def test_an_unknown_mode_is_refused_before_dovi_tool_runs(
|
||||
quiet: None, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
with pytest.raises(ValueError, match="v29, v40, legacy"):
|
||||
run_conversion(monkeypatch, tmp_path, "v50")
|
||||
|
||||
|
||||
def test_a_failed_generate_still_removes_the_temp_files(
|
||||
quiet: None, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
(tmp_path / "HDR10Plus.json").write_text(json.dumps(metadata([(2, histogram({}))])))
|
||||
hybrid = make_hybrid()
|
||||
monkeypatch.setattr(hybrid, "probe_hdr_metadata", lambda: (1000, 1, 0, 0))
|
||||
|
||||
def failed_generate(*args: Any, **kwargs: Any) -> bytes:
|
||||
raise RuntimeError("dovi_tool failed")
|
||||
|
||||
monkeypatch.setattr(hybrid_mod.dovi, "generate", failed_generate)
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
hybrid.convert_hdr10plus_to_dv()
|
||||
assert not (tmp_path / "HDR10Plus.json").exists() and not (tmp_path / "generate.json").exists()
|
||||
|
||||
|
||||
def test_a_conversion_run_removes_its_rpus_so_the_next_base_makes_its_own(
|
||||
quiet: None, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
"""Each base's RPU comes from its own HDR10+, and level_6 reuses an RPU_L6.bin that it finds."""
|
||||
base = tmp_path / "base.mkv"
|
||||
base.write_bytes(b"")
|
||||
(tmp_path / "RPU_L6.bin").write_bytes(b"stale")
|
||||
seen: dict[str, Any] = {}
|
||||
|
||||
def convert(self: Hybrid, active_fraction: float = 1.0) -> None:
|
||||
seen["stale_l6"] = (tmp_path / "RPU_L6.bin").exists()
|
||||
seen["active_fraction"] = active_fraction
|
||||
(tmp_path / "RPU.bin").write_bytes(b"rpu")
|
||||
|
||||
def level_6(self: Hybrid) -> None:
|
||||
(tmp_path / "RPU_L6.bin").write_bytes(b"rpu")
|
||||
|
||||
for name in ("extract_stream", "extract_hdr10plus", "level_5", "injecting"):
|
||||
monkeypatch.setattr(Hybrid, name, lambda self, *a, **k: None)
|
||||
monkeypatch.setattr(Hybrid, "detect_active_area", lambda self, path: ((0, 100, 0, 100), 0.9))
|
||||
monkeypatch.setattr(Hybrid, "convert_hdr10plus_to_dv", convert)
|
||||
monkeypatch.setattr(Hybrid, "level_6", level_6)
|
||||
monkeypatch.setattr(hybrid_mod.console, "print", lambda *a, **k: None)
|
||||
|
||||
Hybrid([SimpleNamespace(id="b", path=base, height=2160, range=Video.Range.HDR10P)], "TEST")
|
||||
|
||||
assert seen == {"stale_l6": False, "active_fraction": 0.9}
|
||||
assert not (tmp_path / "RPU.bin").exists() and not (tmp_path / "RPU_L6.bin").exists()
|
||||
|
||||
|
||||
def test_a_failed_resolution_removes_its_crop_edits_and_keeps_the_dv_files(
|
||||
quiet: None, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
"""level_5 reuses an RPU_L5.bin it finds, so a failure must not leave the crop of another resolution.
|
||||
|
||||
The RPU and DV stream come from the same DV track for every resolution, so those stay.
|
||||
"""
|
||||
for name in ("base.mkv", "dv.mkv", "DV.hevc"):
|
||||
(tmp_path / name).write_bytes(b"")
|
||||
writes = {"extract_rpu": "RPU.bin", "level_6": "RPU_L6.bin", "level_5": "RPU_L5.bin"}
|
||||
for name, written in writes.items():
|
||||
monkeypatch.setattr(Hybrid, name, lambda self, *a, _w=written, **k: (tmp_path / _w).write_bytes(b"x"))
|
||||
monkeypatch.setattr(Hybrid, "extract_stream", lambda self, *a, **k: (tmp_path / "HDR10.hevc").write_bytes(b""))
|
||||
monkeypatch.setattr(Hybrid, "detect_active_area", lambda self, path: ((0, 100, 0, 100), 0.9))
|
||||
|
||||
def failed_inject(self: Hybrid) -> None:
|
||||
(tmp_path / "L5.json").write_text("{}")
|
||||
raise ValueError("dovi_tool failed")
|
||||
|
||||
monkeypatch.setattr(Hybrid, "injecting", failed_inject)
|
||||
monkeypatch.setattr(hybrid_mod.console, "print", lambda *a, **k: None)
|
||||
videos = [
|
||||
SimpleNamespace(id="b", path=tmp_path / "base.mkv", height=2160, range=Video.Range.HDR10),
|
||||
SimpleNamespace(id="d", path=tmp_path / "dv.mkv", height=1080, range=Video.Range.DV),
|
||||
]
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
Hybrid(videos, "TEST")
|
||||
|
||||
left = sorted(p.name for p in tmp_path.iterdir())
|
||||
assert left == ["DV.hevc", "RPU.bin", "RPU_L6.bin", "base.mkv", "dv.mkv"]
|
||||
|
||||
|
||||
def test_a_failed_extract_raises_instead_of_exiting(quiet: None, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""A SystemExit would kill a REST worker; the title must be able to go on without this resolution."""
|
||||
|
||||
def failed_step(*args: Any, **kwargs: Any) -> None:
|
||||
raise RuntimeError("ffmpeg failed")
|
||||
|
||||
monkeypatch.setattr(hybrid_mod, "run_step", failed_step)
|
||||
|
||||
with pytest.raises(ValueError, match="HDR10 stream"):
|
||||
make_hybrid().extract_stream(Path("base.mkv"), "HDR10")
|
||||
|
||||
|
||||
def test_an_empty_hdr10plus_file_is_removed_before_the_error(
|
||||
quiet: None, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
"""The next resolution returns early on a metadata file it finds, so an empty one must not stay."""
|
||||
monkeypatch.setattr(hybrid_mod, "HDR10PlusTool", "hdr10plus_tool")
|
||||
monkeypatch.setattr(hybrid_mod, "run_step", lambda *a, output, **k: output.write_bytes(b""))
|
||||
|
||||
with pytest.raises(ValueError, match="No HDR10\\+ metadata"):
|
||||
make_hybrid().extract_hdr10plus(SimpleNamespace(id="b"))
|
||||
assert not (tmp_path / "HDR10Plus.json").exists()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"error", [ValueError("bad"), KeyError("duration"), FileNotFoundError("tool"), RuntimeError("x")]
|
||||
)
|
||||
def test_a_hybrid_failure_is_reported_so_the_title_goes_on(monkeypatch: pytest.MonkeyPatch, error: Exception) -> None:
|
||||
def failed_hybrid(*args: Any, **kwargs: Any) -> None:
|
||||
raise error
|
||||
|
||||
monkeypatch.setattr(dl_module, "Hybrid", failed_hybrid)
|
||||
|
||||
assert hybrid_step_error([], "TEST", "HDR10-DV-2160p.hevc") == str(error)
|
||||
|
||||
|
||||
def test_a_working_hybrid_reports_no_error_and_an_exit_passes_through(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
seen: dict[str, Any] = {}
|
||||
monkeypatch.setattr(dl_module, "Hybrid", lambda tracks, source, output_name: seen.update(name=output_name))
|
||||
assert hybrid_step_error([], "TEST", "HDR10-DV-2160p.hevc") is None and seen == {"name": "HDR10-DV-2160p.hevc"}
|
||||
|
||||
def exiting(*args: Any, **kwargs: Any) -> None:
|
||||
raise SystemExit(1)
|
||||
|
||||
monkeypatch.setattr(dl_module, "Hybrid", exiting)
|
||||
with pytest.raises(SystemExit):
|
||||
hybrid_step_error([], "TEST", "HDR10-DV-2160p.hevc")
|
||||
|
||||
|
||||
def test_a_title_without_a_hybrid_file_fails() -> None:
|
||||
bases = [video("h2160", Video.Range.HDR10), video("h1080", Video.Range.HDR10, 1080)]
|
||||
assert missing_hybrid_error(bases, 1) is None
|
||||
assert missing_hybrid_error([], 0) == "No HEVC HDR10 or HDR10+ track to make a hybrid from"
|
||||
assert missing_hybrid_error(bases, 0) == ("Every hybrid failed (2160p, 1080p), so the title has no hybrid file")
|
||||
|
||||
|
||||
def test_only_hevc_tracks_are_hybrid_bases() -> None:
|
||||
av1 = SimpleNamespace(id="av1", height=2160, range=Video.Range.HDR10P, codec=Video.Codec.AV1)
|
||||
hevc = SimpleNamespace(id="hevc", height=1080, range=Video.Range.HDR10P, codec=Video.Codec.HEVC)
|
||||
|
||||
assert [v.id for v in select_hybrid_bases([av1, hevc])] == ["hevc"] # type: ignore[list-item]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"dv_available, to_dv, hdr10_gives, fetched, kept",
|
||||
[
|
||||
(True, True, ["HDR10"], ["HDR10", "DV"], ["DV", "HDR10"]),
|
||||
(False, False, ["HDR10"], ["HDR10", "DV"], ["HDR10"]),
|
||||
(False, True, ["HDR10"], ["HDR10", "DV", "HDR10P"], ["HDR10", "HDR10P"]),
|
||||
(False, True, ["HDR10", "HDR10P"], ["HDR10", "DV"], ["HDR10", "HDR10P"]),
|
||||
],
|
||||
)
|
||||
def test_hybrid_asks_for_hdr10plus_only_for_a_conversion_without_dv(
|
||||
dv_available: bool, to_dv: bool, hdr10_gives: list[str], fetched: list[str], kept: list[str]
|
||||
) -> None:
|
||||
"""The HDR10 request may not return HDR10+, and the conversion needs it as its base.
|
||||
|
||||
When that request already returned HDR10+, there is nothing more to ask for.
|
||||
"""
|
||||
calls: list[str] = []
|
||||
|
||||
def fetch(title: Any, codec: Any, range_: Video.Range) -> Tracks:
|
||||
calls.append(range_.name)
|
||||
if range_ == Video.Range.DV and not dv_available:
|
||||
raise ValueError("no DV")
|
||||
tracks = Tracks()
|
||||
for name in hdr10_gives if range_ == Video.Range.HDR10 else [range_.name]:
|
||||
tracks.add(Video(url="https://example.invalid/v", language="en", id_=name, range_=Video.Range[name]))
|
||||
return tracks
|
||||
|
||||
service = SimpleNamespace(
|
||||
track_request=TrackRequest(ranges=[Video.Range.HYBRID], generate_dv=to_dv),
|
||||
log=logging.getLogger("variants-test"),
|
||||
)
|
||||
|
||||
tracks = Service.get_tracks_for_variants(service, None, fetch) # type: ignore[arg-type]
|
||||
|
||||
assert calls == fetched
|
||||
assert sorted(v.range.name for v in tracks.videos) == kept
|
||||
|
||||
|
||||
def test_the_hybrid_step_gets_the_base_alone_when_there_is_no_dv(tmp_path: Path) -> None:
|
||||
base = SimpleNamespace(id="base", path=tmp_path / "base.mp4", height=2160, range=Video.Range.HDR10P)
|
||||
|
||||
tracks = hybrid_source_tracks(base, None) # type: ignore[arg-type]
|
||||
|
||||
assert len(tracks) == 1 and tracks[0].range == Video.Range.HDR10P and tracks[0].needs_duration_fix
|
||||
|
||||
|
||||
def test_the_api_needs_hybrid_for_generate_dv() -> None:
|
||||
assert "HYBRID" in (validate_download_parameters({"generate_dv": True}) or "")
|
||||
assert "HYBRID" in (validate_download_parameters({"generate_dv": True, "range": ["HDR10P"]}) or "")
|
||||
assert validate_download_parameters({"generate_dv": True, "range": ["HYBRID"]}) is None
|
||||
assert "boolean" in (validate_download_parameters({"generate_dv": "yes", "range": ["HYBRID"]}) or "")
|
||||
|
||||
|
||||
def test_the_api_takes_hybrid_from_the_serve_default_range(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(config, "serve", {"range": ["HYBRID"]})
|
||||
assert validate_download_parameters({"generate_dv": True}) is None
|
||||
assert "HYBRID" in (validate_download_parameters({"generate_dv": True, "range": ["HDR10P"]}) or "")
|
||||
|
||||
|
||||
class _Captured(Exception):
|
||||
pass
|
||||
|
||||
|
||||
@pytest.mark.parametrize("to_dv", [False, True])
|
||||
def test_a_rest_job_gives_the_flag_to_the_service_ctx_and_to_dl(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, to_dv: bool
|
||||
) -> None:
|
||||
"""The REST manager calls dl.result() directly, so the flag must take both of the CLI's paths."""
|
||||
import click
|
||||
|
||||
class FakeDl:
|
||||
cli = click.Command("dl")
|
||||
|
||||
def __init__(self, ctx: click.Context, **kwargs: object) -> None:
|
||||
self.ctx = ctx
|
||||
|
||||
def result(self, **kwargs: object) -> None:
|
||||
raise _Captured(self.ctx.params["generate_dv"], kwargs["generate_dv"])
|
||||
|
||||
class FakeService:
|
||||
def __init__(self, ctx: click.Context) -> None:
|
||||
pass
|
||||
|
||||
monkeypatch.setattr(dl_module, "dl", FakeDl)
|
||||
monkeypatch.setattr(Services, "get_path", staticmethod(lambda s: tmp_path))
|
||||
monkeypatch.setattr(Services, "load", staticmethod(lambda s: FakeService))
|
||||
monkeypatch.setattr(handlers, "load_full_cdm", lambda *a: None)
|
||||
params = {"generate_dv": True} if to_dv else {}
|
||||
|
||||
with pytest.raises(_Captured) as exc:
|
||||
download_manager.perform_download("job-1", "EXAMPLE", "t1", params)
|
||||
|
||||
assert exc.value.args == (to_dv, to_dv)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("to_dv", [False, True])
|
||||
def test_the_remote_client_forwards_the_flag_only_when_set(to_dv: bool) -> None:
|
||||
data = _sent_create_data({"generate_dv": to_dv}, {})
|
||||
assert data.get("generate_dv", False) is to_dv
|
||||
|
||||
|
||||
@pytest.mark.parametrize("to_dv", [False, True])
|
||||
def test_a_remote_session_puts_the_flag_in_the_service_ctx(monkeypatch: pytest.MonkeyPatch, to_dv: bool) -> None:
|
||||
seen: dict[str, Any] = {}
|
||||
|
||||
def fake_instantiate(parent_ctx: Any, *args: Any) -> object:
|
||||
seen["ctx"] = parent_ctx
|
||||
return object()
|
||||
|
||||
monkeypatch.setattr(handlers, "load_service_yaml", lambda service: {})
|
||||
monkeypatch.setattr(handlers, "load_full_cdm", lambda *args: None)
|
||||
monkeypatch.setattr(handlers.Services, "load", lambda service: None)
|
||||
monkeypatch.setattr(handlers, "instantiate_service", fake_instantiate)
|
||||
data = {"generate_dv": True} if to_dv else {}
|
||||
|
||||
handlers.create_service_instance("EXAMPLE", "title", data, None, [], None)
|
||||
|
||||
assert seen["ctx"].params["generate_dv"] is to_dv
|
||||
|
||||
|
||||
def video(id_: str, range_: Video.Range, height: int = 2160, codec: Optional[Video.Codec] = Video.Codec.HEVC) -> Any:
|
||||
return SimpleNamespace(id=id_, height=height, range=range_, codec=codec)
|
||||
|
||||
|
||||
def test_hybrid_fails_with_neither_a_base_nor_dv() -> None:
|
||||
assert hybrid_gate_failure([video("sdr", Video.Range.SDR)], True) == (
|
||||
"HYBRID mode requires both HDR10/HDR10+ and DV tracks, but neither is available",
|
||||
"Available ranges: SDR",
|
||||
)
|
||||
assert hybrid_gate_failure([], True) == (
|
||||
"HYBRID mode requires both HDR10/HDR10+ and DV tracks, but neither is available",
|
||||
"Available ranges: none",
|
||||
)
|
||||
|
||||
|
||||
def test_hybrid_fails_with_dv_only() -> None:
|
||||
assert hybrid_gate_failure([video("dv", Video.Range.DV)], True) == (
|
||||
"HYBRID mode requires both HDR10/HDR10+ and DV tracks, but only DV is available",
|
||||
"Available ranges: DV",
|
||||
)
|
||||
|
||||
|
||||
def test_hybrid_goes_on_without_dv_when_the_flag_has_an_hdr10plus_base() -> None:
|
||||
assert hybrid_gate_failure([video("p", Video.Range.HDR10P), video("h", Video.Range.HDR10)], True) is None
|
||||
|
||||
|
||||
def test_hybrid_goes_on_with_a_base_and_dv_whatever_the_flag() -> None:
|
||||
videos = [video("h", Video.Range.HDR10), video("dv", Video.Range.DV)]
|
||||
assert hybrid_gate_failure(videos, False) is None
|
||||
assert hybrid_gate_failure(videos, True) is None
|
||||
|
||||
|
||||
def test_hybrid_fails_up_front_when_no_base_is_hevc() -> None:
|
||||
"""dovi_tool injects into HEVC, so an all-AV1 title fails before it downloads anything."""
|
||||
videos = [video("av1", Video.Range.HDR10P, codec=Video.Codec.AV1), video("dv", Video.Range.DV)]
|
||||
assert hybrid_gate_failure(videos, False) == (
|
||||
"HYBRID mode requires an HEVC HDR10/HDR10+ base, but none is available",
|
||||
"Available HDR10/HDR10+ codecs: AV1",
|
||||
)
|
||||
assert hybrid_gate_failure(videos + [video("h", Video.Range.HDR10, codec=None)], False) is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"ranges, to_dv, detail",
|
||||
[
|
||||
(
|
||||
[Video.Range.HDR10P],
|
||||
False,
|
||||
"Available ranges: HDR10P. Add --generate-dv to generate Dolby Vision from HDR10+.",
|
||||
),
|
||||
([Video.Range.HDR10], False, "Available ranges: HDR10"),
|
||||
(
|
||||
[Video.Range.HDR10],
|
||||
True,
|
||||
"Available ranges: HDR10. --generate-dv needs an HEVC HDR10+ track, and the service returned none.",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_hybrid_fails_without_dv_and_hints_at_the_flag_only_when_it_is_off(
|
||||
ranges: list[Video.Range], to_dv: bool, detail: str
|
||||
) -> None:
|
||||
"""A service that fetches HYBRID by hand may never return HDR10+; the hint must not blame the user then."""
|
||||
videos = [video(r.name, r) for r in ranges]
|
||||
assert hybrid_gate_failure(videos, to_dv) == (
|
||||
"HYBRID mode requires both HDR10/HDR10+ and DV tracks, but only HDR10 is available",
|
||||
detail,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("to_dv", [False, True])
|
||||
def test_real_dv_takes_every_base_and_the_lowest_dv(to_dv: bool) -> None:
|
||||
videos = [
|
||||
video("h2160", Video.Range.HDR10),
|
||||
video("p1080", Video.Range.HDR10P, 1080),
|
||||
video("dv2160", Video.Range.DV),
|
||||
video("dv1080", Video.Range.DV, 1080),
|
||||
]
|
||||
|
||||
bases, dv = hybrid_plan(videos, to_dv)
|
||||
|
||||
assert [v.id for v in bases] == ["h2160", "p1080"]
|
||||
assert dv is not None and dv.id == "dv1080"
|
||||
|
||||
|
||||
def test_the_flag_without_dv_takes_only_hdr10plus_bases() -> None:
|
||||
videos = [video("h2160", Video.Range.HDR10), video("p1080", Video.Range.HDR10P, 1080)]
|
||||
|
||||
bases, dv = hybrid_plan(videos, True)
|
||||
|
||||
assert [v.id for v in bases] == ["p1080"] and dv is None
|
||||
|
||||
|
||||
def test_no_dv_and_no_flag_gives_no_bases() -> None:
|
||||
assert hybrid_plan([video("h", Video.Range.HDR10), video("p", Video.Range.HDR10P)], False) == ([], None)
|
||||
@@ -66,14 +66,11 @@ 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:
|
||||
def test_crop_detection_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 make_hybrid().detect_active_area(tmp_path / "base.dv.hevc") is None
|
||||
assert len(calls) == 1
|
||||
|
||||
|
||||
@@ -207,11 +204,11 @@ def test_level_5_returns_early_when_the_cropped_rpu_exists(monkeypatch: pytest.M
|
||||
(tmp_path / "RPU_L5.bin").write_bytes(b"")
|
||||
|
||||
def boom(*args: Any, **kwargs: Any) -> None:
|
||||
raise AssertionError("level_5 must not run ffprobe when RPU_L5.bin is already there")
|
||||
raise AssertionError("level_5 must not run dovi_tool when RPU_L5.bin is already there")
|
||||
|
||||
monkeypatch.setattr(subprocess, "run", boom)
|
||||
monkeypatch.setattr(hybrid_mod.dovi, "editor", boom)
|
||||
|
||||
obj = make_hybrid()
|
||||
obj.level_5(tmp_path / "HDR10.hevc")
|
||||
obj.level_5((0, 276, 0, 276))
|
||||
|
||||
assert obj.rpu_file == "RPU_L5.bin"
|
||||
|
||||
@@ -95,6 +95,19 @@ def test_select_hybrid_base_picks_highest_bitrate_then_worst_flips() -> None:
|
||||
assert ids(worst) == {"hdr10p-2160-lo", "dv-360"}
|
||||
|
||||
|
||||
def test_select_hybrid_skips_a_base_that_is_not_hevc() -> None:
|
||||
"""dovi_tool injects into HEVC, so a higher-bitrate AV1 base is not a hybrid ingredient."""
|
||||
tracks = [
|
||||
make_video(
|
||||
"hdr10p-2160-av1", range_=Video.Range.HDR10P, height=2160, bitrate=30_000_000, codec=Video.Codec.AV1
|
||||
),
|
||||
make_video("hdr10p-2160", range_=Video.Range.HDR10P, height=2160, bitrate=20_000_000, codec=Video.Codec.HEVC),
|
||||
make_video("dv-360", range_=Video.Range.DV, height=360, bitrate=1_000_000, codec=Video.Codec.HEVC),
|
||||
]
|
||||
chosen = list(filter(Tracks().select_hybrid(tracks, [2160]), tracks))
|
||||
assert ids(chosen) == {"hdr10p-2160", "dv-360"}
|
||||
|
||||
|
||||
def test_select_hybrid_no_dv_selects_only_base() -> None:
|
||||
H = Video.Codec.HEVC
|
||||
tracks = [make_video("hdr10p-2160", range_=Video.Range.HDR10P, height=2160, bitrate=20_000_000, codec=H)]
|
||||
|
||||
+110
-33
@@ -214,10 +214,17 @@ def group_videos_by_variant(videos: list[Video], *, merge: bool) -> list[list[Vi
|
||||
|
||||
|
||||
def select_hybrid_bases(videos: Sequence[Video]) -> list[Video]:
|
||||
"""The HDR10 or HDR10+ track that is the hybrid base for each height, tallest first."""
|
||||
"""The HDR10 or HDR10+ track that is the hybrid base for each height, tallest first.
|
||||
|
||||
Only HEVC tracks qualify, because dovi_tool injects the RPU into an HEVC stream.
|
||||
"""
|
||||
bases: dict[int, Video] = {}
|
||||
for video in sorted(
|
||||
(v for v in videos if v.range in (Video.Range.HDR10P, Video.Range.HDR10)),
|
||||
(
|
||||
v
|
||||
for v in videos
|
||||
if v.range in (Video.Range.HDR10P, Video.Range.HDR10) and v.codec in (Video.Codec.HEVC, None)
|
||||
),
|
||||
key=lambda v: v.height or 0,
|
||||
reverse=True,
|
||||
):
|
||||
@@ -225,19 +232,85 @@ def select_hybrid_bases(videos: Sequence[Video]) -> list[Video]:
|
||||
return list(bases.values())
|
||||
|
||||
|
||||
def hybrid_source_tracks(base: Video, dv: Video, base_container: Optional[Path] = None) -> list[Video]:
|
||||
def hybrid_gate_failure(videos: Sequence[Video], generate_dv: bool) -> Optional[tuple[str, str]]:
|
||||
"""The (message, detail) that stops HYBRID for these selected videos, or None when it can go on."""
|
||||
hdr_tracks = [v for v in videos if v.range in (Video.Range.HDR10, Video.Range.HDR10P)]
|
||||
base_tracks = [v for v in hdr_tracks if v.codec in (Video.Codec.HEVC, None)]
|
||||
dv_tracks = [v for v in videos if v.range == Video.Range.DV]
|
||||
has_hdr10p = any(v.range == Video.Range.HDR10P for v in base_tracks)
|
||||
available_ranges = sorted(set(v.range.name for v in videos))
|
||||
|
||||
if not hdr_tracks and not dv_tracks:
|
||||
return (
|
||||
"HYBRID mode requires both HDR10/HDR10+ and DV tracks, but neither is available",
|
||||
f"Available ranges: {', '.join(available_ranges) if available_ranges else 'none'}",
|
||||
)
|
||||
if not hdr_tracks:
|
||||
return (
|
||||
"HYBRID mode requires both HDR10/HDR10+ and DV tracks, but only DV is available",
|
||||
f"Available ranges: {', '.join(available_ranges)}",
|
||||
)
|
||||
if not base_tracks:
|
||||
codecs = sorted(set(v.codec.name for v in hdr_tracks if v.codec))
|
||||
return (
|
||||
"HYBRID mode requires an HEVC HDR10/HDR10+ base, but none is available",
|
||||
f"Available HDR10/HDR10+ codecs: {', '.join(codecs)}",
|
||||
)
|
||||
if not dv_tracks and not (generate_dv and has_hdr10p):
|
||||
msg_detail = f"Available ranges: {', '.join(available_ranges)}"
|
||||
if generate_dv:
|
||||
msg_detail += ". --generate-dv needs an HEVC HDR10+ track, and the service returned none."
|
||||
elif has_hdr10p:
|
||||
msg_detail += ". Add --generate-dv to generate Dolby Vision from HDR10+."
|
||||
return "HYBRID mode requires both HDR10/HDR10+ and DV tracks, but only HDR10 is available", msg_detail
|
||||
return None
|
||||
|
||||
|
||||
def hybrid_plan(videos: Sequence[Video], generate_dv: bool) -> tuple[list[Video], Optional[Video]]:
|
||||
"""The hybrid bases and the DV track to pair with them.
|
||||
|
||||
With no DV track, only HDR10+ bases qualify, and only with ``--generate-dv``.
|
||||
"""
|
||||
dv_tracks = [v for v in videos if v.range == Video.Range.DV]
|
||||
base_pool = [v for v in videos if dv_tracks or (generate_dv and v.range == Video.Range.HDR10P)]
|
||||
matching_dv = min(dv_tracks, key=lambda v: v.height or 0) if dv_tracks else None
|
||||
return select_hybrid_bases(base_pool), matching_dv
|
||||
|
||||
|
||||
def hybrid_source_tracks(base: Video, dv: Optional[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.
|
||||
permits neither. With no DV track, the hybrid step makes Dolby Vision from the base's HDR10+.
|
||||
"""
|
||||
tracks = [deepcopy(base), deepcopy(dv)]
|
||||
tracks = [deepcopy(track) for track in (base, dv) if track]
|
||||
tracks[0].path = base_container or base.path
|
||||
for track in tracks:
|
||||
track.needs_duration_fix = True
|
||||
return tracks
|
||||
|
||||
|
||||
def hybrid_step_error(tracks: list[Video], source: str, output_name: str) -> Optional[str]:
|
||||
"""Run the hybrid step for one resolution; the error that failed it, or None.
|
||||
|
||||
A failure here is this resolution's alone: the title goes on with its other resolutions.
|
||||
"""
|
||||
try:
|
||||
Hybrid(tracks, source, output_name=output_name)
|
||||
except (ValueError, KeyError, OSError, RuntimeError) as e:
|
||||
return str(e) or type(e).__name__
|
||||
return None
|
||||
|
||||
|
||||
def missing_hybrid_error(bases: Sequence[Video], made: int) -> Optional[str]:
|
||||
"""Why the title has no hybrid file, or None when at least one was made."""
|
||||
if made:
|
||||
return None
|
||||
if not bases:
|
||||
return "No HEVC HDR10 or HDR10+ track to make a hybrid from"
|
||||
return f"Every hybrid failed ({', '.join(f'{v.height or 0}p' for v in bases)}), so the title has no hybrid file"
|
||||
|
||||
|
||||
def title_wanted(candidate: Any, wanted: Collection[str]) -> bool:
|
||||
"""Whether ``-w`` selects this title. A title type without selection keys is always kept.
|
||||
|
||||
@@ -1122,6 +1195,12 @@ class dl:
|
||||
default=False,
|
||||
help="Warn instead of failing when a requested resolution, range, or language is missing, and continue with what is available.",
|
||||
)
|
||||
@click.option(
|
||||
"--generate-dv",
|
||||
is_flag=True,
|
||||
default=False,
|
||||
help="With -r HYBRID, make Dolby Vision from the HDR10+ metadata when a title has no Dolby Vision track.",
|
||||
)
|
||||
@click.option(
|
||||
"--remote",
|
||||
is_flag=True,
|
||||
@@ -1647,6 +1726,7 @@ class dl:
|
||||
postscript: Sequence[str] = (),
|
||||
no_postscript: bool = False,
|
||||
all_drm: bool = False,
|
||||
generate_dv: bool = False,
|
||||
*_: Any,
|
||||
**__: Any,
|
||||
) -> None:
|
||||
@@ -1843,6 +1923,13 @@ class dl:
|
||||
self.log.error("Unable to run hybrid mode: dovi_tool not detected")
|
||||
self.log.error("Please install dovi_tool from https://github.com/quietvoid/dovi_tool")
|
||||
sys.exit(1)
|
||||
if generate_dv:
|
||||
from unshackle.core.binaries import HDR10PlusTool
|
||||
|
||||
if not HDR10PlusTool:
|
||||
self.log.error("Unable to run --generate-dv: hdr10plus_tool not detected")
|
||||
self.log.error("Please install hdr10plus_tool from https://github.com/quietvoid/hdr10plus_tool")
|
||||
sys.exit(1)
|
||||
|
||||
if cdm_only is None:
|
||||
vaults_only = None
|
||||
@@ -2733,31 +2820,12 @@ class dl:
|
||||
title.tracks.videos = selected_videos
|
||||
|
||||
if keep_videos and any(r == Video.Range.HYBRID for r in range_):
|
||||
base_tracks = [
|
||||
v for v in title.tracks.videos if v.range in (Video.Range.HDR10, Video.Range.HDR10P)
|
||||
]
|
||||
dv_tracks = [v for v in title.tracks.videos if v.range == Video.Range.DV]
|
||||
gate_failure = hybrid_gate_failure(title.tracks.videos, generate_dv)
|
||||
if not gate_failure and not any(v.range == Video.Range.DV for v in title.tracks.videos):
|
||||
self.log.info("No Dolby Vision track: Dolby Vision will be generated from HDR10+")
|
||||
|
||||
hybrid_failed = False
|
||||
if not base_tracks and not dv_tracks:
|
||||
available_ranges = sorted(set(v.range.name for v in title.tracks.videos))
|
||||
msg = "HYBRID mode requires both HDR10/HDR10+ and DV tracks, but neither is available"
|
||||
msg_detail = (
|
||||
f"Available ranges: {', '.join(available_ranges) if available_ranges else 'none'}"
|
||||
)
|
||||
hybrid_failed = True
|
||||
elif not base_tracks:
|
||||
available_ranges = sorted(set(v.range.name for v in title.tracks.videos))
|
||||
msg = "HYBRID mode requires both HDR10/HDR10+ and DV tracks, but only DV is available"
|
||||
msg_detail = f"Available ranges: {', '.join(available_ranges)}"
|
||||
hybrid_failed = True
|
||||
elif not dv_tracks:
|
||||
available_ranges = sorted(set(v.range.name for v in title.tracks.videos))
|
||||
msg = "HYBRID mode requires both HDR10/HDR10+ and DV tracks, but only HDR10 is available"
|
||||
msg_detail = f"Available ranges: {', '.join(available_ranges)}"
|
||||
hybrid_failed = True
|
||||
|
||||
if hybrid_failed:
|
||||
if gate_failure:
|
||||
msg, msg_detail = gate_failure
|
||||
other_ranges = [r for r in range_ if r != Video.Range.HYBRID]
|
||||
if best_available and other_ranges:
|
||||
self.log.warning(msg)
|
||||
@@ -3583,16 +3651,15 @@ 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_bases, matching_dv = hybrid_plan(title.tracks.videos, generate_dv)
|
||||
hybrid_base_ids.update(v.id for v in hybrid_bases)
|
||||
hybrids_made = 0
|
||||
|
||||
for group in sorted(standalone_groups, key=group_file_size, reverse=True):
|
||||
mux_video_group(group)
|
||||
|
||||
for hdr10_track in hybrid_bases:
|
||||
resolution = hdr10_track.height or 0
|
||||
matching_dv = min(dv_tracks, key=lambda v: v.height)
|
||||
container = container_paths.get(hdr10_track.id)
|
||||
resolution_tracks = hybrid_source_tracks(hdr10_track, matching_dv, container)
|
||||
|
||||
@@ -3605,7 +3672,12 @@ class dl:
|
||||
"--sample cut the HDR10 and Dolby Vision tracks at different "
|
||||
"points, so a hybrid of them would not line up."
|
||||
)
|
||||
Hybrid(resolution_tracks, self.service, output_name=hybrid_filename)
|
||||
step_error = hybrid_step_error(resolution_tracks, self.service, hybrid_filename)
|
||||
if step_error:
|
||||
# the other files of the title are already queued for muxing
|
||||
self.log.error(f"Hybrid {resolution}p failed, skipping it: {step_error}")
|
||||
continue
|
||||
hybrids_made += 1
|
||||
|
||||
if container:
|
||||
try:
|
||||
@@ -3628,6 +3700,10 @@ class dl:
|
||||
|
||||
enqueue_mux_tasks(task_description, task_tracks, hybrid=True)
|
||||
|
||||
no_hybrid = missing_hybrid_error(hybrid_bases, hybrids_made)
|
||||
if no_hybrid:
|
||||
raise click.ClickException(no_hybrid)
|
||||
|
||||
console.print()
|
||||
else:
|
||||
groups: list[list[Optional[Video]]] = [
|
||||
@@ -3731,6 +3807,7 @@ class dl:
|
||||
"RPU_L5.bin",
|
||||
"RPU_L6.bin",
|
||||
"L5.json",
|
||||
"HDR10Plus.json",
|
||||
):
|
||||
leftover_path = config.directories.temp / leftover
|
||||
try:
|
||||
|
||||
@@ -495,6 +495,7 @@ def perform_download(
|
||||
"vcodec": params.get("vcodec", []),
|
||||
"range_": params.get("range", [Video.Range.SDR]),
|
||||
"best_available": params.get("best_available", False),
|
||||
"generate_dv": params.get("generate_dv", False),
|
||||
# services read these here, not from dl.result(), to pick the manifests they fetch
|
||||
"acodec": params.get("acodec") or [],
|
||||
"lang": params.get("lang", ["orig"]),
|
||||
@@ -659,6 +660,7 @@ def perform_download(
|
||||
downloads=params.get("downloads", 1),
|
||||
worst=params.get("worst", False),
|
||||
best_available=params.get("best_available", False),
|
||||
generate_dv=params.get("generate_dv", False),
|
||||
split_audio=params.get("split_audio"),
|
||||
progress_sink=progress_callback,
|
||||
)
|
||||
|
||||
@@ -108,6 +108,7 @@ DEFAULT_DOWNLOAD_PARAMS = {
|
||||
"downloads": 1,
|
||||
"worst": False,
|
||||
"best_available": False,
|
||||
"generate_dv": False,
|
||||
"repack": False,
|
||||
"tag": None,
|
||||
"tmdb_id": None,
|
||||
@@ -2289,14 +2290,20 @@ def validate_download_parameters(data: Dict[str, Any]) -> Optional[str]:
|
||||
if value is not None and (not isinstance(value, list) or not all(isinstance(v, str) for v in value)):
|
||||
return f"{key} must be an array of language strings"
|
||||
|
||||
if "range" in data and data["range"]:
|
||||
# "HDR10P" is the canonical range value ("+" is awkward in scripts); "HDR10+" stays valid.
|
||||
valid_ranges = ["SDR", "HDR10", "HDR10P", "DV", "HLG", "HYBRID"]
|
||||
accepted = {*valid_ranges, "HDR10+"}
|
||||
values = data["range"] if isinstance(data["range"], list) else [data["range"]]
|
||||
for r in values:
|
||||
if r.upper() not in accepted:
|
||||
return f"Invalid range value: {r}. Must be one of: {', '.join(valid_ranges)}"
|
||||
ranges = data.get("range") or []
|
||||
ranges = ranges if isinstance(ranges, list) else [ranges]
|
||||
# "HDR10P" is the canonical range value ("+" is awkward in scripts); "HDR10+" stays valid.
|
||||
valid_ranges = ["SDR", "HDR10", "HDR10P", "DV", "HLG", "HYBRID"]
|
||||
accepted = {*valid_ranges, "HDR10+"}
|
||||
for r in ranges:
|
||||
if r.upper() not in accepted:
|
||||
return f"Invalid range value: {r}. Must be one of: {', '.join(valid_ranges)}"
|
||||
if data.get("generate_dv"):
|
||||
# a request without a range runs with the serve: default range
|
||||
job_ranges = ranges or (config.serve or {}).get("range") or []
|
||||
job_ranges = job_ranges if isinstance(job_ranges, list) else [job_ranges]
|
||||
if not any(str(r).upper() == "HYBRID" for r in job_ranges):
|
||||
return "generate_dv needs HYBRID in range."
|
||||
|
||||
return None
|
||||
|
||||
@@ -3574,6 +3581,7 @@ SESSION_TRANSPORT_KEYS = {
|
||||
"vcodec",
|
||||
"quality",
|
||||
"best_available",
|
||||
"generate_dv",
|
||||
"dl_params",
|
||||
}
|
||||
|
||||
@@ -3758,6 +3766,7 @@ def create_service_instance(
|
||||
"vcodec": vcodec_values,
|
||||
"quality": data.get("quality"),
|
||||
"best_available": data.get("best_available", False),
|
||||
"generate_dv": data.get("generate_dv", False),
|
||||
**forwarded_dl_params(data),
|
||||
}
|
||||
|
||||
|
||||
@@ -1083,6 +1083,11 @@ async def download(request: web.Request) -> web.Response:
|
||||
description: >-
|
||||
Continue with best available if requested quality unavailable (default - false).
|
||||
An API key with a server_cdm_max_height for the service cannot set it.
|
||||
generate_dv:
|
||||
type: boolean
|
||||
description: >-
|
||||
With HYBRID in range, make Dolby Vision from the HDR10+ metadata when a title has no
|
||||
Dolby Vision track (default - false). Needs hdr10plus_tool on the server.
|
||||
worst:
|
||||
type: boolean
|
||||
description: Select the lowest bitrate track within the specified quality. Requires `quality` (default - false)
|
||||
|
||||
@@ -199,6 +199,7 @@ class Config:
|
||||
self.language_tags: dict = kwargs.get("language_tags") or {}
|
||||
self.tag_rules: list = kwargs.get("tag_rules") or []
|
||||
self.dual_multi_mode: str = (kwargs.get("dual_multi_mode") or "strict").lower()
|
||||
self.generate_dv_mode: str = str(kwargs.get("generate_dv_mode") or "v29").lower()
|
||||
self.output_template: dict = kwargs.get("output_template") or {}
|
||||
folder_cfg = self.output_template.pop("folder", "")
|
||||
self.folder_template: str = ""
|
||||
|
||||
@@ -1213,6 +1213,8 @@ class RemoteService:
|
||||
create_data["quality"] = list(quality)
|
||||
if self.ctx.parent.params.get("best_available"):
|
||||
create_data["best_available"] = True
|
||||
if self.ctx.parent.params.get("generate_dv"):
|
||||
create_data["generate_dv"] = True
|
||||
# Services pick the manifests to fetch from these. They go in their own
|
||||
# object: flat, a service option with the same name would take them (or override them).
|
||||
# An empty list is sent as-is, because only an absent key falls back to the dl default.
|
||||
|
||||
@@ -126,6 +126,7 @@ class TrackRequest:
|
||||
codecs: list[Video.Codec] = field(default_factory=list)
|
||||
ranges: list[Video.Range] = field(default_factory=lambda: [Video.Range.SDR])
|
||||
best_available: bool = False
|
||||
generate_dv: bool = False
|
||||
|
||||
|
||||
def sanitize_proxy_for_log(uri: Optional[str], mask_host: bool = False) -> Optional[str]:
|
||||
@@ -194,10 +195,12 @@ class Service(metaclass=ABCMeta):
|
||||
vcodec = ctx.parent.params.get("vcodec") if ctx.parent else None
|
||||
range_ = ctx.parent.params.get("range_") if ctx.parent else None
|
||||
best_available = ctx.parent.params.get("best_available", False) if ctx.parent else False
|
||||
generate_dv = ctx.parent.params.get("generate_dv", False) if ctx.parent else False
|
||||
self.track_request = TrackRequest(
|
||||
codecs=list(vcodec) if vcodec else [],
|
||||
ranges=list(range_) if range_ else [Video.Range.SDR],
|
||||
best_available=bool(best_available),
|
||||
generate_dv=bool(generate_dv),
|
||||
)
|
||||
|
||||
whose = "the server's" if ctx.parent and ctx.parent.params.get("served") else "your"
|
||||
@@ -353,7 +356,8 @@ class Service(metaclass=ABCMeta):
|
||||
The fetch_fn signature should be: (title, codec, range_) -> Tracks
|
||||
|
||||
For HYBRID range, this helper calls fetch_fn with HDR10 and DV separately,
|
||||
then merges the DV video tracks into the HDR10 result.
|
||||
then merges the DV video tracks into the HDR10 result. With no DV and
|
||||
``generate_dv`` requested, it merges the HDR10P video tracks instead.
|
||||
|
||||
Args:
|
||||
title: The title to process.
|
||||
@@ -386,11 +390,19 @@ class Service(metaclass=ABCMeta):
|
||||
all_tracks.add(video, warn_only=True)
|
||||
|
||||
try:
|
||||
dv_tracks = fetch_fn(title, codec=codec_val, range_=Video.Range.DV)
|
||||
for video in dv_tracks.videos:
|
||||
all_tracks.add(video, warn_only=True)
|
||||
dv_videos = fetch_fn(title, codec=codec_val, range_=Video.Range.DV).videos
|
||||
except (ValueError, SystemExit):
|
||||
self.log.info(" - No DolbyVision manifest available for HYBRID")
|
||||
dv_videos = []
|
||||
has_hdr10p = any(v.range == Video.Range.HDR10P for v in all_tracks.videos)
|
||||
if not dv_videos and self.track_request.generate_dv and not has_hdr10p:
|
||||
# the HDR10 request may not return HDR10+, which the conversion needs as its base
|
||||
try:
|
||||
dv_videos = fetch_fn(title, codec=codec_val, range_=Video.Range.HDR10P).videos
|
||||
except (ValueError, SystemExit):
|
||||
self.log.info(" - No HDR10+ manifest available for HYBRID")
|
||||
for video in dv_videos:
|
||||
all_tracks.add(video, warn_only=True)
|
||||
else:
|
||||
for codec_val in codecs:
|
||||
try:
|
||||
|
||||
+364
-213
@@ -1,10 +1,9 @@
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import random
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from collections import Counter
|
||||
from concurrent import futures
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
@@ -19,6 +18,97 @@ from unshackle.core.utilities import log_event
|
||||
from unshackle.core.utils import dovi
|
||||
from unshackle.core.utils.subprocess import run_step
|
||||
|
||||
# HDR10+ histogram slots that form the maxRGB distribution, as (DistributionIndex, fraction of pixels).
|
||||
# Index 99 holds the 99.98% percentile. Index 5 (a luminance-Y value) and index 10 (the share of pixels
|
||||
# at or under 100 nits) are not part of the distribution.
|
||||
# config.generate_dv_mode values: unshackle's L1 with the CM v2.9 or the CM v4.0 avg floor, or dovi_tool's own L1.
|
||||
GENERATE_DV_MODES = ("v29", "v40", "legacy")
|
||||
|
||||
HDR10PLUS_PERCENTILES = ((1, 0.01), (25, 0.25), (50, 0.5), (75, 0.75), (90, 0.9), (95, 0.95), (99, 0.9998))
|
||||
|
||||
|
||||
def nits_to_pq(nits: float) -> float:
|
||||
"""SMPTE ST 2084 inverse EOTF, as a 12-bit code value (0-4095)."""
|
||||
m1, m2, c1, c2, c3 = 0.1593017578125, 78.84375, 0.8359375, 18.8515625, 18.6875
|
||||
y = (max(nits, 0.0) / 10000) ** m1
|
||||
return 4095 * ((c1 + c2 * y) / (1 + c3 * y)) ** m2
|
||||
|
||||
|
||||
def hdr10plus_mean_pq(values: dict[int, float], active_fraction: float = 1.0) -> Optional[float]:
|
||||
"""Mean PQ code of the maxRGB distribution in one HDR10+ histogram, or None if a slot is missing.
|
||||
|
||||
Dolby Vision L1 avg is a mean of PQ-coded maxRGB over the active area. HDR10+ AverageMaxRGB is a
|
||||
linear mean over the whole frame, so the mean comes from the percentiles instead. The darkest
|
||||
``1 - active_fraction`` of the pixels are the letterbox bars, and the mean leaves them out. A scene
|
||||
whose 1st percentile is above black has no bars, so its whole frame counts.
|
||||
"""
|
||||
if any(index not in values for index, _ in HDR10PLUS_PERCENTILES):
|
||||
return None
|
||||
if values[1] > 0:
|
||||
active_fraction = 1.0
|
||||
knots = [(0.0, nits_to_pq(values[1]))]
|
||||
knots += [(fraction, nits_to_pq(values[index])) for index, fraction in HDR10PLUS_PERCENTILES]
|
||||
knots.append((1.0, knots[-1][1]))
|
||||
|
||||
low = 1.0 - active_fraction
|
||||
area = 0.0
|
||||
for (x1, y1), (x2, y2) in zip(knots, knots[1:]):
|
||||
if x2 <= low:
|
||||
continue
|
||||
if x1 < low:
|
||||
y1 += (y2 - y1) * (low - x1) / (x2 - x1)
|
||||
x1 = low
|
||||
area += (x2 - x1) * (y1 + y2) / 2
|
||||
return area / active_fraction
|
||||
|
||||
|
||||
def hdr10plus_shots(metadata: dict, active_fraction: float = 1.0) -> list[dict]:
|
||||
"""One dovi_tool generator shot with L1 metadata per HDR10+ scene.
|
||||
|
||||
Each scene keeps its own cut. max_pq is the histogram peak, as in dovi_tool's default peak source,
|
||||
without index 10: that slot is the share of pixels at or under 100 nits, not a brightness, and would
|
||||
put a near-black scene at 10 nits. Index 5, the 99.99% luminance, stays: against decoded pixels it
|
||||
tracked the true peak better than index 99 alone. MaxScl is the peak when the histogram is empty.
|
||||
A scene whose histogram has no complete distribution falls back to dovi_tool's average,
|
||||
PQ(AverageMaxRGB), capped at max_pq.
|
||||
"""
|
||||
scenes = metadata["SceneInfo"]
|
||||
summary = metadata["SceneInfoSummary"]
|
||||
first_frames = summary["SceneFirstFrameIndex"]
|
||||
offset = first_frames[0]
|
||||
|
||||
shots = []
|
||||
fallbacks = 0
|
||||
for first_frame, length in zip(first_frames, summary["SceneFrameNumbers"]):
|
||||
start = first_frame - offset
|
||||
luminance = scenes[start]["LuminanceParameters"]
|
||||
distribution = luminance["LuminanceDistributions"]
|
||||
values = {
|
||||
index: value / 10
|
||||
for index, value in zip(distribution["DistributionIndex"], distribution["DistributionValues"])
|
||||
}
|
||||
peaks = [value for index, value in values.items() if index != 10]
|
||||
max_pq = nits_to_pq(max(peaks or [v / 10 for v in luminance["MaxScl"]]))
|
||||
avg_pq = hdr10plus_mean_pq(values, active_fraction)
|
||||
if avg_pq is None:
|
||||
fallbacks += 1
|
||||
avg_pq = min(nits_to_pq(luminance["AverageRGB"] / 10), max_pq)
|
||||
shots.append(
|
||||
{
|
||||
"start": start,
|
||||
"duration": length,
|
||||
"metadata_blocks": [{"Level1": {"min_pq": 0, "max_pq": round(max_pq), "avg_pq": round(avg_pq)}}],
|
||||
}
|
||||
)
|
||||
if fallbacks:
|
||||
log_event(
|
||||
"hybrid_convert_hdr10plus",
|
||||
level="DEBUG",
|
||||
message="HDR10+ scenes without a full histogram use PQ(AverageMaxRGB) for L1 avg",
|
||||
context={"scenes": fallbacks, "total": len(shots)},
|
||||
)
|
||||
return shots
|
||||
|
||||
|
||||
class Hybrid:
|
||||
def __init__(self, videos, source, output_name: Optional[str] = None) -> None:
|
||||
@@ -80,40 +170,55 @@ class Hybrid:
|
||||
console.status("Already Injected")
|
||||
return
|
||||
|
||||
for video in videos:
|
||||
save_path = video.path
|
||||
if not save_path or not os.path.exists(save_path):
|
||||
raise ValueError(f"Video track {video.id} was not downloaded or path not found: {save_path}")
|
||||
try:
|
||||
for video in videos:
|
||||
save_path = video.path
|
||||
if not save_path or not os.path.exists(save_path):
|
||||
raise ValueError(f"Video track {video.id} was not downloaded or path not found: {save_path}")
|
||||
|
||||
if video.range == Video.Range.HDR10:
|
||||
self.extract_stream(save_path, "HDR10")
|
||||
elif video.range == Video.Range.HDR10P:
|
||||
self.extract_stream(save_path, "HDR10")
|
||||
self.hdr_type = "HDR10+"
|
||||
elif video.range == Video.Range.DV:
|
||||
if not os.path.isfile(config.directories.temp / "DV.hevc"):
|
||||
self.extract_stream(save_path, "DV")
|
||||
if video.range == Video.Range.HDR10:
|
||||
self.extract_stream(save_path, "HDR10")
|
||||
elif video.range == Video.Range.HDR10P:
|
||||
self.extract_stream(save_path, "HDR10")
|
||||
self.hdr_type = "HDR10+"
|
||||
elif video.range == Video.Range.DV:
|
||||
if not os.path.isfile(config.directories.temp / "DV.hevc"):
|
||||
self.extract_stream(save_path, "DV")
|
||||
|
||||
if self.hdr10plus_to_dv:
|
||||
hdr10p_video = next(v for v in videos if v.range == Video.Range.HDR10P)
|
||||
self.extract_hdr10plus(hdr10p_video)
|
||||
self.convert_hdr10plus_to_dv()
|
||||
else:
|
||||
dv_video = next(v for v in videos if v.range == Video.Range.DV)
|
||||
self.extract_rpu(dv_video)
|
||||
if os.path.isfile(config.directories.temp / "RPU_UNT.bin"):
|
||||
self.rpu_file = "RPU_UNT.bin"
|
||||
# Mode 3 conversion already done during extraction when not untouched
|
||||
elif os.path.isfile(config.directories.temp / "RPU.bin"):
|
||||
# RPU already extracted with mode 3
|
||||
pass
|
||||
base_video = next((v for v in videos if v.range in (Video.Range.HDR10, Video.Range.HDR10P)), None)
|
||||
area = self.detect_active_area(base_video.path) if base_video and base_video.path else None
|
||||
|
||||
self.level_6()
|
||||
base_video = next((v for v in videos if v.range in (Video.Range.HDR10, Video.Range.HDR10P)), None)
|
||||
if base_video and base_video.path:
|
||||
self.level_5(base_video.path)
|
||||
if self.hdr10plus_to_dv:
|
||||
# a base that failed earlier in this title can leave its edited RPUs, which level_5/level_6 reuse
|
||||
for stale in ("RPU_L5.bin", "RPU_L6.bin"):
|
||||
Path.unlink(config.directories.temp / stale, missing_ok=True)
|
||||
hdr10p_video = next(v for v in videos if v.range == Video.Range.HDR10P)
|
||||
self.extract_hdr10plus(hdr10p_video)
|
||||
self.convert_hdr10plus_to_dv(area[1] if area else 1.0)
|
||||
else:
|
||||
dv_video = next(v for v in videos if v.range == Video.Range.DV)
|
||||
self.extract_rpu(dv_video)
|
||||
if os.path.isfile(config.directories.temp / "RPU_UNT.bin"):
|
||||
self.rpu_file = "RPU_UNT.bin"
|
||||
# Mode 3 conversion already done during extraction when not untouched
|
||||
elif os.path.isfile(config.directories.temp / "RPU.bin"):
|
||||
# RPU already extracted with mode 3
|
||||
pass
|
||||
|
||||
self.injecting()
|
||||
self.level_6()
|
||||
if area:
|
||||
self.level_5(area[0])
|
||||
|
||||
self.injecting()
|
||||
finally:
|
||||
# The title goes on after a failed resolution, and level_5 reuses an RPU_L5.bin it finds,
|
||||
# so the files of this resolution's crop must not outlive it.
|
||||
for own in ("HDR10.hevc", "RPU_L5.bin", "L5.json", "L6.json"):
|
||||
Path.unlink(config.directories.temp / own, missing_ok=True)
|
||||
if self.hdr10plus_to_dv:
|
||||
# The RPU comes from this base's own HDR10+, so the next resolution must not reuse it.
|
||||
Path.unlink(config.directories.temp / "RPU.bin", missing_ok=True)
|
||||
Path.unlink(config.directories.temp / "RPU_L6.bin", missing_ok=True)
|
||||
|
||||
log_event(
|
||||
"hybrid_complete",
|
||||
@@ -131,10 +236,6 @@ class Hybrid:
|
||||
if self.source == ("itunes" or "appletvplus"):
|
||||
Path.unlink(config.directories.temp / "hdr10.mkv")
|
||||
Path.unlink(config.directories.temp / "dv.mkv")
|
||||
Path.unlink(config.directories.temp / "HDR10.hevc", missing_ok=True)
|
||||
Path.unlink(config.directories.temp / "RPU_L5.bin", missing_ok=True)
|
||||
Path.unlink(config.directories.temp / "L5.json", missing_ok=True)
|
||||
Path.unlink(config.directories.temp / "L6.json", missing_ok=True)
|
||||
|
||||
def extract_stream(self, save_path, type_):
|
||||
output = Path(config.directories.temp / f"{type_}.hevc")
|
||||
@@ -152,8 +253,7 @@ class Hybrid:
|
||||
message=f"Failed extracting {type_} stream",
|
||||
context={"type": type_, "input": str(save_path), "output": str(output), "error": str(e)},
|
||||
)
|
||||
self.log.error(f"x Failed extracting {type_} stream")
|
||||
sys.exit(1)
|
||||
raise ValueError(f"Failed extracting {type_} stream")
|
||||
|
||||
log_event(
|
||||
"hybrid_extract_stream",
|
||||
@@ -199,8 +299,8 @@ class Hybrid:
|
||||
success=True,
|
||||
)
|
||||
|
||||
def level_5(self, input_video):
|
||||
"""Make Level 5 active area metadata through crop detection on the HDR10 track.
|
||||
def level_5(self, crop: tuple[int, int, int, int]) -> None:
|
||||
"""Write the active area of the HDR10 track, as (left, top, right, bottom), to Level 5.
|
||||
|
||||
This resolves mismatches where DV has no black bars but HDR10 does (or vice versa)
|
||||
by telling the display the correct active area.
|
||||
@@ -209,136 +309,7 @@ class Hybrid:
|
||||
self.rpu_file = "RPU_L5.bin"
|
||||
return
|
||||
|
||||
ffprobe_bin = str(FFProbe) if FFProbe else "ffprobe"
|
||||
ffmpeg_bin = str(FFMPEG) if FFMPEG else "ffmpeg"
|
||||
|
||||
with console.status("Detecting active area (crop detection)...", spinner="dots"):
|
||||
result_duration = subprocess.run(
|
||||
[ffprobe_bin, "-v", "error", "-show_entries", "format=duration", "-of", "json", str(input_video)],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
encoding="utf-8",
|
||||
errors="replace",
|
||||
)
|
||||
|
||||
if result_duration.returncode != 0:
|
||||
log_event(
|
||||
"hybrid_level5",
|
||||
level="WARNING",
|
||||
message="Could not probe video duration",
|
||||
context={"returncode": result_duration.returncode, "stderr": (result_duration.stderr or "")},
|
||||
)
|
||||
self.log.warning("Could not probe video duration, skipping L5 crop detection")
|
||||
return
|
||||
|
||||
duration_value = json.loads(result_duration.stdout).get("format", {}).get("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(
|
||||
[
|
||||
ffprobe_bin,
|
||||
"-v",
|
||||
"error",
|
||||
"-select_streams",
|
||||
"v:0",
|
||||
"-show_entries",
|
||||
"stream=width,height",
|
||||
"-of",
|
||||
"json",
|
||||
str(input_video),
|
||||
],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
encoding="utf-8",
|
||||
errors="replace",
|
||||
)
|
||||
|
||||
if result_streams.returncode != 0:
|
||||
log_event(
|
||||
"hybrid_level5",
|
||||
level="WARNING",
|
||||
message="Could not probe video resolution",
|
||||
context={"returncode": result_streams.returncode, "stderr": (result_streams.stderr or "")},
|
||||
)
|
||||
self.log.warning("Could not probe video resolution, skipping L5 crop detection")
|
||||
return
|
||||
|
||||
stream_info = json.loads(result_streams.stdout)
|
||||
original_width = int(stream_info["streams"][0]["width"])
|
||||
original_height = int(stream_info["streams"][0]["height"])
|
||||
|
||||
random_times = sorted(random.uniform(0, duration) for _ in range(10))
|
||||
|
||||
def sample_crop(t: float) -> Optional[tuple[int, int, int, int]]:
|
||||
result_cropdetect = subprocess.run(
|
||||
[
|
||||
ffmpeg_bin,
|
||||
"-y",
|
||||
"-nostdin",
|
||||
"-loglevel",
|
||||
"info",
|
||||
"-ss",
|
||||
f"{t:.2f}",
|
||||
"-i",
|
||||
str(input_video),
|
||||
"-vf",
|
||||
"cropdetect=round=2",
|
||||
"-vframes",
|
||||
"4",
|
||||
"-f",
|
||||
"null",
|
||||
"-",
|
||||
],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
encoding="utf-8",
|
||||
errors="replace",
|
||||
)
|
||||
|
||||
# cropdetect outputs crop=w:h:x:y
|
||||
crop_match = re.search(
|
||||
r"crop=(\d+):(\d+):(\d+):(\d+)",
|
||||
(result_cropdetect.stdout or "") + (result_cropdetect.stderr or ""),
|
||||
)
|
||||
if not crop_match:
|
||||
return None
|
||||
w, h = int(crop_match.group(1)), int(crop_match.group(2))
|
||||
x, y = int(crop_match.group(3)), int(crop_match.group(4))
|
||||
return (x, y, original_width - w - x, original_height - h - y)
|
||||
|
||||
with futures.ThreadPoolExecutor(max_workers=len(random_times)) as pool:
|
||||
crop_results = [crop for crop in pool.map(sample_crop, random_times) if crop]
|
||||
|
||||
if not crop_results:
|
||||
log_event(
|
||||
"hybrid_level5",
|
||||
level="WARNING",
|
||||
message="No crop data detected, skipping L5",
|
||||
context={"samples": len(random_times)},
|
||||
)
|
||||
self.log.warning("No crop data detected, skipping L5")
|
||||
return
|
||||
|
||||
crop_counts = {}
|
||||
for crop in crop_results:
|
||||
crop_counts[crop] = crop_counts.get(crop, 0) + 1
|
||||
most_common = max(crop_counts, key=crop_counts.get)
|
||||
left, top, right, bottom = most_common
|
||||
left, top, right, bottom = crop
|
||||
|
||||
l5_json = {
|
||||
"active_area": {
|
||||
@@ -373,13 +344,160 @@ class Hybrid:
|
||||
"hybrid_level5",
|
||||
level="DEBUG",
|
||||
message="Edited RPU Level 5 active area",
|
||||
context={"crop": {"left": left, "right": right, "top": top, "bottom": bottom}},
|
||||
success=True,
|
||||
)
|
||||
self.rpu_file = "RPU_L5.bin"
|
||||
|
||||
def detect_active_area(self, input_video: Path) -> Optional[tuple[tuple[int, int, int, int], float]]:
|
||||
"""The picture's crop as (left, top, right, bottom) and its share of the frame, or None.
|
||||
|
||||
None also when the picture is under 40% of the frame: cropdetect on dark samples can report a tiny
|
||||
picture, which would be wrong in L5 and would divide the HDR10+ conversion's mean by a tiny share.
|
||||
"""
|
||||
ffprobe_bin = str(FFProbe) if FFProbe else "ffprobe"
|
||||
ffmpeg_bin = str(FFMPEG) if FFMPEG else "ffmpeg"
|
||||
|
||||
with console.status("Detecting active area (crop detection)...", spinner="dots"):
|
||||
result_duration = subprocess.run(
|
||||
[ffprobe_bin, "-v", "error", "-show_entries", "format=duration", "-of", "json", str(input_video)],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
encoding="utf-8",
|
||||
errors="replace",
|
||||
)
|
||||
|
||||
if result_duration.returncode != 0:
|
||||
log_event(
|
||||
"hybrid_level5",
|
||||
level="WARNING",
|
||||
message="Could not probe video duration",
|
||||
context={"returncode": result_duration.returncode, "stderr": (result_duration.stderr or "")},
|
||||
)
|
||||
self.log.warning("Could not probe video duration, skipping crop detection")
|
||||
return None
|
||||
|
||||
duration_value = json.loads(result_duration.stdout).get("format", {}).get("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 crop detection. "
|
||||
"The RPU keeps the active area of the DV track."
|
||||
)
|
||||
return None
|
||||
duration = float(duration_value)
|
||||
|
||||
result_streams = subprocess.run(
|
||||
[
|
||||
ffprobe_bin,
|
||||
"-v",
|
||||
"error",
|
||||
"-select_streams",
|
||||
"v:0",
|
||||
"-show_entries",
|
||||
"stream=width,height",
|
||||
"-of",
|
||||
"json",
|
||||
str(input_video),
|
||||
],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
encoding="utf-8",
|
||||
errors="replace",
|
||||
)
|
||||
|
||||
if result_streams.returncode != 0:
|
||||
log_event(
|
||||
"hybrid_level5",
|
||||
level="WARNING",
|
||||
message="Could not probe video resolution",
|
||||
context={"returncode": result_streams.returncode, "stderr": (result_streams.stderr or "")},
|
||||
)
|
||||
self.log.warning("Could not probe video resolution, skipping crop detection")
|
||||
return None
|
||||
|
||||
stream_info = json.loads(result_streams.stdout)
|
||||
original_width = int(stream_info["streams"][0]["width"])
|
||||
original_height = int(stream_info["streams"][0]["height"])
|
||||
|
||||
sample_times = [duration * (i + 0.5) / 10 for i in range(10)]
|
||||
|
||||
def sample_crop(t: float) -> Optional[tuple[int, int, int, int]]:
|
||||
result_cropdetect = subprocess.run(
|
||||
[
|
||||
ffmpeg_bin,
|
||||
"-y",
|
||||
"-nostdin",
|
||||
"-loglevel",
|
||||
"info",
|
||||
"-ss",
|
||||
f"{t:.2f}",
|
||||
"-i",
|
||||
str(input_video),
|
||||
"-vf",
|
||||
"cropdetect=round=2",
|
||||
# cropdetect skips its first 2 frames and only ever grows the picture, so 12 frames
|
||||
# give each sample 10 evaluated ones
|
||||
"-vframes",
|
||||
"12",
|
||||
"-f",
|
||||
"null",
|
||||
"-",
|
||||
],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
encoding="utf-8",
|
||||
errors="replace",
|
||||
)
|
||||
|
||||
# cropdetect prints crop=w:h:x:y per frame and settles over the frames, so the last line counts
|
||||
crop_matches = re.findall(
|
||||
r"crop=(\d+):(\d+):(\d+):(\d+)",
|
||||
(result_cropdetect.stdout or "") + (result_cropdetect.stderr or ""),
|
||||
)
|
||||
if not crop_matches:
|
||||
return None
|
||||
w, h, x, y = map(int, crop_matches[-1])
|
||||
return (x, y, original_width - w - x, original_height - h - y)
|
||||
|
||||
with futures.ThreadPoolExecutor(max_workers=len(sample_times)) as pool:
|
||||
crop_results = [crop for crop in pool.map(sample_crop, sample_times) if crop]
|
||||
|
||||
if not crop_results:
|
||||
log_event(
|
||||
"hybrid_level5",
|
||||
level="WARNING",
|
||||
message="No crop data detected",
|
||||
context={"samples": len(sample_times)},
|
||||
)
|
||||
self.log.warning("No crop data detected, skipping the active area")
|
||||
return None
|
||||
|
||||
left, top, right, bottom = Counter(crop_results).most_common(1)[0][0]
|
||||
fraction = (
|
||||
(original_width - left - right) * (original_height - top - bottom) / (original_width * original_height)
|
||||
)
|
||||
if fraction < 0.4:
|
||||
self.log.warning("The detected picture is under 40% of the frame, skipping the active area")
|
||||
return None
|
||||
log_event(
|
||||
"hybrid_level5",
|
||||
level="DEBUG",
|
||||
message="Detected active area",
|
||||
context={
|
||||
"crop": {"left": left, "right": right, "top": top, "bottom": bottom},
|
||||
"samples": len(crop_results),
|
||||
},
|
||||
success=True,
|
||||
)
|
||||
self.rpu_file = "RPU_L5.bin"
|
||||
return (left, top, right, bottom), fraction
|
||||
|
||||
@staticmethod
|
||||
def sanitize_l6(
|
||||
@@ -537,7 +655,6 @@ class Hybrid:
|
||||
"hdr_type": self.hdr_type,
|
||||
"rpu_file": self.rpu_file,
|
||||
"output": self.hevc_file,
|
||||
"drop_hdr10plus": self.hdr10plus_to_dv,
|
||||
},
|
||||
success=True,
|
||||
)
|
||||
@@ -574,6 +691,8 @@ class Hybrid:
|
||||
|
||||
file_size = os.path.getsize(config.directories.temp / self.hdr10plus_file)
|
||||
if file_size == 0:
|
||||
# left in place, the next resolution would take the empty file for its metadata
|
||||
Path.unlink(config.directories.temp / self.hdr10plus_file, missing_ok=True)
|
||||
log_event(
|
||||
"hybrid_extract_hdr10plus",
|
||||
level="ERROR",
|
||||
@@ -655,51 +774,83 @@ class Hybrid:
|
||||
|
||||
return max_mdl, min_mdl, max_cll, max_fall
|
||||
|
||||
def convert_hdr10plus_to_dv(self):
|
||||
"""Convert HDR10+ metadata to Dolby Vision RPU"""
|
||||
if os.path.isfile(config.directories.temp / "RPU.bin"):
|
||||
return
|
||||
|
||||
with console.status("Converting HDR10+ metadata to Dolby Vision...", spinner="dots"):
|
||||
max_mdl, min_mdl, max_cll, max_fall = self.probe_hdr_metadata()
|
||||
max_mdl, min_mdl, max_cll, max_fall = self.sanitize_l6(max_mdl, min_mdl, max_cll, max_fall)
|
||||
|
||||
extra_metadata = {
|
||||
"cm_version": "V29",
|
||||
"length": 0, # dovi_tool will figure this out
|
||||
"level6": {
|
||||
"max_display_mastering_luminance": max_mdl,
|
||||
"min_display_mastering_luminance": min_mdl,
|
||||
"max_content_light_level": max_cll,
|
||||
"max_frame_average_light_level": max_fall,
|
||||
},
|
||||
}
|
||||
|
||||
with open(config.directories.temp / "extra.json", "w") as f:
|
||||
json.dump(extra_metadata, f, indent=2)
|
||||
def convert_hdr10plus_to_dv(self, active_fraction: float = 1.0) -> None:
|
||||
"""Generate a Dolby Vision RPU with one shot per HDR10+ scene.
|
||||
|
||||
By default unshackle builds the L1 of each shot itself (see hdr10plus_shots), because dovi_tool's
|
||||
--hdr10plus-json sets L1 avg from the linear AverageMaxRGB and so makes most scenes too bright.
|
||||
config.generate_dv_mode picks the L1 avg floor: v29 (about 2.4 nits, closest to the pixels) or v40
|
||||
(about 10 nits, as in studio CM v4.0 RPUs). legacy hands the HDR10+ JSON to dovi_tool instead.
|
||||
"""
|
||||
mode = config.generate_dv_mode
|
||||
if mode not in GENERATE_DV_MODES:
|
||||
raise ValueError(f"generate_dv_mode must be one of {', '.join(GENERATE_DV_MODES)}, not {mode!r}")
|
||||
hdr10plus_path = config.directories.temp / self.hdr10plus_file
|
||||
config_path = config.directories.temp / "generate.json"
|
||||
try:
|
||||
dovi.generate_from_hdr10plus(
|
||||
config.directories.temp / "extra.json",
|
||||
config.directories.temp / self.hdr10plus_file,
|
||||
config.directories.temp / "RPU.bin",
|
||||
label="dovi_tool generate",
|
||||
)
|
||||
except RuntimeError as e:
|
||||
log_event(
|
||||
"hybrid_convert_hdr10plus",
|
||||
level="ERROR",
|
||||
message="Failed converting HDR10+ to Dolby Vision",
|
||||
context={"error": str(e)},
|
||||
)
|
||||
raise ValueError("Failed converting HDR10+ to Dolby Vision")
|
||||
with console.status("Converting HDR10+ metadata to Dolby Vision...", spinner="dots"):
|
||||
with open(hdr10plus_path, encoding="utf-8") as f:
|
||||
metadata = json.load(f)
|
||||
profile = metadata.get("JSONInfo", {}).get("HDR10plusProfile")
|
||||
shots = hdr10plus_shots(metadata, active_fraction)
|
||||
# one entry per frame: a feature film is hundreds of MB of objects, freed before dovi_tool runs
|
||||
del metadata
|
||||
|
||||
max_mdl, min_mdl, max_cll, max_fall = self.probe_hdr_metadata()
|
||||
max_mdl, min_mdl, max_cll, max_fall = self.sanitize_l6(max_mdl, min_mdl, max_cll, max_fall)
|
||||
generator_config = {
|
||||
"cm_version": "V29",
|
||||
"level6": {
|
||||
"max_display_mastering_luminance": max_mdl,
|
||||
"min_display_mastering_luminance": min_mdl,
|
||||
"max_content_light_level": max_cll,
|
||||
"max_frame_average_light_level": max_fall,
|
||||
},
|
||||
}
|
||||
if mode == "legacy":
|
||||
generator_config["length"] = 0 # dovi_tool takes the length from the HDR10+ JSON
|
||||
else:
|
||||
generator_config["shots"] = shots
|
||||
if mode == "v40":
|
||||
generator_config["l1_avg_pq_cm_version"] = "V40"
|
||||
with open(config_path, "w", encoding="utf-8") as f:
|
||||
json.dump(generator_config, f)
|
||||
|
||||
if profile == "A":
|
||||
self.log.warning(
|
||||
"The HDR10+ metadata is Profile A, which is often less reliable than Profile B. "
|
||||
"Check the Dolby Vision result before you keep it."
|
||||
)
|
||||
|
||||
try:
|
||||
dovi.generate(
|
||||
config_path,
|
||||
config.directories.temp / "RPU.bin",
|
||||
hdr10plus_json=hdr10plus_path if mode == "legacy" else None,
|
||||
label="dovi_tool generate",
|
||||
)
|
||||
except RuntimeError as e:
|
||||
log_event(
|
||||
"hybrid_convert_hdr10plus",
|
||||
level="ERROR",
|
||||
message="Failed converting HDR10+ to Dolby Vision",
|
||||
context={"error": str(e)},
|
||||
)
|
||||
raise ValueError("Failed converting HDR10+ to Dolby Vision")
|
||||
finally:
|
||||
config_path.unlink(missing_ok=True)
|
||||
hdr10plus_path.unlink(missing_ok=True)
|
||||
|
||||
log_event(
|
||||
"hybrid_convert_hdr10plus",
|
||||
level="DEBUG",
|
||||
message="Converted HDR10+ metadata to Dolby Vision Profile 8",
|
||||
message="Converted HDR10+ metadata to Dolby Vision Profile 8.1",
|
||||
context={
|
||||
"shots": len(shots),
|
||||
"mode": mode,
|
||||
"hdr10plus_profile": profile,
|
||||
"active_fraction": round(active_fraction, 3),
|
||||
},
|
||||
success=True,
|
||||
)
|
||||
|
||||
Path.unlink(config.directories.temp / "extra.json")
|
||||
Path.unlink(config.directories.temp / self.hdr10plus_file)
|
||||
self.log.info(f"Dolby Vision generated from HDR10+ metadata ({len(shots)} scenes, {mode})")
|
||||
|
||||
@@ -440,14 +440,17 @@ class Tracks:
|
||||
video.hybrid_base_only = True
|
||||
|
||||
def select_hybrid(self, tracks, quality, worst: bool = False):
|
||||
# Prefer HDR10+ over HDR10 as the base layer (preserves dynamic metadata)
|
||||
# Prefer HDR10+ over HDR10 as the base layer (preserves dynamic metadata).
|
||||
# Only an HEVC base qualifies, because dovi_tool injects the RPU into an HEVC stream.
|
||||
base_ranges = (Video.Range.HDR10P, Video.Range.HDR10)
|
||||
base_tracks = []
|
||||
for range_type in base_ranges:
|
||||
base_tracks = [
|
||||
v
|
||||
for v in tracks
|
||||
if v.range == range_type and (v.height in quality or (v.width and int(v.width * 9 / 16) in quality))
|
||||
if v.range == range_type
|
||||
and v.codec in (Video.Codec.HEVC, None)
|
||||
and (v.height in quality or (v.width and int(v.width * 9 / 16) in quality))
|
||||
]
|
||||
if base_tracks:
|
||||
break
|
||||
|
||||
@@ -127,18 +127,19 @@ def info_frame(rpu: Path, frame: int = 0) -> dict:
|
||||
raise RuntimeError(f"dovi_tool info returned invalid JSON: {e}")
|
||||
|
||||
|
||||
def generate_from_hdr10plus(
|
||||
extra_json: Path,
|
||||
hdr10plus_json: Path,
|
||||
def generate(
|
||||
config_json: Path,
|
||||
output: Path,
|
||||
*,
|
||||
status: Optional[str] = "Generating DV RPU from HDR10+ metadata...",
|
||||
hdr10plus_json: Optional[Path] = None,
|
||||
status: Optional[str] = "Generating DV RPU...",
|
||||
label: str = "dovi_tool generate",
|
||||
) -> bytes:
|
||||
"""Assemble a DV RPU from extracted HDR10+ metadata + an extra JSON descriptor."""
|
||||
"""Build a DV RPU from a dovi_tool generator config, and from hdr10plus_tool JSON when given."""
|
||||
tool = require_dovi_tool()
|
||||
extra = ["--hdr10plus-json", hdr10plus_json] if hdr10plus_json else []
|
||||
return run_step(
|
||||
[tool, "generate", "-j", extra_json, "--hdr10plus-json", hdr10plus_json, "-o", output],
|
||||
[tool, "generate", "-j", config_json, *extra, "-o", output],
|
||||
status=status,
|
||||
output=output,
|
||||
label=label,
|
||||
@@ -166,5 +167,5 @@ __all__ = (
|
||||
"editor",
|
||||
"info_summary",
|
||||
"info_frame",
|
||||
"generate_from_hdr10plus",
|
||||
"generate",
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user