fix(mux): address embedded audio by the file's real mkvmerge track IDs

A TS file can hold its audio first, which put the audio name on the video.

tests/conftest.py

@@ -4,10 +4,12 @@ from __future__ import annotations
 
 import socket
 import socketserver
-from typing import Any
+from pathlib import Path
+from typing import Any, Callable, Optional
 
 import pytest
 
+from unshackle.core import binaries
 from unshackle.core.config import config
 from unshackle.core.update_checker import UpdateChecker
 
@@ -45,3 +47,27 @@ def no_update_check(monkeypatch: pytest.MonkeyPatch) -> None:
 def default_muxing(monkeypatch: pytest.MonkeyPatch) -> None:
     """Mux with the default options, whatever the muxing block of the local unshackle.yaml sets."""
     monkeypatch.setattr(config, "muxing", {})
+
+
+@pytest.fixture
+def mux_args(monkeypatch: pytest.MonkeyPatch) -> Callable[..., list[str]]:
+    """Run Tracks.mux up to mkvmerge and return its arguments; ``ids`` is what the track ID probe reports."""
+
+    class Stop(Exception):
+        pass
+
+    def run(tracks: Any, ids: Optional[dict[str, int]] = None, **kwargs: Any) -> list[str]:
+        captured: list[list[str]] = []
+
+        def fake_popen(args: list[str], **_: object) -> None:
+            captured.append(args)
+            raise Stop
+
+        monkeypatch.setattr("unshackle.core.tracks.tracks.subprocess.Popen", fake_popen)
+        monkeypatch.setattr("unshackle.core.tracks.tracks.file_track_ids", lambda path: ids or {"video": 0})
+        monkeypatch.setattr(binaries, "MKVToolNix", Path("mkvmerge"))
+        with pytest.raises(Stop):
+            tracks.mux("Title", delete=False, **kwargs)
+        return captured[0]
+
+    return run

tests/test_sample.py

@@ -8,7 +8,7 @@ import struct
 import threading
 from pathlib import Path
 from types import SimpleNamespace
-from typing import Optional
+from typing import Callable, Optional
 from unittest.mock import MagicMock
 
 import click
@@ -16,7 +16,6 @@ import pytest
 from requests import Session
 
 from unshackle.commands.dl import sample_name
-from unshackle.core import binaries
 from unshackle.core.api.handlers import validate_download_parameters
 from unshackle.core.config import config
 from unshackle.core.manifests.dash import DASH
@@ -231,19 +230,7 @@ def test_subtitle_trim_leaves_an_unreadable_file_whole(tmp_path: Path) -> None:
     assert subtitle.path.read_bytes() == b"\x00\x01not a subtitle"
 
 
-def mux_command(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> list[str]:
-    class Stop(Exception):
-        pass
-
-    captured: list[list[str]] = []
-
-    def fake_popen(args: list[str], **kwargs: object) -> None:
-        captured.append(args)
-        raise Stop
-
-    monkeypatch.setattr("unshackle.core.tracks.tracks.subprocess.Popen", fake_popen)
-    monkeypatch.setattr(binaries, "MKVToolNix", Path("mkvmerge"))
-
+def mux_command(tmp_path: Path, mux_args: Callable[..., list[str]]) -> list[str]:
     video, audio = make_video(), make_audio()
     trimmed = Subtitle(id_="s1", url="https://x/s1", codec=Subtitle.Codec.SubRip, language="en")
     whole = Subtitle(id_="s2", url="https://x/s2", codec=Subtitle.Codec.SubRip, language="fr")
@@ -254,14 +241,14 @@ def mux_command(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> list[str]:
     tracks = Tracks([video, audio, trimmed, whole])
     tracks.chapters.add(Chapter("00:02:05.000", "Scene"))
 
-    with pytest.raises(Stop):
-        tracks.mux("Title", delete=False, progress=lambda **k: None, output_path=tmp_path / "out.mkv")
-    return captured[0]
+    return mux_args(tracks, progress=lambda **k: None, output_path=tmp_path / "out.mkv")
 
 
-def test_the_mux_shifts_each_track_and_cuts_to_the_window(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+def test_the_mux_shifts_each_track_and_cuts_to_the_window(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, mux_args: Callable[..., list[str]]
+) -> None:
     monkeypatch.setattr(config, "sample", (121, 181))
-    command = mux_command(tmp_path, monkeypatch)
+    command = mux_command(tmp_path, mux_args)
 
     def sync_before(name: str) -> str:
         file_at = command.index(str(tmp_path / name))
@@ -273,17 +260,19 @@ def test_the_mux_shifts_each_track_and_cuts_to_the_window(tmp_path: Path, monkey
         return command[sync_at + 1]
 
     assert command[command.index("--split") + 1] == "parts:1.500s-61.500s"
-    assert sync_before("v.mp4") == "0:500"
+    assert sync_before("v.mp4") == "-1:500"
     assert "--sync" not in command[: command.index(str(tmp_path / "a.mp4"))][-12:], "the earliest track is not moved"
-    assert sync_before("s1.srt") == "0:1500"
-    assert sync_before("s2.srt") == "0:-119500"
+    assert sync_before("s1.srt") == "-1:1500"
+    assert sync_before("s2.srt") == "-1:-119500"
     assert command[command.index("--chapter-sync") + 1] == "-119500"
     assert command.index("--chapter-sync") < command.index("--chapters")
 
 
-def test_a_full_download_adds_no_sample_options(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+def test_a_full_download_adds_no_sample_options(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, mux_args: Callable[..., list[str]]
+) -> None:
     monkeypatch.setattr(config, "sample", None)
-    command = mux_command(tmp_path, monkeypatch)
+    command = mux_command(tmp_path, mux_args)
     assert not {"--split", "--sync", "--chapter-sync"} & set(command)
 
 

tests/tracks/test_mux_embedded_audio_metadata.py

@@ -1,7 +1,11 @@
+import json
+from types import SimpleNamespace
+
 import pytest
 
 from unshackle.core import binaries
 from unshackle.core.tracks import Tracks, Video
+from unshackle.core.tracks.tracks import file_track_ids
 
 
 def make_video(track_id: str, **data) -> Video:
@@ -19,6 +23,12 @@ def make_video(track_id: str, **data) -> Video:
     return video
 
 
+def downloaded(video: Video, tmp_path) -> Video:
+    video.path = tmp_path / "v.ts"
+    video.path.write_bytes(b"\x00" * 16)
+    return video
+
+
 def mux_until_mkvmerge_check(tracks: Tracks) -> None:
     # the embedded-audio metadata pass runs before the mkvmerge lookup, so a missing
     # binary stops the mux right after the state under test has been written
@@ -35,7 +45,7 @@ def test_names_undeclared_video(monkeypatch):
 
     assert video.needs_repack
     assert video.data["audio_language"] == "zh"
-    assert video.data["audio_language_name"] == "Chinese"
+    assert video.embedded_audio_name() == "Chinese"
 
 
 def test_service_declaration_is_preserved(monkeypatch):
@@ -50,3 +60,36 @@ def test_service_declaration_is_preserved(monkeypatch):
     assert declared.data["audio_language_name"] == "Thai [Original]"
     assert not declared.needs_repack
     assert undeclared.data["audio_language"] == "zh"
+
+
+def test_embedded_audio_without_a_name_gets_the_language_name(tmp_path, mux_args):
+    args = mux_args(Tracks(downloaded(make_video("v1", audio_language="pt-BR"), tmp_path)), {"video": 0, "audio": 1})
+    assert args[args.index("--track-name") + 1] == "1:Portuguese"
+
+
+def test_embedded_audio_options_use_the_file_track_ids(tmp_path, mux_args):
+    video = make_video("v1", audio_language="ko", audio_language_name="Korean")
+    args = mux_args(Tracks(downloaded(video, tmp_path)), {"audio": 0, "video": 1})
+    assert args[args.index("--track-name") + 1] == "0:Korean"
+    original = f"{video.is_original_lang}"
+    assert [args[i + 1] for i, arg in enumerate(args) if arg == "--original-flag"] == [f"1:{original}", f"0:{original}"]
+    assert "0:ko" in args and "1:zh" in args
+
+
+def test_embedded_audio_options_are_left_out_when_the_file_has_no_audio(tmp_path, mux_args):
+    args = mux_args(Tracks(downloaded(make_video("v1", audio_language="ko"), tmp_path)), {"video": 0})
+    assert "--track-name" not in args
+
+
+def test_repack_and_mux_share_the_embedded_audio_name():
+    assert make_video("v1", audio_language="zh").embedded_audio_name() == "Chinese"
+    assert make_video("v1", audio_language="zh", audio_language_name="Mandarin").embedded_audio_name() == "Mandarin"
+
+
+def test_file_track_ids_reads_the_first_id_of_each_type(monkeypatch, tmp_path):
+    tracks = [{"id": 0, "type": "audio"}, {"id": 1, "type": "video"}, {"id": 2, "type": "audio"}]
+    reply = SimpleNamespace(stdout=json.dumps({"tracks": tracks}))
+    monkeypatch.setattr("unshackle.core.tracks.tracks.subprocess.run", lambda *a, **k: reply)
+    assert file_track_ids(tmp_path / "v.ts") == {"audio": 0, "video": 1}
+    monkeypatch.setattr("unshackle.core.tracks.tracks.subprocess.run", lambda *a, **k: SimpleNamespace(stdout=""))
+    assert file_track_ids(tmp_path / "v.ts") == {"video": 0, "audio": 1}

tests/tracks/test_track_language_names.py

@@ -1,10 +1,9 @@
 from pathlib import Path
-from typing import Optional
+from typing import Callable, Optional
 
 import pytest
 from langcodes import Language
 
-from unshackle.core import binaries
 from unshackle.core.config import config
 from unshackle.core.tracks import Audio, Subtitle, Tracks, Video
 from unshackle.core.utilities import language_name, multi_script_languages, with_script
@@ -130,18 +129,7 @@ def test_a_service_name_wins() -> None:
     assert sub("zh", name="简体中文").get_track_name() == "简体中文"
 
 
-def test_mkvmerge_gets_the_subtitle_script_and_name(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
-    class Stop(Exception):
-        pass
-
-    captured: list[list[str]] = []
-
-    def fake_popen(args: list[str], **kwargs: object) -> None:
-        captured.append(args)
-        raise Stop
-
-    monkeypatch.setattr("unshackle.core.tracks.tracks.subprocess.Popen", fake_popen)
-    monkeypatch.setattr(binaries, "MKVToolNix", Path("mkvmerge"))
+def test_mkvmerge_gets_the_subtitle_script_and_name(tmp_path: Path, mux_args: Callable[..., list[str]]) -> None:
     video = Video(id_="v", url="https://x/v", codec=Video.Codec.AVC, language="zh-TW", width=1920, height=1080)
     audio = Audio(id_="a", url="https://x/a", codec=Audio.Codec.AAC, language="zh-TW", bitrate=1)
     subtitle = sub("zh-TW")
@@ -149,12 +137,11 @@ def test_mkvmerge_gets_the_subtitle_script_and_name(tmp_path: Path, monkeypatch:
         track.path = tmp_path / file
         track.path.write_bytes(b"\x00" * 16)
 
-    with pytest.raises(Stop):
-        Tracks([video, audio, subtitle]).mux("T", delete=False, output_path=tmp_path / "o.mkv")
+    args = mux_args(Tracks([video, audio, subtitle]), output_path=tmp_path / "o.mkv")
 
     def last_option(file: str, flag: str) -> str:
         """Return the value of the last flag given before the file, which is the one mkvmerge applies to it."""
-        before = captured[0][: captured[0].index(str(tmp_path / file))]
+        before = args[: args.index(str(tmp_path / file))]
         return before[len(before) - before[::-1].index(flag)]
 
     assert last_option("s.vtt", "--language") == "0:zh-Hant-TW"

unshackle/core/tracks/track.py

@@ -1012,6 +1012,11 @@ class Track:
         """The language tag written to the output file."""
         return self.language
 
+    def embedded_audio_name(self) -> str:
+        """Name the audio a service keeps inside this track's file, by its language when the service set none."""
+        lang = self.data["audio_language"]
+        return self.data.get("audio_language_name") or language_name(Language.get(lang)) or lang
+
     def get_track_name(self) -> Optional[str]:
         """Get the Track Name, or the name of its language when the service set none."""
         return self.name if self.name is not None else language_name(self.mux_language)
@@ -1276,7 +1281,7 @@ class Track:
 
             if hasattr(self, "data") and self.data.get("audio_language"):
                 audio_lang = self.data["audio_language"]
-                audio_name = self.data.get("audio_language_name", audio_lang)
+                audio_name = self.embedded_audio_name()
                 args.extend(
                     [
                         "-metadata:s:a:0",

unshackle/core/tracks/tracks.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 import logging
 import os
 import re
@@ -28,7 +29,6 @@ from unshackle.core.tracks.track import Track, has_dts_uhd_sample_entry, strip_d
 from unshackle.core.tracks.video import Video
 from unshackle.core.utilities import (
     is_close_match,
-    language_name,
     log_event,
     matching_languages,
     original_tag,
@@ -47,6 +47,21 @@ def at_resolution(video: Video, res: int) -> bool:
     return video.height == res or bool(video.width and int(video.width * 9 / 16) == res)
 
 
+def file_track_ids(path: Path) -> dict[str, int]:
+    """Return the mkvmerge ID of the first track of each type in a file. A TS file can hold its audio first."""
+    res = subprocess.run(
+        [str(binaries.MKVToolNix), "-J", str(path)], capture_output=True, text=True, encoding="utf-8", errors="replace"
+    )
+    ids: dict[str, int] = {}
+    for track in json.loads(res.stdout or "{}").get("tracks") or []:
+        ids.setdefault(track["type"], track["id"])
+    if not ids:
+        logging.getLogger("Tracks").debug(
+            f"mkvmerge found no tracks in {path.name}, so it is addressed as video 0, audio 1"
+        )
+    return ids or {"video": 0, "audio": 1}
+
+
 class Tracks:
     """
     Video, Audio, Subtitle, Chapter, and Attachment Track Store.
@@ -766,14 +781,11 @@ class Tracks:
             video_track = video_track or self.videos[0]
             if video_track.language.is_valid():
                 lang_code = str(video_track.language)
-                lang_name = language_name(video_track.language) or lang_code
-
                 for video in self.videos:
                     if video.data.get("audio_language"):
                         continue
                     video.needs_repack = True
                     video.data["audio_language"] = lang_code
-                    video.data["audio_language_name"] = lang_name
 
         if any(at.path and at.path.exists() and has_dts_uhd_sample_entry(at.path) for at in self.audio):
             return self._mux_mp4(
@@ -796,7 +808,7 @@ class Tracks:
 
         def sample_sync(start: float) -> list[str]:
             shift = round((start - sample_base) * 1000)
-            return ["--sync", f"0:{shift}"] if config.sample and shift else []
+            return ["--sync", f"-1:{shift}"] if config.sample and shift else []
 
         if config.sample:
             start, end = (seconds - sample_base for seconds in config.sample)
@@ -831,15 +843,17 @@ class Tracks:
             else:
                 is_default = i == 0
 
+            ids = file_track_ids(vt.path)
+            v = ids.get("video", 0)
             video_args = [
                 "--language",
-                f"0:{vt.mux_language}",
+                f"{v}:{vt.mux_language}",
                 "--default-track",
-                f"0:{is_default}",
+                f"{v}:{is_default}",
                 "--original-flag",
-                f"0:{vt.is_original_lang}",
+                f"{v}:{vt.is_original_lang}",
                 "--compression",
-                "0:none",
+                f"{v}:none",
                 *sample_sync(vt.sample_start),
             ]
 
@@ -848,9 +862,9 @@ class Tracks:
                 video_args.extend(
                     [
                         "--default-duration",
-                        f"0:{vt.fps}fps" if isinstance(vt.fps, str) else f"0:{vt.fps:.3f}fps",
+                        f"{v}:{vt.fps}fps" if isinstance(vt.fps, str) else f"{v}:{vt.fps:.3f}fps",
                         "--fix-bitstream-timing-information",
-                        "0:1",
+                        f"{v}:1",
                     ]
                 )
 
@@ -858,19 +872,20 @@ class Tracks:
                 video_args.extend(
                     [
                         "--color-transfer-characteristics",
-                        "0:18",  # ARIB STD-B67 (HLG)
+                        f"{v}:18",  # ARIB STD-B67 (HLG)
                     ]
                 )
 
-            if hasattr(vt, "data") and vt.data.get("audio_language"):
-                audio_lang = vt.data["audio_language"]
-                audio_name = vt.data.get("audio_language_name", audio_lang)
+            a = ids.get("audio")
+            if vt.data.get("audio_language") and a is not None:
                 video_args.extend(
                     [
                         "--language",
-                        f"1:{audio_lang}",
+                        f"{a}:{vt.data['audio_language']}",
                         "--track-name",
-                        f"1:{audio_name}",
+                        f"{a}:{vt.embedded_audio_name()}",
+                        "--original-flag",
+                        f"{a}:{vt.is_original_lang}",
                     ]
                 )