mirror of
https://github.com/unshackle-dl/unshackle.git
synced 2026-11-14 04:55:07 +00:00
fix(manifests): match dash representations across periods by shape, not id
Some packagers renumber AdaptationSet and Representation ids in every Period, so audio took video segments after the first Period. A match never crosses codec, size or language.
This commit is contained in:
imSp4rky
@@ -0,0 +1,165 @@
|
||||
"""DASH.resolve_representation must continue a track into a later Period without crossing codec, size or language."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from tests.core.test_manifest_corpus import CORPUS, INDEX, KNOWN_FAILURES, parse
|
||||
from unshackle.core.manifests import DASH
|
||||
from unshackle.core.utils.xml import load_xml
|
||||
|
||||
RENUMBERED = b"""<MPD type="static" mediaPresentationDuration="PT20S">
|
||||
<Period id="0" duration="PT10S">
|
||||
<AdaptationSet id="0" contentType="text" lang="en">
|
||||
<Representation id="0" bandwidth="1000" codecs="wvtt" mimeType="application/mp4"/>
|
||||
</AdaptationSet>
|
||||
<AdaptationSet id="1" contentType="video">
|
||||
<Representation id="1" bandwidth="2000000" codecs="avc1" width="720" height="540"/>
|
||||
<Representation id="2" bandwidth="500000" codecs="avc1" width="640" height="480"/>
|
||||
</AdaptationSet>
|
||||
<AdaptationSet id="2" contentType="audio" lang="en">
|
||||
<Representation id="3" bandwidth="128000" codecs="mp4a.40.2" audioSamplingRate="48000"/>
|
||||
</AdaptationSet>
|
||||
</Period>
|
||||
<Period id="1" duration="PT10S">
|
||||
<AdaptationSet id="0" contentType="audio" lang="en">
|
||||
<Representation id="0" bandwidth="129000" codecs="mp4a.40.2" audioSamplingRate="48000"/>
|
||||
</AdaptationSet>
|
||||
<AdaptationSet id="1" contentType="video">
|
||||
<Representation id="1" bandwidth="480000" codecs="avc1" width="640" height="480"/>
|
||||
<Representation id="2" bandwidth="1900000" codecs="avc1" width="720" height="540"/>
|
||||
<Representation id="3" bandwidth="3000000" codecs="avc1" width="720" height="540"/>
|
||||
</AdaptationSet>
|
||||
</Period>
|
||||
</MPD>"""
|
||||
|
||||
|
||||
def video(rep_id: str, bandwidth: str, codecs: str = "avc1.640028") -> str:
|
||||
return f'<Representation id="{rep_id}" bandwidth="{bandwidth}" codecs="{codecs}" width="1920" height="1080" mimeType="video/mp4"/>'
|
||||
|
||||
|
||||
def audio(rep_id: str) -> str:
|
||||
return f'<Representation id="{rep_id}" bandwidth="128000" codecs="mp4a.40.2" audioSamplingRate="48000" mimeType="audio/mp4"/>'
|
||||
|
||||
|
||||
def two_periods(first: str, second: str) -> bytes:
|
||||
return f'<MPD><Period id="0">{first}</Period><Period id="1">{second}</Period></MPD>'.encode()
|
||||
|
||||
|
||||
def resolve(mpd: bytes, set_index: int, rep_id: str) -> tuple[str | None, str | None, str | None] | None:
|
||||
p0, p1 = load_xml(mpd).findall("Period")
|
||||
found = DASH.resolve_representation(p1, rep_id, p0.findall("AdaptationSet")[set_index])
|
||||
if found is None:
|
||||
return None
|
||||
adaptation_set, rep = found
|
||||
return adaptation_set.get("lang"), rep.get("id"), DASH.get_attr("codecs", adaptation_set, rep)
|
||||
|
||||
|
||||
def test_renumbered_audio_continues_into_audio_not_a_video_rep_with_its_id():
|
||||
assert resolve(RENUMBERED, 2, "3") == ("en", "0", "mp4a.40.2")
|
||||
|
||||
|
||||
def test_renumbered_video_continues_into_the_same_size():
|
||||
assert resolve(RENUMBERED, 1, "1") == (None, "2", "avc1")
|
||||
assert resolve(RENUMBERED, 1, "2") == (None, "1", "avc1")
|
||||
|
||||
|
||||
def test_track_missing_from_the_period_is_not_matched_to_another_type():
|
||||
assert resolve(RENUMBERED, 0, "0") is None
|
||||
|
||||
|
||||
def test_untyped_sets_that_swap_order_keep_audio_on_audio():
|
||||
mpd = two_periods(
|
||||
f"<AdaptationSet>{video('v', '5000000')}</AdaptationSet><AdaptationSet>{audio('a')}</AdaptationSet>",
|
||||
f"<AdaptationSet>{audio('a')}</AdaptationSet><AdaptationSet>{video('v', '5000000')}</AdaptationSet>",
|
||||
)
|
||||
assert resolve(mpd, 1, "a") == (None, "a", "mp4a.40.2")
|
||||
|
||||
|
||||
def test_codec_missing_from_the_period_is_skipped_not_swapped():
|
||||
mpd = two_periods(
|
||||
f'<AdaptationSet contentType="video">{video("avc", "5000000")}</AdaptationSet>'
|
||||
f'<AdaptationSet contentType="video">{video("hevc", "3000000", "hvc1.2.4.L150")}</AdaptationSet>',
|
||||
f'<AdaptationSet contentType="video">{video("avc", "5000000")}</AdaptationSet>',
|
||||
)
|
||||
assert resolve(mpd, 1, "hevc") is None
|
||||
|
||||
|
||||
def test_renumbered_sets_that_swap_order_keep_the_codec():
|
||||
mpd = two_periods(
|
||||
f'<AdaptationSet id="0" contentType="video">{video("0", "8000000", "hvc1.2.4.L150")}</AdaptationSet>'
|
||||
f'<AdaptationSet id="1" contentType="video">{video("1", "8000000", "dvh1.08.06")}</AdaptationSet>',
|
||||
f'<AdaptationSet id="0" contentType="video">{video("0", "8000000", "dvh1.08.06")}</AdaptationSet>'
|
||||
f'<AdaptationSet id="1" contentType="video">{video("1", "8000000", "hvc1.2.4.L150")}</AdaptationSet>',
|
||||
)
|
||||
assert resolve(mpd, 1, "1") == (None, "0", "dvh1.08.06")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("content", ["audio", "text"])
|
||||
def test_language_missing_from_the_period_is_skipped_not_swapped(content: str):
|
||||
def rep(rep_id: str) -> str:
|
||||
return (
|
||||
audio(rep_id) if content == "audio" else f'<Representation id="{rep_id}" bandwidth="1000" codecs="wvtt"/>'
|
||||
)
|
||||
|
||||
mpd = two_periods(
|
||||
f'<AdaptationSet id="1" contentType="{content}" lang="en">{rep("x")}</AdaptationSet>',
|
||||
f'<AdaptationSet id="2" contentType="{content}" lang="de">{rep("y")}</AdaptationSet>',
|
||||
)
|
||||
assert resolve(mpd, 0, "x") is None
|
||||
|
||||
|
||||
def test_regional_tag_continues_into_the_same_language():
|
||||
mpd = two_periods(
|
||||
f'<AdaptationSet id="1" contentType="audio" lang="en">{audio("a")}</AdaptationSet>',
|
||||
f'<AdaptationSet id="5" contentType="audio" lang="en-US">{audio("b")}</AdaptationSet>',
|
||||
)
|
||||
assert resolve(mpd, 0, "a") == ("en-US", "b", "mp4a.40.2")
|
||||
|
||||
|
||||
def test_same_identity_sets_that_swap_order_follow_the_rep_id():
|
||||
dialog = f'<AdaptationSet contentType="audio" lang="en">{audio("dialog")}</AdaptationSet>'
|
||||
described = f'<AdaptationSet contentType="audio" lang="en">{audio("ad")}</AdaptationSet>'
|
||||
assert resolve(two_periods(dialog + described, described + dialog), 0, "dialog") == ("en", "dialog", "mp4a.40.2")
|
||||
|
||||
|
||||
def test_stable_ladder_follows_bandwidth_rank_when_every_rung_drifts():
|
||||
mpd = two_periods(
|
||||
f'<AdaptationSet id="1" contentType="video">{video("v1", "5000000")}{video("v2", "6000000")}</AdaptationSet>',
|
||||
f'<AdaptationSet id="1" contentType="video">{video("v1", "3000000")}{video("v2", "4500000")}</AdaptationSet>',
|
||||
)
|
||||
assert resolve(mpd, 0, "v1") == (None, "v1", "avc1.640028")
|
||||
assert resolve(mpd, 0, "v2") == (None, "v2", "avc1.640028")
|
||||
|
||||
|
||||
def test_non_integer_bandwidth_does_not_raise():
|
||||
mpd = two_periods(
|
||||
f'<AdaptationSet id="1" contentType="video">{video("v1", "5000000")}</AdaptationSet>',
|
||||
f'<AdaptationSet id="1" contentType="video">{video("v1", "5e6")}</AdaptationSet>',
|
||||
)
|
||||
assert resolve(mpd, 0, "v1") == (None, "v1", "avc1.640028")
|
||||
|
||||
|
||||
def multi_period(entry: dict[str, Any]) -> bool:
|
||||
periods = load_xml((CORPUS / entry["file"]).read_bytes()).findall("Period")
|
||||
return sum(DASH.is_content_period(p, []) for p in periods) > 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"entry",
|
||||
[e for e in INDEX if e["kind"] == "dash" and e["file"] not in KNOWN_FAILURES and multi_period(e)],
|
||||
ids=lambda e: e["file"],
|
||||
)
|
||||
def test_corpus_matches_never_cross_codec_or_size(entry: dict[str, Any]):
|
||||
shape = ("codecs", "width", "height")
|
||||
for track in parse(entry):
|
||||
data = track.data["dash"]
|
||||
want = tuple(DASH.get_attr(a, data["adaptation_set"], data["representation"]) for a in shape)
|
||||
for period in data["manifest"].findall("Period"):
|
||||
if period is data["period"] or not DASH.is_content_period(period, []):
|
||||
continue
|
||||
found = DASH.resolve_representation(period, data["representation_id"], data["adaptation_set"])
|
||||
if found is not None:
|
||||
assert tuple(DASH.get_attr(a, *found) for a in shape) == want
|
||||
@@ -147,6 +147,25 @@ class RollingMerge:
|
||||
return self.cursor
|
||||
|
||||
|
||||
def rep_shape(adaptation_set: Element, rep: Element) -> tuple[Optional[str], ...]:
|
||||
"""The attributes a Representation must keep to continue a track into another Period."""
|
||||
return tuple(DASH.get_attr(a, adaptation_set, rep) for a in ("codecs", "width", "height", "audioSamplingRate"))
|
||||
|
||||
|
||||
def rep_bandwidth(rep: Element) -> float:
|
||||
try:
|
||||
return float(rep.get("bandwidth") or 0)
|
||||
except ValueError:
|
||||
return 0.0
|
||||
|
||||
|
||||
def kind_and_lang(adaptation_set: Element) -> tuple[Optional[str], str]:
|
||||
"""Content type and language subtag, without the region, of an AdaptationSet."""
|
||||
mime = DASH.get_attr("mimeType", adaptation_set, adaptation_set.find("Representation")) or ""
|
||||
kind = adaptation_set.get("contentType") or mime.split("/")[0] or None
|
||||
return kind, (adaptation_set.get("lang") or "").split("-")[0].lower()
|
||||
|
||||
|
||||
class DASH:
|
||||
def __init__(self, manifest: Element, url: str) -> None:
|
||||
if manifest is None:
|
||||
@@ -503,7 +522,7 @@ class DASH:
|
||||
|
||||
if matched_rep is None or matched_as is None:
|
||||
period_id = content_period.get("id", period_idx)
|
||||
log.warning(f"Representation '{rep_id}' not found in period '{period_id}', skipping")
|
||||
log.debug(f"Representation '{rep_id}' not found in period '{period_id}', skipping")
|
||||
continue
|
||||
|
||||
p_init, p_segments, p_timescale, p_durations, p_kid = DASH.get_period_segments(
|
||||
@@ -1316,40 +1335,48 @@ class DASH:
|
||||
period: Element, rep_id: Optional[str], adaptation_set: Element
|
||||
) -> Optional[tuple[Element, Element]]:
|
||||
"""
|
||||
Find the (AdaptationSet, Representation) in another Period matching rep_id.
|
||||
Find the (AdaptationSet, Representation) in another Period that continues rep_id.
|
||||
|
||||
Representation ids are only unique within an AdaptationSet, so the set matching
|
||||
`adaptation_set`'s identity wins over an earlier sibling that reuses the same id
|
||||
for a different track (for example, audio description vs dialog).
|
||||
Ids are not trusted alone, because some packagers renumber them per Period. A match never
|
||||
crosses codec, size or language; within the set, bandwidth rank decides.
|
||||
"""
|
||||
as_key = DASH.adaptation_set_key(adaptation_set)
|
||||
source = next((r for r in adaptation_set.findall("Representation") if r.get("id") == rep_id), None)
|
||||
if source is None:
|
||||
return None
|
||||
want = rep_shape(adaptation_set, source)
|
||||
|
||||
def candidates(within: Element) -> list[tuple[Element, Element]]:
|
||||
found = []
|
||||
for as_ in within.findall("AdaptationSet"):
|
||||
if DASH.is_trick_mode(as_) or DASH.adaptation_set_key(as_) != as_key:
|
||||
continue
|
||||
for rep in as_.findall("Representation"):
|
||||
if rep.get("id") == rep_id:
|
||||
found.append((as_, rep))
|
||||
break
|
||||
return found
|
||||
def same_shape(as_: Element) -> list[Element]:
|
||||
return sorted((r for r in as_.findall("Representation") if rep_shape(as_, r) == want), key=rep_bandwidth)
|
||||
|
||||
matches = candidates(period)
|
||||
if matches:
|
||||
parent = adaptation_set.getparent()
|
||||
siblings = [as_ for as_, _ in candidates(parent)] if parent is not None else []
|
||||
def sets(within: Optional[Element], key: Callable[[Element], Any]) -> list[Element]:
|
||||
found = within.findall("AdaptationSet") if within is not None else []
|
||||
target = key(adaptation_set)
|
||||
return [a for a in found if key(a) == target and not DASH.is_trick_mode(a) and same_shape(a)]
|
||||
|
||||
keys: tuple[Callable[[Element], Any], ...] = (
|
||||
DASH.adaptation_set_key,
|
||||
lambda as_: DASH.adaptation_set_key(as_)[:-1],
|
||||
kind_and_lang,
|
||||
)
|
||||
for key in keys:
|
||||
if matches := sets(period, key):
|
||||
break
|
||||
else:
|
||||
return None
|
||||
with_id = [a for a in matches if any(r.get("id") == rep_id for r in same_shape(a))]
|
||||
if len(with_id) == 1:
|
||||
matched_as = with_id[0]
|
||||
else:
|
||||
siblings = sets(adaptation_set.getparent(), key)
|
||||
ordinal = siblings.index(adaptation_set) if adaptation_set in siblings else 0
|
||||
return matches[min(ordinal, len(matches) - 1)]
|
||||
matched_as = matches[min(ordinal, len(matches) - 1)]
|
||||
|
||||
for as_ in period.findall("AdaptationSet"):
|
||||
if DASH.is_trick_mode(as_):
|
||||
continue
|
||||
for rep in as_.findall("Representation"):
|
||||
if rep.get("id") == rep_id:
|
||||
return as_, rep
|
||||
|
||||
return None
|
||||
group, source_group = same_shape(matched_as), same_shape(adaptation_set)
|
||||
if len(group) == len(source_group):
|
||||
return matched_as, group[source_group.index(source)]
|
||||
return matched_as, min(
|
||||
group, key=lambda r: (abs(rep_bandwidth(r) - rep_bandwidth(source)), r.get("id") != rep_id)
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def is_descriptive(adaptation_set: Element) -> bool:
|
||||
|
||||
Reference in New Issue
Block a user