fix(manifests): decode audio channels by scheme in dash, hls and ism

DASH reads the AudioChannelConfiguration scheme (Dolby mask, CICP index incl. immersive layouts, channel count); HLS and ISM counts map 6/7/8 to 5.1/6.1/7.1.

tests/core/test_dash_audio_channels.py

@@ -0,0 +1,70 @@
+import xml.etree.ElementTree as ET
+from typing import Optional
+
+import pytest
+
+from unshackle.core.manifests.dash import DASH
+
+COUNT = "urn:mpeg:dash:23003:3:audio_channel_configuration:2011"
+CICP = "urn:mpeg:mpegB:cicp:ChannelConfiguration"
+DOLBY = "tag:dolby.com,2014:dash:audio_channel_configuration:2011"
+DOLBY_LEGACY = "urn:dolby:dash:audio_channel_configuration:2011"
+
+
+def config(*descriptors: tuple[str, str], tag: str = "AdaptationSet") -> ET.Element:
+    children = "".join(f'<AudioChannelConfiguration schemeIdUri="{s}" value="{v}"/>' for s, v in descriptors)
+    return ET.fromstring(f"<{tag}>{children}</{tag}>")
+
+
+@pytest.mark.parametrize(
+    ("scheme", "value", "expected"),
+    [
+        (COUNT, "2", 2.0),
+        (COUNT, "6", 5.1),
+        (COUNT, "7", 6.1),
+        (COUNT, "8", 7.1),
+        ("urn:dts:dash:audio_channel_configuration:2012", "6", 5.1),
+        (CICP, "2", 2.0),
+        (CICP, "6", 5.1),
+        (CICP, "7", 7.1),
+        (CICP, "8", 7.1),
+        (CICP, "12", 7.1),
+        (CICP, "16", "5.1.4"),
+        (CICP, "17", "5.1.6"),
+        (CICP, "18", "7.1.6"),
+        (CICP, "19", "7.1.4"),
+        (CICP, "20", "9.1.4"),
+        (DOLBY, "F801", 5.1),
+        (DOLBY_LEGACY, "FA01", 7.1),
+        # an all-digit mask is still a mask under a Dolby scheme
+        (DOLBY, "4000", 1.0),
+    ],
+)
+def test_a_channel_value_is_decoded_by_its_scheme(scheme: str, value: str, expected: object) -> None:
+    assert DASH.get_audio_channels(config((scheme, value)), None) == expected
+
+
+@pytest.mark.parametrize(
+    ("scheme", "value"),
+    [
+        ("tag:dolby.com,2015:dash:audio_channel_configuration:2015", "000047"),
+        ("urn:mpeg:dash:outputChannelPositionList:2012", "L R C LFE"),
+        (CICP, "13"),
+        (COUNT, "x"),
+        (DOLBY, "zz"),
+    ],
+)
+def test_an_unknown_scheme_or_value_gives_no_channels(scheme: str, value: str) -> None:
+    assert DASH.get_audio_channels(config((scheme, value)), None) is None
+
+
+def test_the_first_decodable_descriptor_wins() -> None:
+    element = config(("urn:example:unknown", "9"), (DOLBY, "F801"), (CICP, "2"))
+    assert DASH.get_audio_channels(element, None) == 5.1
+
+
+@pytest.mark.parametrize(
+    ("representation", "expected"), [(config((COUNT, "6"), tag="Representation"), 5.1), (None, 2.0)]
+)
+def test_the_representation_overrides_the_adaptation_set(representation: Optional[ET.Element], expected: float) -> None:
+    assert DASH.get_audio_channels(config((COUNT, "2")), representation) == expected

tests/tracks/test_audio_channel_layouts.py

@@ -30,6 +30,15 @@ def test_a_dolby_channel_mask_counts_its_channels(value: str, expected: object)
     assert Audio.parse_channels(value) == expected
 
 
+@pytest.mark.parametrize(
+    ("count", "expected"),
+    [("2", 2.0), ("6", 5.1), ("7", 6.1), ("8", 7.1), (6, 5.1), ("1", 1.0), (None, None), ("2ch", "2ch")],
+)
+def test_a_bare_channel_count_takes_its_usual_layout(count: object, expected: object) -> None:
+    """HLS CHANNELS and Smooth Streaming Channels give a count, which names no LFE."""
+    assert Audio.channels_from_count(count) == expected
+
+
 @pytest.mark.parametrize("value", ["stereo", "5.1.4.2", "", "..", "5.x.4"])
 def test_a_value_that_is_no_layout_still_raises(value: str) -> None:
     with pytest.raises(NotImplementedError):

unshackle/core/manifests/dash.py

@@ -46,6 +46,36 @@ from unshackle.core.utils.bitrate import ranged_get
 from unshackle.core.utils.redact import safe_display_url
 from unshackle.core.utils.xml import load_xml
 
+DOLBY_CHANNEL_SCHEMES = (
+    "tag:dolby.com,2014:dash:audio_channel_configuration:2011",
+    "urn:dolby:dash:audio_channel_configuration:2011",
+)
+COUNT_CHANNEL_SCHEMES = (
+    "urn:mpeg:dash:23003:3:audio_channel_configuration:2011",
+    "urn:dts:dash:audio_channel_configuration:2012",
+)
+CICP_CHANNEL_SCHEME = "urn:mpeg:mpegB:cicp:ChannelConfiguration"
+CICP_CHANNEL_LAYOUTS: dict[int, Union[float, str]] = {
+    1: 1.0,
+    2: 2.0,
+    3: 3.0,
+    4: 4.0,
+    5: 5.0,
+    6: 5.1,
+    7: 7.1,
+    8: 7.1,
+    9: 3.0,
+    10: 4.0,
+    11: 6.1,
+    12: 7.1,
+    14: "5.1.2",
+    16: "5.1.4",
+    17: "5.1.6",
+    18: "7.1.6",
+    19: "7.1.4",
+    20: "9.1.4",
+}
+
 
 class MissingLanguageError(ValueError):
     """A manifest track has no language, and the fallback language is missing or invalid."""
@@ -282,13 +312,7 @@ class DASH:
                         track_codec = Audio.Codec.from_codecs(codecs or "")
                         track_args = dict(
                             bitrate=get("bandwidth") or None,
-                            channels=next(
-                                iter(
-                                    rep.xpath("AudioChannelConfiguration/@value")
-                                    or adaptation_set.xpath("AudioChannelConfiguration/@value")
-                                ),
-                                None,
-                            ),
+                            channels=self.get_audio_channels(adaptation_set, rep),
                             joc=self.get_ddp_complexity_index(adaptation_set, rep),
                             descriptive=self.is_descriptive(adaptation_set),
                         )
@@ -1365,6 +1389,27 @@ class DASH:
             for x in adaptation_set.findall("Role")
         )
 
+    @staticmethod
+    def get_audio_channels(adaptation_set: Element, representation: Optional[Element]) -> Union[float, str, None]:
+        """
+        Channel layout from the first AudioChannelConfiguration whose scheme is known, decoded by that scheme.
+
+        The 23003:3 value is a 23003-3 index by the DASH-IF registry, but packagers write the number of channels
+        and players read it so; both readings agree up to 6 channels.
+        """
+        for descriptor in DASH.find_elements("AudioChannelConfiguration", adaptation_set, representation):
+            scheme, value = descriptor.get("schemeIdUri"), (descriptor.get("value") or "").strip()
+            try:
+                if scheme in DOLBY_CHANNEL_SCHEMES:
+                    return Audio.dolby_channel_layout(int(value, 16))
+                if scheme == CICP_CHANNEL_SCHEME and (layout := CICP_CHANNEL_LAYOUTS.get(int(value))):
+                    return layout
+                if scheme in COUNT_CHANNEL_SCHEMES and value.isdigit():
+                    return Audio.channels_from_count(value)
+            except ValueError:
+                continue
+        return None
+
     @staticmethod
     def get_ddp_complexity_index(adaptation_set: Element, representation: Optional[Element]) -> Optional[int]:
         """Get the DD+ Complexity Index (if any) from the AdaptationSet or Representation."""

unshackle/core/manifests/hls.py

@@ -295,7 +295,7 @@ class HLS:
                         joc = int(count)
                         media.channels = "5.1"
                     else:
-                        media.channels = count
+                        media.channels = Audio.channels_from_count(count)
             else:
                 track_type = Subtitle
                 codec = Subtitle.Codec.WebVTT  # assuming WebVTT, codec info isn't shown

unshackle/core/manifests/ism.py

@@ -354,7 +354,7 @@ class ISM:
                             language=track_lang or language,
                             is_original_lang=bool(language and track_lang and str(track_lang) == str(language)),
                             bitrate=ql.get("Bitrate"),
-                            channels=ql.get("Channels"),
+                            channels=Audio.channels_from_count(ql.get("Channels")),
                             extra={"atmos": True} if (ql.get("HasAtmos") or "").lower() == "true" else None,
                             descriptor=Track.Descriptor.ISM,
                             drm=drm,

unshackle/core/tracks/audio.py

@@ -29,6 +29,8 @@ DOLBY_CHANNEL_BITS = (
     (1, "lfe"),  # LFE
 )
 
+COUNT_CHANNEL_LAYOUTS: dict[int, float] = {6: 5.1, 7: 6.1, 8: 7.1}
+
 
 class Audio(Track):
     class Codec(str, Enum):
@@ -219,16 +221,9 @@ class Audio(Track):
         It does not convert the value to assume a sub speaker channel layout, e.g. 5.1->6.0.
         """
         if isinstance(channels, str):
-            # TODO: decode a DASH AudioChannelConfiguration by its schemeIdUri (CICP, Dolby), not by its shape
+            # a manifest parser decodes a Dolby mask by its scheme; this shape guess serves a service passing one raw
             if re.fullmatch(r"[0-9A-Fa-f]{4}", channels) and not channels.isdigit():
-                mask = int(channels, 16)
-                counts = {"main": 0, "lfe": 0, "height": 0}
-                for bit, (count, kind) in enumerate(DOLBY_CHANNEL_BITS):
-                    if mask & (0x8000 >> bit):
-                        counts[kind] += count
-                if counts["height"]:
-                    return f"{counts['main']}.{counts['lfe']}.{counts['height']}"
-                return float(f"{counts['main']}.{counts['lfe']}")
+                return Audio.dolby_channel_layout(int(channels, 16))
             layout = channels.replace("ch", "")
             if layout.count(".") == 2 and all(part.isdigit() for part in layout.split(".")):
                 return layout
@@ -238,6 +233,29 @@ class Audio(Track):
 
         return float(channels)
 
+    @staticmethod
+    def dolby_channel_layout(mask: int) -> Union[float, str]:
+        """Channel layout of a Dolby DASH channel configuration mask, e.g. 0xF801 -> 5.1."""
+        counts = {"main": 0, "lfe": 0, "height": 0}
+        for bit, (count, kind) in enumerate(DOLBY_CHANNEL_BITS):
+            if mask & (0x8000 >> bit):
+                counts[kind] += count
+        if counts["height"]:
+            return f"{counts['main']}.{counts['lfe']}.{counts['height']}"
+        return float(f"{counts['main']}.{counts['lfe']}")
+
+    @staticmethod
+    def channels_from_count(count: Union[str, int, None]) -> Union[float, str, None]:
+        """
+        Channel layout for a bare number of channels, as HLS CHANNELS, Smooth Streaming and most DASH schemes give it.
+
+        A count does not say which channel is the LFE, so 6, 7 and 8 take the 5.1, 6.1 and 7.1 layouts those streams
+        carry in practice. Any other value goes back as-is, for parse_channels to read.
+        """
+        if count is None or (isinstance(count, str) and not count.isdigit()):
+            return count
+        return COUNT_CHANNEL_LAYOUTS.get(int(count), float(count))
+
     @staticmethod
     def channel_total(channels: Union[str, int, float]) -> float:
         """

unshackle/services/EXAMPLE/__init__.py

@@ -388,9 +388,6 @@ class EXAMPLE(Service):
         # Drop "clear"/unencrypted decoy renditions the API sometimes returns.
         tracks.audio = [a for a in tracks.audio if "clear" not in (a.data["dash"]["representation"].get("id") or "")]
         for audio in tracks.audio:
-            # Normalize odd channel counts (6.0 -> 5.1) for correct filename tokens.
-            if audio.channels == 6.0:
-                audio.channels = 5.1
             # Mark descriptive / audio-description tracks from the AdaptationSet label.
             label = audio.data["dash"]["adaptation_set"].get("label") or ""
             if "Audio Description" in label or "description" in label.lower():