From dc188b727633534d4b1ca00f564936697b7ffeaf Mon Sep 17 00:00:00 2001 From: imSp4rky Date: Mon, 2 Nov 2026 17:50:35 -0700 Subject: [PATCH] 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. --- tests/core/test_dash_period_match.py | 165 +++++++++++++++++++++++++++ unshackle/core/manifests/dash.py | 85 +++++++++----- 2 files changed, 221 insertions(+), 29 deletions(-) create mode 100644 tests/core/test_dash_period_match.py diff --git a/tests/core/test_dash_period_match.py b/tests/core/test_dash_period_match.py new file mode 100644 index 00000000..5a4436ab --- /dev/null +++ b/tests/core/test_dash_period_match.py @@ -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""" + + + + + + + + + + + + + + + + + + + + + + +""" + + +def video(rep_id: str, bandwidth: str, codecs: str = "avc1.640028") -> str: + return f'' + + +def audio(rep_id: str) -> str: + return f'' + + +def two_periods(first: str, second: str) -> bytes: + return f'{first}{second}'.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"{video('v', '5000000')}{audio('a')}", + f"{audio('a')}{video('v', '5000000')}", + ) + 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'{video("avc", "5000000")}' + f'{video("hevc", "3000000", "hvc1.2.4.L150")}', + f'{video("avc", "5000000")}', + ) + assert resolve(mpd, 1, "hevc") is None + + +def test_renumbered_sets_that_swap_order_keep_the_codec(): + mpd = two_periods( + f'{video("0", "8000000", "hvc1.2.4.L150")}' + f'{video("1", "8000000", "dvh1.08.06")}', + f'{video("0", "8000000", "dvh1.08.06")}' + f'{video("1", "8000000", "hvc1.2.4.L150")}', + ) + 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'' + ) + + mpd = two_periods( + f'{rep("x")}', + f'{rep("y")}', + ) + assert resolve(mpd, 0, "x") is None + + +def test_regional_tag_continues_into_the_same_language(): + mpd = two_periods( + f'{audio("a")}', + f'{audio("b")}', + ) + 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'{audio("dialog")}' + described = f'{audio("ad")}' + 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'{video("v1", "5000000")}{video("v2", "6000000")}', + f'{video("v1", "3000000")}{video("v2", "4500000")}', + ) + 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'{video("v1", "5000000")}', + f'{video("v1", "5e6")}', + ) + 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 diff --git a/unshackle/core/manifests/dash.py b/unshackle/core/manifests/dash.py index 21e982b0..b22ed183 100644 --- a/unshackle/core/manifests/dash.py +++ b/unshackle/core/manifests/dash.py @@ -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: