mirror of
https://github.com/unshackle-dl/unshackle.git
synced 2026-11-14 21:15:06 +00:00
feat(audio): support DTS:X Profile 2 with MP4 muxing
Route titles with a dtsx/dtsy sample entry to MP4 via MP4Box, since Matroska has no CodecID for DTS-UHD. Add per-rendition HLS audio codec selection, fix merge_segments progress and its cleanup path.
This commit is contained in:
imSp4rky
@@ -0,0 +1,53 @@
|
||||
"""Audio codec and channel layout in a filename."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import types
|
||||
|
||||
import pytest
|
||||
|
||||
from unshackle.core.titles.movie import Movie
|
||||
|
||||
|
||||
class DummyService:
|
||||
pass
|
||||
|
||||
|
||||
def named(fmt: str, layout: str | None, channel_s: int = 6) -> dict:
|
||||
"""Build the naming context from a MediaInfo audio track alone."""
|
||||
audio = types.SimpleNamespace(
|
||||
language="en",
|
||||
format=fmt,
|
||||
channel_layout=layout,
|
||||
channellayout_original=None,
|
||||
channel_s=channel_s,
|
||||
channels=channel_s,
|
||||
format_additionalfeatures=None,
|
||||
joc=None,
|
||||
)
|
||||
movie = Movie(id_="movie-0001", service=DummyService, name="The Film", year=2024, language="en")
|
||||
return movie.build_base_template_context(types.SimpleNamespace(video_tracks=[], audio_tracks=[audio]))
|
||||
|
||||
|
||||
def test_a_height_layout_is_named_bed_lfe_heights() -> None:
|
||||
"""
|
||||
DTS:X and Atmos carry height channels. Folding them into the bed gives 9.1, which names
|
||||
nine ear-level speakers: a different layout from five plus four overhead.
|
||||
"""
|
||||
context = named("DTS-UHD", "L C R LFE Ls Rs Tfl Tfr Tbl Tbr", channel_s=10)
|
||||
assert context["audio_channels"] == "5.1.4"
|
||||
assert context["audio"] == "DTS-X", "the filename carries the brand, not the bitstream name"
|
||||
assert context["audio_full"] == "DTS-X 5.1.4"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("layout", "expected"),
|
||||
[
|
||||
("L R C LFE Ls Rs", "5.1"),
|
||||
("L R C LFE Ls Rs Lb Rb", "7.1"),
|
||||
("L R", "2.0"),
|
||||
(None, "6.0"),
|
||||
],
|
||||
)
|
||||
def test_layouts_without_heights_are_unchanged(layout: str | None, expected: str) -> None:
|
||||
assert named("E-AC-3", layout)["audio_channels"] == expected
|
||||
@@ -0,0 +1,140 @@
|
||||
"""HLS.merge_segments must keep the ffmpeg child off the terminal.
|
||||
|
||||
merge_segments runs while the download progress tree is live. A child that inherits the
|
||||
terminal moves the cursor out from under rich, which leaves an orphaned copy of the tree
|
||||
on screen. Real trigger: ffmpeg refuses DTS-UHD audio with "Cannot map stream #0:0".
|
||||
"""
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from unshackle.core.manifests.hls import HLS
|
||||
|
||||
FFMPEG_ERROR = "[out#0/mp4 @ 0x0] Cannot map stream #0:0 - unsupported type."
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def segments(tmp_path: Path) -> list[Path]:
|
||||
save_dir = tmp_path / "Audio_segments"
|
||||
save_dir.mkdir()
|
||||
paths = []
|
||||
for i in range(3):
|
||||
path = save_dir / f"{i:05}.mp4"
|
||||
path.write_bytes(f"segment-{i}".encode())
|
||||
paths.append(path)
|
||||
return paths
|
||||
|
||||
|
||||
def _capture_fd(fd: int):
|
||||
"""Redirect a raw file descriptor to a temp file, returning (restore, read)."""
|
||||
sink = tempfile.TemporaryFile()
|
||||
saved = os.dup(fd)
|
||||
os.dup2(sink.fileno(), fd)
|
||||
|
||||
def restore() -> bytes:
|
||||
os.dup2(saved, fd)
|
||||
os.close(saved)
|
||||
sink.seek(0)
|
||||
data = sink.read()
|
||||
sink.close()
|
||||
return data
|
||||
|
||||
return restore
|
||||
|
||||
|
||||
def test_failed_ffmpeg_concat_writes_nothing_to_the_terminal(
|
||||
tmp_path: Path, segments: list[Path], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
def fake_run(args, **kwargs):
|
||||
# emulate the child: with no capture it writes straight to the inherited terminal
|
||||
if not kwargs.get("capture_output") and kwargs.get("stderr") is None:
|
||||
os.write(2, (FFMPEG_ERROR + "\n").encode())
|
||||
raise subprocess.CalledProcessError(1, args)
|
||||
raise subprocess.CalledProcessError(1, args, output="", stderr=FFMPEG_ERROR)
|
||||
|
||||
monkeypatch.setattr("unshackle.core.manifests.hls.binaries.FFMPEG", Path("ffmpeg"))
|
||||
monkeypatch.setattr("unshackle.core.manifests.hls.subprocess.run", fake_run)
|
||||
|
||||
save_path = tmp_path / "Audio.mp4"
|
||||
read_stderr = _capture_fd(2)
|
||||
try:
|
||||
HLS.merge_segments(segments=segments, save_path=save_path)
|
||||
finally:
|
||||
leaked = read_stderr()
|
||||
|
||||
assert leaked == b"", f"ffmpeg output reached the terminal: {leaked!r}"
|
||||
assert save_path.read_bytes() == b"segment-0segment-1segment-2"
|
||||
|
||||
|
||||
def test_failed_ffmpeg_concat_logs_the_reason(
|
||||
tmp_path: Path, segments: list[Path], monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
def fake_run(args, **kwargs):
|
||||
raise subprocess.CalledProcessError(1, args, output="", stderr=FFMPEG_ERROR)
|
||||
|
||||
monkeypatch.setattr("unshackle.core.manifests.hls.binaries.FFMPEG", Path("ffmpeg"))
|
||||
monkeypatch.setattr("unshackle.core.manifests.hls.subprocess.run", fake_run)
|
||||
|
||||
with caplog.at_level("DEBUG", logger="HLS"):
|
||||
HLS.merge_segments(segments=segments, save_path=tmp_path / "Audio.mp4")
|
||||
|
||||
assert FFMPEG_ERROR in caplog.text
|
||||
|
||||
|
||||
def test_binary_concat_advances_progress_per_segment(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""FFmpeg concat reports nothing, so the fallback is the only place a count can come from."""
|
||||
segments = []
|
||||
for i in range(4):
|
||||
seg = tmp_path / f"seg{i}.mp4"
|
||||
seg.write_bytes(b"\x00" * 16)
|
||||
segments.append(seg)
|
||||
|
||||
calls: list[dict] = []
|
||||
# no ffmpeg binary means merge_segments takes the binary fallback
|
||||
monkeypatch.setattr("unshackle.core.manifests.hls.binaries.FFMPEG", None)
|
||||
HLS.merge_segments(
|
||||
segments=segments,
|
||||
save_path=tmp_path / "out.mp4",
|
||||
progress=lambda **kwargs: calls.append(kwargs),
|
||||
)
|
||||
|
||||
assert calls.count({"advance": 1}) == len(segments), f"expected one advance per segment, got {calls}"
|
||||
assert (tmp_path / "out.mp4").stat().st_size == 4 * 16
|
||||
|
||||
|
||||
def test_merge_segments_without_progress_still_works(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""progress stays optional; other callers pass nothing."""
|
||||
seg = tmp_path / "seg0.mp4"
|
||||
seg.write_bytes(b"\x01" * 8)
|
||||
monkeypatch.setattr("unshackle.core.manifests.hls.binaries.FFMPEG", None)
|
||||
size = HLS.merge_segments(segments=[seg], save_path=tmp_path / "out.mp4")
|
||||
assert size == 8
|
||||
|
||||
|
||||
def test_cleanup_never_climbs_past_the_segment_store(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""
|
||||
A download directory may itself say "_segments". Only the track's own segment store
|
||||
may be removed, never a parent that merely contains the word.
|
||||
"""
|
||||
library = tmp_path / "media_segments"
|
||||
store = library / "temp" / "task_x" / "Audio_1.mp4_segments" / "segments"
|
||||
store.mkdir(parents=True)
|
||||
keep = library / "Some Other Movie.mkv"
|
||||
keep.write_bytes(b"do not delete me")
|
||||
|
||||
segments = []
|
||||
for i in range(2):
|
||||
seg = store / f"{i:05}.mp4"
|
||||
seg.write_bytes(b"\x00" * 8)
|
||||
segments.append(seg)
|
||||
|
||||
monkeypatch.setattr("unshackle.core.manifests.hls.binaries.FFMPEG", None)
|
||||
HLS.merge_segments(segments=segments, save_path=library / "out.mp4")
|
||||
|
||||
assert keep.exists(), "cleanup deleted an unrelated file in the download directory"
|
||||
assert library.exists()
|
||||
assert not (library / "temp" / "task_x" / "Audio_1.mp4_segments").exists(), "the segment store was left behind"
|
||||
@@ -0,0 +1,246 @@
|
||||
"""Container selection for audio Matroska cannot name.
|
||||
|
||||
DTS:X ships in two shapes. Profile 2 (DTS-UHD, sample entry dtsx/dtsy) has no Matroska
|
||||
CodecID, so mkvmerge stores it as an A_QUICKTIME passthrough and no reader can name the
|
||||
codec; those titles have to go to MP4. Profile 1 rides inside DTS-HD Master Audio
|
||||
(dtsc/dtse/dtsh/
|
||||
dtsl), which Matroska maps to A_DTS and stores properly, so it must stay in Matroska.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import struct
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from unshackle.core import binaries
|
||||
from unshackle.core.tracks import Audio, Subtitle, Tracks, Video
|
||||
from unshackle.core.tracks.track import (
|
||||
has_dts_uhd_sample_entry,
|
||||
iter_top_level_boxes,
|
||||
strip_duplicate_init_boxes,
|
||||
)
|
||||
|
||||
|
||||
def box(box_type: bytes, payload: bytes = b"") -> bytes:
|
||||
return struct.pack(">I", 8 + len(payload)) + box_type + payload
|
||||
|
||||
|
||||
def write_mp4(path: Path, sample_entry: bytes) -> Path:
|
||||
"""An MP4 whose moov holds one audio sample entry of the given 4CC."""
|
||||
entry = box(sample_entry, b"\x00" * 20)
|
||||
path.write_bytes(box(b"ftyp", b"isom") + box(b"moov", box(b"trak", box(b"stsd", entry))))
|
||||
return path
|
||||
|
||||
|
||||
@pytest.mark.parametrize("fourcc", [b"dtsx", b"dtsy"])
|
||||
def test_dts_uhd_sample_entries_are_detected(tmp_path: Path, fourcc: bytes) -> None:
|
||||
assert has_dts_uhd_sample_entry(write_mp4(tmp_path / "a.mp4", fourcc)) is True
|
||||
|
||||
|
||||
@pytest.mark.parametrize("fourcc", [b"dtsc", b"dtse", b"dtsh", b"dtsl", b"ec-3", b"mp4a"])
|
||||
def test_other_dts_profiles_stay_in_matroska(tmp_path: Path, fourcc: bytes) -> None:
|
||||
"""
|
||||
DTS:X Profile 1 is DTS-HD Master Audio underneath and Matroska stores it as
|
||||
A_DTS, so it must not be pulled out of Matroska by the DTS:X container rule.
|
||||
"""
|
||||
assert has_dts_uhd_sample_entry(write_mp4(tmp_path / "a.mp4", fourcc)) is False
|
||||
|
||||
|
||||
def test_a_file_with_no_moov_is_not_detected(tmp_path: Path) -> None:
|
||||
path = tmp_path / "a.mp4"
|
||||
path.write_bytes(box(b"ftyp", b"isom") + box(b"mdat", b"dtsx" * 8))
|
||||
assert has_dts_uhd_sample_entry(path) is False, "a 4CC in mdat is not a sample entry"
|
||||
|
||||
|
||||
def build_tracks(tmp_path: Path, audio_entry: bytes) -> Tracks:
|
||||
video = Video(id_="v1", url="https://x/v", codec=Video.Codec.AVC, language="en", width=1920, height=1080, fps=24)
|
||||
audio = Audio(id_="a1", url="https://x/a", codec=Audio.Codec.DTSX, language="en", bitrate=448000)
|
||||
video.path = write_mp4(tmp_path / "v.mp4", b"avc1")
|
||||
audio.path = write_mp4(tmp_path / "a.mp4", audio_entry)
|
||||
return Tracks([video, audio])
|
||||
|
||||
|
||||
class FakePopen:
|
||||
"""Stands in for MP4Box, recording the command it was given."""
|
||||
|
||||
stdout_text = ""
|
||||
|
||||
def __init__(self, args, **kwargs):
|
||||
self.args = args
|
||||
self.kwargs = kwargs
|
||||
self.stdout = io.StringIO(FakePopen.stdout_text or "ISO File Writing: |== | (50/100)\n")
|
||||
|
||||
def wait(self) -> int:
|
||||
return 0
|
||||
|
||||
|
||||
def test_dts_uhd_audio_routes_the_mux_to_mp4(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
captured: list[FakePopen] = []
|
||||
|
||||
def fake_popen(args, **kwargs):
|
||||
p = FakePopen(args, **kwargs)
|
||||
captured.append(p)
|
||||
Path(args[-1]).write_bytes(b"muxed")
|
||||
return p
|
||||
|
||||
monkeypatch.setattr(binaries, "MP4Box", Path("MP4Box"), raising=False)
|
||||
monkeypatch.setattr(subprocess, "Popen", fake_popen)
|
||||
|
||||
tracks = build_tracks(tmp_path, b"dtsx")
|
||||
out, returncode, errors = tracks.mux("Title", delete=False, output_path=tmp_path / "out.muxed.mkv")
|
||||
|
||||
assert out.suffix == ".mp4", "a DTS-UHD title must not be delivered as Matroska"
|
||||
assert returncode == 0 and not errors
|
||||
command = captured[0].args
|
||||
assert command[0] == "MP4Box"
|
||||
assert command[-2:] == ["-new", str(out)]
|
||||
assert f"{tmp_path / 'a.mp4'}:lang=en" in command
|
||||
# the child must not inherit the terminal that the live progress tree is drawing on
|
||||
assert captured[0].kwargs["stdout"] is subprocess.PIPE
|
||||
assert captured[0].kwargs["stderr"] is subprocess.STDOUT
|
||||
|
||||
|
||||
def test_other_audio_still_muxes_with_mkvmerge(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(binaries, "MP4Box", None, raising=False)
|
||||
monkeypatch.setattr("unshackle.core.tracks.tracks.binaries.MKVToolNix", None)
|
||||
|
||||
tracks = build_tracks(tmp_path, b"dtsh")
|
||||
# reaching the mkvmerge requirement proves the DTS-UHD branch was not taken; MP4Box
|
||||
# is None here too, so the MP4 branch would have raised about MP4Box instead
|
||||
with pytest.raises(RuntimeError, match="MKVToolNix"):
|
||||
tracks.mux("Title", delete=False, output_path=tmp_path / "out.muxed.mkv")
|
||||
|
||||
|
||||
def test_a_missing_mp4box_names_the_install(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(binaries, "MP4Box", None, raising=False)
|
||||
tracks = build_tracks(tmp_path, b"dtsx")
|
||||
with pytest.raises(RuntimeError, match="gpac"):
|
||||
tracks.mux("Title", delete=False, output_path=tmp_path / "out.muxed.mkv")
|
||||
|
||||
|
||||
def test_substation_alpha_subtitles_are_left_out(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""MP4 has no ASS format and tx3g would drop the styling that picked it."""
|
||||
captured: list[FakePopen] = []
|
||||
|
||||
def fake_popen(args, **kwargs):
|
||||
p = FakePopen(args, **kwargs)
|
||||
captured.append(p)
|
||||
Path(args[-1]).write_bytes(b"muxed")
|
||||
return p
|
||||
|
||||
monkeypatch.setattr(binaries, "MP4Box", Path("MP4Box"), raising=False)
|
||||
monkeypatch.setattr(subprocess, "Popen", fake_popen)
|
||||
|
||||
tracks = build_tracks(tmp_path, b"dtsx")
|
||||
ass = Subtitle(id_="s1", url="https://x/s", codec=Subtitle.Codec.SubStationAlphav4, language="en")
|
||||
ass.path = tmp_path / "s.ass"
|
||||
ass.path.write_text("[Script Info]\n")
|
||||
vtt = Subtitle(id_="s2", url="https://x/s2", codec=Subtitle.Codec.WebVTT, language="es")
|
||||
vtt.path = tmp_path / "s.vtt"
|
||||
vtt.path.write_text("WEBVTT\n")
|
||||
tracks.add([ass, vtt])
|
||||
|
||||
tracks.mux("Title", delete=False, output_path=tmp_path / "out.muxed.mkv")
|
||||
command = " ".join(captured[0].args)
|
||||
assert "s.ass" not in command
|
||||
assert "s.vtt:lang=es" in command
|
||||
|
||||
|
||||
def test_mp4box_errors_come_back_to_the_caller(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""A failed mux does not raise; the caller inspects the return code and lines."""
|
||||
|
||||
class FailingPopen(FakePopen):
|
||||
def __init__(self, args, **kwargs):
|
||||
super().__init__(args, **kwargs)
|
||||
self.stdout = io.StringIO("Error importing track: not supported\n")
|
||||
|
||||
def wait(self) -> int:
|
||||
return 1
|
||||
|
||||
monkeypatch.setattr(binaries, "MP4Box", Path("MP4Box"), raising=False)
|
||||
monkeypatch.setattr(subprocess, "Popen", FailingPopen)
|
||||
|
||||
tracks = build_tracks(tmp_path, b"dtsx")
|
||||
_, returncode, errors = tracks.mux("Title", delete=False, output_path=tmp_path / "out.muxed.mkv")
|
||||
assert returncode == 1
|
||||
assert errors == ["Error importing track: not supported"]
|
||||
|
||||
|
||||
def init_pair(sample_entry: bytes, stamp: bytes = b"\x00\x00\x00\x01") -> bytes:
|
||||
"""An ftyp+moov pair; the stamp stands in for the fetch time a packager writes."""
|
||||
stsd = box(b"stsd", box(sample_entry, b"\x00" * 20))
|
||||
return box(b"ftyp", b"isom") + box(b"moov", box(b"mvhd", stamp) + box(b"trak", stsd))
|
||||
|
||||
|
||||
def fragment(payload: bytes) -> bytes:
|
||||
return box(b"moof", b"\x00" * 8) + box(b"mdat", payload)
|
||||
|
||||
|
||||
def test_repeated_inits_are_dropped_when_the_stsd_matches(tmp_path: Path) -> None:
|
||||
"""
|
||||
An HLS discontinuity gives each period its own init, so a merged track holds several.
|
||||
Only the stsd decides how samples decode, and a packager stamps each init with the
|
||||
fetch time, so copies that differ only in that stamp are safe to drop.
|
||||
"""
|
||||
src = tmp_path / "a.mp4"
|
||||
src.write_bytes(
|
||||
init_pair(b"dtsx", b"\x00\x00\x00\x01")
|
||||
+ fragment(b"one")
|
||||
+ init_pair(b"dtsx", b"\x00\x00\x00\x99")
|
||||
+ fragment(b"two")
|
||||
)
|
||||
dst = tmp_path / "clean.mp4"
|
||||
|
||||
assert strip_duplicate_init_boxes(src, dst) == 2, "the second ftyp and moov should be dropped"
|
||||
types = [t for t, _, _ in iter_top_level_boxes(dst)]
|
||||
assert types == [b"ftyp", b"moov", b"moof", b"mdat", b"moof", b"mdat"]
|
||||
assert dst.read_bytes().count(b"mdat") == 2, "no sample data may be lost"
|
||||
|
||||
|
||||
def test_a_changed_stsd_is_never_dropped(tmp_path: Path) -> None:
|
||||
"""A later init describing a different format is a real change; dropping it corrupts."""
|
||||
src = tmp_path / "a.mp4"
|
||||
src.write_bytes(init_pair(b"dtsx") + fragment(b"one") + init_pair(b"ec-3") + fragment(b"two"))
|
||||
dst = tmp_path / "clean.mp4"
|
||||
|
||||
assert strip_duplicate_init_boxes(src, dst) == 0
|
||||
assert not dst.exists(), "the caller must keep the original when the format changes"
|
||||
|
||||
|
||||
def test_a_single_init_is_left_alone(tmp_path: Path) -> None:
|
||||
src = tmp_path / "a.mp4"
|
||||
src.write_bytes(init_pair(b"dtsx") + fragment(b"one"))
|
||||
dst = tmp_path / "clean.mp4"
|
||||
assert strip_duplicate_init_boxes(src, dst) == 0
|
||||
assert not dst.exists()
|
||||
|
||||
|
||||
def test_an_unreadable_input_aborts_instead_of_shipping_a_placeholder(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""
|
||||
MP4Box answers a file its ISOM reader refused by importing it through ffmpeg, which
|
||||
writes the samples under a placeholder 4CC. mkvmerge-style return codes treat 1 as a
|
||||
warning, so that shipped a whole title with an unnameable audio track.
|
||||
"""
|
||||
|
||||
class RefusingPopen(FakePopen):
|
||||
def __init__(self, args, **kwargs):
|
||||
super().__init__(args, **kwargs)
|
||||
self.stdout = io.StringIO("[IsoMedia] error while opening /tmp/a.mp4, error=Invalid IsoMedia File\n")
|
||||
|
||||
def wait(self) -> int:
|
||||
return 0
|
||||
|
||||
monkeypatch.setattr(binaries, "MP4Box", Path("MP4Box"), raising=False)
|
||||
monkeypatch.setattr(subprocess, "Popen", RefusingPopen)
|
||||
|
||||
tracks = build_tracks(tmp_path, b"dtsx")
|
||||
_, returncode, errors = tracks.mux("Title", delete=False, output_path=tmp_path / "out.muxed.mkv")
|
||||
|
||||
assert returncode >= 2, "a refused input must abort the run, not pass as a warning"
|
||||
assert errors
|
||||
@@ -132,7 +132,8 @@ def test_dolby_digital_plus_falls_back_to_the_mp4_muxer(tmp_path: Path) -> None:
|
||||
|
||||
def test_a_failed_remux_keeps_the_source_and_removes_its_partial_file(tmp_path: Path) -> None:
|
||||
"""FFmpeg creates the output before it fails, and nothing else sweeps the temp directory."""
|
||||
source = encode(tmp_path / "track.mp4", "pcm_s16le", "-f", "mov") # no MP4 muxer takes PCM
|
||||
# FFmpeg 9.0 writes PCM to MP4 as ipcm, so PCM no longer fails the remux.
|
||||
source = encode(tmp_path / "track.mp4", "adpcm_ms", "-f", "mov") # no MP4 muxer takes ADPCM
|
||||
|
||||
track = Audio(id_="a1", url="https://example.test/a1.mp4", language="en", codec=None)
|
||||
track.path = source
|
||||
|
||||
@@ -0,0 +1,117 @@
|
||||
import textwrap
|
||||
|
||||
import pytest
|
||||
|
||||
from unshackle.core.constants import SPACED_AUDIO_CODECS
|
||||
from unshackle.core.manifests.hls import HLS
|
||||
from unshackle.core.tracks.audio import Audio
|
||||
|
||||
# The codecs one variant declares, in manifest order. One GROUP-ID holds renditions of
|
||||
# all three, which is why a group-wide codec cannot be right.
|
||||
DTS_X_GROUP = [
|
||||
("dtsx", Audio.Codec.DTSX),
|
||||
("ec-3", Audio.Codec.EC3),
|
||||
("mp4a", Audio.Codec.AAC),
|
||||
]
|
||||
|
||||
|
||||
def test_dtsc_maps_to_dts() -> None:
|
||||
assert Audio.Codec.from_mime("dtsc") == Audio.Codec.DTS
|
||||
|
||||
|
||||
@pytest.mark.parametrize("mime", ["dtsx", "dtsy"])
|
||||
def test_dts_uhd_sample_entries_map_to_dtsx(mime: str) -> None:
|
||||
"""DTS-UHD is a separate bitstream, so it must not share the DTS name."""
|
||||
assert Audio.Codec.from_mime(mime) == Audio.Codec.DTSX
|
||||
assert Audio.Codec.from_codecs(f"avc1.64001f,{mime},ec-3,mp4a.40.2") == Audio.Codec.DTSX
|
||||
assert Audio.Codec.DTSX.value == "DTS-X", "':' is not a legal path character on Windows"
|
||||
|
||||
|
||||
def test_unknown_mime_still_raises() -> None:
|
||||
with pytest.raises(ValueError):
|
||||
Audio.Codec.from_mime("dtsz")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("uri", "expected"),
|
||||
[
|
||||
# Rendition URIs that name the codec in one "_" delimited field.
|
||||
("r/composite_448k_dtsx_en_PRIMARY_de37f606-8b62-4116-bdec-0de711f2b7d0_x.m3u8", Audio.Codec.DTSX),
|
||||
("r/composite_256k_ec-3_cs_PRIMARY_b4556be9-7eb3-47a9-b103-d97263a99ec1_x.m3u8", Audio.Codec.EC3),
|
||||
("r/composite_128k_mp4a.40.2_en_NARRATION_3891a283-94c0-4b89-9dc8-a70fcec25661_x.m3u8", Audio.Codec.AAC),
|
||||
# A UUID can spell "ec-3", so a substring match would call this rendition EC3.
|
||||
("r/composite_128k_mp4a.40.2_en_PRIMARY_b4556be9-7eb3-47a9-b103-d97263a9ec-3_x.m3u8", Audio.Codec.AAC),
|
||||
# Packagers that name the codec as a directory, hyphen or not.
|
||||
("https://cdn/x/dtsx-448k/index.m3u8", Audio.Codec.DTSX),
|
||||
("https://cdn/x/dtsx/index.m3u8", Audio.Codec.DTSX),
|
||||
# Nothing names a codec, so the first codec the variant declares wins.
|
||||
("r/audio-en.m3u8", Audio.Codec.DTSX),
|
||||
],
|
||||
)
|
||||
def test_select_audio_codec_matches_delimited_codecs(uri: str, expected: Audio.Codec) -> None:
|
||||
assert HLS.select_audio_codec(DTS_X_GROUP, uri) == expected
|
||||
|
||||
|
||||
def test_select_audio_codec_will_not_read_one_codec_out_of_another() -> None:
|
||||
""" "ac-3" sits inside the "eac-3" that names an E-AC-3 rendition."""
|
||||
candidates = [("ec-3", Audio.Codec.EC3), ("ac-3", Audio.Codec.AC3)]
|
||||
assert HLS.select_audio_codec(candidates, "https://cdn/x/audio_ac-3_192k.m3u8") == Audio.Codec.AC3
|
||||
# No codec matches, so this falls back to the first declared rather than to AC3.
|
||||
assert HLS.select_audio_codec(candidates, "https://cdn/x/audio_eac-3_768k.m3u8") == Audio.Codec.EC3
|
||||
|
||||
|
||||
def test_select_audio_codec_ignores_the_query_string() -> None:
|
||||
assert HLS.select_audio_codec(DTS_X_GROUP, "r/composite_448k_dtsx_en.m3u8?fallback=mp4a") == Audio.Codec.DTSX
|
||||
|
||||
|
||||
def test_select_audio_codec_holds_a_single_declaration() -> None:
|
||||
"""One declared codec needs no URI, so a group of one is never guessed at."""
|
||||
assert HLS.select_audio_codec([("mp4a", Audio.Codec.AAC)], "r/anything_dtsx.m3u8") == Audio.Codec.AAC
|
||||
assert HLS.select_audio_codec([], "r/x.m3u8") is None
|
||||
assert HLS.select_audio_codec(None, None) is None
|
||||
|
||||
|
||||
def test_spaced_codec_constants_name_real_codecs() -> None:
|
||||
"""The constant holds bare strings, so a rename must not silently stop matching."""
|
||||
for value in SPACED_AUDIO_CODECS:
|
||||
assert value in {c.value for c in Audio.Codec}, f"{value} is not an Audio.Codec value"
|
||||
|
||||
|
||||
def test_a_short_ladder_rung_cannot_hide_a_group_codec() -> None:
|
||||
"""
|
||||
One GROUP-ID is reached from every variant that references it, and a low rung may
|
||||
declare fewer codecs than the group holds. The union has to survive the short rung.
|
||||
"""
|
||||
manifest = textwrap.dedent("""\
|
||||
#EXTM3U
|
||||
#EXT-X-MEDIA:TYPE=AUDIO,GROUP-ID="A",NAME="English",LANGUAGE="en",CHANNELS="10",URI="composite_448k_dtsx_en.m3u8"
|
||||
#EXT-X-MEDIA:TYPE=AUDIO,GROUP-ID="A",NAME="Czech",LANGUAGE="cs",CHANNELS="6",URI="composite_256k_ec-3_cs.m3u8"
|
||||
#EXT-X-STREAM-INF:BANDWIDTH=9000000,CODECS="hvc1.2.4.L150.90,dtsx,ec-3",AUDIO="A"
|
||||
v_high.m3u8
|
||||
#EXT-X-STREAM-INF:BANDWIDTH=400000,CODECS="avc1.64001f,ec-3",AUDIO="A"
|
||||
v_low.m3u8
|
||||
""")
|
||||
tracks = HLS.from_text(manifest, "https://example.test/master.m3u8").to_tracks("en")
|
||||
by_lang = {str(t.language): t.codec for t in tracks.audio}
|
||||
assert by_lang["en"] == Audio.Codec.DTSX, "the low rung dropped dtsx from the group"
|
||||
assert by_lang["cs"] == Audio.Codec.EC3
|
||||
|
||||
|
||||
def test_the_group_fallback_does_not_depend_on_variant_order() -> None:
|
||||
"""
|
||||
A rendition URI that names no codec falls back to a codec the group declares. Which
|
||||
one must not depend on where the packager put its variants: ascending by BANDWIDTH is
|
||||
as common as descending, and picking the lossy label makes the premium rendition
|
||||
unreachable through --acodec.
|
||||
"""
|
||||
variants = [
|
||||
'#EXT-X-STREAM-INF:BANDWIDTH=9000000,CODECS="hvc1.2.4.L150.90,dtsx,ec-3",AUDIO="A"\nv_high.m3u8',
|
||||
'#EXT-X-STREAM-INF:BANDWIDTH=400000,CODECS="avc1.64001f,ec-3",AUDIO="A"\nv_low.m3u8',
|
||||
]
|
||||
header = (
|
||||
'#EXTM3U\n#EXT-X-MEDIA:TYPE=AUDIO,GROUP-ID="A",NAME="English",LANGUAGE="en",CHANNELS="10",URI="audio_en.m3u8"\n'
|
||||
)
|
||||
for order in (variants, list(reversed(variants))):
|
||||
manifest = header + "\n".join(order) + "\n"
|
||||
tracks = HLS.from_text(manifest, "https://example.test/master.m3u8").to_tracks("en")
|
||||
assert [t.codec for t in tracks.audio] == [Audio.Codec.DTSX], f"variant order decided the codec: {order[0]!r}"
|
||||
Reference in New Issue
Block a user