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: