From 3522de1e9c572f1b27cc8bbfce5f5894a96edf8b Mon Sep 17 00:00:00 2001 From: imSp4rky Date: Sun, 15 Nov 2026 12:43:15 -0700 Subject: [PATCH] feat(mux): keep Audio Vivid beside the output and add wide_output listings Audio Vivid skips the mux and its .ts is saved beside the output; a user's avs3Decoder adds a PCM copy. wide_output prints dl listings at the terminal width; cfg saves lowercase true/false as booleans. --- docs/guide/output-and-naming.md | 13 +++ docs/reference/configuration/index.md | 2 +- docs/reference/configuration/misc.md | 1 + tests/commands/test_cfg.py | 8 ++ tests/core/test_audio_vivid.py | 113 ++++++++++++++++++++++++++ tests/core/test_print_listing.py | 31 +++++++ unshackle/commands/cfg.py | 2 +- unshackle/commands/dl.py | 69 +++++++++++----- unshackle/core/config.py | 1 + unshackle/core/console.py | 18 +++- unshackle/core/tracks/audio.py | 2 +- unshackle/core/utils/audio_vivid.py | 53 ++++++++++++ unshackle/core/utils/hdr10plus.py | 33 +++++--- 13 files changed, 310 insertions(+), 36 deletions(-) create mode 100644 tests/core/test_audio_vivid.py create mode 100644 tests/core/test_print_listing.py create mode 100644 unshackle/core/utils/audio_vivid.py diff --git a/docs/guide/output-and-naming.md b/docs/guide/output-and-naming.md index 51cd48ca..3494d9d5 100644 --- a/docs/guide/output-and-naming.md +++ b/docs/guide/output-and-naming.md @@ -450,6 +450,19 @@ An MP4 title loses: - SubStation Alpha (`.ass`) subtitles - container tags (`mkvpropedit`) +### Audio Vivid output + +No muxer carries Audio Vivid (AVS3-P3): Matroska has no CodecID for it, and FFmpeg and MP4Box cannot +write it to MP4. unshackle leaves an Audio Vivid track out of the mux and saves its original MPEG-TS file +beside the output, named like a `--no-mux` audio file (`..AV3A.ts`). + +With an Audio Vivid decoder named `avs3Decoder` in a `binaries` folder or on `PATH`, unshackle also +decodes the track to PCM and muxes that into the output. unshackle does not ship the decoder: the UWA +reference decoder's licence limits its use. A service that delivers Audio Vivid can declare the decoder so +that `unshackle env check` lists it. + +Without the decoder, a title whose only tracks are Audio Vivid is not muxed. + ### 3D output A stereoscopic 3D (MV-HEVC) video track also muxes to MP4 with MP4Box, because Matroska has no diff --git a/docs/reference/configuration/index.md b/docs/reference/configuration/index.md index 1141e4e0..e842090f 100644 --- a/docs/reference/configuration/index.md +++ b/docs/reference/configuration/index.md @@ -39,6 +39,6 @@ Where the file lives, how unshackle discovers it, and how to edit it: | [Post-scripts](post-scripts.md) | `post_scripts` | | [External API keys](misc.md#external-api-keys) | `omdb_api_key`, `tmdb_api_key`, `tvdb_api_key`, `tvdb_pin`, `tvdb_order`, `simkl_client_id`, `metadata_providers`, `anilist_title_language`, `disable_metadata`, `decrypt_labs_api_key`, `ipinfo_api_key` | | [Caching & updates](misc.md#caching-updates) | `title_cache_enabled`, `title_cache_time`, `title_cache_max_retention`, `update_checks`, `update_check_interval` | -| [Appearance](misc.md#appearance) | `theme` | +| [Appearance](misc.md#appearance) | `theme`, `wide_output` | | [Logging, privacy & debug](misc.md#logging-privacy-debug) | `redact_paths`, `debug`, `debug_keys`, `debug_requests`, `set_terminal_bg` | | [Deprecated & removed](misc.md#deprecated-removed-keys) | `curl_impersonate`, `downloader`, `scene_naming` | diff --git a/docs/reference/configuration/misc.md b/docs/reference/configuration/misc.md index a4481de2..9ff17e3f 100644 --- a/docs/reference/configuration/misc.md +++ b/docs/reference/configuration/misc.md @@ -146,6 +146,7 @@ calls. | Key | Type | Default | Description | |-----|------|---------|-------------| | `theme` | str | `catppuccin-mocha` | Named colour theme for all CLI output: banner, logs, progress bars, and help screens. One of `catppuccin-mocha` (aliases `default`, `catppuccin`, `mocha`), `dracula`, `nord`, `gruvbox`, `one-dark` (alias `onedark`), `tokyo-night` (alias `tokyonight`), `monokai`, `flexoki`, `solarized-dark` (alias `solarized`). Case-insensitive; underscores work in place of hyphens. Unknown names fall back to the default with a warning. Preview them all with `unshackle env theme`. | +| `wide_output` | bool | `false` | Print the `dl` title and track listings at the terminal width. When `false`, they fold at 80 columns, the width that logs and progress bars always use. | --- diff --git a/tests/commands/test_cfg.py b/tests/commands/test_cfg.py index 5327e1c0..d1dc0355 100644 --- a/tests/commands/test_cfg.py +++ b/tests/commands/test_cfg.py @@ -58,3 +58,11 @@ def test_a_duplicate_only_ruamel_sees_still_gives_its_line(config_file: Path, ca result = CliRunner().invoke(cfg_module.cfg, ["a"]) assert result.exit_code == 1 assert caplog.messages == [f"{config_file}: duplicate key on line 3. Unshackle uses the later one."] + + +def test_lowercase_true_and_false_are_written_as_booleans_and_no_stays_a_language(config_file: Path) -> None: + config_file.write_text("theme: rainbow\n", encoding="utf-8") + for key, value in (("wide_output", "false"), ("debug_requests", "TRUE"), ("lang", "no")): + assert CliRunner().invoke(cfg_module.cfg, [key, value]).exit_code == 0 + text = config_file.read_text(encoding="utf-8") + assert "wide_output: false\n" in text and "debug_requests: true\n" in text and "lang: no\n" in text diff --git a/tests/core/test_audio_vivid.py b/tests/core/test_audio_vivid.py new file mode 100644 index 00000000..14ad75ed --- /dev/null +++ b/tests/core/test_audio_vivid.py @@ -0,0 +1,113 @@ +from __future__ import annotations + +import logging +import stat +import sys +from pathlib import Path +from typing import Any + +import pytest + +from tests.core.test_hdr10plus_probe import ts +from unshackle.commands.dl import take_unmuxable_audio, track_file_suffix +from unshackle.core.tracks import Audio, Tracks, Video +from unshackle.core.utils import audio_vivid + +LOG = logging.getLogger("test") + + +def media_packet(pid: int, payload: bytes, start: bool) -> bytes: + """One TS packet carrying ``payload``, padded with adaptation-field stuffing as packagers do.""" + stuffing = 184 - len(payload) + flags = 0x40 if start else 0x00 + if stuffing: + field = bytes([stuffing - 1]) + (b"\x00" + b"\xff" * (stuffing - 2) if stuffing > 1 else b"") + return bytes([0x47, flags | pid >> 8, pid & 0xFF, 0x30]) + field + payload + return bytes([0x47, flags | pid >> 8, pid & 0xFF, 0x10]) + payload + + +def audio_vivid_ts(frames: list[bytes]) -> bytes: + """A PAT, a PMT with one Audio Vivid stream on PID 0x100, and one PES per frame.""" + data = ts((0xD5, b"")) + for frame in frames: + pes = b"\x00\x00\x01\xc0\x00\x00\x80\x80\x05" + b"\x21\x00\x01\x00\x01" + frame + data += b"".join(media_packet(0x100, pes[i : i + 184], i == 0) for i in range(0, len(pes), 184)) + return data + + +def test_the_elementary_stream_is_every_pes_payload_in_order(tmp_path: Path) -> None: + frames = [bytes([n]) * size for n, size in ((1, 40), (2, 400), (3, 175))] + source, out = tmp_path / "a.ts", tmp_path / "a.av3a" + source.write_bytes(audio_vivid_ts(frames)) + assert audio_vivid.extract_audio_vivid(source, out) + assert out.read_bytes() == b"".join(frames) + + +def test_a_ts_without_audio_vivid_extracts_nothing(tmp_path: Path) -> None: + source = tmp_path / "a.ts" + source.write_bytes(ts((0x0F, b""))) + assert not audio_vivid.extract_audio_vivid(source, tmp_path / "a.av3a") + + +def fake_decoder(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, *, fail: bool = False) -> None: + """A stand-in decoder that copies its input to its output, or exits 3; a .cmd on Windows.""" + if sys.platform == "win32": + decoder = tmp_path / "avs3Decoder.cmd" + decoder.write_text("@exit /b 3\r\n" if fail else '@copy /y "%~1" "%~2" >nul\r\n') + else: + decoder = tmp_path / "avs3Decoder" + decoder.write_text("#!/bin/sh\nexit 3\n" if fail else '#!/bin/sh\ncp "$1" "$2"\n') + decoder.chmod(decoder.stat().st_mode | stat.S_IEXEC) + monkeypatch.setattr(audio_vivid.binaries, "avs3Decoder", decoder, raising=False) + + +def test_the_decoder_gets_the_stream_and_writes_the_wav(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + fake_decoder(tmp_path, monkeypatch) + source = tmp_path / "a.ts" + source.write_bytes(audio_vivid_ts([b"frame"])) + wav = audio_vivid.decode_audio_vivid(source, LOG) + assert wav == tmp_path / "a.wav" and wav.read_bytes() == b"frame" + assert not (tmp_path / "a.av3a").exists() + + +def test_a_failing_decoder_warns_once_and_leaves_no_files( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + fake_decoder(tmp_path, monkeypatch, fail=True) + source = tmp_path / "a.ts" + source.write_bytes(audio_vivid_ts([b"frame"])) + with caplog.at_level(logging.WARNING): + assert audio_vivid.decode_audio_vivid(source, LOG) is None + assert caplog.messages == ["Audio Vivid decode of a.ts failed"] + assert sorted(p.name for p in tmp_path.iterdir() if not p.name.startswith("avs3")) == ["a.ts"] + + +def tracks_with(tmp_path: Path, *codecs: Any) -> Tracks: + tracks = Tracks() + tracks.videos = [Video(url="https://x.example/v.m3u8", language="zh", codec=Video.Codec.HEVC)] + for n, codec in enumerate(codecs): + audio = Audio(url=f"https://x.example/a{n}.m3u8", language="zh", codec=codec, id_=f"a{n}") + audio.path = tmp_path / f"a{n}.ts" + tracks.audio.append(audio) + return tracks + + +def test_without_a_decoder_audio_vivid_leaves_the_mux(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("unshackle.commands.dl.decode_audio_vivid", lambda path, log: None) + tracks = tracks_with(tmp_path, Audio.Codec.EC3, Audio.Codec.AV3A) + kept = take_unmuxable_audio(tracks, LOG) + assert [a.id for a in kept] == ["a1"] + assert [a.codec for a in tracks.audio] == [Audio.Codec.EC3] + + +def test_with_a_decoder_a_wav_copy_joins_the_mux(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("unshackle.commands.dl.decode_audio_vivid", lambda path, log: path.with_suffix(".wav")) + tracks = tracks_with(tmp_path, Audio.Codec.AV3A) + kept = take_unmuxable_audio(tracks, LOG) + assert [a.id for a in kept] == ["a0"] and kept[0].codec == Audio.Codec.AV3A + assert [(a.id, a.codec, a.path) for a in tracks.audio] == [("a0", Audio.Codec.PCM, tmp_path / "a0.wav")] + + +def test_a_kept_track_is_named_like_a_no_mux_track(tmp_path: Path) -> None: + tracks = tracks_with(tmp_path, Audio.Codec.AV3A) + assert [track_file_suffix(t) for t in (tracks.videos[0], tracks.audio[0])] == [".HEVC", ".zh.AV3A"] diff --git a/tests/core/test_print_listing.py b/tests/core/test_print_listing.py new file mode 100644 index 00000000..61b99dfb --- /dev/null +++ b/tests/core/test_print_listing.py @@ -0,0 +1,31 @@ +import os + +import pytest + +from unshackle.core import console as console_module +from unshackle.core.config import config +from unshackle.core.console import console, print_listing + + +@pytest.mark.parametrize("wide, lines", [(True, 1), (False, 2)]) +def test_wide_output_prints_a_listing_at_the_terminal_width( + monkeypatch: pytest.MonkeyPatch, wide: bool, lines: int +) -> None: + monkeypatch.setattr(config, "wide_output", wide) + monkeypatch.setattr(console_module.shutil, "get_terminal_size", lambda: os.terminal_size((140, 40))) + monkeypatch.setattr(console, "width", 80) + with console.capture() as capture: + print_listing("x" * 100) + assert len(capture.get().strip("\n").splitlines()) == lines + assert console.width == 80 + + +def test_print_wide_keeps_a_legacy_windows_console_at_its_width(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(console_module.shutil, "get_terminal_size", lambda: os.terminal_size((140, 40))) + monkeypatch.setattr(console, "legacy_windows", True) + monkeypatch.setattr(console, "_width", 80) + monkeypatch.setattr(console, "_height", 25) + with console.capture(): + for _ in range(3): + console_module.print_wide("x") + assert console._width == 80 diff --git a/unshackle/commands/cfg.py b/unshackle/commands/cfg.py index 2cdc9983..b2d9f506 100644 --- a/unshackle/commands/cfg.py +++ b/unshackle/commands/cfg.py @@ -33,7 +33,7 @@ def cfg(ctx: click.Context, key: str, value: str, unset: bool, list_: bool) -> N try: value = ast.literal_eval(value) except (ValueError, SyntaxError): - pass # probably a str without quotes or similar, assume it's a string value + value = ast.literal_eval(value.title()) if value.lower() in ("true", "false") else value log = logging.getLogger("cfg") diff --git a/unshackle/commands/dl.py b/unshackle/commands/dl.py index eda94562..f0173d13 100644 --- a/unshackle/commands/dl.py +++ b/unshackle/commands/dl.py @@ -1,5 +1,6 @@ from __future__ import annotations +import copy import html import logging import math @@ -45,7 +46,7 @@ from unshackle.core import __version__, binaries, providers from unshackle.core.cdm import DecryptLabsRemoteCDM from unshackle.core.cdm.detect import cdm_type_stub, is_playready_cdm, is_remote_cdm, is_widevine_cdm from unshackle.core.config import cdm_entry_names_device, config, load_yaml, resolve_cdm_name, resolve_decryption -from unshackle.core.console import GradientPulseBarColumn, SyncLive, console, listing_panel +from unshackle.core.console import GradientPulseBarColumn, SyncLive, console, listing_panel, print_listing from unshackle.core.constants import ( DEFAULT_SAMPLE, DOWNLOAD_ALL_DRM, @@ -135,6 +136,7 @@ from unshackle.core.utilities import ( time_elapsed_since, ) from unshackle.core.utils import tags +from unshackle.core.utils.audio_vivid import decode_audio_vivid from unshackle.core.utils.bitrate import apply_real_bitrates from unshackle.core.utils.click_types import ( AUDIO_CODEC_LIST, @@ -523,6 +525,35 @@ def mux_skipped(no_mux: bool, title: Any) -> bool: return no_mux and not isinstance(title, Song) +def take_unmuxable_audio(tracks: Tracks, log: logging.Logger) -> list[Audio]: + """Remove and return the Audio Vivid tracks, which no muxer carries, to be saved beside the output. + + With the user's decoder, each one also goes back into the mux as a WAV copy. + """ + kept = [a for a in tracks.audio if a.codec == Audio.Codec.AV3A] + tracks.audio = [a for a in tracks.audio if a.codec != Audio.Codec.AV3A] + for track in kept: + wav = decode_audio_vivid(track.path, log) if track.path else None + if wav: + decoded = copy.copy(track) + decoded.path = wav + decoded.codec = Audio.Codec.PCM + tracks.audio.append(decoded) + return kept + + +def track_file_suffix(track: Any) -> str: + """The name part a track file saved on its own adds after the title: language and codec, or subtitle flags.""" + if isinstance(track, Video): + return f".{track.codec.name if track.codec is not None else 'video'}" + lang_suffix = f".{track.language}" if track.language else "" + if isinstance(track, Audio): + return f"{lang_suffix}.{track.codec.name if track.codec is not None else 'audio'}" + if isinstance(track, Subtitle): + return f"{lang_suffix}{'.forced' if track.forced else ''}{'.sdh' if track.sdh else ''}" + return "" + + def no_mux_extension(track: Any, path: Path, media_info: MediaInfo) -> str: """The extension of a --no-mux track file: an audio track in an MPEG-4 file is saved as .m4a.""" container = media_info.general_tracks[0].format if media_info.general_tracks else None @@ -2162,7 +2193,7 @@ class dl: titles = self.apply_continuous_episodes(titles) console.print(Padding(Rule(f"[rule.text]{titles.__class__.__name__}: {titles}"), (1, 2))) - console.print(Padding(titles.tree(verbose=list_titles), (0, 5))) + print_listing(titles.tree(verbose=list_titles)) if list_titles: return @@ -2598,7 +2629,7 @@ class dl: if list_: available_tracks, _ = title.tracks.tree() - console.print(Padding(listing_panel(available_tracks, "Available Tracks"), (0, 5))) + print_listing(listing_panel(available_tracks, "Available Tracks")) continue keep_videos = True @@ -3589,6 +3620,11 @@ class dl: tagged_paths: set[Path] = set() append_audio_codec_suffix = True skip_mux = mux_skipped(no_mux, title) + kept_audio = [] if skip_mux else take_unmuxable_audio(title.tracks, self.log) + if kept_audio and not title.tracks: + title.tracks.audio = kept_audio + kept_audio = [] + skip_mux = True if skip_mux: for track in title.tracks: @@ -3948,23 +3984,8 @@ class dl: track = next((t for t in title.tracks if t.path == track_path), None) if track: - if isinstance(track, Video): - track_suffix = f".{track.codec.name if track.codec is not None else 'video'}" - elif isinstance(track, Audio): - lang_suffix = f".{track.language}" if track.language else "" - track_suffix = ( - f"{lang_suffix}.{track.codec.name if track.codec is not None else 'audio'}" - ) - elif isinstance(track, Subtitle): - lang_suffix = f".{track.language}" if track.language else "" - forced_suffix = ".forced" if track.forced else "" - sdh_suffix = ".sdh" if track.sdh else "" - track_suffix = f"{lang_suffix}{forced_suffix}{sdh_suffix}" - else: - track_suffix = "" - ext = no_mux_extension(track, track_path, media_info) - final_path = final_dir / f"{base_filename}{track_suffix}{ext}" + final_path = final_dir / f"{base_filename}{track_file_suffix(track)}{ext}" else: final_path = final_dir / f"{base_filename}{track_path.suffix}" @@ -4010,6 +4031,16 @@ class dl: shutil.move(muxed_path, final_path) used_final_paths.add(final_path) self.completed_files.append(final_path) + for kept in kept_audio: + assert kept.path is not None + kept_path = final_path.with_name( + f"{final_path.stem}{track_file_suffix(kept)}{kept.path.suffix}" + ) + shutil.move(kept.path, kept_path) + self.completed_files.append(kept_path) + sidecar_files.setdefault(muxed_path, []).append(kept_path) + self.log.info(f"Saved {kept_path.name}: no muxer carries Audio Vivid") + kept_audio = [] if muxed_path not in tagged_paths: tags.tag_file( final_path, diff --git a/unshackle/core/config.py b/unshackle/core/config.py index b1a74e8f..b07a9f46 100644 --- a/unshackle/core/config.py +++ b/unshackle/core/config.py @@ -176,6 +176,7 @@ class Config: self.merge_segments: bool = bool(kwargs.get("merge_segments", False)) self.theme: str = kwargs.get("theme") or "default" + self.wide_output: bool = kwargs.get("wide_output", False) self.set_terminal_bg: bool = kwargs.get("set_terminal_bg", False) self.tag: str = kwargs.get("tag") or "" self.tag_group_name: bool = kwargs.get("tag_group_name", True) diff --git a/unshackle/core/console.py b/unshackle/core/console.py index 809bb8cf..a37bb5c9 100644 --- a/unshackle/core/console.py +++ b/unshackle/core/console.py @@ -492,14 +492,23 @@ def print_wide(renderable: RenderableType, pad: PaddingDimensions = (1, 3)) -> N """Print a listing at the terminal width, indented like the rich-click help panels. The shared console holds 80 columns to keep logs and progress compact. Long values such as paths - and URLs fold badly at that width, so this widens the console for one print and restores it after. + and URLs fold badly at that width, so this widens the console for one print and restores it after. It + restores ``_width``, because on a legacy Windows console the ``width`` property reads one column less. """ - fixed_width = console.width - console.width = max(fixed_width, shutil.get_terminal_size().columns) + fixed_width = console._width + console.width = max(console.width, shutil.get_terminal_size().columns - console.legacy_windows) try: console.print(Padding(renderable, pad)) finally: - console.width = fixed_width + console._width = fixed_width + + +def print_listing(renderable: RenderableType) -> None: + """Print a listing at the terminal width when ``wide_output`` is on, else at the shared console width.""" + if config.wide_output: + print_wide(renderable, (0, 5)) + else: + console.print(Padding(renderable, (0, 5))) _prompt_handler: Optional[Callable[[str], str]] = None @@ -545,6 +554,7 @@ __all__ = ( "console", "listing_panel", "listing_table", + "print_listing", "print_wide", "prompt_user", "set_notice_handler", diff --git a/unshackle/core/tracks/audio.py b/unshackle/core/tracks/audio.py index de188ba9..4aacba87 100644 --- a/unshackle/core/tracks/audio.py +++ b/unshackle/core/tracks/audio.py @@ -103,7 +103,7 @@ class Audio(Track): @staticmethod def from_ts(stream_type: int, registration: bytes) -> Optional[Audio.Codec]: """The codec an MPEG-TS PMT entry names by its registration descriptor or ``stream_type``, else None.""" - if registration == b"AVSA": + if registration in (b"AVSA", b"av3a"): return Audio.Codec.AV3A return {0x0F: Audio.Codec.AAC, 0x81: Audio.Codec.AC3, 0x87: Audio.Codec.EC3, 0xD5: Audio.Codec.AV3A}.get( stream_type diff --git a/unshackle/core/utils/audio_vivid.py b/unshackle/core/utils/audio_vivid.py new file mode 100644 index 00000000..813cab23 --- /dev/null +++ b/unshackle/core/utils/audio_vivid.py @@ -0,0 +1,53 @@ +from __future__ import annotations + +import logging +from pathlib import Path +from typing import Optional + +from unshackle.core import binaries +from unshackle.core.utils.hdr10plus import PROBE_BYTES, TS_PACKET, ts_payload, ts_pid, ts_pmt +from unshackle.core.utils.subprocess import run_step + + +def extract_audio_vivid(ts_path: Path, out_path: Path) -> bool: + """Write the Audio Vivid elementary stream of an MPEG-TS file to ``out_path``; False when it holds none.""" + from unshackle.core.tracks import Audio + + with ts_path.open("rb") as ts: + entries = ts_pmt(ts.read(PROBE_BYTES)) or [] + pid = next((p for p, kind, reg in entries if Audio.Codec.from_ts(kind, reg) == Audio.Codec.AV3A), None) + if pid is None: + return False + ts.seek(0) + with out_path.open("wb") as out: + while packet := ts.read(TS_PACKET): + if len(packet) < TS_PACKET or packet[0] != 0x47 or ts_pid(packet) != pid: + continue + payload = ts_payload(packet) + if packet[1] & 0x40 and payload[:3] == b"\x00\x00\x01" and len(payload) > 8: + payload = payload[9 + payload[8] :] + out.write(payload) + return out_path.stat().st_size > 0 + + +def decode_audio_vivid(ts_path: Path, log: logging.Logger) -> Optional[Path]: + """Decode an Audio Vivid MPEG-TS file to WAV with the user's ``avs3Decoder``, or None without one or on failure. + + unshackle does not ship the decoder: the UWA reference decoder's licence limits its use and redistribution. + A service that delivers Audio Vivid declares it in ``get_binaries()``, so ``env check`` lists it. + """ + decoder: Optional[Path] = getattr(binaries, "avs3Decoder", None) + if not decoder: + return None + stream, wav = ts_path.with_suffix(".av3a"), ts_path.with_suffix(".wav") + try: + if not extract_audio_vivid(ts_path, stream): + raise ValueError("no Audio Vivid stream in the file") + run_step([decoder, stream, wav], status="Decoding Audio Vivid...", output=wav, label="Audio Vivid decode") + except (OSError, ValueError, RuntimeError) as e: + log.warning(f"Audio Vivid decode of {ts_path.name} failed") + log.debug(f"Audio Vivid decode of {ts_path.name}: {e}") + return None + finally: + stream.unlink(missing_ok=True) + return wav diff --git a/unshackle/core/utils/hdr10plus.py b/unshackle/core/utils/hdr10plus.py index 30188b8b..d02879e6 100644 --- a/unshackle/core/utils/hdr10plus.py +++ b/unshackle/core/utils/hdr10plus.py @@ -40,7 +40,24 @@ def is_ts(data: bytes) -> bool: def ts_streams(data: bytes) -> Optional[list[tuple[int, bytes]]]: - """The ``stream_type`` and registration format identifier of each PMT entry, or None without a PMT. + """The ``stream_type`` and registration format identifier of each PMT entry, or None without a PMT.""" + entries = ts_pmt(data) + return None if entries is None else [(stream_type, registration) for _, stream_type, registration in entries] + + +def ts_pid(b: bytes, i: int = 1) -> int: + """The 13-bit PID at ``b[i]``: a packet header at offset 1, or a PAT or PMT entry.""" + return ((b[i] & 0x1F) << 8) | b[i + 1] + + +def ts_payload(packet: bytes) -> bytes: + """The payload of a TS packet after any adaptation field, or b"" when it carries none.""" + adaptation = (packet[3] >> 4) & 3 + return packet[5 + packet[4] if adaptation & 2 else 4 :] if adaptation & 1 else b"" + + +def ts_pmt(data: bytes) -> Optional[list[tuple[int, int, bytes]]]: + """The PID, ``stream_type`` and registration format identifier of each PMT entry, or None without a PMT. PAT program 0 names the network information table, so it is skipped. Lengths that run past the packet are cut to it, so bad bytes give fewer entries, never an exception. @@ -48,16 +65,12 @@ def ts_streams(data: bytes) -> Optional[list[tuple[int, bytes]]]: # reads only a PAT and PMT that fit in one packet, as packagers write them; reassemble # sections across packets if a source with a longer PMT appears - def pid(b: bytes, i: int) -> int: - return ((b[i] & 0x1F) << 8) | b[i + 1] - def length(b: bytes, i: int) -> int: return ((b[i] & 0x0F) << 8) | b[i + 1] def section(packet: bytes, table_id: int) -> bytes: """The packet's ``table_id`` section up to its CRC, or b"" when the packet holds no such section.""" - adaptation = (packet[3] >> 4) & 3 - payload = packet[5 + packet[4] if adaptation & 2 else 4 :] if adaptation & 1 else b"" + payload = ts_payload(packet) body = payload[1 + payload[0] :] if payload else b"" body = body[: max(0, 3 + length(body, 1) - 4)] if len(body) >= 3 else b"" return body if len(body) >= 12 and body[0] == table_id else b"" @@ -67,10 +80,10 @@ def ts_streams(data: bytes) -> Optional[list[tuple[int, bytes]]]: packet = data[pos : pos + TS_PACKET] if packet[0] != 0x47 or not packet[1] & 0x40: continue - if pid(packet, 1) == 0 and pmt_pid is None: + if ts_pid(packet) == 0 and pmt_pid is None: pat = section(packet, 0x00) - pmt_pid = next((pid(pat, k + 2) for k in range(8, len(pat) - 3, 4) if pat[k] or pat[k + 1]), None) - elif pid(packet, 1) == pmt_pid: + pmt_pid = next((ts_pid(pat, k + 2) for k in range(8, len(pat) - 3, 4) if pat[k] or pat[k + 1]), None) + elif ts_pid(packet) == pmt_pid: pmt = section(packet, 0x02) if not pmt: return None @@ -82,7 +95,7 @@ def ts_streams(data: bytes) -> Optional[list[tuple[int, bytes]]]: if pmt[j] == 0x05 and j + 6 <= info_end: registration = pmt[j + 2 : j + 6] j += 2 + pmt[j + 1] - streams.append((pmt[i], registration)) + streams.append((ts_pid(pmt, i + 1), pmt[i], registration)) i = info_end return streams return None