#!/usr/bin/env python3 # -*- coding: utf-8 -*- # # TrackShepherd # Copyright (c) 2026 J. Philipp de Graaff, www.playsheep.de # # Licensed under the Playsheep Source-Available Licence 1.0. # Run "python3 trackshepherd.py --license" to print the full licence text. # Use is free for any purpose. Selling the software is not permitted. # """TrackShepherd - audio track processing for video files. TrackShepherd reprocesses the audio tracks of a video file and leaves the video stream untouched. For each selected audio track it: - downmixes multichannel audio to stereo with a layout-aware matrix - normalises the level to a true-peak target - encodes the result to AAC-LC with neroAacEnc - remuxes the new tracks with the original video, subtitles and chapters - verifies timing, stream structure and packet integrity before it replaces anything The source file is never overwritten. Output is written to a candidate file and committed only after verification has passed. External programs: ffmpeg, ffprobe, neroAacEnc Python: 3.10 or newer, standard library only """ from __future__ import annotations import argparse import stat import json import logging import math import os import random import re import shlex import shutil import subprocess import tempfile import sys import time import unicodedata from collections import deque from datetime import datetime from dataclasses import dataclass, field, fields from pathlib import Path from typing import Any if sys.version_info < (3, 10): sys.stderr.write("Error. Python 3.10 or newer is required.\n") raise SystemExit(2) # =========================================================================== # 01 Header, constants, tables # =========================================================================== # Full licence text, zlib-compressed and base64-encoded to keep the source # short. Printed by --license. LICENSE_BLOB = """ eNqFWE2P2zYQvftXEHtJAyhOsknTNr00SIqiRVoETYqiR1oa2cxSpEpSdvTv+2ZISvJig+bktcn5 ePPmzTAfrJ7jiWhUH/0UWnry5qyN1QdL6r1pybWknu+f7d76cQ7meErqm/axun12+0r9tlcfTsaa cVQdqV+C1n3fqMvlsh+rzX1Hu93zvXpHvXEmGe/iTil189H36aID3aiBtIsqnUjF8p3SrlMmRdX5 dhrIJc33VJzG0Rrq1OQ6CrhhxJTNMTbK4IgkoHxQB+N0mFXvw7BXn2Bct60fRu1m446qXXJxPhm5 wabG4D9Tm2CJ72WvpkMApjeUY2z9mQJiqLHuJZt3FMwZ589LPnCkLj7cqYOOOA5DfLtmzRG2HokB E4Sj2UicDjFp+NJWjTok5XuAkO3/46ctUCOFCIsw4sNROxNzqPSFQmsiWyynBhMjQ64OZP1FbH06 mdA9YQezYkS8Q35xTbNTOqpIOKAT2TnDyykEGhCuOjIAjjq2dZjZkUEYF1frEBluE+ufqCGQA8q4 H1MwQJdjE/CjOgYkvLHlI612drtb8GtJQiWvpkh8FnCoQc/85zWsKJsgP05hhC1w0cDolFR70uFI JTT6MiKYaGe2ZVxrpw4xsjHjIoBQWhWuNLCm9AQbwQCvYt3D5zX0DVti1rIR4KcRrLfqguaQOjCd uCyjNl2mhUTKHA6dVHtGurDxrjA+irXsyLgRGXQ6aYApUB4I8REzRugKYNIGcrYzgKQoHg7qAtKs GCAwHNByORmZK/D2W1gbaRFJa/Cd6YEEWG+4gocp5Q6NZO3SopFNd1P7kOVmxZlByL2V+GNMswX0 +M2jYMy4jonAvY8ieTRBP7lWEK6lZFvCoEInobcZuNxdJtGKRKk3bjGC9KWlURqlcpJtCQUrH43L JVn6A+zEr6iVNBOqUCFgM2hQRCKlZ2Vqi2AkANMLC1zJApFVnFDmF/dZLVAXnAXZLdRbvvPBa2Tr LYR2/6IcBG+PUEIr1BWCLUrVqD4QcRK5ORqu4Bn073BTp9eS1+MrPMVHVsGqn/xNbpB4MmOR0yhc C8T6BnPGsS386ycwJscpWnKGWPGYMa7oP6v3j6IGj3NmrEe9scXioMNdFqj6YwMKaaHS5eQ5EXeU +GkQP5cTObGHiaU3uSuTNYl1gFiB2OYVXnFBuOZcv3kUSzZOD4uhKUqecHLRs+CH7jwewdF4D7CT tzy9yHWAjLnPlHi5V3+ufOZpEbXNcKPTB5Z1xLQlH8fwdxG3MRiU9wKJAls3so8C++FB942aC6UQ fK209PP9KaWvGIMvKhXAuKxhPS0lu0e+r9kB2GIKmJdBB5SLfqQyinkgUjjzbBY0uY37XkYvux31 zN3UcIXDPW+FxjET/iuTVYwWl48iR1Pc4Y+qOfztWduJKoUkgv9LDX9jjjGn8NugnWZGFuMygT9S tv6S9yhRDsSKPMpUET4J4Ur+D6d+Zeq28tAaqNRGCGM+sFd/LWvBEvo6p5JnYxgu63zKHRqL7sGk +Pu1F+Jc/GQ7+Lojli/cixhH8CceVq/r7NcW6wcExpKWAXlXos9bxpbjdfpLv+u6iKJFvt2rPzDn gu6IZUA0WYs6b8be1eYhys7+N7e2Or9taWnmRr7kT2zrQ11hmzr3rT/6rAewWSzlXVR2DokGkb6S SGGX/c85rA3sJq5Ce4MamHiz7in1FlOUjd4ZB40rCwubQigYdlakbx2paLmr8qfFP3MLthheiGPS B6zrKQ/23iQHq6WNuTFMO1kd6vZUMnVPsA4HeCGh3XUykBDpjbzLKgoBqiYXWVzKSiDzHUFTkn6R aR94cUFSk01RyPA7LunksbOzok/yqWBcpsXk8LzANebi7rs9HidIVtdJbE3JTQD3UkqeN5BV4CdE WbjM7VFGNo8fqy/NgzIpa/kJ5C1blzix65ppfRR0eTdrlr+QNaBtUBT0XCmbwZAJYRqrrKxrZGcY Ci7m+qmVbVxbTLeRpC2yKALmSP9OlKWs0wO0hTsl7/zMn7ySyBqKjwJ2gaWsz3VjWbe+92Y5wZrY aleSxb7Em3F9b20AG2qt5P2lL4tabEqUe7Fu1/I6WynWrOUSMI3sWU3exxhGR0drjvldJ1NwsqPO q8LnKUgy1vT48eA72ctPpG06gRjfwzPX2ek6JrE9het2vdppKG8wnt97LWqNZSqr3CGQbk88oJnz kmtea5BpklW1HGynMoLKDW5mpPviGUo0CyVkYZeXXh4a+WW3+2EPDcRzqkLBmVqsMBPqWuRWlIXl QgY6wj6R5YoQP1FbYkyasg8P5TW5HI/cSnNZv+R47t6fHcCN2NHRF2xSQmf+4TKiiTIw3sAxdNNF W9baVbaKOabjmZwRFL2z/IR5y0LQpvLztaoLSbM0oAlDngz1PX1vc3+9K+/On7b/kyAD2WgJ+EKH CK1Tp5TG+Prp0+25p7v/AELM7bw= """ def licence_text() -> str: import base64 import zlib return zlib.decompress(base64.b64decode("".join(LICENSE_BLOB.split()))).decode("utf-8") APP_NAME = "trackshepherd" DISPLAY_NAME = "TrackShepherd" VERSION = "0.7.2" JSON_SCHEMA_VERSION = 1 EXIT_OK = 0 EXIT_FILE_FAILED = 1 EXIT_USAGE = 2 EXIT_TOOL_MISSING = 3 EXIT_INTERRUPTED = 130 SQRT_HALF = 0.7071067811865476 MEDIA_SUFFIXES = {".mkv", ".mp4", ".m4v", ".avi", ".mov", ".webm", ".ts", ".m2ts"} # Status: legacy profile, empirically characterized. # Measured on the real binary, 20 s pink noise, stereo: # SQ 32000 Hz, q 0.40 -> 63493 bit/s # hq 44100 Hz, q 0.36 -> 82223 bit/s PROFILES = { "sq": {"sample_rate": 32000, "nero_quality": 0.40, "measured_bitrate": 63493}, "hq": {"sample_rate": 44100, "nero_quality": 0.36, "measured_bitrate": 82223}, } # Historical CLI name, kept as an alias. PROFILE_ALIASES = {"normal": "sq"} SD_MAX_WIDTH = 720 SD_MAX_HEIGHT = 576 OVERDRIVE_QUOTAS = { "off": None, "light": 1.0 / 100000.0, "medium": 1.0 / 50000.0, "hard": 1.0 / 25000.0, "brutal": 1.0 / 12500.0, } OVERDRIVE_CAP_DB = 6.0 SILENCE_THRESHOLD_DB = -60.0 PEAK_RETRIES_MAX = 3 # Automatic protection against isolated digital full-scale outliers. # It deliberately acts only on an extremely small upper sample fraction and # a clear level step to the remaining programme material. It detects rare # high samples; it cannot identify the cause of a transient. ISOLATED_PEAK_NEAR_FS_DB = -0.20 ISOLATED_PEAK_MIN_GAP_DB = 4.0 ISOLATED_PEAK_QUANTILE = 1.0 / 1_000_000.0 ISOLATED_PEAK_MAX_BUDGET = 256 ISOLATED_PEAK_MAX_RECOVERY_DB = 12.0 LAYOUT_CHANNELS = { "mono": ["FC"], "stereo": ["FL", "FR"], "2.1": ["FL", "FR", "LFE"], "3.0": ["FL", "FR", "FC"], "3.0(back)": ["FL", "FR", "BC"], "4.0": ["FL", "FR", "FC", "BC"], "quad": ["FL", "FR", "BL", "BR"], "quad(side)": ["FL", "FR", "SL", "SR"], "5.0": ["FL", "FR", "FC", "BL", "BR"], "5.0(side)": ["FL", "FR", "FC", "SL", "SR"], "5.1": ["FL", "FR", "FC", "LFE", "BL", "BR"], "5.1(side)": ["FL", "FR", "FC", "LFE", "SL", "SR"], "6.1": ["FL", "FR", "FC", "LFE", "BC", "SL", "SR"], "6.1(back)": ["FL", "FR", "FC", "LFE", "BL", "BR", "BC"], "7.1": ["FL", "FR", "FC", "LFE", "BL", "BR", "SL", "SR"], } UNSUPPORTED_LAYOUTS = {"7.1(wide)", "7.1(wide-side)", "hexagonal", "octagonal", "22.2"} LEGACY_DERIVED_LAYOUTS = {"quad", "5.0", "5.1"} LEGACY_ADAPTED_LAYOUTS = {"quad(side)", "5.0(side)", "5.1(side)"} LANG_CANONICAL = { "ger": "de", "deu": "de", "eng": "en", "fre": "fr", "fra": "fr", "spa": "es", "ita": "it", "dut": "nl", "nld": "nl", "jpn": "ja", "chi": "zh", "zho": "zh", "rus": "ru", "pol": "pl", "por": "pt", "swe": "sv", "dan": "da", "nor": "no", "fin": "fi", "cze": "cs", "ces": "cs", "hun": "hu", "tur": "tr", "ara": "ar", "kor": "ko", } LANG_LETTER = {"de": "D", "en": "E"} REQUIRED_FILTERS = {"pan", "channelmap", "aresample", "loudnorm", "volumedetect", "alimiter", "asplit", "aformat", "ebur128"} MATROSKA_FORMATS = ("matroska", "webm") MP4_FORMATS = ("mov", "mp4", "m4a", "3gp", "3g2", "mj2", "ipod") NERO_SHA256_TESTED = { "linux-x86": "c1258eb3f4c4eb278a51ad2fc4203424a442a012a8b1028a3b5c7c810bd49ea9", "windows-x86": "24c678ced0040014ccaf365be57ba856fdacd2fc1fc68d9c36b64c7aeea722bf", } NERO_CHANNEL_ORDER = { "5.1": "pan=5.1|c0=FC|c1=FL|c2=FR|c3=LFE|c4=BL|c5=BR", } NERO_VERIFIED_LAYOUTS = frozenset({"5.1"}) NERO_TESTED_PRIMING_SAMPLES_48K = 2624 NERO_32BIT_HINT = ( "The Linux binary of neroAacEnc is ELF 32-bit x86. The file exists " "and is executable, but start-up fails because the 32-bit loader or " "the 32-bit runtime is missing. On Debian and Ubuntu the packages " "libc6:i386 and libstdc++6:i386 suffice. Other distributions use other names." ) TIMELINE_TOLERANCE = 0.005 MUX_OFFSET_DEADBAND = 0.001 LIMITER_HEADROOM_DB = 1.0 PACKET_HASH_SAMPLE = 500 PACKET_HASH_ALGORITHM = "SHA256" ZERO_DELAY_AUDIO_CODECS = frozenset({"flac"}) # Technical tags are not copied from the source for newly encoded audio # tracks. All other tags count as semantic metadata. STALE_AUDIO_TAGS = { "encoder", "bps", "duration", "number_of_frames", "number_of_bytes", "source_id", "handler_name", "vendor_id", "_statistics_writing_app", "_statistics_writing_date_utc", "_statistics_tags", } log = logging.getLogger(APP_NAME) PROGRESS_ENABLED = True PROGRESS_REFRESH = 0.25 # =========================================================================== # 02 Data model # =========================================================================== ANCHOR_METHODS = ("presentation", "start_time", "packet_pts") @dataclass(frozen=True) class RawTiming: presentation: float | None = None start_time: float | None = None packet_pts: float | None = None initial_padding: int | None = None skip_samples: int | None = None discard_padding: int | None = None sample_rate: int | None = None def get(self, method: str) -> float | None: if method == "presentation": return self.presentation if method == "start_time": return self.start_time if method == "packet_pts": return self.packet_pts return None def padding_seconds(self) -> float | None: if self.initial_padding is None or not self.sample_rate: return None return self.initial_padding / self.sample_rate def skip_seconds(self) -> float | None: if self.skip_samples is None or not self.sample_rate: return None return self.skip_samples / self.sample_rate @dataclass(frozen=True) class ContentAnchor: seconds: float | None method: str = "unknown" provenance: str = "none" confidence: str = "not-verifiable" correction: float = 0.0 @property def known(self) -> bool: return self.seconds is not None and self.confidence == "verified" UNKNOWN_CONTENT = ContentAnchor(None) @dataclass(frozen=True) class AnchorComparison: delta: float | None method: str @property def known(self) -> bool: return self.delta is not None and self.method != "unknown" @dataclass(frozen=True) class EncoderTiming: raw: RawTiming priming_samples: int | None priming_seconds: float | None signalling: str status: str @dataclass(frozen=True) class StreamInfo: index: int kind: str codec: str tags_raw: dict[str, str] disposition: dict[str, bool] language_raw: str | None language_canonical: str | None title: str | None time: RawTiming content: ContentAnchor duration: float | None bit_rate: int | None @dataclass(frozen=True) class VideoInfo(StreamInfo): width: int | None height: int | None sar_num: int | None sar_den: int | None @property def display_width(self) -> float | None: if self.width is None: return None if not self.sar_num or not self.sar_den: return float(self.width) return self.width * (self.sar_num / self.sar_den) @dataclass(frozen=True) class AudioInfo(StreamInfo): audio_index: int channels: int layout: str | None layout_source: str sample_rate: int @dataclass(frozen=True) class MediaFile: path: Path format_name: str | None duration: float | None start_time: float | None streams: tuple[StreamInfo, ...] chapter_count: int def audio(self) -> list[AudioInfo]: return [s for s in self.streams if isinstance(s, AudioInfo)] def video(self) -> list[VideoInfo]: return [s for s in self.streams if isinstance(s, VideoInfo)] def of_kind(self, kind: str) -> list[StreamInfo]: return [s for s in self.streams if s.kind == kind] @dataclass(frozen=True) class Profile: name: str sample_rate: int nero_quality: float selection: str = "explicit" @dataclass(frozen=True) class DownmixPlan: source_layout: str target_layout: str pan_expr: str | None weights_left: tuple[tuple[str, float], ...] weights_right: tuple[tuple[str, float], ...] provenance: str note: str @dataclass(frozen=True) class NormalizationPlan: mode: str target_tp: float max_gain: float overdrive_step: str overdrive_db: float isolated_peak_guard: str limiter: str tolerance: float retries: int @dataclass(frozen=True) class Analysis: n_samples: int max_volume: float mean_volume: float histogram: dict[int, int] input_i: float input_tp: float input_lra: float input_thresh: float @dataclass class GainResult: peak_gain: float sample_peak_gain: float g_hist: float | None overdrive_auto: float overdrive_manual: float total_gain: float limiter_active: bool notes: list[str] = field(default_factory=list) @dataclass class TrackPlan: source: AudioInfo profile: Profile downmix: DownmixPlan target_sample_rate: int normalization: NormalizationPlan analysis: Analysis | None = None gain: GainResult | None = None encoded_path: Path | None = None encoded_time: RawTiming | None = None encoder_timing: EncoderTiming | None = None mux_offset: float | None = None applied_mux_offset: float | None = None mux_offset_note: str = "" mux_offset_method: str = "unknown" measured_tp: float | None = None attempts: int = 0 peak_history: list[tuple[float, float]] = field(default_factory=list) conform: bool | None = None source_timeline: AudioFrameTimeline | None = None preserve_gaps: bool = False @dataclass class JobPlan: source: MediaFile output: Path reference: StreamInfo | None tracks: list[TrackPlan] passthrough_audio: list[AudioInfo] unmapped_streams: list[StreamInfo] = field(default_factory=list) temp_dir: Path | None = None candidate_output: Path | None = None job_log_output: Path | None = None job_log_candidate: Path | None = None @dataclass class StepResult: argv: list[str] returncode: int stderr: str seconds: float stdout: str = "" @property def commandline(self) -> str: return " ".join(shlex.quote(a) for a in self.argv) @dataclass class FileResult: source: Path output: Path | None ok: bool messages: list[str] = field(default_factory=list) steps: list[StepResult] = field(default_factory=list) summary: str = "" gains: str = "" peak_attempts: tuple[int, ...] = () @dataclass(frozen=True) class TimelineCheck: audio_index: int expected: float | None measured: float | None deviation: float | None status: str note: str @dataclass(frozen=True) class PacketTimelineStats: first: float last: float count: int payload: float = 0.0 # sum of reported packet durations; not a payload parse max_gap: float = 0.0 # largest jump between consecutive packets missing_durations: int = 0 @property def span(self) -> float: return max(0.0, self.last - self.first) # =========================================================================== # 03 Process layer # =========================================================================== class ToolMissing(Exception): pass class StepFailed(Exception): def __init__(self, message: str, result: StepResult | None = None, *, compact_message: str | None = None): super().__init__(message) self.result = result self.compact_message = compact_message class TrackPlanningError(Exception): pass def run(argv: list[str], *, timeout: float | None = None) -> StepResult: log.debug("start %s", " ".join(shlex.quote(a) for a in argv)) started = time.monotonic() try: proc = subprocess.run( argv, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=False, timeout=timeout, ) except OSError as exc: raise ToolMissing(f"Program cannot be started. {argv[0]}: {exc}") from exc except subprocess.TimeoutExpired as exc: raise StepFailed(f"Timeout after {timeout} s. {argv[0]}") from exc elapsed = time.monotonic() - started return StepResult( argv=list(argv), returncode=proc.returncode, stderr=proc.stderr.decode("utf-8", "replace"), seconds=elapsed, stdout=proc.stdout.decode("utf-8", "replace"), ) def run_checked(argv: list[str], *, timeout: float | None = None) -> StepResult: result = run(argv, timeout=timeout) if result.returncode != 0: raise StepFailed(f"Returncode {result.returncode} at {Path(argv[0]).name}", result) return result COMPACT_MODE = False CONSOLE_VERBOSITY = 0 def _verbosity_required(record: logging.LogRecord) -> int: """Console level of a log record. -v : progress lines and the track overview only -vv : processing details -vvv : additional diagnostics, but no program calls -vvvv : full debug output including FFmpeg, ffprobe and Nero calls The job log is always complete, independent of this level. """ if record.levelno >= logging.WARNING: return 0 explicit = getattr(record, "ts_verbosity", None) if explicit is not None: return int(explicit) if record.levelno <= logging.DEBUG: return 4 return 2 def log_detail(message: str, *args: Any) -> None: """Information that appears on the console from -vvv upwards.""" log.info(message, *args, extra={"ts_verbosity": 3}) def log_track_line(message: str, *args: Any) -> None: """Track overview. It is the only INFO line shown from -v upwards, because the user needs it while watching the progress.""" log.info(message, *args, extra={"ts_verbosity": 1}) def setup_logging(verbosity: int, quiet: bool, logfile: Path | None, compact: bool = False) -> None: global COMPACT_MODE, CONSOLE_VERBOSITY COMPACT_MODE = bool(compact) CONSOLE_VERBOSITY = max(0, min(4, int(verbosity or 0))) log.setLevel(logging.DEBUG) log.handlers.clear() console = ReporterAwareHandler(sys.stderr) # ReporterAwareHandler performs the actual selection, so that INFO # messages can appear only from -vv upwards. console.setLevel(logging.DEBUG if not quiet else logging.ERROR) console.setFormatter(logging.Formatter("%(levelname)s %(message)s")) log.addHandler(console) if logfile is not None: handler = logging.FileHandler(logfile, encoding="utf-8") handler.setLevel(logging.DEBUG) handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s")) log.addHandler(handler) def configure_progress(enabled: bool) -> None: global PROGRESS_ENABLED PROGRESS_ENABLED = bool(enabled and sys.stderr.isatty()) TEMP_OWNER_MARKER = ".trackshepherd-owner.json" TEMP_KEEP_MARKER = ".trackshepherd-keep" def _process_is_alive(pid: int) -> bool: """Checks whether a PID is alive, using only the Python standard library. On Windows ``os.kill(pid, 0)`` is deliberately not used, because its semantics there do not match the POSIX check. """ if type(pid) is not int or not 0 < pid <= 0x7fffffff: return True # An invalid PID is no proof of a dead process. if pid == os.getpid(): return True if os.name == "nt": try: import ctypes from ctypes import wintypes PROCESS_QUERY_LIMITED_INFORMATION = 0x1000 STILL_ACTIVE = 259 kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) kernel32.OpenProcess.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD] kernel32.OpenProcess.restype = wintypes.HANDLE kernel32.GetExitCodeProcess.argtypes = [wintypes.HANDLE, ctypes.POINTER(wintypes.DWORD)] kernel32.GetExitCodeProcess.restype = wintypes.BOOL kernel32.CloseHandle.argtypes = [wintypes.HANDLE] kernel32.CloseHandle.restype = wintypes.BOOL handle = kernel32.OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, False, pid) if not handle: # Only a non-existent PID counts as dead. Access denied # and unknown API errors protect the directory. return ctypes.get_last_error() != 87 # ERROR_INVALID_PARAMETER try: exit_code = wintypes.DWORD() if not kernel32.GetExitCodeProcess(handle, ctypes.byref(exit_code)): return True return exit_code.value == STILL_ACTIVE finally: kernel32.CloseHandle(handle) except Exception: # When in doubt, do not delete. Clean-up must never endanger an active # job just because the liveness check fails on this system. return True try: os.kill(pid, 0) except ProcessLookupError: return False except PermissionError: return True except OSError: return True return True def create_managed_tempdir(*, prefix: str | None = None) -> Path: path = Path(tempfile.mkdtemp(prefix=prefix or f"{APP_NAME}-")) owner = { "pid": os.getpid(), "version": VERSION, "started": datetime.now().astimezone().isoformat(), "script": str(Path(__file__).resolve()), } try: (path / TEMP_OWNER_MARKER).write_text( json.dumps(owner, ensure_ascii=False, indent=2) + "\n", encoding="utf-8" ) except OSError: # The job itself must not fail on a diagnostic marker. A directory # without a marker is deliberately not deleted automatically later. pass return path def mark_temp_keep(path: Path, reason: str) -> None: try: (path / TEMP_KEEP_MARKER).write_text( f"{datetime.now().astimezone().isoformat()} {reason}\n", encoding="utf-8" ) except OSError: pass def cleanup_stale_tempdirs() -> tuple[int, int]: """Removes orphaned TrackShepherd working directories safely. Only directories with our owner marker are touched. A keep marker protects them permanently. If the PID recorded there is still alive, the directory counts as active. The next TrackShepherd start therefore also cleans up after a hard abort, without touching running jobs. """ root = Path(tempfile.gettempdir()) removed = 0 preserved = 0 try: candidates = list(root.glob(f"{APP_NAME}-*")) except OSError: return 0, 0 for path in candidates: try: info = path.lstat() except OSError: continue # Windows junctions are reparse points, but not always symlinks. if (not stat.S_ISDIR(info.st_mode) or getattr(info, "st_file_attributes", 0) & 0x400): continue if hasattr(os, "getuid"): try: if path.stat().st_uid != os.getuid(): continue except OSError: continue owner_path = path / TEMP_OWNER_MARKER keep_path = path / TEMP_KEEP_MARKER if owner_path.is_symlink() or not owner_path.is_file(): # Old versions or foreign directories with the same prefix # are not touched, for safety reasons. continue if os.path.lexists(keep_path): preserved += 1 continue try: payload = json.loads(owner_path.read_text("utf-8")) if not isinstance(payload, dict): raise ValueError("Owner marker is not an object") pid = payload.get("pid") if type(pid) is not int or not 0 < pid <= 0x7fffffff: raise ValueError("Owner marker contains no valid PID") except (OSError, ValueError, TypeError, json.JSONDecodeError): # A damaged marker is not sufficient proof for deletion. preserved += 1 continue if _process_is_alive(pid): preserved += 1 continue try: shutil.rmtree(path) removed += 1 except OSError: preserved += 1 if removed or preserved: log_detail( "Temp clean-up: %d orphaned working directories removed, %d active or keep directories preserved.", removed, preserved, ) return removed, preserved def _format_clock(seconds: float | None) -> str: if seconds is None or not math.isfinite(seconds) or seconds < 0: return "--:--" total = int(round(seconds)) hours, rem = divmod(total, 3600) minutes, secs = divmod(rem, 60) return f"{hours:d}:{minutes:02d}:{secs:02d}" if hours else f"{minutes:02d}:{secs:02d}" def _progress_seconds(path: Path) -> float | None: try: lines = path.read_text("utf-8", "replace").splitlines() except OSError: return None for line in reversed(lines): if not line.startswith("out_time="): continue value = line.partition("=")[2].strip() if value in ("", "N/A"): continue try: hh, mm, ss = value.split(":", 2) return int(hh) * 3600 + int(mm) * 60 + float(ss) except (ValueError, TypeError): return None return None def _show_progress(label: str, duration: float | None, current: float | None, *, elapsed: float | None = None, done: bool = False) -> None: reporter = globals().get("REPORTER") if reporter is not None and reporter.enabled: reporter.set_phase(label) reporter.set_progress(current=current, duration=duration, phase_elapsed=elapsed, done=done) return if not PROGRESS_ENABLED: return shown_current = current if done and duration is not None and duration > 0: shown_current = duration if duration is not None and duration > 0 and shown_current is not None: pct = max(0.0, min(100.0, shown_current / duration * 100.0)) text = f"{label:<22} {pct:6.1f}% media {_format_clock(shown_current)} / {_format_clock(duration)}" else: text = f"{label:<22} media {_format_clock(shown_current)}" if elapsed is not None and elapsed >= 0: text += f" elapsed {_format_clock(elapsed)}" speed_base = shown_current if shown_current is not None else duration if speed_base is not None and speed_base > 0 and elapsed > 0: text += f" {speed_base / elapsed:5.1f}x" if done: text += " done" # Progress lines deliberately carry no INFO or DEBUG tag. In the # verbose modes they are therefore indented. text = " " + text sys.stderr.write("\r" + text.ljust(120)) sys.stderr.flush() if done: sys.stderr.write("\n") sys.stderr.flush() def _with_ffmpeg_progress(argv: list[str], progress_path: Path) -> list[str]: # The FFmpeg file protocol does not decode %20 or %C3... URL escapes. # An explicit file: plus the unchanged absolute path also works # with spaces, umlauts and Windows drive letters. return [argv[0], "-progress", "file:" + progress_path.resolve().as_posix(), "-nostats", *argv[1:]] def run_ffmpeg_progress(argv: list[str], *, duration: float | None, label: str, workdir: Path | None = None) -> StepResult: """Runs FFmpeg and shows its machine-readable progress live. stderr is written to a file, so that analysis output is kept complete and no pipe can block. """ base = workdir or Path(tempfile.gettempdir()) base.mkdir(parents=True, exist_ok=True) fd_p, name_p = tempfile.mkstemp(prefix=f".{APP_NAME}-progress-", suffix=".txt", dir=base) os.close(fd_p) fd_e, name_e = tempfile.mkstemp(prefix=f".{APP_NAME}-stderr-", suffix=".txt", dir=base) os.close(fd_e) progress_path = Path(name_p) err_path = Path(name_e) actual = _with_ffmpeg_progress(argv, progress_path) log.debug("start %s", " ".join(shlex.quote(a) for a in actual)) started = time.monotonic() try: with open(err_path, "wb") as err_file: try: proc = subprocess.Popen(actual, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=err_file, shell=False) except FileNotFoundError as exc: raise ToolMissing(f"Program not found. {argv[0]}") from exc try: while proc.poll() is None: _show_progress(label, duration, _progress_seconds(progress_path), elapsed=time.monotonic() - started) time.sleep(PROGRESS_REFRESH) except KeyboardInterrupt: proc.terminate() try: proc.wait(timeout=5) except subprocess.TimeoutExpired: proc.kill() proc.wait() raise rc = proc.returncode elapsed = time.monotonic() - started current = _progress_seconds(progress_path) _show_progress(label, duration, current, elapsed=elapsed, done=True) media_seconds = duration if duration is not None and duration > 0 else current speed = (media_seconds / elapsed) if media_seconds is not None and elapsed > 0 else None log.info( "%s finished, elapsed %s%s", label, _format_clock(elapsed), f", {speed:.1f}x realtime" if speed is not None else "", ) telemetry = globals().get("TIMING") if telemetry is not None: telemetry.step(label, media_seconds, elapsed) stderr = err_path.read_text("utf-8", "replace") return StepResult(list(argv), rc, stderr, elapsed, "") finally: progress_path.unlink(missing_ok=True) err_path.unlink(missing_ok=True) # =========================================================================== # 0X Compact line output # =========================================================================== STATUS_COLUMN = 50 # 1-based target column for status and phase ANSI_YELLOW = "\033[33m" ANSI_RED = "\033[31m" ANSI_GREEN = "\033[32m" ANSI_PATTERN = re.compile(r"\033\[[0-9;]*m") ANSI_RESET = "\033[0m" def visible_width(text: str) -> int: """Column width without SGR colour codes. Combining characters do not count. Unicode grapheme clusters (for example emoji ZWJ) depend on the terminal. Live lines are limited conservatively, so that a previous phase does not wrap. """ return sum( 0 if unicodedata.combining(ch) or unicodedata.category(ch) == "Cf" else 2 if unicodedata.east_asian_width(ch) in ("W", "F") else 1 for ch in ANSI_PATTERN.sub("", text) ) def terminal_line(text: str, previous_width: int, *, final: bool = False) -> int: """Replaces a live line without fixed padding width or ANSI erase command.""" limit = terminal_columns() if limit is not None: previous_width = min(previous_width, limit) if not final and limit is not None and visible_width(text) > limit: parts: list[str] = [] width = 0 for token in re.findall(r"\033\[[0-9;]*m|[^\033]", text): columns = visible_width(token) if width + columns > max(0, limit - 1): break parts.append(token) width += columns text = "".join(parts) + "…" + (ANSI_RESET if ANSI_PATTERN.search(text) else "") width = visible_width(text) sys.stderr.write("\r" + text + " " * max(0, previous_width - width)) if final: sys.stderr.write("\n") sys.stderr.flush() return 0 if final else max(width, previous_width) # =========================================================================== # 0Z Runtime telemetry, remaining-time estimate and two-line live display # =========================================================================== TIMING_SCHEMA = 1 # Rates are seconds of processing time per second of media. The built-in # values come from real runs and are refined from the local timing log # once enough files have been recorded. PHASE_KEYS = ( ("Audio-Spool", "spool"), ("Analysis", "analysis"), ("Encoding", "encode"), ("True Peak", "peak"), ("Mux", "mux"), ("Timing", "verify"), ("Verification", "verify"), ("Full decode check", "verify"), ) def phase_key(label: str) -> str | None: for prefix, key in PHASE_KEYS: if label.startswith(prefix): return key return None def prescan_durations(tools, paths: list[Path]) -> list["JobItem"]: """Reads media duration and audio stream count for every input file. One ffprobe call per file, without decoding. It takes a fraction of a second and feeds the remaining-time estimate. A file without a readable duration gets no estimate. The prescan never fails a job. """ items: list[JobItem] = [] for path in paths: media: float | None = None tracks: int | None = None try: result = run([tools.ffprobe, "-hide_banner", "-v", "error", "-show_entries", "format=duration:stream=codec_type", "-of", "json", str(path)]) except Exception as exc: # noqa: BLE001 log.debug("Duration of %s not readable. %s", path.name, exc) else: if result.returncode == 0: try: data = json.loads(result.stdout or "{}") except json.JSONDecodeError: data = {} value = _as_float((data.get("format") or {}).get("duration")) if value and value > 0: media = value streams = data.get("streams") if isinstance(streams, list): tracks = sum(1 for entry in streams if isinstance(entry, dict) and entry.get("codec_type") == "audio") items.append(JobItem(path=path, media=media, tracks=tracks)) return items class TimingLog: """Minimal timing log in JSON Lines format. It records only what a later statistical adjustment of the remaining-time estimate needs. No file names, no paths, no titles, no languages. Records {"v":1,"k":"step","s":,"m":,"w":, ...} {"v":1,"k":"file","m":,"w":,"n":, ...} {"v":1,"k":"job","f":,"m":,"w":} """ def __init__(self, path: Path | None): self.path = path self.job = f"{random.getrandbits(32):08x}" self.failed = False self.file_index: int | None = None self.round_no = 0 def begin_file(self, index: int) -> None: self.file_index = int(index) self.round_no = 0 def end_file(self) -> None: self.file_index = None def _write(self, record: dict) -> None: if self.path is None or self.failed: return if self.file_index is not None and record.get("k") not in ("start", "job"): record = {"i": self.file_index, **record} record = {"v": TIMING_SCHEMA, "t": int(time.time()), "j": self.job, **record} try: self.path.parent.mkdir(parents=True, exist_ok=True) # Deliberately append: a log collects data over any number of # program starts. It is never emptied at start-up. with open(self.path, "a", encoding="utf-8") as handle: handle.write(json.dumps(record, separators=(",", ":"), sort_keys=True) + "\n") except OSError as exc: self.failed = True log.debug("Timing log not writable, skipped. %s", exc) def start(self) -> None: """Marks every program start in the continuous log.""" self._write({"k": "start", "av": VERSION}) def retry_round(self, round_no: int, track_count: int) -> None: """Records a peak retry round that is actually about to run. ``n`` is the number of tracks re-encoded in this round. This later distinguishes one parallel round over several tracks from several serial rounds. """ if round_no <= 0 or track_count <= 0: return # Remember the current round, so that the following step records # carry the same number and not only a yes-no flag. self.round_no = int(round_no) self._write({"k": "retry", "r": int(round_no), "n": int(track_count)}) def step(self, label: str, media: float | None, wall: float, **extra) -> None: key = phase_key(label) if key is None or not media or media <= 0 or wall <= 0: return # Encoding and peak measurements always carry a round number. The # first pass is r=0. Real retries take the round set earlier with # retry_round(). This keeps the schema unambiguous in the serial # single-track path, and later statistics need not interpret a missing r # in different ways. if "r" not in extra and key in ("encode", "peak"): extra["r"] = (self.round_no or 1) if "Retry" in label else 0 self._write({"k": "step", "s": key, "m": round(media, 3), "w": round(wall, 3), **extra}) def file(self, media: float | None, wall: float, **extra) -> None: if not media or media <= 0 or wall <= 0: return self._write({"k": "file", "m": round(media, 3), "w": round(wall, 3), **extra}) def job_summary(self, files: int, media: float, wall: float) -> None: if files <= 0 or wall <= 0: return self._write({"k": "job", "f": files, "m": round(media, 3), "w": round(wall, 3)}) class TimingModel: """Predicts job duration from media length, audio track count and turbo. Rates are seconds of processing time per second of media. The built-in values are derived from 402 successful real-world files processed with versions 0.6.3 to 0.6.7 (trimmed mean, 5 % cut on each side). Retries are right-skewed, so the mean rather than the median is used: the expected total of many files includes the average retry cost. A local timing log refines a group once it holds enough normal files. """ # (audio tracks, turbo) -> seconds of processing per second of media PRIOR_RATE = {(1, 0): 0.0980, (1, 1): 0.0770, (2, 0): 0.1234} # Share of each phase in the processing time, per audio track count. PRIOR_SHARE = { 1: {"analysis": 0.53, "encode": 0.29, "peak": 0.17, "mux": 0.01}, 2: {"analysis": 0.46, "encode": 0.34, "peak": 0.17, "spool": 0.02, "mux": 0.02}, } MIN_GROUP = 8 # local files needed before a group overrides the prior NORMAL_RATE = 0.03 # below this a file finished abnormally fast LIMIT = 20000 # newest log lines evaluated def __init__(self, path: Path | None): self.rate_table = dict(self.PRIOR_RATE) self.samples = 0 self._load(path) @staticmethod def trimmed_mean(values: list[float], cut: float = 0.05) -> float | None: if not values: return None ordered = sorted(values) drop = int(len(ordered) * cut) kept = ordered[drop:len(ordered) - drop] or ordered return sum(value / len(kept) for value in kept) @classmethod def is_normal(cls, record: dict) -> bool: """A file that ran its full processing. Instant failures and files without processed audio do not describe processing speed.""" media = record.get("m") wall = record.get("w") if not isinstance(media, (int, float)) or not isinstance(wall, (int, float)): return False tracks = record.get("n") if (isinstance(media, bool) or isinstance(wall, bool) or media <= 0 or wall <= 0 or isinstance(tracks, bool) or not isinstance(tracks, int) or tracks < 1): return False try: rate = wall / media return (record.get("ok") == 1 and math.isfinite(media) and math.isfinite(wall) and math.isfinite(rate) and rate > cls.NORMAL_RATE) except (OverflowError, ValueError): return False def _load(self, path: Path | None) -> None: if path is None or not path.is_file(): return try: with path.open("r", encoding="utf-8", errors="replace") as handle: lines = deque(handle, maxlen=self.LIMIT) except OSError: return groups: dict[tuple[int, int], list[float]] = {} for line in lines: line = line.strip() if not line: continue try: record = json.loads(line) except (ValueError, RecursionError): continue if not isinstance(record, dict): continue if record.get("v") != TIMING_SCHEMA or record.get("k") != "file": continue if not self.is_normal(record): continue key = (min(int(record["n"]), 2), 1 if record.get("tr") else 0) groups.setdefault(key, []).append(record["w"] / record["m"]) self.samples += 1 for key, values in groups.items(): if len(values) >= self.MIN_GROUP: value = self.trimmed_mean(values) if value is not None: self.rate_table[key] = value def rate(self, tracks: int | None, turbo: bool) -> float: """Processing seconds per media second for a file.""" mode = 1 if turbo else 0 one = self.rate_table.get((1, mode)) if one is None: # Turbo without data: scale the normal rate by the measured ratio. one = self.rate_table[(1, 0)] * (self.PRIOR_RATE[(1, 1)] / self.PRIOR_RATE[(1, 0)]) two = self.rate_table.get((2, mode)) if two is None: two = self.rate_table[(2, 0)] * (one / self.rate_table[(1, 0)]) count = max(1, int(tracks or 1)) if count == 1: return one # Each additional track beyond two costs as much as the second one. return two + (count - 2) * max(0.0, two - one) def expected_seconds(self, media: float | None, tracks: int | None, turbo: bool) -> float: if not media or media <= 0: return 0.0 if tracks is not None and tracks <= 0: return 0.0 return media * self.rate(tracks, turbo) def phase_share(self, tracks: int | None) -> dict[str, float]: return dict(self.PRIOR_SHARE[1 if (tracks or 1) <= 1 else 2]) @dataclass class JobItem: path: Path media: float | None = None tracks: int | None = None # audio streams found by the prescan wall: float | None = None done: bool = False normal: bool = False # finished with full processing class JobProgress: """Overall progress of a job across all input files. The job line stays on screen after the program ends. The remaining time combines the model prediction for the open files with a damped correction learned from the files already finished in this job. """ SHRINK = 3.0 # finished files needed for half the observed correction def __init__(self, items: list[JobItem], model: TimingModel, *, enabled: bool, turbo: bool = False): self.items = items self.model = model self.enabled = enabled self.turbo = turbo self.started = time.monotonic() self.index = 0 self.done_media = 0.0 self.done_wall = 0.0 self.current_fraction = 0.0 self.failed = 0 self.warned = 0 self._share: dict[str, float] = {} self._phase_done: set[str] = set() self._phase_now: str | None = None @property def total_media(self) -> float: return sum(i.media for i in self.items if i.media) def expected(self, item: JobItem) -> float: return self.model.expected_seconds(item.media, item.tracks, self.turbo) def observed_rate(self) -> float | None: if self.done_media > 0 and self.done_wall > 0: return self.done_wall / self.done_media return None def correction(self) -> float: """Damped ratio of actual to predicted time over normal finished files. Files that failed or finished abnormally fast are excluded. They say nothing about processing speed and would pull the estimate down. """ done = [i for i in self.items if i.done and i.normal and i.wall] predicted = sum(self.expected(i) for i in done) if not done or predicted <= 0: return 1.0 ratio = sum(i.wall for i in done) / predicted weight = len(done) / (len(done) + self.SHRINK) return 1.0 + (ratio - 1.0) * weight def start_file(self, index: int) -> None: self.index = index self.current_fraction = 0.0 self._share = self.model.phase_share(self.items[index].tracks) self._phase_done = set() self._phase_now = None def set_phase(self, label: str, fraction: float | None) -> None: key = phase_key(label) if key is None: return if self._phase_now and self._phase_now != key: self._phase_done.add(self._phase_now) self._phase_now = key base = sum(self._share.get(k, 0.0) for k in self._phase_done) running = self._share.get(key, 0.0) * (fraction if fraction is not None else 0.0) # A retry revisits encode and peak. Progress never runs backwards. self.current_fraction = max(self.current_fraction, min(0.99, base + running)) def finish_file(self, item: JobItem, wall: float, status: str) -> None: item.wall = wall item.done = True item.normal = (status != "Failed" and bool(item.media) and (item.tracks or 0) >= 1 and wall / item.media > TimingModel.NORMAL_RATE) if item.media: self.done_media += item.media self.done_wall += wall if status == "Failed": self.failed += 1 elif status == "Warning": self.warned += 1 self.current_fraction = 0.0 self._phase_done = set() self._phase_now = None def remaining_seconds(self) -> float | None: factor = self.correction() remaining = 0.0 known = False for position, item in enumerate(self.items): if item.done or not item.media: continue known = True share = (1.0 - self.current_fraction) if position == self.index else 1.0 remaining += self.expected(item) * factor * share return remaining if known else None def line(self, *, final: bool = False) -> str: elapsed = time.monotonic() - self.started total = len(self.items) done = sum(1 for i in self.items if i.done) parts = [f"Total {done}/{total}"] rate = self.observed_rate() if rate and rate > 0: parts.append(f"{1.0 / rate:5.1f}x") parts.append(f"Elapsed {_format_clock(elapsed)}") if not final: remaining = self.remaining_seconds() if remaining is not None: # "Remaining" is the estimated remaining duration of the whole # job. No additional end time is shown. parts.append(f"Remaining {_format_clock(remaining)}") else: media = sum(i.media for i in self.items if i.done and i.media) if media: parts.append(f"Media {_format_clock(media)}") if self.failed: parts.append(f"{self.failed} failed") if self.warned: parts.append(f"{self.warned} with warning") return " ".join(parts) def terminal_columns() -> int | None: """Usable console width, or None if it is unusable. Some environments report 0 or very small values. Without this bound the status line would shrink to a single ellipsis there. """ try: columns = os.get_terminal_size(sys.stderr.fileno()).columns except (OSError, AttributeError, ValueError): return None if columns < 20: return None return columns - 1 def fit_line(text: str, previous_width: int, *, clip: bool) -> tuple[str, int]: """Truncates a line to the terminal width and pads it to the previous width.""" limit = terminal_columns() if limit is not None: previous_width = min(previous_width, limit) if clip and limit is not None and visible_width(text) > limit: parts: list[str] = [] width = 0 for token in re.findall(r"\033\[[0-9;]*m|[^\033]", text): columns = visible_width(token) if width + columns > max(0, limit - 1): break parts.append(token) width += columns text = "".join(parts) + "…" + (ANSI_RESET if ANSI_PATTERN.search(text) else "") width = visible_width(text) return text + " " * max(0, previous_width - width), max(width, previous_width) class LiveBlock: """Two lines updated together at the bottom of the screen. The top line shows the current file, the bottom line the overall progress. Permanent lines are written above the block, and the block moves down. Without a terminal everything stays on one line and without control codes. """ def __init__(self, interactive: bool): self.interactive = interactive self.widths: list[int] = [] def _to_top(self) -> None: """Moves the cursor to the start of the first block line. draw already parks the cursor there, so a carriage return suffices. An additional cursor-up would go one line too far and overwrite the permanent line above. """ sys.stderr.write("\r") def _render(self, lines: list[str], *, clip: bool) -> list[str]: """Pads lines to the previous width. When the block shrinks, the freed lines below are explicitly overwritten.""" rendered: list[str] = [] widths: list[int] = [] for position in range(max(len(lines), len(self.widths))): line = lines[position] if position < len(lines) else "" previous = self.widths[position] if position < len(self.widths) else 0 text, width = fit_line(line, previous, clip=clip) rendered.append(text) widths.append(width if position < len(lines) else 0) self.widths = widths[:len(lines)] return rendered def draw(self, lines: list[str]) -> None: if not self.interactive or not lines: return self._to_top() rendered = self._render(lines, clip=True) sys.stderr.write("\n".join(rendered)) sys.stderr.write("\r") if len(rendered) > 1: sys.stderr.write("\033[A" * (len(rendered) - 1)) # The cursor now sits at the start of the first block line. Surplus # lines below have been cleared and no longer belong to the block. sys.stderr.flush() def emit(self, line: str) -> None: """Permanent line above the block.""" if not self.interactive: sys.stderr.write(line.rstrip() + "\n") sys.stderr.flush() return self._to_top() text, _ = fit_line(line, self.widths[0] if self.widths else 0, clip=False) # Keep the padding: exactly these spaces erase the remains of a # previously longer live line before the line becomes permanent. sys.stderr.write(text + "\n") sys.stderr.flush() # The block moves up one line. The widths of the remaining lines # must be kept, otherwise the rest of a previously longer line # stays on the screen. self.widths = self.widths[1:] def close(self, lines: list[str]) -> None: """Final state. Afterwards the block stays on screen permanently.""" if not self.interactive: for line in lines: sys.stderr.write(line.rstrip() + "\n") sys.stderr.flush() return self._to_top() rendered = self._render(lines, clip=False) sys.stderr.write("\n".join(rendered) + "\n") sys.stderr.flush() self.widths = [] LIVE_BLOCK: "LiveBlock | None" = None JOB_PROGRESS: "JobProgress | None" = None TIMING: "TimingLog | None" = None PHASE_NAMES = ( ("Audio-Spool", "Prepare"), ("Analysis", "Analysis"), ("Encoding", "Encoding"), ("True Peak", "Peak"), ("Mux", "Mux"), ("Timing", "Timing"), ("Verification", "Verify"), ) def _use_colour() -> bool: if os.environ.get("NO_COLOR"): return False return sys.stderr.isatty() def _phase_label(raw: str) -> str: for prefix, name in PHASE_NAMES: if raw.startswith(prefix): return name return raw.split()[0] if raw else "" class CompactReporter: """Exactly one status line per file in normal mode. In interactive mode the same line is updated during processing. It shows phase, progress, media position and elapsed time. The final line starts its status information in column 50 where possible. Long file names are not wrapped artificially. """ def __init__(self, position: int, total: int, name: str, *, enabled: bool): self.position = position self.total = total self.name = name self.enabled = bool(enabled) self.interactive = bool(enabled) and PROGRESS_ENABLED and sys.stderr.isatty() self.colour = _use_colour() self.started = time.monotonic() self.phase = "" self.percent: float | None = None self.media_current: float | None = None self.media_duration: float | None = None self.reason = "" self.dirty = False self.written_width = 0 self.raw_label = "" self.phase_started = self.started self.media_processed = 0.0 self.warnings: list[str] = [] self.track_gains: dict[int, float] = {} def prefix(self) -> str: if self.total <= 1: return "" width = len(str(self.total)) return f"({self.position:{width}d}/{self.total}) " def head(self) -> str: return self.prefix() + self.name def _compose(self, tail: str) -> str: head = self.head() # STATUS_COLUMN is 1-based: with 49 visible characters the # Tail in Zeichenposition 50. target = max(0, STATUS_COLUMN - 1) if visible_width(head) < target: head += " " * (target - visible_width(head)) else: head += " " return head + tail def _paint(self, text: str, colour: str) -> str: if not self.colour or not colour: return text return colour + text + ANSI_RESET def _write(self, text: str, *, final: bool = False) -> None: if not self.interactive: return block = globals().get("LIVE_BLOCK") job = globals().get("JOB_PROGRESS") if block is not None and job is not None and job.enabled: # Two lines: the running file at the top, the overall progress below. if final: block.emit(text) block.draw([job.line()]) else: block.draw([text, job.line()]) self.dirty = not final return self.written_width = terminal_line(text, self.written_width, final=final) self.dirty = not final def set_phase(self, raw_label: str) -> None: label = _phase_label(raw_label) reason = "" if "Retry" in raw_label: reason = raw_label.split("Retry", 1)[1].strip(" :-") if not reason: reason = "retry" self.raw_label = raw_label if label != self.phase: if self.media_current: self.media_processed += self.media_current self.phase = label self.phase_started = time.monotonic() self.percent = None self.media_current = None self.reason = reason elif reason: self.reason = reason job = globals().get("JOB_PROGRESS") if job is not None: job.set_phase(raw_label, None) def set_progress(self, *, current: float | None, duration: float | None, phase_elapsed: float | None = None, done: bool = False) -> None: self.media_duration = duration if done and duration is not None and duration > 0: self.media_current = duration self.percent = 100.0 else: self.media_current = current if duration is not None and duration > 0 and current is not None: self.percent = max(0.0, min(100.0, current / duration * 100.0)) else: self.percent = None job = globals().get("JOB_PROGRESS") if job is not None and self.raw_label: fraction = None if self.percent is None else self.percent / 100.0 job.set_phase(self.raw_label, 1.0 if done else fraction) self.render() def render(self) -> None: if not self.interactive: return elapsed = _format_clock(time.monotonic() - self.started) if self.percent is None: tail = f"{self.phase:<11} ..." else: tail = f"{self.phase:<11} {self.percent:5.1f}%" if self.media_current is not None: if self.media_duration is not None and self.media_duration > 0: tail += f" {_format_clock(self.media_current)}/{_format_clock(self.media_duration)}" else: tail += f" {_format_clock(self.media_current)}" tail += f" {elapsed}" speed = self.speed_text() if speed: tail += f" {speed}" if self.reason: tail += " " + self.reason gains = self.gain_text() if gains: tail += " " + gains self._write(self._compose(tail)) def speed_text(self, *, overall: bool = False) -> str: """Processing speed relative to the playing time of the material. During a step this is the speed of that step, as the verbose view shows it. In the final line it is the speed over the whole file. """ if overall: elapsed = time.monotonic() - self.started media = self.media_duration else: elapsed = time.monotonic() - self.phase_started media = self.media_current if not media or elapsed <= 0.5: return "" return f"{media / elapsed:.1f}x" def interrupt(self) -> None: # In compact mode the handler suppresses console logs during a job. # This method remains available for explicit output. if self.interactive and self.dirty: sys.stderr.write("\n") sys.stderr.flush() self.dirty = False self.written_width = 0 def note_warning(self, text: str) -> None: if text and text not in self.warnings: self.warnings.append(text) def set_track_gain(self, audio_index: int, gain_db: float | None) -> None: if gain_db is None or not math.isfinite(gain_db): return self.track_gains[int(audio_index)] = float(gain_db) if self.interactive: self.render() def gain_text(self) -> str: if not self.track_gains: return "" values = [self.track_gains[k] for k in sorted(self.track_gains)] return "/".join(f"{value:.1f}dB" for value in values) def finish(self, status: str, detail: str = "", reason: str = "") -> None: seconds = time.monotonic() - self.started colour = {"OK": ANSI_GREEN, "Warning": ANSI_YELLOW, "Failed": ANSI_RED}.get(status, "") tail = f"{self._paint(f'{status:<8}', colour)} {_format_clock(seconds)}" overall = self.speed_text(overall=True) if overall: tail += f" {overall}" gains = self.gain_text() if detail: tail += " " + detail if gains and gains not in detail: tail += " " + gains if reason: tail += " " + reason if not self.enabled: text_detail = detail if gains and gains not in text_detail: text_detail = (text_detail + " " if text_detail else "") + gains if reason: text_detail = (text_detail + " " if text_detail else "") + reason log_track_line("%s %s %s", self.head().strip(), status, text_detail) return line = self._compose(tail).rstrip() if self.interactive: self._write(line, final=True) else: sys.stderr.write(line + "\n") sys.stderr.flush() self.dirty = False REPORTER: CompactReporter | None = None class ReporterAwareHandler(logging.StreamHandler): """Ends the progress line before a message is written, and colours warnings yellow and errors red.""" def emit(self, record: logging.LogRecord) -> None: # In normal mode a running job owns exactly one line. All details # still go to the automatic job log. Before and after jobs, # warnings and errors remain visible as usual. if COMPACT_MODE and REPORTER is not None and REPORTER.enabled: return if _verbosity_required(record) > CONSOLE_VERBOSITY: return if REPORTER is not None: REPORTER.interrupt() try: message = self.format(record) if _use_colour(): if record.levelno >= logging.ERROR: message = ANSI_RED + message + ANSI_RESET elif record.levelno >= logging.WARNING: message = ANSI_YELLOW + message + ANSI_RESET self.stream.write(message + self.terminator) self.flush() except Exception: # pragma: no cover self.handleError(record) def _layout_short(layout: str) -> str: return {"stereo": "2.0", "mono": "1.0"}.get(layout, layout) def describe_conversion(plan: "JobPlan", args: argparse.Namespace) -> str: """Compact description of the actual output standard.""" if not plan.tracks: return "" names = {t.profile.name for t in plan.tracks} rates = {t.target_sample_rate for t in plan.tracks} qualities = {round(t.profile.nero_quality, 3) for t in plan.tracks} layouts = [_layout_short(t.downmix.source_layout) for t in plan.tracks] layout_text = "/".join(layouts) label = {"normal": "SQ", "sq": "SQ", "hq": "HQ"}.get( next(iter(names)).lower(), next(iter(names)).upper() ) if len(names) == 1 else "mixed" profile_default = len(names) == 1 and len(rates) == 1 and len(qualities) == 1 if profile_default: expected = PROFILES.get(next(iter(names)), {}) matches = ( expected.get("sample_rate") == next(iter(rates)) and abs(float(expected.get("nero_quality", -1)) - next(iter(qualities))) < 1e-6 ) if matches: non_stereo = any(x != "2.0" for x in layouts) base = f"{label} {layout_text}->2.0" if non_stereo else label return base + (" TURBO" if getattr(args, "turbo", False) else "") rate = "/".join(f"{r / 1000:g}kHz" for r in sorted(rates)) quality = "/".join(f"q{q:.2f}" for q in sorted(qualities)) text = f"m4a {quality}, {rate}, {layout_text}->2.0" return text + (" TURBO" if getattr(args, "turbo", False) else "") def describe_gains(plan: "JobPlan") -> str: values = [] for track in plan.tracks: if track.gain is None: values.append("?") else: values.append(f"{track.gain.total_gain:+.1f}") return "/".join(values) + " dB" if values else "" # =========================================================================== # 04 Tool discovery and capability check # =========================================================================== @dataclass(frozen=True) class Tools: ffmpeg: str ffprobe: str nero: str | None ffmpeg_version: str ffprobe_version: str def _script_directory() -> Path: """Directory of the started TrackShepherd file, not of the CWD.""" return Path(__file__).resolve().parent def _tool_is_executable(path: Path) -> bool: return path.is_file() and (os.name == "nt" or os.access(path, os.X_OK)) def _local_tool_candidates(name: str) -> list[Path]: """Local candidates for the current operating system only. Linux and POSIX deliberately ignore Windows .exe files. Windows prefers the .exe variant. The script directory takes precedence over PATH. """ base = _script_directory() if os.name == "nt": # Portable Windows tools are PE .exe files. A Linux ELF file without # extension next to them is deliberately ignored. names = [name] if name.lower().endswith(".exe") else [name + ".exe"] else: # A stray Windows version must neither mark a Linux run as ready # nor hide the system-wide native version. bare = name[:-4] if name.lower().endswith(".exe") else name names = [bare] exact = [base / candidate for candidate in names] wanted = {candidate.casefold() for candidate in names} folded: list[Path] = [] try: for entry in sorted(base.iterdir(), key=lambda item: item.name.casefold()): if entry.name.casefold() in wanted and entry not in exact: folded.append(entry) except OSError: pass return exact + folded def find_tool(explicit: str | None, name: str, required: bool) -> str | None: if explicit: path = Path(explicit).expanduser() if _tool_is_executable(path): return str(path.resolve()) found = shutil.which(explicit) if found: return found if required: raise ToolMissing(f"Specified program is not executable. {explicit}") return None # For a portable single-file application the script directory is the # first search location. The system-wide PATH follows after it. for candidate in _local_tool_candidates(name): if _tool_is_executable(candidate): return str(candidate) found = shutil.which(name) if found is None and os.name == "nt" and not name.lower().endswith(".exe"): found = shutil.which(name + ".exe") if found is None and required: raise ToolMissing( f"Program not found. Searched next to TrackShepherd and in PATH: {name}" ) return found def _version_line(path: str, display: str) -> str: result = run([path, "-hide_banner", "-version"]) if result.returncode != 0: raise ToolMissing(f"{display} cannot be started.") line = result.stdout.splitlines()[0] if result.stdout else "" if not line: raise ToolMissing(f"{display} returns no version information.") return line def check_ffmpeg_capabilities(ffmpeg: str) -> None: result = run_checked([ffmpeg, "-hide_banner", "-filters"]) available = set() for line in result.stdout.splitlines(): parts = line.split() if len(parts) >= 2 and parts[0] and all(c in ".TSCAVN|" for c in parts[0]): available.add(parts[1]) missing = sorted(REQUIRED_FILTERS - available) if missing: raise ToolMissing("FFmpeg lacks required filters: " + ", ".join(missing)) def check_nero(path: str) -> str: binary = Path(path) try: result = run([path]) except ToolMissing: if os.name != "nt" and binary.is_file() and os.access(binary, os.X_OK): raise ToolMissing(NERO_32BIT_HINT) from None raise text = (result.stdout or "") + (result.stderr or "") if "Nero AAC Encoder" not in text: raise ToolMissing(f"The program at {path} does not identify itself as Nero AAC Encoder.") for line in text.splitlines(): if "Package version" in line: # The Nero banner is framed with asterisks. Without clean-up the # right frame ends up in the displayed version number. return line.split(":", 1)[-1].strip().strip("*").strip() return "unknown" @dataclass(frozen=True) class DependencyStatus: name: str ok: bool version: str path: str | None detail: str = "" def _compact_version(line: str) -> str: """Keeps the meaningful first version line without unnecessary length.""" return line.strip() or "unknown" def dependency_status(ffmpeg_override: str | None = None, ffprobe_override: str | None = None, nero_override: str | None = None) -> list[DependencyStatus]: """Diagnostics without abort. Used by --help and --version.""" result: list[DependencyStatus] = [ DependencyStatus( "Python", sys.version_info >= (3, 10), f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}", sys.executable, "requires >= 3.10", ) ] ffmpeg = find_tool(ffmpeg_override, "ffmpeg", False) if ffmpeg is None: result.append(DependencyStatus("ffmpeg", False, "missing", None, "not found next to TrackShepherd or in PATH")) else: try: version = _compact_version(_version_line(ffmpeg, "ffmpeg")) check_ffmpeg_capabilities(ffmpeg) result.append(DependencyStatus("ffmpeg", True, version, ffmpeg, "required filters present")) except (ToolMissing, StepFailed) as exc: result.append(DependencyStatus("ffmpeg", False, "not usable", ffmpeg, str(exc))) ffprobe = find_tool(ffprobe_override, "ffprobe", False) if ffprobe is None: result.append(DependencyStatus("ffprobe", False, "missing", None, "not found next to TrackShepherd or in PATH")) else: try: version = _compact_version(_version_line(ffprobe, "ffprobe")) result.append(DependencyStatus("ffprobe", True, version, ffprobe)) except (ToolMissing, StepFailed) as exc: result.append(DependencyStatus("ffprobe", False, "not usable", ffprobe, str(exc))) nero = find_tool(nero_override, "neroAacEnc", False) if nero is None: result.append(DependencyStatus("neroAacEnc", False, "missing", None, "not found next to TrackShepherd or in PATH")) else: try: version = check_nero(nero) result.append(DependencyStatus("neroAacEnc", True, version, nero)) except (ToolMissing, StepFailed) as exc: result.append(DependencyStatus("neroAacEnc", False, "not usable", nero, str(exc))) return result def print_dependency_report(ffmpeg_override: str | None = None, ffprobe_override: str | None = None, nero_override: str | None = None) -> bool: statuses = dependency_status(ffmpeg_override, ffprobe_override, nero_override) print("Dependencies") for item in statuses: state = "OK" if item.ok else "MISSING/ERROR" where = f" [{item.path}]" if item.path else "" print(f" {item.name:<11} {state:<12} {item.version}{where}") if item.detail and not item.ok: print(f" {item.detail}") ready = all(item.ok for item in statuses) print(f" Processing {'ready' if ready else 'NOT ready'}") return ready def _early_tool_overrides(argv: list[str]) -> tuple[str | None, str | None, str | None]: """Reads tool paths only, for the early --help and --version diagnostics.""" parser = argparse.ArgumentParser(add_help=False) parser.add_argument("--ffmpeg") parser.add_argument("--ffprobe") parser.add_argument("--nero") known, _ = parser.parse_known_args(argv) return known.ffmpeg, known.ffprobe, known.nero def discover_tools(args: argparse.Namespace, need_nero: bool) -> Tools: ffmpeg = find_tool(args.ffmpeg, "ffmpeg", True) ffprobe = find_tool(args.ffprobe, "ffprobe", True) assert ffmpeg is not None and ffprobe is not None ffmpeg_version = _version_line(ffmpeg, "ffmpeg") ffprobe_version = _version_line(ffprobe, "ffprobe") check_ffmpeg_capabilities(ffmpeg) nero = find_tool(args.nero, "neroAacEnc", need_nero) if need_nero and nero is None: raise ToolMissing("neroAacEnc was not found. Specify the path with --nero.") if nero is not None and (need_nero or args.nero): version = check_nero(nero) log.info("neroAacEnc found, version %s", version) return Tools(ffmpeg, ffprobe, nero, ffmpeg_version, ffprobe_version) # =========================================================================== # 05 Source analysis and timing model # =========================================================================== def _as_float(value: Any) -> float | None: try: result = float(value) except (TypeError, ValueError): return None if not math.isfinite(result): return None return result def _as_int(value: Any) -> int | None: try: return int(value) except (TypeError, ValueError): return None def raw_tags(mapping: Any) -> dict[str, str]: if not isinstance(mapping, dict): return {} return {str(k): str(v) for k, v in mapping.items()} def tag_value(tags: dict[str, str], key: str) -> str | None: wanted = key.casefold() for name, value in tags.items(): if name.casefold() == wanted: return value return None def canonical_language(raw: str | None) -> str | None: if not raw: return None key = raw.strip().lower() if key in ("and", "unknown", ""): return None return LANG_CANONICAL.get(key, key if len(key) == 2 else key) def parse_sar(value: Any) -> tuple[int | None, int | None]: if not isinstance(value, str) or ":" not in value: return None, None left, _, right = value.partition(":") num = _as_int(left) den = _as_int(right) if num is None or den is None or den == 0: return None, None return num, den def _first_number(text: str) -> float | None: for line in text.splitlines(): value = _as_float(line.strip().rstrip(",")) if value is not None: return value return None def container_family(format_name: str | None) -> str: if not format_name: return "unknown" names = {part.strip().lower() for part in format_name.split(",")} if names & set(MATROSKA_FORMATS): return "matroska" if names & set(MP4_FORMATS): return "mp4" return "other" def probe_time_candidates(tools: Tools, path: Path, selector: str, start_time: float | None, initial_padding: int | None, sample_rate: int | None) -> RawTiming: presentation: float | None = None packet_pts: float | None = None skip_samples: int | None = None discard_padding: int | None = None # The earlier probe requested best_effort_timestamp_time from exactly one # packet. For older or unusually muxed Matroska audio tracks ffprobe may # return no value there, although the decoder has a valid frame PTS. # The source timing was then marked unverifiable without need. # "not-verifiable" is discarded. # # We therefore read several initial packets, also request the real # frame PTS, and use only an actually decoded frame as the # presentation anchor. Packet PTS alone is still NOT promoted to a # verified content anchor. Fail-closed therefore stays in place. # frames = run([ tools.ffprobe, "-hide_banner", "-v", "error", "-select_streams", selector, "-show_frames", "-show_entries", "frame=best_effort_timestamp_time,pts_time", "-read_intervals", "%+#16", "-of", "json", str(path), ]) if frames.returncode == 0: try: frame_payload = json.loads(frames.stdout or "{}") except json.JSONDecodeError: frame_payload = {} frame_entries = ( frame_payload.get("frames") if isinstance(frame_payload.get("frames"), list) else [] ) for frame in frame_entries: if not isinstance(frame, dict): continue presentation = _as_float(frame.get("best_effort_timestamp_time")) if presentation is None: presentation = _as_float(frame.get("pts_time")) if presentation is not None: break # Second, independent decoder fallback for audio: ashowinfo sees the # PTS of the PCM frame that is also fed to the later transcode. # -copyts is mandatory, otherwise ffmpeg normalises the first frame to 0 # and destroys exactly the source offset we want to measure. if presentation is None and selector.startswith("a:"): decoded = run([ tools.ffmpeg, "-hide_banner", "-v", "info", "-nostdin", "-copyts", "-i", str(path), "-map", f"0:{selector}", "-frames:a", "1", "-af", "ashowinfo", "-f", "null", "-", ]) if decoded.returncode == 0: match = re.search( r"\bpts_time:([+-]?(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][+-]?\d+)?)", decoded.stderr, ) if match: presentation = _as_float(match.group(1)) packets = run([ tools.ffprobe, "-hide_banner", "-v", "error", "-select_streams", selector, "-show_packets", "-read_intervals", "%+#1", "-of", "json", str(path), ]) if packets.returncode == 0: try: payload = json.loads(packets.stdout or "{}") except json.JSONDecodeError: payload = {} entries = payload.get("packets") if isinstance(payload.get("packets"), list) else [] if entries and isinstance(entries[0], dict): first = entries[0] packet_pts = _as_float(first.get("pts_time")) for side in first.get("side_data_list") or []: if not isinstance(side, dict): continue if str(side.get("side_data_type", "")).lower().startswith("skip samples"): skip_samples = _as_int(side.get("skip_samples")) discard_padding = _as_int(side.get("discard_padding")) return RawTiming( presentation=presentation, start_time=start_time, packet_pts=packet_pts, initial_padding=initial_padding, skip_samples=skip_samples, discard_padding=discard_padding, sample_rate=sample_rate, ) def compare_time_candidates(left: RawTiming, right: RawTiming) -> AnchorComparison: for method in ANCHOR_METHODS: lv = left.get(method) rv = right.get(method) if lv is not None and rv is not None: return AnchorComparison(lv - rv, method) return AnchorComparison(None, "unknown") def is_zero_delay_audio(codec: str | None) -> bool: if not codec: return False name = codec.lower() return name in ZERO_DELAY_AUDIO_CODECS or name.startswith("pcm_") def resolve_content_anchor(family: str, raw: RawTiming, kind: str = "audio", codec: str | None = None, origin: str = "external") -> ContentAnchor: if family == "matroska": padding = raw.initial_padding or 0 if padding: if not raw.sample_rate: correction, provenance, confidence = 0.0, "raw-start_time", "not-verifiable" else: correction = padding / raw.sample_rate provenance, confidence = "matroska-codecdelay", "verified" else: correction = 0.0 if kind == "video": provenance, confidence = "matroska-video-no-delay", "verified" elif kind == "audio" and raw.presentation is not None: # A missing CodecDelay does NOT prove that an MP3 or AAC codec has no # priming. For a transcode, however, the timestamp of the actually # decoded PCM start is directly observable, and exactly this PCM # stream is processed further. Historical Matroska material without # CodecDelay therefore keeps its existing playback timing, # without guessing a codec delay. return ContentAnchor(raw.presentation, "presentation", "matroska-decoded-presentation-no-codecdelay", "verified", 0.0) elif is_zero_delay_audio(codec): provenance, confidence = "matroska-zero-delay-codec", "verified" else: provenance, confidence = "matroska-no-codecdelay", "not-verifiable" for method in ANCHOR_METHODS: value = raw.get(method) if value is not None: return ContentAnchor(value + correction, method, provenance, confidence, correction) return UNKNOWN_CONTENT if family == "mp4": skip = raw.skip_seconds() if raw.packet_pts is not None and skip is not None: return ContentAnchor(raw.packet_pts + skip, "packet_pts", "mp4-skip-samples", "verified", skip) if raw.presentation is not None: return ContentAnchor(raw.presentation, "presentation", "mp4-presentation", "verified", 0.0) if raw.start_time is not None: confidence = "verified" if origin == "generated" else "not-verifiable" return ContentAnchor(raw.start_time, "start_time", "mp4-start_time", confidence, 0.0) return UNKNOWN_CONTENT for method in ANCHOR_METHODS: value = raw.get(method) if value is not None: return ContentAnchor(value, method, "raw-" + method, "not-verifiable", 0.0) return UNKNOWN_CONTENT def mux_offset_for(track_content: ContentAnchor, reference_content: ContentAnchor) -> float | None: if track_content.seconds is None or reference_content.seconds is None: return None if track_content.confidence != "verified" or reference_content.confidence != "verified": return None return track_content.seconds - reference_content.seconds def mux_input_offset_for(track_content: ContentAnchor, source_container_start: float | None) -> float | None: """Start value for the input offset of the separately encoded replacement track. FFmpeg shifts every input by its own container start time, in both directions. The source container therefore does not stay on its time axis. The replacement track must be placed at the content anchor of its source track minus the container start time. Measured directly with neroAacEnc 1.5.4.0 from the anchor fields of the finished file, without a decoding heuristic: Case S A A A-max(S,0) A-S base +0.000 +0.000 ok ok ok audio +100 ms +0.000 +0.100 ok ok ok video +120 ms +0.000 +0.000 ok ok ok global +500 ms +0.500 +0.500 +499.7 ms ok ok S=+0.3 A=+0.4 +0.320 +0.400 +319.7 ms ok ok AC-3 CodecDelay -0.005 +0.000 -5.7 ms -5.7 ms ok MP3 CodecDelay -0.023 +0.000 -23.4 ms -23.4 ms ok The max(S, 0) variant from 0.3.7 relied on a faulty comparison measurement that counted the encoder priming twice. It is refuted. The measurement on the finished file remains authoritative. """ if track_content.seconds is None or track_content.confidence != "verified": return None return track_content.seconds - (source_container_start or 0.0) def evaluate_encoder_timing(tools: Tools, path: Path) -> EncoderTiming: media = probe_media(tools, path, full_time_anchors=True, origin="generated") audio = media.audio() if not audio: return EncoderTiming(RawTiming(), None, None, "none", "unverified") raw = audio[0].time if raw.skip_samples is not None and raw.skip_samples > 0: seconds = raw.skip_seconds() return EncoderTiming(raw, raw.skip_samples, seconds, "mp4-skip-samples", "verified") if raw.initial_padding is not None and raw.initial_padding > 0: seconds = raw.padding_seconds() return EncoderTiming(raw, raw.initial_padding, seconds, "container-padding", "verified") if (raw.start_time is not None and raw.packet_pts is not None and raw.start_time > raw.packet_pts): seconds = raw.start_time - raw.packet_pts samples = int(round(seconds * raw.sample_rate)) if raw.sample_rate else None return EncoderTiming(raw, samples, seconds, "mp4-edit-list", "verified") return EncoderTiming(raw, None, None, "none", "unverified") def probe_media(tools: Tools, path: Path, *, full_time_anchors: bool = True, origin: str = "external") -> MediaFile: argv = [ tools.ffprobe, "-hide_banner", "-v", "error", "-print_format", "json", "-show_streams", "-show_format", "-show_chapters", str(path), ] result = run_checked(argv) try: data = json.loads(result.stdout or "{}") except json.JSONDecodeError as exc: raise StepFailed(f"ffprobe returned no valid JSON. {path}", result) from exc fmt = data.get("format") if isinstance(data.get("format"), dict) else {} entries = data.get("streams") if isinstance(data.get("streams"), list) else [] raw_format = fmt.get("format_name") family = container_family(str(raw_format) if raw_format is not None else None) streams: list[StreamInfo] = [] audio_counter = 0 type_counters: dict[str, int] = {} for entry in entries: if not isinstance(entry, dict): continue index = _as_int(entry.get("index")) kind = entry.get("codec_type") codec = entry.get("codec_name") if index is None: raise StepFailed("ffprobe stream without a valid stream index.") if not isinstance(kind, str) or not kind: kind = "unknown" if not isinstance(codec, str) or not codec: codec = "unknown" tags = raw_tags(entry.get("tags")) language_raw = tag_value(tags, "language") start_time = _as_float(entry.get("start_time")) stream_rate = _as_int(entry.get("sample_rate")) stream_padding = _as_int(entry.get("initial_padding")) selector_pos = type_counters.get(kind, 0) type_counters[kind] = selector_pos + 1 selector_letter = {"video": "v", "audio": "a"}.get(kind) if full_time_anchors and selector_letter is not None: time_info = probe_time_candidates( tools, path, f"{selector_letter}:{selector_pos}", start_time, stream_padding, stream_rate, ) else: time_info = RawTiming(start_time=start_time, initial_padding=stream_padding, sample_rate=stream_rate) content_info = resolve_content_anchor(family, time_info, kind, codec, origin) common = dict( index=index, kind=kind, codec=codec, tags_raw=tags, disposition={str(k): bool(v) for k, v in (entry.get("disposition") or {}).items()}, language_raw=language_raw, language_canonical=canonical_language(language_raw), title=tag_value(tags, "title"), time=time_info, content=content_info, duration=_as_float(entry.get("duration")), bit_rate=_as_int(entry.get("bit_rate")), ) if kind == "video": sar_num, sar_den = parse_sar(entry.get("sample_aspect_ratio")) streams.append(VideoInfo( **common, width=_as_int(entry.get("width")), height=_as_int(entry.get("height")), sar_num=sar_num, sar_den=sar_den, )) elif kind == "audio": channels = _as_int(entry.get("channels")) sample_rate = _as_int(entry.get("sample_rate")) if channels is None or channels <= 0: raise StepFailed(f"Audio track {audio_counter} without a valid channel count.") if sample_rate is None or sample_rate <= 0: raise StepFailed(f"Audio track {audio_counter} without a valid sample rate.") layout = entry.get("channel_layout") if not isinstance(layout, str) or layout in ("", "unknown"): layout = None layout_source = "unknown" else: layout_source = "reported" streams.append(AudioInfo( **common, audio_index=audio_counter, channels=channels, layout=layout, layout_source=layout_source, sample_rate=sample_rate, )) audio_counter += 1 else: streams.append(StreamInfo(**common)) return MediaFile( path=path, format_name=str(fmt.get("format_name")) if fmt.get("format_name") is not None else None, duration=_as_float(fmt.get("duration")), start_time=_as_float(fmt.get("start_time")), streams=tuple(streams), chapter_count=len(data.get("chapters") or []) if isinstance(data.get("chapters"), list) else 0, ) def packet_payload_hashes(tools: Tools, path: Path, selector: str, limit: int | None = None) -> list[str] | None: """Reads packet payload hashes. Missing hash values are no proof of identity.""" argv = [ tools.ffprobe, "-hide_banner", "-v", "error", "-select_streams", selector, "-show_packets", "-show_data_hash", PACKET_HASH_ALGORITHM, "-of", "json", ] if limit: argv += ["-read_intervals", f"%+#{limit}"] argv.append(str(path)) result = run(argv) if result.returncode != 0: return None try: payload = json.loads(result.stdout or "{}") except json.JSONDecodeError: return None packets = payload.get("packets") if not isinstance(packets, list) or not packets: return None values: list[str] = [] for item in packets: if not isinstance(item, dict): return None value = item.get("data_hash") if not isinstance(value, str) or not value: return None values.append(value) return values[:limit] if limit else values def packet_timeline_stats(tools: Tools, path: Path, selector: str | None = None) -> PacketTimelineStats | None: """First and last occupied packet time of a container or single stream. Used only as a verification fallback. No decoding takes place. ``selector`` uses ffprobe syntax, for example ``v:0`` or ``a:0``. """ argv = [tools.ffprobe, "-hide_banner", "-v", "error"] if selector: argv += ["-select_streams", selector] argv += [ "-show_packets", "-show_entries", "packet=pts_time,dts_time,duration_time", "-of", "compact=p=0:nk=0", str(path), ] log.debug("start %s", " ".join(shlex.quote(a) for a in argv)) try: proc = subprocess.Popen( argv, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, encoding="utf-8", errors="replace", shell=False, ) except OSError: return None if proc.stdout is None: proc.kill() proc.wait() return None first: float | None = None last: float | None = None count = 0 payload = 0.0 max_gap = 0.0 previous_end: float | None = None missing_durations = 0 try: for raw_line in proc.stdout: if not raw_line.strip(): continue fields: dict[str, str] = {} for token in raw_line.strip().split("|"): if "=" in token: key, value = token.split("=", 1) fields[key] = value pts = _as_float(fields.get("pts_time")) dts = _as_float(fields.get("dts_time")) packet_start = pts if pts is not None else dts if packet_start is None or not math.isfinite(packet_start): missing_durations += 1 continue duration = _as_float(fields.get("duration_time")) if duration is None or not math.isfinite(duration) or duration <= 0: duration = 0.0 missing_durations += 1 packet_end = packet_start + duration first = packet_start if first is None else min(first, packet_start) last = packet_end if last is None else max(last, packet_end) payload += duration if previous_end is not None: max_gap = max(max_gap, packet_start - previous_end) previous_end = packet_end count += 1 except BaseException: _terminate_processes([proc]) raise finally: proc.stdout.close() rc = proc.wait() if rc != 0 or first is None or last is None or last < first or count <= 0: return None return PacketTimelineStats(first, last, count, payload, max_gap, missing_durations) def packet_timeline_span(tools: Tools, path: Path) -> float | None: stats = packet_timeline_stats(tools, path) return stats.span if stats is not None else None @dataclass(frozen=True) class AudioFrameTimeline: first: float last: float samples: int rate: int frames: int min_drift: float max_drift: float min_step: float @property def decoded_duration(self) -> float: return self.samples / self.rate @property def span(self) -> float: return self.last - self.first def summarize_audio_frames(lines, rate: int) -> AudioFrameTimeline: """Measure occupied presentation times, independently of the encoding pipe.""" first = last = None samples = frames = 0 min_drift = max_drift = min_step = 0.0 if rate <= 0: raise StepFailed("Source sample rate is unavailable for the frame timeline.") for line in lines: fields = dict(token.split("=", 1) for token in line.strip().split("|") if "=" in token) if "nb_samples" not in fields: continue count = _as_int(fields.get("nb_samples")) pts = _as_float(fields.get("best_effort_timestamp_time")) if pts is None: pts = _as_float(fields.get("pts_time")) if pts is None or count is None or count <= 0: raise StepFailed("Source audio frame has no usable sample count or timestamp.") if first is None: first = pts drift = pts - first - samples / rate min_drift = min(min_drift, drift) max_drift = max(max_drift, drift) if last is not None: min_step = min(min_step, pts - last) last = pts + count / rate samples += count frames += 1 if first is None or last is None or frames == 0: raise StepFailed("Source audio frame timeline is unavailable.") return AudioFrameTimeline(first, last, samples, rate, frames, min_drift, max_drift, min_step) def prepare_audio_timeline(tools: Tools, path: Path, track: TrackPlan) -> None: """Preserve forward gaps; reject overlapping frames instead of dropping samples.""" argv = [tools.ffprobe, "-v", "error", "-select_streams", f"a:{track.source.audio_index}", "-show_frames", "-show_entries", "frame=pts_time,best_effort_timestamp_time,nb_samples", "-of", "compact=p=0:nk=0", str(path)] log.debug("start %s", " ".join(shlex.quote(a) for a in argv)) started = time.monotonic() label = f"Timing source audio {track.source.audio_index}" _show_progress(label, None, None, elapsed=0.0) with tempfile.TemporaryFile(mode="w+t", encoding="utf-8") as err: proc = subprocess.Popen(argv, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=err, text=True, encoding="utf-8", errors="replace") try: timeline = summarize_audio_frames(proc.stdout, track.source.sample_rate or 0) rc = proc.wait() except BaseException: _terminate_processes([proc]) raise finally: if proc.stdout is not None: proc.stdout.close() err.seek(0) diagnostics = err.read() elapsed = time.monotonic() - started _show_progress(label, timeline.span, timeline.span, elapsed=elapsed, done=True) if rc != 0: raise StepFailed(f"Source audio timeline scan failed for track {track.source.audio_index}.") if min(timeline.min_drift, timeline.min_step) < -TIMELINE_TOLERANCE: raise StepFailed(f"Audio {track.source.audio_index}: overlapping or backward source " "timestamps; automatic sample removal is not allowed.") anchor = track.source.content.seconds if anchor is None or abs(timeline.first + track.source.content.correction - anchor) > TIMELINE_TOLERANCE: raise StepFailed(f"Audio {track.source.audio_index}: full frame scan disagrees with the source anchor.") track.source_timeline = timeline track.preserve_gaps = timeline.max_drift > TIMELINE_TOLERANCE if diagnostics.strip(): log.warning("Audio %d: source decoder reports damaged data; only decodable frames " "can be retained. Details are in the job log.", track.source.audio_index) log.debug("Source decoder diagnostics:\n%s", diagnostics) if track.preserve_gaps: log.warning("Audio %d: preserving source timestamp gaps with silence; " "PCM %.6f s, occupied span %.6f s, maximum drift %.3f ms.", track.source.audio_index, timeline.decoded_duration, timeline.span, timeline.max_drift * 1000) telemetry = globals().get("TIMING") if telemetry is not None: telemetry.step(label, timeline.span, elapsed) def _decoded_program_duration(track: TrackPlan) -> float | None: """Duration of analysed stereo PCM, or the audited source when gain is off.""" if track.analysis is not None and track.analysis.n_samples > 0: return track.analysis.n_samples / (2.0 * max(track.target_sample_rate, 1)) timeline = track.source_timeline if timeline is None: return None return timeline.span if track.preserve_gaps else timeline.decoded_duration def _malformed_processed_audio_tail_is_safe( tools: Tools, plan: JobPlan, output: MediaFile, output_path: Path, allowed: float) -> str | None: """Proves a faulty timestamp tail of the source audio track. A large global duration deviation is accepted only if (1) the stream-copied video keeps the same real packet span, (2) the generated AAC track matches the actually decoded PCM duration, (3) the source audio packets run clearly beyond the picture, while the decoded content ends at about the same time as the picture, and (4) the packet-duration sum matches raw decoded samples within the codec framing allowance, and the packet span exceeds that sum by the tail, (5) analysed PCM matches the complete audited frame timeline, and (6) every copied video packet matches. Packet durations are metadata, not an independent parse of codec payload. This is a bounded consistency check; decoder damage is reported separately. An internal gap is preserved by the resampler rather than collapsed. """ if not plan.tracks or not plan.source.video() or not output.video(): return None src_v = packet_timeline_stats(tools, plan.source.path, "v:0") out_v = packet_timeline_stats(tools, output_path, "v:0") if src_v is None or out_v is None: return None if abs(src_v.span - out_v.span) > allowed: return None output_audio = {a.audio_index: a for a in output.audio()} tail_notes: list[str] = [] saw_bad_tail = False endpoint_tolerance = max(0.50, allowed * 3.0) audio_duration_tolerance = max(0.25, allowed * 2.0) for track in plan.tracks: target = output_audio.get(track.source.audio_index) decoded = _decoded_program_duration(track) if target is None or decoded is None or track.source.content.seconds is None: return None src_a = packet_timeline_stats( tools, plan.source.path, f"a:{track.source.audio_index}" ) out_a = packet_timeline_stats( tools, output_path, f"a:{target.audio_index}" ) if src_a is None or out_a is None: return None priming = 0.0 if track.encoder_timing is not None and track.encoder_timing.priming_seconds: priming = max(0.0, track.encoder_timing.priming_seconds) expected_out_span = decoded + priming if abs(out_a.span - expected_out_span) > audio_duration_tolerance: return None timeline = track.source_timeline if timeline is None or src_a.missing_durations: return None expected_pcm = timeline.span if track.preserve_gaps else timeline.decoded_duration if abs(decoded - expected_pcm) > TIMELINE_TOLERANCE + 1 / track.target_sample_rate: return None decoded_end = track.source.content.seconds + decoded raw_tail = src_a.last - decoded_end video_end_delta = decoded_end - src_v.last if raw_tail > 1.0: # A real audio ending beyond the picture must not be reinterpreted # as a broken timestamp. if abs(video_end_delta) > endpoint_tolerance: return None # Packet-duration consistency is an additional bounded check. # A timestamp jump inflates the span but not the duration sum. # Malformed packets can still carry a reported duration. unbacked = src_a.span - src_a.payload if unbacked < raw_tail - audio_duration_tolerance: # The tail is backed by frames. If they were not decoded, # audio would be lost. Fail closed. return None # Compare raw decoded samples, not silence inserted into PTS gaps. # Otherwise real frames exist that did not reach the output. if abs(timeline.decoded_duration - src_a.payload) > audio_duration_tolerance: return None saw_bad_tail = True tail_notes.append( f"audio {track.source.audio_index}: packet end {src_a.last:.3f} s, " f"decoded end {decoded_end:.3f} s, tail {raw_tail:.3f} s, " f"packet duration sum {src_a.payload:.3f} s, largest timestamp gap " f"{src_a.max_gap:.3f} s" ) if not saw_bad_tail: return None # A duration exception requires every copied video packet, even in turbo. for index in range(len(plan.source.video())): if not streamcopy_is_identical(tools, plan.source.path, f"v:{index}", output_path, f"v:{index}", limit=None): return None return ( f"video packet span source/output {src_v.span:.3f}/{out_v.span:.3f} s; " + "; ".join(tail_notes) ) def streamcopy_is_identical(tools: Tools, source: Path, source_selector: str, target: Path, target_selector: str, limit: int | None = PACKET_HASH_SAMPLE) -> bool: """Proves identical packet payloads. For timing probes a limited sample suffices by default. For the final proof of newly generated AAC tracks ``limit=None`` is used, which compares the complete track. """ left = packet_payload_hashes(tools, source, source_selector, limit) right = packet_payload_hashes(tools, target, target_selector, limit) if not left or not right: return False return left == right def recover_output_content_anchor(tools: Tools, output: MediaFile, target: AudioInfo, encoded_path: Path | None, timing: EncoderTiming | None) -> ContentAnchor: if target.time.initial_padding or target.time.skip_samples: return target.content if timing is None or timing.status != "verified": return ContentAnchor(None, target.content.method, "encoder-timing-unverified", "not-verifiable", 0.0) # The skip and padding samples belong to the generated AAC track. Because # it is stream-copied, encoder file and target track must report the same # sample rate. A contradiction must never be converted into seconds with a # silently chosen rate. This area was suspected as a source of error # after the 0.3.7 review. encoder_rate = timing.raw.sample_rate if encoder_rate and target.sample_rate and encoder_rate != target.sample_rate: return ContentAnchor(None, target.content.method, f"encoder-output-samplerate-mismatch-{encoder_rate}-{target.sample_rate}", "not-verifiable", 0.0) if timing.priming_samples is not None and timing.priming_samples > 0: rate = target.sample_rate or encoder_rate if not rate: return ContentAnchor(None, target.content.method, "encoder-priming-rate-missing", "not-verifiable", 0.0) priming_seconds = timing.priming_samples / rate else: priming_seconds = timing.priming_seconds if priming_seconds is None: return ContentAnchor(None, target.content.method, "encoder-timing-unverified", "not-verifiable", 0.0) if encoded_path is None: return ContentAnchor(None, target.content.method, "encoded-file-missing", "not-verifiable", 0.0) identical = streamcopy_is_identical(tools, encoded_path, "a:0", output.path, f"a:{target.audio_index}") if not identical: return ContentAnchor(None, target.content.method, "streamcopy-not-proven", "not-verifiable", 0.0) if target.content.seconds is None: return ContentAnchor(None, target.content.method, "container-anchor-missing", "not-verifiable", 0.0) return ContentAnchor( target.content.seconds + priming_seconds, target.content.method, "recovered-from-" + timing.signalling, "verified", priming_seconds, ) def verify_output_timeline(tools: Tools, output: MediaFile, plan: JobPlan, tolerance: float = TIMELINE_TOLERANCE) -> list[TimelineCheck]: checks: list[TimelineCheck] = [] out_reference = output.video()[0] if output.video() else (output.audio()[0] if output.audio() else None) src_reference = plan.reference out_audio = {a.audio_index: a for a in output.audio()} for track in plan.tracks: index = track.source.audio_index expected = mux_offset_for(track.source.content, src_reference.content) if src_reference else None target = out_audio.get(index) if target is None or out_reference is None or out_reference.content.seconds is None: checks.append(TimelineCheck(index, expected, None, None, "not-verifiable", "output track or reference anchor missing")) continue effective = recover_output_content_anchor( tools, output, target, track.encoded_path, track.encoder_timing) if effective.seconds is None or expected is None: checks.append(TimelineCheck(index, expected, None, None, "not-verifiable", f"provenance {effective.provenance}")) continue measured = effective.seconds - out_reference.content.seconds deviation = measured - expected status = "passed" if abs(deviation) <= tolerance else "deviation" checks.append(TimelineCheck(index, expected, measured, deviation, status, f"provenance {effective.provenance}")) return checks def log_timeline_diagnostics(plan: JobPlan, output: MediaFile, checks: list[TimelineCheck]) -> None: """Writes detailed timing values to the debug and job log. Normal console messages stay short. On a deviation the key values are additionally written as WARNING, so that real tests can diagnose the open CodecDelay case without a special build. """ out_audio = {a.audio_index: a for a in output.audio()} by_index = {c.audio_index: c for c in checks} for track in plan.tracks: idx = track.source.audio_index target = out_audio.get(idx) check = by_index.get(idx) timing = track.encoder_timing log.debug( "Timing diagnostics track %d: src_raw={%s}; src_content=%s; source_padding=%s@%sHz; " "encoder_priming=%s@%sHz=%s; out_raw={%s}; out_content=%s; " "expected=%s measured=%s deviation=%s applied_itsoffset=%s", idx, summarize_time(track.source.time), summarize_content(track.source.content), track.source.time.initial_padding, track.source.time.sample_rate, timing.priming_samples if timing else None, timing.raw.sample_rate if timing else None, timing.priming_seconds if timing else None, summarize_time(target.time) if target else "missing", summarize_content(target.content) if target else "missing", check.expected if check else None, check.measured if check else None, check.deviation if check else None, track.applied_mux_offset, ) if check is not None and check.status == "deviation": source_delay = track.source.time.padding_seconds() log.warning( "Timing diagnostics track %d: deviation %+.2f ms; source CodecDelay %s; " "encoder priming %s samples at %s Hz; target rate %s Hz.", idx, (check.deviation or 0.0) * 1000.0, "unknown" if source_delay is None else f"{source_delay * 1000:.2f} ms", timing.priming_samples if timing else None, timing.raw.sample_rate if timing else None, target.sample_rate if target else None, ) # =========================================================================== # 06 Downmix # =========================================================================== def resolve_layout(track: AudioInfo, assumed: dict[int, str]) -> tuple[str, str]: if track.audio_index in assumed: layout = assumed[track.audio_index] if layout not in LAYOUT_CHANNELS: raise TrackPlanningError(f"Unknown layout in --assume-layout. {layout}") if len(LAYOUT_CHANNELS[layout]) != track.channels: raise TrackPlanningError( f"Assumed layout {layout} has {len(LAYOUT_CHANNELS[layout])} channels, " f"the track reports {track.channels}." ) return layout, "user" if track.layout in UNSUPPORTED_LAYOUTS: raise TrackPlanningError(f"Layout not supported in this version. {track.layout}") if track.layout: if track.layout not in LAYOUT_CHANNELS: raise TrackPlanningError(f"Reported layout not supported in this version. {track.layout}") if len(LAYOUT_CHANNELS[track.layout]) != track.channels: raise TrackPlanningError( f"Reported layout {track.layout} does not match channel count {track.channels}." ) return track.layout, "reported" if track.channels == 1: return "mono", "assumed-mono" if track.channels == 2: return "stereo", "assumed-stereo" raise TrackPlanningError( f"No channel layout for {track.channels} channels. " f"Specify it with --assume-layout {track.audio_index}= required." ) def downmix_weights(channels: list[str]) -> tuple[list[tuple[str, float]], list[tuple[str, float]]]: has_bc = "BC" in channels surround_left = [c for c in channels if c in ("SL", "BL")] surround_right = [c for c in channels if c in ("SR", "BR")] n_left = len(surround_left) + (1 if has_bc else 0) n_right = len(surround_right) + (1 if has_bc else 0) def build(front: str, surrounds: list[str], n: int) -> list[tuple[str, float]]: row: list[tuple[str, float]] = [] if front in channels: row.append((front, 1.0)) if "FC" in channels: row.append(("FC", SQRT_HALF)) if n > 0: share = SQRT_HALF / math.sqrt(n) row.extend((channel, share) for channel in surrounds) if has_bc: row.append(("BC", share * SQRT_HALF)) return row return build("FL", surround_left, n_left), build("FR", surround_right, n_right) def _term(weight: float, reference: str) -> str: return reference if abs(weight - 1.0) < 1e-12 else f"{weight:.9g}*{reference}" def build_downmix(layout: str, layout_source: str, mode: str) -> DownmixPlan: channels = LAYOUT_CHANNELS[layout] positional = layout_source == "user" def ref(channel: str) -> str: return f"c{channels.index(channel)}" if positional else channel if mode == "ffmpeg": # --assume-layout defines the meaning of the physical cN positions, # also in FFmpeg mode. aresample alone would otherwise see the decoder layout. relabel = ( f"channelmap=map={'|'.join(str(i) for i in range(len(channels)))}:channel_layout={layout}" if positional else None ) return DownmixPlan(layout, "stereo", relabel, (), (), "ffmpeg", "FFmpeg rematrixing in the aresample filter") if layout == "stereo": return DownmixPlan(layout, "stereo", None, (("FL", 1.0),), (("FR", 1.0),), "passthrough", "Stereoquelle") if layout == "mono": expr = f"pan=stereo|FL<{ref('FC')}|FR<{ref('FC')}" return DownmixPlan(layout, "stereo", expr, (("FC", 1.0),), (("FC", 1.0),), "project-defined", "mono is duplicated to stereo") left, right = downmix_weights(channels) if not left or not right: raise TrackPlanningError(f"Layout {layout} cannot be mapped to a stereo matrix.") left_expr = "+".join(_term(w, ref(c)) for c, w in left) right_expr = "+".join(_term(w, ref(c)) for c, w in right) expr = f"pan=stereo|FL<{left_expr}|FR<{right_expr}" if layout in LEGACY_DERIVED_LAYOUTS: provenance = "legacy-derived" elif layout in LEGACY_ADAPTED_LAYOUTS: provenance = "legacy-adapted" else: provenance = "project-defined" note = "LFE discarded" if "LFE" in channels else "" return DownmixPlan(layout, "stereo", expr, tuple(left), tuple(right), provenance, note) def normalized_coefficients(weights: tuple[tuple[str, float], ...]) -> list[tuple[str, float]]: total = sum(w for _, w in weights) return [] if total <= 0 else [(c, w / total) for c, w in weights] # =========================================================================== # 07 Analysis pass, overdrive, gain # =========================================================================== RE_N_SAMPLES = re.compile(r"n_samples:\s*(\d+)") RE_MAX_VOLUME = re.compile(r"max_volume:\s*(-?\d+(?:\.\d+)?)\s*dB") RE_MEAN_VOLUME = re.compile(r"mean_volume:\s*(-?\d+(?:\.\d+)?)\s*dB") RE_HISTOGRAM = re.compile(r"histogram_(\d+)db:\s*(\d+)") RE_JSON_BLOCK = re.compile(r'\{[^{}]*"input_i"[^{}]*\}', re.S) def filter_chain(plan: TrackPlan, *, with_gain: bool, gain_placeholder: bool = False) -> str: parts: list[str] = [] if plan.downmix.pan_expr: parts.append(plan.downmix.pan_expr) # Only a measured forward discontinuity enables timestamp compensation. # async=1 fills gaps without stretching; the default 100 ms hard-compensation # threshold misses the real source's 88 ms drift. Keep the 5 ms timing bound. compensation = ":async=1:min_hard_comp=0.005" if getattr(plan, "preserve_gaps", False) else "" if plan.downmix.provenance == "ffmpeg": parts.append( f"aresample=out_chlayout=stereo:osr={plan.target_sample_rate}{compensation}" ) else: parts.append(f"aresample={plan.target_sample_rate}{compensation}") if with_gain: if gain_placeholder: parts.append("volume=dB") else: value = plan.gain.total_gain if plan.gain is not None else 0.0 parts.append(f"volume={value:.3f}dB") if plan.gain is not None and plan.gain.limiter_active: limit = 10.0 ** ((plan.normalization.target_tp - LIMITER_HEADROOM_DB) / 20.0) parts.append(f"alimiter=limit={limit:.9f}:level=0:latency=1") return ",".join(parts) def analysis_command(tools: Tools, source: Path, plan: TrackPlan, input_audio_index: int | None = None) -> list[str]: chain = filter_chain(plan, with_gain=False) audio_index = plan.source.audio_index if input_audio_index is None else input_audio_index graph = ( f"[0:a:{audio_index}]{chain},asplit=2[a][b];" f"[a]aformat=sample_fmts=s16,volumedetect[o1];" f"[b]loudnorm=print_format=json[o2]" ) return [ tools.ffmpeg, "-hide_banner", "-v", "info", "-nostdin", "-i", str(source), "-filter_complex", graph, "-map", "[o1]", "-f", "null", "-", "-map", "[o2]", "-f", "null", "-", ] def parse_analysis(stderr: str) -> Analysis: samples = [int(m) for m in RE_N_SAMPLES.findall(stderr)] max_values = [float(m) for m in RE_MAX_VOLUME.findall(stderr)] mean_values = [float(m) for m in RE_MEAN_VOLUME.findall(stderr)] histogram = {int(k): int(v) for k, v in RE_HISTOGRAM.findall(stderr)} payload: dict[str, Any] = {} matches = list(RE_JSON_BLOCK.finditer(stderr)) if matches: try: payload = json.loads(matches[-1].group(0)) except json.JSONDecodeError: payload = {} if not samples or not max_values: raise StepFailed("Analysis pass returned no usable volumedetect values.") def number(key: str, fallback: float) -> float: value = _as_float(payload.get(key)) return fallback if value is None else value max_volume = max_values[-1] return Analysis( n_samples=max(samples), max_volume=max_volume, mean_volume=mean_values[-1] if mean_values else float("nan"), histogram=histogram, input_i=number("input_i", float("nan")), input_tp=number("input_tp", max_volume), input_lra=number("input_lra", float("nan")), input_thresh=number("input_thresh", float("nan")), ) def overdrive_candidate(histogram: dict[int, int], n_samples: int, quota: float | None) -> tuple[float | None, str]: if quota is None: return None, "off" if not histogram or n_samples <= 0: return None, "histogram missing" budget = n_samples * quota cumulative = 0 for level, count in sorted(histogram.items()): if count <= 0: continue if cumulative + count > budget: fraction = (budget - cumulative) / count return level + fraction, "estimated, interpolated" cumulative += count return None, "histogram unexpectedly insufficient for quantile estimate" def isolated_peak_candidate(analysis: Analysis) -> tuple[float | None, int, str]: """Determines an extremely small upper quantile for the outlier guard. The volumedetect histogram value is an attenuation below full scale. Unlike overdrive, this path is not meant as a loudness option: it may only ignore single or very short full-scale outliers. """ if not analysis.histogram or analysis.n_samples <= 0: return None, 0, "histogram missing" budget = min(ISOLATED_PEAK_MAX_BUDGET, int(analysis.n_samples * ISOLATED_PEAK_QUANTILE)) if budget < 1: return None, 0, "recording too short for the outlier budget" cumulative = 0 for level, count in sorted(analysis.histogram.items()): if count <= 0: continue if cumulative + count > budget: # A bin does not reveal the distribution within its 1 dB interval. # Use its louder edge; interpolation can invent almost 1 dB of gain. return float(level), budget, "estimated, 1 dB histogram edge" cumulative += count return None, budget, "histogram insufficient for outlier quantile" def compute_gain(plan: TrackPlan) -> GainResult: analysis = plan.analysis rules = plan.normalization if analysis is None: raise StepFailed("Gain cannot be calculated without analysis.") notes: list[str] = [] silence_guard = analysis.max_volume <= SILENCE_THRESHOLD_DB isolated_guard_active = False if rules.mode == "peak": if silence_guard: peak_gain = 0.0 notes.append(f"Peak <= {SILENCE_THRESHOLD_DB:.0f} dBFS: no automatic gain.") else: peak_gain = rules.target_tp - analysis.input_tp if (rules.isolated_peak_guard == "auto" and rules.overdrive_step == "off" and rules.overdrive_db == 0.0 and rules.limiter != "off" and analysis.max_volume >= ISOLATED_PEAK_NEAR_FS_DB): candidate, budget, candidate_note = isolated_peak_candidate(analysis) sample_peak_gain_local = -analysis.max_volume if candidate is not None: gap = candidate - sample_peak_gain_local # The robust candidate is not pulled up to the true-peak target but # only to the limiter headroom. The few ignored peaks are then caught # by the limiter. The post-AAC control loop stays # unchanged. robust_gain = candidate + rules.target_tp - LIMITER_HEADROOM_DB recovery = robust_gain - peak_gain if gap >= ISOLATED_PEAK_MIN_GAP_DB and recovery > 0.0: recovery = min(recovery, ISOLATED_PEAK_MAX_RECOVERY_DB) peak_gain += recovery isolated_guard_active = True notes.append( "Isolated peak guard active: top quantile up to " f"{budget} samples tolerated; robust peak edge about " f"{-candidate:.2f} dBFS, gap {gap:.2f} dB; " f"additional recovery gain {recovery:+.2f} dB " f"({candidate_note})." ) elif (rules.isolated_peak_guard == "auto" and rules.limiter == "off" and analysis.max_volume >= ISOLATED_PEAK_NEAR_FS_DB): notes.append( "Isolated peak guard not applied, because the limiter " "is explicitly switched off." ) else: peak_gain = 0.0 if plan.downmix.pan_expr: coeffs = normalized_coefficients(plan.downmix.weights_left) front = next((w for c, w in coeffs if c == "FL"), None) if front is not None and front > 0: notes.append( f"Normalisation off. Front contribution of the left matrix row: " f"{20 * math.log10(front):.2f} dB. The overall level depends on the programme material." ) sample_peak_gain = -analysis.max_volume quota = OVERDRIVE_QUOTAS[rules.overdrive_step] g_hist: float | None = None overdrive_auto = 0.0 if quota is not None: if silence_guard: notes.append("Automatic overdrive disabled because of the silence lock.") else: g_hist, note = overdrive_candidate(analysis.histogram, analysis.n_samples, quota) if g_hist is None: notes.append(f"Overdrive {rules.overdrive_step}: {note}; automatic extra gain 0 dB.") else: overdrive_auto = min(OVERDRIVE_CAP_DB, max(0.0, g_hist - sample_peak_gain)) notes.append(f"Overdrive {rules.overdrive_step}: {note}. The value is an estimate.") total = peak_gain + overdrive_auto + rules.overdrive_db if total > rules.max_gain: notes.append(f"Positive gain limited to {rules.max_gain:.1f} dB.") total = rules.max_gain positive_overdrive = overdrive_auto > 0.0 or rules.overdrive_db > 0.0 if rules.limiter == "on": limiter = True elif rules.limiter == "off": limiter = False else: limiter = positive_overdrive or isolated_guard_active if rules.limiter == "auto" and limiter: if isolated_guard_active and not positive_overdrive: notes.append("Limiter activated by the isolated peak guard.") elif isolated_guard_active: notes.append("Limiter activated by overdrive and the isolated peak guard.") else: notes.append("Limiter activated by positive overdrive.") if rules.mode == "off" and rules.overdrive_db != 0.0: notes.append("Normalisation off, manual gain explicitly active.") result = GainResult( peak_gain=peak_gain, sample_peak_gain=sample_peak_gain, g_hist=g_hist, overdrive_auto=overdrive_auto, overdrive_manual=rules.overdrive_db, total_gain=total, limiter_active=limiter, notes=notes, ) reporter = globals().get("REPORTER") if reporter is not None: reporter.set_track_gain(plan.source.audio_index, result.total_gain) return result def analysis_required(track: TrackPlan) -> bool: rules = track.normalization return rules.mode != "off" or rules.overdrive_step != "off" def gain_without_analysis(track: TrackPlan) -> GainResult: """Fast path for normalize=off and automatic overdrive=off. max_gain applies here as well to positive manual gain. """ rules = track.normalization total = rules.overdrive_db notes = ["without analysis pass, mode off"] if total > rules.max_gain: total = rules.max_gain notes.append(f"Positive gain limited to {rules.max_gain:.1f} dB.") if rules.overdrive_db != 0.0: notes.append("manual gain explicitly requested") limiter = rules.limiter == "on" or (rules.limiter == "auto" and total > 0.0) result = GainResult(0.0, 0.0, None, 0.0, rules.overdrive_db, total, limiter, notes) reporter = globals().get("REPORTER") if reporter is not None: reporter.set_track_gain(track.source.audio_index, result.total_gain) return result def analyse_track(tools: Tools, source: Path, plan: TrackPlan, duration: float | None = None, input_audio_index: int | None = None) -> None: result = run_ffmpeg_progress( analysis_command(tools, source, plan, input_audio_index), duration=duration, label=f"Analysis Audio {plan.source.audio_index}", ) if result.returncode != 0: raise StepFailed(f"Analysis pass failed for track {plan.source.audio_index}.", result) plan.analysis = parse_analysis(result.stderr) plan.gain = compute_gain(plan) def _terminate_processes(processes: list[subprocess.Popen[Any]]) -> None: for proc in processes: if proc.poll() is None: try: proc.terminate() except OSError: pass deadline = time.monotonic() + 5.0 for proc in processes: if proc.poll() is None: try: proc.wait(timeout=max(0.0, deadline - time.monotonic())) except subprocess.TimeoutExpired: try: proc.kill() except OSError: pass for proc in processes: try: proc.wait(timeout=1) except Exception: pass # --------------------------------------------------------------------------- # 07B Multitrack spool: demux once, then truly independent workers # --------------------------------------------------------------------------- def resolved_audio_workers(args: argparse.Namespace, n_tracks: int) -> int: requested = int(getattr(args, "audio_workers", 0) or 0) if requested > 0: return max(1, min(n_tracks, requested)) available = max(1, (os.cpu_count() or 2) - 1) return max(1, min(n_tracks, available)) def create_audio_spool(tools: Tools, source: Path, tracks: list[TrackPlan], workdir: Path, duration: float | None) -> Path: """Demux selected audio once to a small Matroska spool. Original timing anchors are never taken from this spool; it is only an efficient PCM source for analysis/encoding workers. This avoids Python in the data path and lets workers run independently. """ spool = workdir / "audio-spool.mka" argv = [tools.ffmpeg, "-hide_banner", "-v", "error", "-nostdin", "-i", str(source)] for track in tracks: argv += ["-map", f"0:{track.source.index}"] argv += ["-map_metadata", "-1", "-map_chapters", "-1", "-c", "copy", "-f", "matroska", "-y", str(spool)] result = run_ffmpeg_progress(argv, duration=duration, label="Audio-Spool", workdir=workdir) if result.returncode != 0 or not spool.is_file() or spool.stat().st_size == 0: raise StepFailed("Audio spool could not be created.", result) log_detail("Audio-Spool: %s (%.1f MiB)", spool, spool.stat().st_size / (1024 * 1024)) return spool def _run_analysis_batch(tools: Tools, spool: Path, jobs: list[tuple[int, TrackPlan]], *, duration: float | None, workdir: Path) -> list[StepResult]: procs: list[subprocess.Popen[Any]] = [] handles: list[Any] = [] err_paths: list[Path] = [] progress_paths: list[Path] = [] argvs: list[list[str]] = [] started = time.monotonic() label = f"Analysis {len(jobs)} audio tracks in parallel" try: for pos, (local_index, track) in enumerate(jobs): argv = analysis_command(tools, spool, track, local_index) progress = workdir / f"spool-analysis-{local_index}.progress" err = workdir / f"spool-analysis-{local_index}.stderr" handle = open(err, "wb") handles.append(handle) actual = _with_ffmpeg_progress(argv, progress) log.debug("start %s", " ".join(shlex.quote(a) for a in actual)) proc = subprocess.Popen( actual, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=handle, shell=False, ) procs.append(proc) argvs.append(argv) err_paths.append(err) progress_paths.append(progress) while any(proc.poll() is None for proc in procs): for proc in procs: if proc.poll() is not None and proc.returncode not in (0, None): raise StepFailed("Parallel analysis worker failed early.") vals = [_progress_seconds(p) for p in progress_paths] known = [v for v in vals if v is not None] current = min(known) if known else None _show_progress(label, duration, current, elapsed=time.monotonic() - started) time.sleep(PROGRESS_REFRESH) elapsed = time.monotonic() - started _show_progress(label, duration, duration, elapsed=elapsed, done=True) telemetry = globals().get("TIMING") if telemetry is not None: telemetry.step(label, duration, elapsed, n=len(jobs)) results: list[StepResult] = [] for argv, proc, err in zip(argvs, procs, err_paths): text = err.read_text("utf-8", "replace") if err.exists() else "" result = StepResult(argv, proc.returncode or 0, text, elapsed) if proc.returncode != 0: raise StepFailed("Parallel analysis worker failed.", result) results.append(result) return results except BaseException: _terminate_processes(procs) raise finally: for handle in handles: try: handle.close() except OSError: pass for path in progress_paths: path.unlink(missing_ok=True) def analyse_tracks_spool(tools: Tools, spool: Path, tracks: list[TrackPlan], duration: float | None, workdir: Path, workers: int) -> None: jobs: list[tuple[int, TrackPlan]] = [] for local_index, track in enumerate(tracks): if analysis_required(track): jobs.append((local_index, track)) else: track.gain = gain_without_analysis(track) if not jobs: return for first in range(0, len(jobs), workers): batch = jobs[first:first + workers] if len(batch) == 1: local_index, track = batch[0] analyse_track(tools, spool, track, duration, input_audio_index=local_index) continue results = _run_analysis_batch( tools, spool, batch, duration=duration, workdir=workdir ) for (_, track), result in zip(batch, results): track.analysis = parse_analysis(result.stderr) track.gain = compute_gain(track) def _encode_spool_batch(tools: Tools, plan: JobPlan, spool: Path, indices: list[int], workdir: Path, *, retry_reasons: dict[int, str] | None = None) -> None: if tools.nero is None: raise StepFailed("neroAacEnc is required for encoding.") ffmpeg_procs: list[subprocess.Popen[Any]] = [] nero_procs: list[subprocess.Popen[Any]] = [] handles: list[Any] = [] progress_paths: list[Path] = [] ff_errs: list[Path] = [] nero_errs: list[Path] = [] started = time.monotonic() label = f"Encoding {len(indices)} audio tracks in parallel" if retry_reasons: reasons = [retry_reasons[i] for i in indices if retry_reasons.get(i)] if reasons: label += " Retry " + ", ".join(reasons) try: for global_index in indices: track = plan.tracks[global_index] target = workdir / f"track{global_index}.m4a" progress = workdir / f"spool-encode-{global_index}.progress" ff_err = workdir / f"spool-encode-{global_index}-ffmpeg.stderr" nero_err = workdir / f"spool-encode-{global_index}-nero.stderr" ff_handle = open(ff_err, "wb") ne_handle = open(nero_err, "wb") handles.extend([ff_handle, ne_handle]) producer = [ tools.ffmpeg, "-hide_banner", "-v", "error", "-nostdin", "-i", str(spool), "-map", f"0:a:{global_index}", "-af", filter_chain(track, with_gain=True), "-c:a", "pcm_s16le", "-f", "wav", "-", ] actual_producer = _with_ffmpeg_progress(producer, progress) log.debug("start %s", " ".join(shlex.quote(a) for a in actual_producer)) ff = subprocess.Popen( actual_producer, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=ff_handle, shell=False, ) ffmpeg_procs.append(ff) assert ff.stdout is not None nero_argv = [ tools.nero, "-ignorelength", "-q", f"{track.profile.nero_quality:.2f}", "-lc", "-if", "-", "-of", str(target), ] log.debug("pipe %s", " ".join(shlex.quote(a) for a in nero_argv)) try: nero = subprocess.Popen( nero_argv, stdin=ff.stdout, stdout=subprocess.DEVNULL, stderr=ne_handle, shell=False, ) except OSError as exc: ff.stdout.close() ff.kill(); ff.wait() raise StepFailed( f"neroAacEnc for track {track.source.audio_index} could not be started." ) from exc ff.stdout.close() nero_procs.append(nero) progress_paths.append(progress) ff_errs.append(ff_err) nero_errs.append(nero_err) while any(p.poll() is None for p in ffmpeg_procs + nero_procs): for global_index, ff, nero in zip(indices, ffmpeg_procs, nero_procs): if nero.poll() is not None and nero.returncode not in (0, None): raise StepFailed( f"neroAacEnc Returncode {nero.returncode} for track " f"{plan.tracks[global_index].source.audio_index}." ) if ff.poll() is not None and ff.returncode not in (0, None): raise StepFailed( f"FFmpeg Returncode {ff.returncode} for track " f"{plan.tracks[global_index].source.audio_index}." ) vals = [_progress_seconds(p) for p in progress_paths] known = [v for v in vals if v is not None] current = min(known) if known else None _show_progress(label, plan.source.duration, current, elapsed=time.monotonic() - started) time.sleep(PROGRESS_REFRESH) elapsed = time.monotonic() - started _show_progress(label, plan.source.duration, plan.source.duration, elapsed=elapsed, done=True) telemetry = globals().get("TIMING") if telemetry is not None: # In a retry round step() fills in the remembered round number. # Only the first pass is marked explicitly with r=0. extra = {} if retry_reasons else {"r": 0} telemetry.step(label, plan.source.duration, elapsed, n=len(indices), **extra) for global_index, ff, nero, ff_err, nero_err in zip( indices, ffmpeg_procs, nero_procs, ff_errs, nero_errs ): track = plan.tracks[global_index] if nero.returncode != 0: text = nero_err.read_text("utf-8", "replace") if nero_err.exists() else "" raise StepFailed( f"neroAacEnc Returncode {nero.returncode} for track {track.source.audio_index}.", StepResult([tools.nero], nero.returncode or -1, text, elapsed), ) if ff.returncode != 0: text = ff_err.read_text("utf-8", "replace") if ff_err.exists() else "" raise StepFailed( f"FFmpeg Returncode {ff.returncode} for track {track.source.audio_index}.", StepResult([tools.ffmpeg], ff.returncode or -1, text, elapsed), ) target = workdir / f"track{global_index}.m4a" if not target.is_file() or target.stat().st_size == 0: raise StepFailed(f"The encoder output is missing or empty. {target}") track.encoded_path = target except BaseException: _terminate_processes(ffmpeg_procs + nero_procs) raise finally: for handle in handles: try: handle.close() except OSError: pass for path in progress_paths: path.unlink(missing_ok=True) def encode_tracks_spool_initial(tools: Tools, plan: JobPlan, spool: Path, workdir: Path, workers: int) -> None: indices = list(range(len(plan.tracks))) for first in range(0, len(indices), workers): batch = indices[first:first + workers] if len(batch) == 1: idx = batch[0] track = plan.tracks[idx] track.encoded_path = encode_track( tools, plan, track, idx, workdir, source_path=spool, input_audio_index=idx, ) track.attempts = 1 else: _encode_spool_batch(tools, plan, spool, batch, workdir) for idx in batch: plan.tracks[idx].attempts = 1 # =========================================================================== # 08 Planning # =========================================================================== def shallow_dataclass_dict(item: Any) -> dict[str, Any]: return {f.name: getattr(item, f.name) for f in fields(item)} def parse_assume_layouts(values: list[str] | None) -> dict[int, str]: result: dict[int, str] = {} for item in values or []: if "=" not in item: raise argparse.ArgumentTypeError( f"--assume-layout expects TRACK=LAYOUT, got: {item!r}" ) left, _, right = item.partition("=") index = _as_int(left.strip()) layout = right.strip() if index is None or index < 0 or not layout: raise argparse.ArgumentTypeError(f"Invalid --assume-layout value: {item!r}") if index in result: raise argparse.ArgumentTypeError(f"Layout for audio track {index} specified more than once.") result[index] = layout return result def select_tracks(media: MediaFile, spec: str, languages: list[str] | None) -> list[AudioInfo]: tracks = media.audio() if spec != "all": wanted: set[int] = set() for part in spec.split(","): value = _as_int(part.strip()) if value is None or value < 0: raise argparse.ArgumentTypeError(f"Invalid track specification: {part!r}") wanted.add(value) unknown = sorted(wanted - {t.audio_index for t in tracks}) if unknown: raise argparse.ArgumentTypeError( "Audio track(s) not present: " + ", ".join(map(str, unknown)) ) tracks = [t for t in tracks if t.audio_index in wanted] if languages: wanted_langs = {lang.strip().lower() for lang in languages if lang.strip()} tracks = [t for t in tracks if (t.language_canonical or "") in wanted_langs] return tracks def is_sd_material(media: MediaFile) -> bool: """SD by coded raster, so that anamorphic 720x576 stays SD.""" videos = media.video() if not videos: return False video = videos[0] return (video.width is not None and video.height is not None and video.width <= SD_MAX_WIDTH and video.height <= SD_MAX_HEIGHT) def automatic_profile_name(media: MediaFile) -> str: # With an unknown or missing picture size, use the higher-quality # setting conservatively. Audio-only input is not the primary use case. return "sq" if is_sd_material(media) else "hq" def resolution_letter(media: MediaFile) -> str: videos = media.video() if not videos: return "" video = videos[0] if is_sd_material(media): return "s" width = video.display_width if width is None: return "U" if width <= 1200: return "1" if width <= 1900: return "h" if width <= 3000: return "2" return "4" def language_letters(output_audio: list[AudioInfo]) -> str: return "".join(LANG_LETTER.get(track.language_canonical or "", "U") for track in output_audio) def build_output_path(media: MediaFile, tracks: list[TrackPlan], passthrough: list[AudioInfo], out_dir: Path | None, template: str) -> Path: ordered = sorted([t.source for t in tracks] + passthrough, key=lambda a: a.audio_index) values = {"stem": media.path.stem, "res": resolution_letter(media), "lang": language_letters(ordered)} try: name = template.format(**values) except (KeyError, ValueError) as exc: raise argparse.ArgumentTypeError(f"Invalid --name-template: {exc}") from exc if not name.strip(): raise argparse.ArgumentTypeError("--name-template produces an empty file name.") directory = out_dir if out_dir is not None else media.path.parent return directory / f"{name}.mkv" def build_plan(media: MediaFile, args: argparse.Namespace) -> JobPlan: requested_profile = PROFILE_ALIASES.get(args.profile, args.profile) if requested_profile == "auto": profile_name = automatic_profile_name(media) selection = "auto" else: profile_name = requested_profile selection = "explicit" profile = Profile( profile_name, PROFILES[profile_name]["sample_rate"], args.quality if args.quality is not None else PROFILES[profile_name]["nero_quality"], selection, ) assumed = parse_assume_layouts(args.assume_layout) selected = select_tracks(media, args.tracks, args.lang) if not selected: raise StepFailed("No matching audio track found.") rules = NormalizationPlan( mode=args.normalize, target_tp=args.target_tp, max_gain=args.max_gain, overdrive_step=args.overdrive, overdrive_db=args.overdrive_db, isolated_peak_guard=args.isolated_peak_guard, limiter=args.limiter, tolerance=args.peak_tolerance, retries=args.peak_retries, ) tracks: list[TrackPlan] = [] for track in selected: try: layout, source_kind = resolve_layout(track, assumed) downmix = build_downmix(layout, source_kind, args.downmix) except TrackPlanningError as exc: raise StepFailed( f"Audio track {track.audio_index} cannot be processed: {exc}" ) from exc resolved = AudioInfo(**{ **shallow_dataclass_dict(track), "layout": layout, "layout_source": source_kind, }) target_rate = ( track.sample_rate if args.samplerate == "source" else (int(args.samplerate) if args.samplerate else profile.sample_rate) ) tracks.append(TrackPlan(resolved, profile, downmix, target_rate, rules)) reference_stream = media.video()[0] if media.video() else media.audio()[0] for entry in tracks: # Verification criterion: relative source offset to the reference track. entry.mux_offset = mux_offset_for(entry.source.content, reference_stream.content) # FFmpeg input offset: absolute content anchor of the source audio track. # The separately generated AAC track starts on a new input time axis. planned_input_offset = mux_input_offset_for(entry.source.content, media.start_time) if entry.mux_offset is None or planned_input_offset is None: entry.applied_mux_offset = None entry.mux_offset_note = "timing not verifiable, no offset applied" log.warning("Audio track %d: %s", entry.source.audio_index, entry.mux_offset_note) elif abs(planned_input_offset) < MUX_OFFSET_DEADBAND: entry.applied_mux_offset = 0.0 entry.mux_offset_note = ( f"Source content anchor within the deadband of " f"{MUX_OFFSET_DEADBAND * 1000:.0f} ms, no -itsoffset" ) else: entry.applied_mux_offset = planned_input_offset entry.mux_offset_note = ( "start value from the source content anchor minus " "container start time, the measurement on the output decides") handled = {t.source.audio_index for t in tracks} passthrough = [a for a in media.audio() if a.audio_index not in handled] unmapped = [ s for s in media.streams if s.kind not in ("video", "audio", "subtitle", "attachment", "data") ] if unmapped: desc = ", ".join(f"Index {s.index} ({s.kind}, {s.codec})" for s in unmapped) if args.on_incompatible_stream == "fail": log.warning("Streams of unknown type are mapped by index. %s", desc) else: log.warning("Streams of unknown type are dropped. %s", desc) unmapped = [] output = build_output_path(media, tracks, passthrough, args.out, args.name_template) if output.resolve() == media.path.resolve(): raise StepFailed("The output name equals the source. Aborting.") if output.exists() and not args.force and not (args.inspect or args.dry_run): raise StepFailed(f"The output file already exists. {output}") return JobPlan(media, output, reference_stream, tracks, passthrough, unmapped) def require_verified_source_timing(plan: JobPlan) -> None: """Production mode: a newly encoded track requires a proven source timing.""" unknown = [ t.source.audio_index for t in plan.tracks if t.mux_offset is None or t.applied_mux_offset is None ] if unknown: raise StepFailed( "Timing of the audio track(s) to be processed is not verifiable: " + ", ".join(map(str, unknown)) + ". Processing aborted. --inspect and --dry-run show the raw data." ) # =========================================================================== # 09 Dry-run commands and mux plan # =========================================================================== def encode_commands(tools: Tools, plan: JobPlan, track: TrackPlan, index: int) -> tuple[list[str], list[str]]: chain = filter_chain(track, with_gain=True, gain_placeholder=track.gain is None) workdir = plan.temp_dir or Path("") target = workdir / f"track{index}.m4a" ffmpeg_argv = [ tools.ffmpeg, "-hide_banner", "-v", "error", "-nostdin", "-i", str(plan.source.path), "-map", f"0:a:{track.source.audio_index}", "-af", chain, "-c:a", "pcm_s16le", "-f", "wav", "-", ] nero_argv = [ tools.nero or "neroAacEnc", "-ignorelength", "-q", f"{track.profile.nero_quality:.2f}", "-lc", "-if", "-", "-of", str(target), ] return ffmpeg_argv, nero_argv def audio_tag_is_semantic(key: str) -> bool: folded = key.casefold() if folded in STALE_AUDIO_TAGS or folded.startswith("_statistics_"): return False return bool(key.strip()) def active_output_path(plan: JobPlan) -> Path: return plan.candidate_output or plan.output def mux_command(tools: Tools, plan: JobPlan) -> list[str]: workdir = plan.temp_dir or Path("") argv = [ tools.ffmpeg, "-hide_banner", "-v", "error", "-nostdin", "-i", str(plan.source.path), ] for index, track in enumerate(plan.tracks): offset = track.applied_mux_offset if offset is not None and abs(offset) >= MUX_OFFSET_DEADBAND: argv += ["-itsoffset", f"{offset:.6f}"] argv += ["-i", str(workdir / f"track{index}.m4a")] argv += ["-map", "0:v?"] encoded = {t.source.audio_index: i for i, t in enumerate(plan.tracks)} for audio in sorted(plan.source.audio(), key=lambda a: a.audio_index): if audio.audio_index in encoded: argv += ["-map", f"{encoded[audio.audio_index] + 1}:a:0"] else: argv += ["-map", f"0:a:{audio.audio_index}"] argv += ["-map", "0:s?", "-map", "0:t?", "-map", "0:d?"] for stream in plan.unmapped_streams: argv += ["-map", f"0:{stream.index}"] argv += ["-map_chapters", "0", "-map_metadata", "0", "-c", "copy"] for position, audio in enumerate(sorted(plan.source.audio(), key=lambda a: a.audio_index)): # Restore semantic source tags; omit technical statistics and encoder tags. for key, value in audio.tags_raw.items(): folded = key.casefold() if folded in ("language", "title") or not audio_tag_is_semantic(key): continue argv += [f"-metadata:s:a:{position}", f"{key}={value}"] if audio.language_raw: argv += [f"-metadata:s:a:{position}", f"language={audio.language_raw}"] if audio.title: argv += [f"-metadata:s:a:{position}", f"title={audio.title}"] flags = sorted(name for name, active in audio.disposition.items() if active) argv += [f"-disposition:a:{position}", "+".join(flags) if flags else "0"] argv += ["-y", str(active_output_path(plan))] return argv # =========================================================================== # 10 Reports # =========================================================================== def fmt_float(value: float | None, digits: int = 3, suffix: str = "") -> str: return "unknown" if value is None else f"{value:.{digits}f}{suffix}" def summarize_time(c: RawTiming) -> str: parts = [] for name in ANCHOR_METHODS: value = c.get(name) if value is not None: parts.append(f"{name}={value:+.6f}s") if c.initial_padding is not None: ms = (c.padding_seconds() or 0.0) * 1000.0 parts.append(f"initial_padding={c.initial_padding} Samples ({ms:.3f} ms)") if c.skip_samples is not None: ms = (c.skip_seconds() or 0.0) * 1000.0 parts.append(f"skip_samples={c.skip_samples} ({ms:.3f} ms)") return ", ".join(parts) if parts else "unknown" def summarize_content(a: ContentAnchor) -> str: if a.seconds is None: return f"unknown (provenance {a.provenance}, {a.confidence})" return ( f"{a.seconds:+.6f}s from {a.method}, correction {a.correction:+.6f}s, " f"provenance {a.provenance}, {a.confidence}" ) def describe_matrix(downmix: DownmixPlan) -> list[str]: if downmix.provenance == "ffmpeg": lines = [f" Matrix FFmpeg rematrixing; {downmix.note}"] if downmix.pan_expr: lines.append(f" Input mapping {downmix.pan_expr}") return lines if downmix.pan_expr is None: if downmix.provenance == "passthrough": return [" Matrix none (stereo, no downmix)"] return [f" Matrix none ({downmix.provenance}; {downmix.note})"] lines = [f" Matrix classic matrix: {downmix.provenance}"] coeffs = normalized_coefficients(downmix.weights_left) if coeffs: lines.append( " Coefficients " + " ".join(f"{c} {w:.3f}" for c, w in coeffs) + " (display)" ) lines.append(f" Filter {downmix.pan_expr}") if downmix.note: lines.append(f" Note {downmix.note}") return lines def inspect_text(plan: JobPlan, analyzed: bool) -> str: media = plan.source out = [ f"Application {DISPLAY_NAME} {VERSION}", f"File {media.path}", f"Container {media.format_name or 'unknown'}", f"Duration {fmt_float(media.duration, 3, ' s')}", f"Chapters {media.chapter_count}", f"Output {plan.output.name}", f"Audio tracks {len(media.audio())} total, {len(plan.tracks)} will be encoded", "", ] for video in media.video(): size = f"{video.width or '?'}x{video.height or '?'}" sar = ( f"{video.sar_num}:{video.sar_den}" if video.sar_num is not None and video.sar_den is not None else "unknown" ) out.append(f"Video Index {video.index} {video.codec} {size} SAR {sar}") out.append(f" Raw timing {summarize_time(video.time)}") out.append(f" Content anchor {summarize_content(video.content)}") for stream in media.streams: if stream.kind not in ("video", "audio"): out.append(f"{stream.kind.capitalize():10} Index {stream.index} {stream.codec}") if media.streams: out.append("") planned = {t.source.audio_index: t for t in plan.tracks} for source_track in media.audio(): entry = planned.get(source_track.audio_index) shown = entry.source if entry else source_track out.append( f"Audio {source_track.audio_index} Index {source_track.index} {source_track.codec} " f"{source_track.channels} channels {source_track.sample_rate} Hz " f"Language {source_track.language_raw or 'none'} " f"[{'processed' if entry else 'unchanged'}]" ) out.append(f" Layout {shown.layout or 'unknown'} provenance {shown.layout_source}") out.append(f" Raw timing {summarize_time(source_track.time)}") out.append(f" Content anchor {summarize_content(source_track.content)}") if entry: out.extend(describe_matrix(entry.downmix)) out.append( f" Encoding AAC-LC, 2 channels, {entry.target_sample_rate} Hz " f"Profile {entry.profile.name.upper()}" f"{' (automatic)' if entry.profile.selection == 'auto' else ''} " f"Nero q {entry.profile.nero_quality:.2f}" ) out.append( f" Normalisation {entry.normalization.mode} " f"Target {entry.normalization.target_tp:.1f} dBTP " f"Overdrive {entry.normalization.overdrive_step} " f"manual {entry.normalization.overdrive_db:+.1f} dB " f"Peak-Guard {entry.normalization.isolated_peak_guard}" ) out.append( f" Source offset " f"{fmt_float(entry.mux_offset, 6, ' s') if entry.mux_offset is not None else 'not verifiable'} " "relative to the reference" ) out.append( f" Mux offset " f"{fmt_float(entry.applied_mux_offset, 6, ' s') if entry.applied_mux_offset is not None else 'not verifiable'} " f"({entry.mux_offset_note})" ) if analyzed and entry.analysis and entry.gain: a, g = entry.analysis, entry.gain out.append( f" Measurement Sample Peak {a.max_volume:.1f} dBFS " f"True Peak {a.input_tp:.2f} dBTP Mean {a.mean_volume:.1f} dB" ) out.append( f" Integrated {a.input_i:.2f} LUFS " f"LRA {a.input_lra:.2f} LU Samples {a.n_samples}" ) candidate = "unknown" if g.g_hist is None else f"{g.g_hist:.3f} dB" out.append( f" Overdrive {g.overdrive_auto:+.2f} dB estimated " f"candidate {candidate}" ) out.append( f" Total gain {g.total_gain:+.2f} dB " f"Limiter {'an' if g.limiter_active else 'off'}" ) out.extend(f" {note}" for note in g.notes) out.append("") return "\n".join(out) def _content_json(a: ContentAnchor) -> dict[str, object]: return { "seconds": a.seconds, "method": a.method, "provenance": a.provenance, "confidence": a.confidence, "correction": a.correction, } def _time_json(c: RawTiming) -> dict[str, object]: return { "presentation": c.presentation, "start_time": c.start_time, "packet_pts": c.packet_pts, "initial_padding_samples": c.initial_padding, "skip_samples": c.skip_samples, "discard_padding": c.discard_padding, "sample_rate": c.sample_rate, } def inspect_json(plan: JobPlan, analyzed: bool) -> dict[str, Any]: planned = {t.source.audio_index: t for t in plan.tracks} audio: list[dict[str, Any]] = [] for track in plan.source.audio(): entry = planned.get(track.audio_index) shown = entry.source if entry else track item: dict[str, Any] = { "audio_index": track.audio_index, "stream_index": track.index, "codec": track.codec, "channels": track.channels, "sample_rate": track.sample_rate, "layout": shown.layout, "layout_source": shown.layout_source, "language_raw": track.language_raw, "language_canonical": track.language_canonical, "title": track.title, "disposition": track.disposition, "tags_raw": track.tags_raw, "time_raw": _time_json(track.time), "content_anchor": _content_json(track.content), "processed": entry is not None, } if entry: item["downmix"] = { "source_layout": entry.downmix.source_layout, "provenance": entry.downmix.provenance, "pan_expr": entry.downmix.pan_expr, "weights_left": [ {"channel": c, "weight": w} for c, w in entry.downmix.weights_left ], } item["target_sample_rate"] = entry.target_sample_rate item["source_timeline"] = (shallow_dataclass_dict(entry.source_timeline) if entry.source_timeline else None) item["preserve_timestamp_gaps"] = entry.preserve_gaps item["profile"] = { "name": entry.profile.name, "nero_quality": entry.profile.nero_quality, } item["normalization"] = { "mode": entry.normalization.mode, "target_tp": entry.normalization.target_tp, "overdrive_step": entry.normalization.overdrive_step, "overdrive_db": entry.normalization.overdrive_db, "isolated_peak_guard": entry.normalization.isolated_peak_guard, "limiter": entry.normalization.limiter, } item["mux_timing"] = { "expected_relative_offset": entry.mux_offset, "input_timestamp_offset": entry.applied_mux_offset, "note": entry.mux_offset_note, } if analyzed and entry.analysis and entry.gain: item["analysis"] = { "n_samples": entry.analysis.n_samples, "sample_peak_dbfs": entry.analysis.max_volume, "true_peak_dbtp": entry.analysis.input_tp, "mean_dbfs": entry.analysis.mean_volume, "integrated_lufs": entry.analysis.input_i, "lra_lu": entry.analysis.input_lra, "threshold_lufs": entry.analysis.input_thresh, "histogram": { str(k): v for k, v in sorted(entry.analysis.histogram.items()) }, } item["gain"] = { "peak_gain_db": entry.gain.peak_gain, "sample_peak_gain_db": entry.gain.sample_peak_gain, "overdrive_candidate_db": entry.gain.g_hist, "overdrive_auto_db_estimated": entry.gain.overdrive_auto, "overdrive_manual_db": entry.gain.overdrive_manual, "total_gain_db": entry.gain.total_gain, "limiter_active": entry.gain.limiter_active, "notes": entry.gain.notes, } audio.append(item) return { "schema_version": JSON_SCHEMA_VERSION, "application": { "name": APP_NAME, "display_name": DISPLAY_NAME, "version": VERSION, }, "source": { "path": str(plan.source.path), "format": plan.source.format_name, "duration": plan.source.duration, "start_time": plan.source.start_time, "chapters": plan.source.chapter_count, }, "output": str(plan.output), "video": [ { "index": v.index, "codec": v.codec, "width": v.width, "height": v.height, "sar_num": v.sar_num, "sar_den": v.sar_den, "display_width": v.display_width, "time_raw": _time_json(v.time), "content_anchor": _content_json(v.content), } for v in plan.source.video() ], "audio": audio, "other_streams": [ {"index": s.index, "kind": s.kind, "codec": s.codec, "tags_raw": s.tags_raw} for s in plan.source.streams if s.kind not in ("video", "audio") ], "analyzed": analyzed, } def dry_run_payload(tools: Tools, plan: JobPlan, analyzed: bool) -> dict[str, Any]: tracks = [] for index, track in enumerate(plan.tracks): ffmpeg_argv, nero_argv = encode_commands(tools, plan, track, index) tracks.append({ "audio_index": track.source.audio_index, "analysis_command": analysis_command(tools, plan.source.path, track), "encode_ffmpeg_command": ffmpeg_argv, "encode_nero_command": nero_argv, "gain_known": track.gain is not None, "expected_relative_offset": track.mux_offset, "mux_input_offset": track.applied_mux_offset, "mux_offset_known": track.mux_offset is not None and track.applied_mux_offset is not None, "mux_offset_note": track.mux_offset_note, }) return { "schema_version": JSON_SCHEMA_VERSION, "application": {"name": APP_NAME, "version": VERSION}, "mode": "dry-run", "analyzed": analyzed, "plan": inspect_json(plan, analyzed), "tracks": tracks, "mux_command": mux_command(tools, plan), } def dry_run_text(tools: Tools, plan: JobPlan, analyzed: bool) -> str: payload = dry_run_payload(tools, plan, analyzed) out = [inspect_text(plan, analyzed), "Planned command lines", ""] for track in payload["tracks"]: out.append(f" Track {track['audio_index']}, analysis") out.append(" " + " ".join(shlex.quote(a) for a in track["analysis_command"])) out.append(f" Track {track['audio_index']}, Encoding") out.append( " " + " ".join(shlex.quote(a) for a in track["encode_ffmpeg_command"]) + " | " + " ".join(shlex.quote(a) for a in track["encode_nero_command"]) ) expected = track.get("expected_relative_offset") applied = track.get("mux_input_offset") fmt = lambda v: "not verifiable" if v is None else f"{v:+.6f} s" out.append(f" Source offset relative to reference {fmt(expected)}") out.append( f" Mux input offset {fmt(applied)} " f"({track.get('mux_offset_note', '')})" ) out.append("") out.append(" Mux") out.append(" " + " ".join(shlex.quote(a) for a in payload["mux_command"])) if analyzed: out.append( "\\n Gain is calculated. Source offset and mux input offset " "do not depend on the gain." ) else: out.append( "\\n The gain is a placeholder and is calculated with --analyze. " "The mux_offset is already fixed." ) return "\n".join(out) # =========================================================================== # 11 CLI # =========================================================================== def collect_inputs(paths: list[str]) -> list[Path]: """Collects files directly or from directories, not recursively.""" files_out: list[Path] = [] seen: set[Path] = set() for item in paths: path = Path(item).expanduser() if path.is_dir(): candidates = [ child for child in sorted(path.iterdir()) if child.is_file() and child.suffix.lower() in MEDIA_SUFFIXES ] elif path.is_file(): candidates = [path] else: log.error("Input not found. %s", path) continue for candidate in candidates: key = candidate.resolve() if key not in seen: seen.add(key) files_out.append(candidate) return files_out def validate_args(parser: argparse.ArgumentParser, args: argparse.Namespace) -> None: if args.channels != 2: parser.error("Only --channels 2 is allowed in this version.") if args.quality is not None and not (0.0 <= args.quality <= 1.0): parser.error("--quality must be between 0 and 1.") if not math.isfinite(args.max_gain) or args.max_gain < 0: parser.error("--max-gain must be finite and >= 0.") if not math.isfinite(args.peak_tolerance) or args.peak_tolerance < 0: parser.error("--peak-tolerance must be finite and >= 0.") if not (0 <= args.peak_retries <= PEAK_RETRIES_MAX): parser.error(f"--peak-retries must be between 0 and {PEAK_RETRIES_MAX}.") if not math.isfinite(args.target_tp) or args.target_tp > 0.0 or args.target_tp < -20.0: parser.error("--target-tp must be between -20 and 0 dBTP.") if not math.isfinite(args.overdrive_db): parser.error("--overdrive-db must be finite.") if args.samplerate not in (None, "source"): rate = _as_int(args.samplerate) if rate is None or rate <= 0: parser.error("--samplerate expects a positive integer or 'source'.") if args.json and not (args.inspect or args.dry_run): parser.error("--json requires --inspect or --dry-run.") if args.analyze and not (args.inspect or args.dry_run): parser.error("--analyze requires --inspect or --dry-run.") if args.audio_workers < 0: parser.error("--audio-workers must be >= 0 (0 = automatic).") if not math.isfinite(args.peak_warn_limit) or args.peak_warn_limit < 0: parser.error("--peak-warn-limit must be finite and >= 0 (0 = no warning release).") if 0 < args.peak_warn_limit < args.peak_tolerance: parser.error("--peak-warn-limit must be 0 or >= --peak-tolerance.") if args.turbo and args.verify == "full": parser.error("--turbo and --verify full are mutually exclusive.") def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog=APP_NAME, description="Processes the audio tracks of video files and verifies the result.", ) parser.add_argument("inputs", nargs="+", metavar="INPUT", help="file or directory, not recursive") group = parser.add_argument_group("Track selection") group.add_argument("-T", "--tracks", default="all", help="all, or comma-separated audio indices") group.add_argument("-l", "--lang", type=lambda s: s.split(","), help="language filter, comma-separated, for example ger,eng") group = parser.add_argument_group("Profile") group.add_argument( "-p", "--profile", choices=("auto", "sq", "hq", "normal"), default="auto", help="auto (SD->SQ, otherwise HQ), sq=32 kHz/q0.40, hq=44.1 kHz/q0.36; normal=alias for sq", ) group.add_argument("-s", "--samplerate", help="positive target rate, or source") group.add_argument("-Q", "--quality", type=float, help="Nero quality value 0..1") group = parser.add_argument_group("Audio") group.add_argument("-c", "--channels", type=int, default=2, help="output channels; this version supports 2 only") group.add_argument("--assume-layout", action="append", metavar="TRACK=LAYOUT", help="declare the channel layout of an audio track, for example 0=5.1(side)") group.add_argument("--downmix", choices=("classic", "ffmpeg"), default="classic", help="classic = layout-aware matrix (default), ffmpeg = FFmpeg standard downmix") group = parser.add_argument_group("Normalisation") group.add_argument("-n", "--normalize", choices=("peak", "off"), default="peak", help="peak = normalise to the true-peak target (default), off = no gain") group.add_argument("--target-tp", type=float, default=-1.0, metavar="DBTP", help="true-peak target after AAC decoding; default -1.0") group.add_argument("--max-gain", type=float, default=20.0, metavar="DB", help="upper limit for positive gain; default 20.0") group.add_argument("-O", "--overdrive", choices=sorted(OVERDRIVE_QUOTAS), default="off", help="extra gain from a small clipping budget: " "off, light, medium, hard or brutal") group.add_argument("-m", dest="overdrive", action="store_const", const="medium", help="shortcut for --overdrive medium") group.add_argument("--overdrive-db", type=float, default=0.0, metavar="DB", help="manual gain offset added to the calculated gain; default 0.0") group.add_argument( "--isolated-peak-guard", choices=("auto", "off"), default="auto", help="automatically ignores only very few isolated full-scale outliers; default auto", ) group.add_argument("--limiter", choices=("on", "off", "auto"), default="auto", help="auto = on when overdrive or the isolated peak guard adds gain (default)") group.add_argument("--peak-tolerance", type=float, default=0.2, metavar="DB", help="allowed excess over the true-peak target without retry; default 0.2") group.add_argument("--peak-retries", type=int, default=3, metavar="N", help="maximum peak retries after the first encode; default 3") group.add_argument("--peak-warn-limit", type=float, default=0.6, metavar="DB", help="a peak excess up to this amount counts as " "a warning instead of an error. Default 0.6") group.add_argument("--peak-nonconform", choices=("error", "warn"), default="error", help="result when the peak target is still missed after all retries; default error") group = parser.add_argument_group("Output") group.add_argument("-o", "--out", type=Path, metavar="DIRECTORY", help="output directory; default is the directory of the source file") group.add_argument("--name-template", default="{stem} {res}{lang}", metavar="TEMPLATE", help="output file name; default \"{stem} {res}{lang}\"") group.add_argument("-f", "--force", action="store_true", help="replace an existing output file after successful verification") group.add_argument("--keep-temp", action="store_true", help="keep the working directory; a keep marker protects it from automatic clean-up") group.add_argument("--on-incompatible-stream", choices=("drop", "fail"), default="fail", help="streams Matroska cannot hold: fail (default) or drop them") group.add_argument("--timing-log", default=None, metavar="FILE", help="opt in to reading and writing a timing log at FILE; " "disabled by default; contains no names or paths") group.add_argument("--no-timing-log", action="store_true", help="disable timing-log reads and writes, including an explicit --timing-log") group = parser.add_argument_group("Tools") group.add_argument("--ffmpeg", metavar="PATH", help="path to ffmpeg") group.add_argument("--ffprobe", metavar="PATH", help="path to ffprobe") group.add_argument("--nero", metavar="PATH", help="path to neroAacEnc") group = parser.add_argument_group("Modes") mode = group.add_mutually_exclusive_group(required=False) mode.add_argument("-i", "--inspect", action="store_true", help="show streams, plan and timing without processing") mode.add_argument("-d", "--dry-run", action="store_true", help="show the planned command lines without running them") group.add_argument("-a", "--analyze", action="store_true", help="with --inspect or --dry-run: also run the level analysis") group.add_argument("-j", "--json", action="store_true", help="with --inspect or --dry-run: output JSON") group.add_argument("--stop-on-error", action="store_true", help="stop the batch after the first failed file") group.add_argument("--log", type=Path, help="additional session log; every output also gets its own .log automatically") group.add_argument("--verify", choices=("fast", "full"), default="fast", help="fast = structure, timing and complete AAC stream-copy proof (default); full = additionally a complete A/V decode") group.add_argument("--no-progress", action="store_true", help="disable the live progress display") group.add_argument("--serial-audio", action="store_true", help="disable the multitrack spool and parallel processing; process audio serially") group.add_argument("--audio-workers", type=int, default=0, metavar="N", help="maximum parallel audio workers; 0 = automatic (CPU cores minus one)") group.add_argument("-t", "--turbo", action="store_true", help="fast: no post-AAC true-peak measurement or retry and no final full-track or video hash checks; structure and timing checks remain") group.add_argument("-v", "--verbose", action="count", default=0, help="-v progress and track overview, -vv processing details, -vvv additional details, -vvvv full debug and command lines; more v act like -vvvv") group.add_argument("-q", "--quiet", action="store_true", help="show errors only") group.add_argument("--license", action="store_true", help="print the licence text and exit") group.add_argument("--version", action="store_true", help="show version and status of the dependencies") return parser # =========================================================================== # 12 Execution, encoder, mux, verification # =========================================================================== def run_pipe(producer: list[str], consumer: list[str], workdir: Path, label: str, *, duration: float | None = None, progress_label: str | None = None) -> tuple[StepResult, StepResult]: err_a = workdir / f"{label}-producer.stderr" err_b = workdir / f"{label}-consumer.stderr" progress_path = workdir / f"{label}-progress.txt" actual_producer = _with_ffmpeg_progress(producer, progress_path) log.debug("start %s", " ".join(shlex.quote(a) for a in actual_producer)) log.debug("pipe %s", " ".join(shlex.quote(a) for a in consumer)) started = time.monotonic() with open(err_a, "wb") as fa, open(err_b, "wb") as fb: first = subprocess.Popen( actual_producer, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=fa, shell=False, ) assert first.stdout is not None try: second = subprocess.Popen( consumer, stdin=first.stdout, stdout=subprocess.DEVNULL, stderr=fb, shell=False, ) except OSError as exc: first.stdout.close() first.kill() first.wait() raise StepFailed(f"Consumer process could not be started: {consumer[0]}") from exc first.stdout.close() try: while second.poll() is None: if progress_label: _show_progress(progress_label, duration, _progress_seconds(progress_path), elapsed=time.monotonic() - started) time.sleep(PROGRESS_REFRESH) except BaseException: _terminate_processes([first, second]) raise rc_second = second.returncode if rc_second != 0 and first.poll() is None: first.terminate() try: first.wait(timeout=5) except subprocess.TimeoutExpired: first.kill() rc_first = first.wait() seconds = time.monotonic() - started current = _progress_seconds(progress_path) if progress_label: _show_progress(progress_label, duration, current, elapsed=seconds, done=True) media_seconds = duration if duration is not None and duration > 0 else current speed = (media_seconds / seconds) if media_seconds is not None and seconds > 0 else None log.info( "%s finished, elapsed %s%s", progress_label, _format_clock(seconds), f", {speed:.1f}x realtime" if speed is not None else "", ) telemetry = globals().get("TIMING") if telemetry is not None: telemetry.step(progress_label, duration, seconds) def read(path: Path) -> str: try: return path.read_text("utf-8", "replace") except OSError: return "" progress_path.unlink(missing_ok=True) return ( StepResult(list(producer), rc_first, read(err_a), seconds), StepResult(list(consumer), rc_second, read(err_b), seconds), ) def encode_track(tools: Tools, plan: JobPlan, track: TrackPlan, index: int, workdir: Path, *, source_path: Path | None = None, input_audio_index: int | None = None, retry_reason: str = "") -> Path: if tools.nero is None: raise StepFailed("neroAacEnc is required for encoding.") target = workdir / f"track{index}.m4a" chain = filter_chain(track, with_gain=True) source = source_path or plan.source.path audio_index = track.source.audio_index if input_audio_index is None else input_audio_index producer = [ tools.ffmpeg, "-hide_banner", "-v", "error", "-nostdin", "-i", str(source), "-map", f"0:a:{audio_index}", "-af", chain, "-c:a", "pcm_s16le", "-f", "wav", "-", ] consumer = [ tools.nero, "-ignorelength", "-q", f"{track.profile.nero_quality:.2f}", "-lc", "-if", "-", "-of", str(target), ] a, b = run_pipe( producer, consumer, workdir, f"encode{index}", duration=plan.source.duration, progress_label=( f"Encoding Audio {track.source.audio_index} Retry {retry_reason}" if retry_reason else f"Encoding Audio {track.source.audio_index}" ), ) if b.returncode != 0: raise StepFailed( f"neroAacEnc Returncode {b.returncode} for track {track.source.audio_index}.", b ) if a.returncode != 0: raise StepFailed( f"FFmpeg Returncode {a.returncode} for track {track.source.audio_index}.", a ) if not target.is_file() or target.stat().st_size == 0: raise StepFailed(f"The encoder output is missing or empty. {target}") return target def parse_true_peak(stderr: str) -> float | None: values: list[float] = [] lines = stderr.splitlines() for position, line in enumerate(lines): if "True peak" in line: for follow in lines[position:position + 3]: match = re.search(r"Peak:\s*(-inf|[+-]?\d+(?:\.\d+)?)\s+dB", follow) if match: values.append(float(match.group(1))) break return values[-1] if values else None def measure_true_peak(tools: Tools, path: Path, *, duration: float | None = None, label: str = "True peak check") -> float | None: # No additional atrim based on skip_samples: FFmpeg already applies the # skip-samples side data while decoding, before the filter graph receives # the samples. A second trim would therefore remove real programme # material at the start. This was verified with an AAC impulse at t=0: # despite 1024 signalled skip samples, the first programme sample was # decoded at index 0 again. result = run_ffmpeg_progress([ tools.ffmpeg, "-hide_banner", "-v", "info", "-nostdin", "-i", str(path), "-af", "ebur128=peak=true", "-f", "null", "-", ], duration=duration, label=label) if result.returncode != 0: return None return parse_true_peak(result.stderr) def measure_true_peaks_parallel(tools: Tools, tracks: list[TrackPlan], *, duration: float | None, workdir: Path, retry_reasons: dict[int, str] | None = None, workers: int | None = None) -> None: ready = [track for track in tracks if track.encoded_path is not None] if not ready: return if workers is not None: workers = max(1, workers) if len(ready) > workers: for first in range(0, len(ready), workers): measure_true_peaks_parallel( tools, ready[first:first + workers], duration=duration, workdir=workdir, retry_reasons=retry_reasons, ) return if len(ready) == 1: track = ready[0] reason = (retry_reasons or {}).get(track.source.audio_index, "") label_one = f"True Peak Audio {track.source.audio_index}" if reason: label_one += f" Retry {reason}" track.measured_tp = measure_true_peak( tools, track.encoded_path, duration=duration, label=label_one, ) return procs: list[subprocess.Popen[Any]] = [] progress_paths: list[Path] = [] stderr_paths: list[Path] = [] handles: list[Any] = [] started = time.monotonic() label = f"True Peak {len(ready)} audio tracks in parallel" if retry_reasons: reasons = [retry_reasons.get(track.source.audio_index, "") for track in ready] reasons = [reason for reason in reasons if reason] if reasons: label += " Retry " + ", ".join(reasons) try: for pos, track in enumerate(ready): progress = workdir / f"truepeak-{pos}.progress" stderr = workdir / f"truepeak-{pos}.stderr" handle = open(stderr, "wb") handles.append(handle) # As in measure_true_peak(): no second trim of skip_samples. # The FFmpeg decoder has already applied them. argv = [ tools.ffmpeg, "-hide_banner", "-v", "info", "-nostdin", "-i", str(track.encoded_path), "-af", "ebur128=peak=true", "-f", "null", "-", ] actual = _with_ffmpeg_progress(argv, progress) log.debug("start %s", " ".join(shlex.quote(a) for a in actual)) procs.append(subprocess.Popen( actual, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=handle, shell=False, )) progress_paths.append(progress) stderr_paths.append(stderr) while any(proc.poll() is None for proc in procs): values: list[float] = [] for proc, progress in zip(procs, progress_paths): if proc.poll() is not None and duration is not None: values.append(duration) else: value = _progress_seconds(progress) if value is not None: values.append(value) current = (sum(values) / len(ready)) if len(values) == len(ready) else None _show_progress(label, duration, current, elapsed=time.monotonic() - started) time.sleep(PROGRESS_REFRESH) elapsed = time.monotonic() - started _show_progress(label, duration, duration, elapsed=elapsed, done=True) speed = (duration / elapsed) if duration is not None and duration > 0 and elapsed > 0 else None log.info( "%s finished, elapsed %s%s", label, _format_clock(elapsed), f", {speed:.1f}x realtime" if speed is not None else "", ) telemetry = globals().get("TIMING") if telemetry is not None: telemetry.step(label, duration, elapsed, n=len(ready)) for track, proc, stderr_path in zip(ready, procs, stderr_paths): stderr = stderr_path.read_text("utf-8", "replace") if stderr_path.exists() else "" track.measured_tp = parse_true_peak(stderr) if proc.returncode == 0 else None except BaseException: _terminate_processes(procs) raise finally: for handle in handles: try: handle.close() except OSError: pass for path in progress_paths: path.unlink(missing_ok=True) def peak_attempt_requires_retry(track: TrackPlan, args: argparse.Namespace) -> bool: rules = track.normalization attempts = track.attempts if track.measured_tp is None: track.conform = None if rules.mode == "peak": raise StepFailed( f"Track {track.source.audio_index}: true peak of the generated AAC track " "could not be measured. Peak compliance cannot be proven." ) log.warning( "Track %d: true peak not measurable. Mode off, no retry.", track.source.audio_index, ) return False if rules.mode == "off": track.conform = None log.info( "Track %d: true peak %.2f dBTP. Mode off, no retry.", track.source.audio_index, track.measured_tp, ) return False excess = track.measured_tp - rules.target_tp if excess <= rules.tolerance + 1e-9: track.conform = True log.info( "Track %d: true peak %.2f dBTP, target %.2f dBTP, compliant.", track.source.audio_index, track.measured_tp, rules.target_tp, ) return False # User rule: small remaining excesses are accepted at once as a # warning. This saves a complete re-encode and prevents an already # usable result from becoming quieter without need through an # aggressive control step. limit = getattr(args, "peak_warn_limit", 0.6) if limit > 0 and excess <= limit + 1e-9: track.conform = False warning = ( f"Track {track.source.audio_index}: True Peak {track.measured_tp:+.2f} dBTP; " f"Target {rules.target_tp:+.2f} dBTP; excess {excess:.2f} dB" ) log.warning("%s; within the warning limit of %.2f dB.", warning, limit) reporter = globals().get("REPORTER") if reporter is not None: reporter.note_warning( f"A{track.source.audio_index} Peak +{excess:.2f}dB above target" ) return False if attempts > rules.retries or track.gain is None: track.conform = False message = ( f"Track {track.source.audio_index}: True Peak {track.measured_tp:+.2f} dBTP; " f"Target {rules.target_tp:+.2f} dBTP; excess {excess:.2f} dB " f"after {attempts} attempts." ) if args.peak_nonconform == "error": raise StepFailed( message, compact_message=( f"A{track.source.audio_index} Peak +{excess:.2f}dB above target " f"after {attempts} attempt{'s' if attempts != 1 else ''}" ), ) log.warning("%s Continuing, because --peak-nonconform warn is set.", message) reporter = globals().get("REPORTER") if reporter is not None: reporter.note_warning(f"A{track.source.audio_index} Peak +{excess:.2f}dB above target") return False # Non-linear codec response: the real problem file showed between the # first two points only 0.22 dB peak response per dB of gain, but about # 0.93 at the next larger step. A local slope is therefore no reliable # model for the next step. It is only logged for diagnostics. # Operationally we correct directly by the current error. track.peak_history.append((track.gain.total_gain, track.measured_tp)) observed_response: float | None = None if len(track.peak_history) >= 2: (gain_before, peak_before) = track.peak_history[-2] (gain_now, peak_now) = track.peak_history[-1] delta_gain = gain_now - gain_before if abs(delta_gain) > 0.05: observed_response = (peak_now - peak_before) / delta_gain needed = (rules.target_tp - 0.5 * rules.tolerance) - track.measured_tp correction = max(needed, -PEAK_CORRECTION_LIMIT_DB) track.gain.total_gain += correction reporter = globals().get("REPORTER") if reporter is not None: reporter.set_track_gain(track.source.audio_index, track.gain.total_gain) track.gain.notes.append( f"Retry {correction:+.2f} dB after measurement {track.measured_tp:.2f} dBTP" ) diagnostic = ( "unknown" if observed_response is None else f"{observed_response:.2f} dB per dB (diagnostics only)" ) log.info( "Track %d: true peak %+.2f dBTP; excess %.2f dB above target %+.2f dBTP; " "retry with %+.2f dB. Last observed response: %s.", track.source.audio_index, track.measured_tp, excess, rules.target_tp, correction, diagnostic, ) return True def encode_with_peak_control(tools: Tools, plan: JobPlan, track: TrackPlan, index: int, workdir: Path, args: argparse.Namespace, *, source_path: Path | None = None, input_audio_index: int | None = None) -> None: track.attempts = 0 retry_reason = "" while True: track.attempts += 1 track.encoded_path = encode_track( tools, plan, track, index, workdir, source_path=source_path, input_audio_index=input_audio_index, retry_reason=retry_reason, ) track.encoder_timing = evaluate_encoder_timing(tools, track.encoded_path) if args.turbo: track.measured_tp = None track.conform = None log.info("Turbo: post-AAC true-peak check for track %d skipped.", track.source.audio_index) return peak_label = f"True Peak Audio {track.source.audio_index}" if retry_reason: peak_label += f" Retry {retry_reason}" track.measured_tp = measure_true_peak( tools, track.encoded_path, duration=plan.source.duration, label=peak_label, ) if not peak_attempt_requires_retry(track, args): return telemetry = globals().get("TIMING") if telemetry is not None: telemetry.retry_round(track.attempts, 1) excess = (track.measured_tp - track.normalization.target_tp) if track.measured_tp is not None else 0.0 retry_reason = f"A{track.source.audio_index} Peak +{excess:.2f}dB" def encode_tracks_spool_with_peak_control(tools: Tools, plan: JobPlan, spool: Path, workdir: Path, args: argparse.Namespace, workers: int) -> None: """Encodes multitrack material and runs peak retries round by round. 0.5.1 processed required retries one track after another. With two problematic tracks this wasted almost half of the retry time on a multicore system. Since 0.5.2 each round collects all tracks that are still too loud, re-encodes them in parallel from the same spool and then measures their true peaks in parallel as well. """ encode_tracks_spool_initial(tools, plan, spool, workdir, workers) for track in plan.tracks: assert track.encoded_path is not None track.encoder_timing = evaluate_encoder_timing(tools, track.encoded_path) if args.turbo: for track in plan.tracks: track.measured_tp = None track.conform = None log.info("Turbo: post-AAC true-peak check and peak retries skipped.") return measure_true_peaks_parallel( tools, plan.tracks, duration=plan.source.duration, workdir=workdir, workers=workers ) pending_indices = list(range(len(plan.tracks))) retry_round_no = 0 while True: retry_indices: list[int] = [] retry_reasons_by_plan: dict[int, str] = {} retry_reasons_by_audio: dict[int, str] = {} # peak_attempt_requires_retry decides and already sets the new gain. # Only then do we increase the attempt count for the re-encode that # actually follows. for index in pending_indices: track = plan.tracks[index] if not peak_attempt_requires_retry(track, args): continue track.attempts += 1 excess = ( track.measured_tp - track.normalization.target_tp if track.measured_tp is not None else 0.0 ) reason = f"A{track.source.audio_index} Peak +{excess:.2f}dB" retry_indices.append(index) retry_reasons_by_plan[index] = reason retry_reasons_by_audio[track.source.audio_index] = reason if not retry_indices: return retry_round_no += 1 telemetry = globals().get("TIMING") if telemetry is not None: telemetry.retry_round(retry_round_no, len(retry_indices)) for first in range(0, len(retry_indices), workers): batch = retry_indices[first:first + workers] if len(batch) == 1: index = batch[0] track = plan.tracks[index] track.encoded_path = encode_track( tools, plan, track, index, workdir, source_path=spool, input_audio_index=index, retry_reason=retry_reasons_by_plan[index], ) else: _encode_spool_batch( tools, plan, spool, batch, workdir, retry_reasons=retry_reasons_by_plan, ) retry_tracks = [plan.tracks[index] for index in retry_indices] for track in retry_tracks: assert track.encoded_path is not None track.encoder_timing = evaluate_encoder_timing(tools, track.encoded_path) measure_true_peaks_parallel( tools, retry_tracks, duration=plan.source.duration, workdir=workdir, retry_reasons=retry_reasons_by_audio, workers=workers, ) pending_indices = retry_indices # Single retry steps must not grow arbitrarily large. The correction # deliberately does not extrapolate a local codec slope, because it # was measured to be strongly non-linear. PEAK_CORRECTION_LIMIT_DB = 3.0 MUX_CORRECTION_ATTEMPTS = 2 def create_candidate_output(final_output: Path) -> Path: """Creates a hidden candidate in the same directory and file system.""" final_output.parent.mkdir(parents=True, exist_ok=True) fd, name = tempfile.mkstemp( prefix=f".{final_output.stem}.trackshepherd-", suffix=".mkv", dir=final_output.parent, ) os.close(fd) return Path(name) def commit_output(plan: JobPlan, force: bool) -> None: candidate = plan.candidate_output if candidate is None or not candidate.is_file() or candidate.stat().st_size == 0: raise StepFailed("Verified candidate file missing. Commit aborted.") if plan.output.exists() and not force: raise StepFailed( f"Target file appeared before the commit. The existing file stays untouched: {plan.output}" ) if force: os.replace(candidate, plan.output) else: # Publish atomically without replacing a target that appeared in the # meantime. exists() plus replace() would leave a time window here. try: os.link(candidate, plan.output) except FileExistsError as exc: raise StepFailed(f"Target file appeared before the commit: {plan.output}") from exc except OSError as exc: raise StepFailed(f"Atomic commit without overwriting is not possible: {exc}") from exc try: candidate.unlink() except OSError as exc: log.warning("Output published. The candidate link could not be removed: %s", exc) plan.candidate_output = None def run_mux(tools: Tools, plan: JobPlan) -> StepResult: argv = mux_command(tools, plan) result = run_ffmpeg_progress(argv, duration=plan.source.duration, label="Mux") if result.returncode != 0: raise StepFailed(f"The mux failed with return code {result.returncode}.", result) output = active_output_path(plan) if not output.is_file() or output.stat().st_size == 0: raise StepFailed(f"The mux output is missing or empty. {output}") return result def probe_and_verify_timeline(tools: Tools, plan: JobPlan) -> tuple[MediaFile, list[TimelineCheck]]: started = time.monotonic() output_path = active_output_path(plan) log_detail("Timing check: output anchors and stream-copy reference are checked.") output = probe_media(tools, output_path, full_time_anchors=True) checks = verify_output_timeline(tools, output, plan) log_timeline_diagnostics(plan, output, checks) elapsed = time.monotonic() - started log_detail("Timing check finished, elapsed %s.", _format_clock(elapsed)) return output, checks def mux_with_timeline_control( tools: Tools, plan: JobPlan ) -> tuple[StepResult, MediaFile, list[TimelineCheck]]: result = run_mux(tools, plan) output, checks = probe_and_verify_timeline(tools, plan) for attempt in range(1, MUX_CORRECTION_ATTEMPTS + 1): deviations = { c.audio_index: c.deviation for c in checks if c.status == "deviation" and c.deviation is not None } if not deviations: return result, output, checks if attempt > MUX_CORRECTION_ATTEMPTS - 1: return result, output, checks # Safety net for CodecDelay sources. The 0.5.0 counter-review found # that the earlier reproducible error came from the wrong start formula # A-max(S,0). The verifier was right in those cases. # With A-S this lock should therefore not trigger in normal operation. # If a deviation still remains, start model and measurement contradict # each other in a historically sensitive case. Fail-closed then applies # instead of a speculative automatic correction. codec_delay_tracks = [ track for track in plan.tracks if track.source.audio_index in deviations and (track.source.time.initial_padding or 0) > 0 ] if codec_delay_tracks: for track in codec_delay_tracks: delay = track.source.time.padding_seconds() track.mux_offset_note += ( "; automatic correction blocked: " "source CodecDelay and verifier not yet reconciled" ) log.error( "Track %d: timing correction is NOT applied. " "The source carries CodecDelay%s; despite A-S an unexplained " "deviation remains. Fail-closed.", track.source.audio_index, "" if delay is None else f" ({delay * 1000:.2f} ms)", ) return result, output, checks # Save the state before the correction. A correction must not make an # already correct file worse. On Matroska sources with CodecDelay the # control loop did exactly that. previous_offsets = {t.source.audio_index: t.applied_mux_offset for t in plan.tracks} previous_notes = {t.source.audio_index: t.mux_offset_note for t in plan.tracks} previous_worst = max(abs(v) for v in deviations.values()) for track in plan.tracks: deviation = deviations.get(track.source.audio_index) if deviation is None: continue base = track.applied_mux_offset or 0.0 track.applied_mux_offset = base - deviation track.mux_offset_note = ( f"corrected by {-deviation * 1000:+.2f} ms after measurement on the output" ) log.warning( "Track %d: timing off by %+.2f ms. Safety net: " "mux is repeated once with %+.6f s.", track.source.audio_index, deviation * 1000, track.applied_mux_offset, ) result = run_mux(tools, plan) output, checks = probe_and_verify_timeline(tools, plan) new_worst = max( (abs(c.deviation) for c in checks if c.status == "deviation" and c.deviation is not None), default=0.0, ) if new_worst >= previous_worst: log.warning( "The correction did not improve the timing " "(%.2f ms against %.2f ms). The previous state is " "restored.", new_worst * 1000, previous_worst * 1000, ) for track in plan.tracks: track.applied_mux_offset = previous_offsets[track.source.audio_index] track.mux_offset_note = ( previous_notes[track.source.audio_index] + "; correction discarded, no improvement" ) result = run_mux(tools, plan) output, checks = probe_and_verify_timeline(tools, plan) return result, output, checks def verify_output(tools: Tools, plan: JobPlan, output: MediaFile, timeline_checks: list[TimelineCheck], *, full_decode: bool = False, turbo: bool = False) -> list[str]: problems: list[str] = [] output_path = active_output_path(plan) for kind in ("video", "audio", "subtitle", "attachment"): want = len(plan.source.of_kind(kind)) have = len(output.of_kind(kind)) if want != have: problems.append(f"Stream count {kind}: expected {want}, found {have}") if plan.source.chapter_count != output.chapter_count: problems.append( f"Chapters: expected {plan.source.chapter_count}, found {output.chapter_count}" ) if plan.source.duration is not None and output.duration is not None: # FFmpeg removes a positive container start time while reading. The # output then starts at zero and is shorter by exactly this amount. # The Matroska format.duration value can also be outdated or larger # than the packets actually present. A notable difference is therefore # not accepted blindly but cross-checked against the real packet time # span of both files. source_span = plan.source.duration - max(plan.source.start_time or 0.0, 0.0) priming = 0.0 for track in plan.tracks: timing = track.encoder_timing if timing is not None and timing.priming_seconds: priming = max(priming, timing.priming_seconds) priming = max(priming, 2048.0 / max(track.target_sample_rate, 1)) allowed = 0.1 + priming delta = abs(output.duration - source_span) if delta > allowed: source_packets = packet_timeline_span(tools, plan.source.path) output_packets = packet_timeline_span(tools, output_path) packet_delta = ( abs(output_packets - source_packets) if source_packets is not None and output_packets is not None else None ) if packet_delta is not None and packet_delta <= allowed: log.warning( "Container duration deviates by %.0f ms; the real packet time span " "matches within %.0f ms (source %.3f s, output %.3f s). " "The duration metadata is not treated as content loss.", delta * 1000, packet_delta * 1000, source_packets, output_packets, ) else: malformed_tail = _malformed_processed_audio_tail_is_safe( tools, plan, output, output_path, allowed ) if malformed_tail is not None: log.warning( "Container or packet duration deviates by %.0f ms, but the " "stream-copied picture timeline and the decoded audio duration " "support a malformed source packet tail within the stated " "duration tolerances. Source decoder warnings still apply. %s", delta * 1000, malformed_tail, ) else: detail = "" if source_packets is not None and output_packets is not None: detail = ( f"; packet time span source {source_packets:.3f} s, " f"output {output_packets:.3f} s, " f"deviation {packet_delta * 1000:.0f} ms" ) problems.append( f"Duration deviates by {delta * 1000:.0f} ms, " f"allowed would be {allowed * 1000:.0f} ms{detail}" ) produced = {a.audio_index: a for a in output.audio()} processed_indices = {t.source.audio_index for t in plan.tracks} for track in plan.tracks: target = produced.get(track.source.audio_index) if target is None: problems.append(f"Audio track {track.source.audio_index} missing in the output") continue if target.codec != "aac": problems.append( f"Audio track {track.source.audio_index}: codec {target.codec} instead of aac" ) if target.channels != 2: problems.append( f"Audio track {track.source.audio_index}: {target.channels} channels instead of 2" ) if target.sample_rate != track.target_sample_rate: problems.append( f"Audio track {track.source.audio_index}: {target.sample_rate} Hz " f"instead of {track.target_sample_rate} Hz" ) timeline = track.source_timeline if timeline is not None: expected_pcm = timeline.span if track.preserve_gaps else timeline.decoded_duration actual_pcm = _decoded_program_duration(track) if actual_pcm is None or abs(actual_pcm - expected_pcm) > TIMELINE_TOLERANCE + 1 / track.target_sample_rate: problems.append(f"Audio track {track.source.audio_index}: analysed PCM duration " "does not match the audited source timeline") # Global container duration can hide a shortened audio track behind # a longer video track. Check every processed AAC track separately. audio_packets = packet_timeline_stats(tools, output_path, f"a:{track.source.audio_index}") priming = (track.encoder_timing.priming_seconds or 0.0) if track.encoder_timing else 0.0 audio_allowed = 0.1 + max(0.0, priming, 2048.0 / track.target_sample_rate) if (audio_packets is None or abs(audio_packets.span - expected_pcm - priming) > audio_allowed): problems.append(f"Audio track {track.source.audio_index}: encoded duration " f"does not match the audited source programme " f"(packets={audio_packets}, expected PCM={expected_pcm:.6f} s, " f"priming={priming:.6f} s, allowance={audio_allowed:.6f} s)") # True peak was measured on the generated M4A. For this proof to hold # for the final MKV as well, the AAC bitstream contained there must be # identical over the ENTIRE track, not just in a 500-packet sample. if not turbo: if track.encoded_path is None or not streamcopy_is_identical( tools, track.encoded_path, "a:0", output_path, f"a:{track.source.audio_index}", limit=None ): problems.append( f"Audio track {track.source.audio_index}: complete " f"{PACKET_HASH_ALGORITHM} proof M4A->MKV failed" ) for source_track in plan.source.audio(): target = produced.get(source_track.audio_index) if target is None: continue if source_track.audio_index not in processed_indices and target.codec != source_track.codec: problems.append( f"Audio track {source_track.audio_index}: Passthrough-Codec " f"{source_track.codec} became {target.codec}" ) if (source_track.language_raw or "") != (target.language_raw or ""): problems.append(f"Audio track {source_track.audio_index}: language changed") if (source_track.title or "") != (target.title or ""): problems.append(f"Audio track {source_track.audio_index}: title changed") want_flags = {k for k, v in source_track.disposition.items() if v} have_flags = {k for k, v in target.disposition.items() if v} if want_flags != have_flags: problems.append( f"Audio track {source_track.audio_index}: Flags {sorted(want_flags)} " f"against {sorted(have_flags)}" ) if full_decode and not turbo: argv = [ tools.ffmpeg, "-hide_banner", "-v", "error", "-nostdin", "-i", str(output_path) ] for index in range(len(output.video())): argv += ["-map", f"0:v:{index}"] for index in range(len(output.audio())): argv += ["-map", f"0:a:{index}"] argv += ["-f", "null", "-"] decoded = run_ffmpeg_progress( argv, duration=output.duration or plan.source.duration, label="Full decode check" ) if decoded.returncode != 0 or decoded.stderr.strip(): first = (decoded.stderr.strip().splitlines() or ["Returncode"])[0] problems.append("Decode check reports errors: " + first) else: log_detail("Verification %s: complete A/V decode skipped.", "turbo" if turbo else "fast") if plan.source.video() and output.video(): if plan.source.video()[0].codec != output.video()[0].codec: problems.append("Video codec has changed") elif container_family(plan.source.format_name) == "matroska" and not turbo: if not streamcopy_is_identical( tools, plan.source.path, "v:0", output_path, "v:0" ): problems.append( f"Video packet data differ; {PACKET_HASH_ALGORITHM} stream-copy proof failed" ) for check in timeline_checks: if check.status == "deviation": problems.append( f"Timing track {check.audio_index}: deviation " f"{check.deviation * 1000:+.2f} ms, {check.note}" ) elif check.status == "not-verifiable": # A production output must not count as successful # while the synchronisation of a newly encoded track stays unproven. problems.append( f"Timing track {check.audio_index} not verifiable: {check.note}" ) else: log_detail( "Timing track %d passed, deviation %+.2f ms, %s", check.audio_index, check.deviation * 1000, check.note, ) return problems def log_plan_summary(plan: JobPlan) -> None: video = plan.source.video()[0] if plan.source.video() else None size = (f"{video.width}x{video.height}" if video and video.width and video.height else "unknown") log.info("Source: %s", plan.source.path) log.info("Picture: %s", size) log.info("Audio tracks: %d total, %d will be encoded", len(plan.source.audio()), len(plan.tracks)) for track in plan.tracks: selection = "automatic" if track.profile.selection == "auto" else "explicit" log_track_line( " Audio %d: %s, %d -> 2 channels, %d Hz, profile %s (%s), Nero q %.2f", track.source.audio_index, track.source.codec, track.source.channels, track.target_sample_rate, track.profile.name.upper(), selection, track.profile.nero_quality, ) if plan.passthrough_audio: log.info("Unchanged audio tracks: %d", len(plan.passthrough_audio)) log.info("Output: %s", plan.output) def _job_log_paths(plan: JobPlan) -> tuple[Path, Path]: final = Path(str(plan.output) + ".log") final.parent.mkdir(parents=True, exist_ok=True) fd, name = tempfile.mkstemp( prefix=f".{plan.output.name}.trackshepherd-log-", suffix=".log", dir=final.parent ) os.close(fd) return final, Path(name) def attach_job_log(plan: JobPlan, tools: Tools) -> logging.Handler: final, candidate = _job_log_paths(plan) plan.job_log_output = final plan.job_log_candidate = candidate with candidate.open("w", encoding="utf-8") as fh: fh.write(f"# {DISPLAY_NAME} {VERSION}\n") fh.write(f"# Start: {datetime.now().astimezone().isoformat()}\n") fh.write(f"# Source: {plan.source.path}\n") fh.write(f"# Target: {plan.output}\n") fh.write(f"# FFmpeg: {tools.ffmpeg_version}\n") fh.write(f"# FFprobe: {tools.ffprobe_version}\n") fh.write("# Status: started\n\n") handler = logging.FileHandler(candidate, mode="a", encoding="utf-8") handler.setLevel(logging.DEBUG) handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s")) log.addHandler(handler) return handler def finalize_job_log(plan: JobPlan, handler: logging.Handler, *, success: bool, detail: str = "", verified: bool = True) -> Path | None: handler.flush() log.removeHandler(handler) handler.close() candidate = plan.job_log_candidate final = plan.job_log_output if candidate is None or not candidate.exists() or final is None: return None try: with candidate.open("a", encoding="utf-8") as fh: fh.write("\n# Completion\n") if success: status_text = "successful and verified" if verified else "successful, turbo check" else: status_text = "ERROR" fh.write(f"# Status: {status_text}\n") if detail: fh.write(f"# Detail: {detail}\n") fh.write(f"# End: {datetime.now().astimezone().isoformat()}\n\n") fh.write(inspect_text(plan, analyzed=any(t.analysis is not None for t in plan.tracks))) fh.write("\n") if success: os.replace(candidate, final) plan.job_log_candidate = None return final stamp = datetime.now().strftime("%Y%m%d-%H%M%S") failed = Path(str(plan.output) + f".failed-{stamp}.log") os.replace(candidate, failed) plan.job_log_candidate = None return failed except OSError as exc: log.warning("Job log could not be finalised: %s", exc) return candidate def process_file(tools: Tools, path: Path, args: argparse.Namespace) -> FileResult: media = probe_media(tools, path, full_time_anchors=True) if not media.audio(): raise StepFailed("The file contains no audio track.") plan = build_plan(media, args) job_handler = attach_job_log(plan, tools) log_plan_summary(plan) workdir: Path | None = None keep = args.keep_temp try: require_verified_source_timing(plan) for track in plan.tracks: prepare_audio_timeline(tools, path, track) workdir = create_managed_tempdir() plan.temp_dir = workdir if args.keep_temp: mark_temp_keep(workdir, "--keep-temp") if args.serial_audio or len(plan.tracks) <= 1: for track in plan.tracks: if analysis_required(track): analyse_track(tools, path, track, media.duration) else: track.gain = gain_without_analysis(track) for index, track in enumerate(plan.tracks): encode_with_peak_control(tools, plan, track, index, workdir, args) else: workers = resolved_audio_workers(args, len(plan.tracks)) try: spool = create_audio_spool(tools, path, plan.tracks, workdir, media.duration) except StepFailed as exc: log.warning("Audio spool not usable, serial fallback: %s", exc) reporter = globals().get("REPORTER") if reporter is not None: reporter.note_warning("spool fallback serial") for track in plan.tracks: if analysis_required(track): analyse_track(tools, path, track, media.duration) else: track.gain = gain_without_analysis(track) for index, track in enumerate(plan.tracks): encode_with_peak_control(tools, plan, track, index, workdir, args) else: log_detail( "Multitrack strategy: audio spool %.1f MiB, %d workers; spool reused for analysis and encoding after the source timing scan.", spool.stat().st_size / (1024 * 1024), workers, ) analyse_tracks_spool(tools, spool, plan.tracks, media.duration, workdir, workers) encode_tracks_spool_with_peak_control( tools, plan, spool, workdir, args, workers ) for track in plan.tracks: if track.analysis is None or track.gain is None: continue if track.measured_tp is None: log.info( "Result audio %d: input true peak %+.2f dBTP, gain %+.2f dB; " "post-AAC peak not measured%s.", track.source.audio_index, track.analysis.input_tp, track.gain.total_gain, " (Turbo)" if args.turbo else "", ) else: excess = track.measured_tp - track.normalization.target_tp log.info( "Result audio %d: input true peak %+.2f dBTP, gain %+.2f dB, " "AAC true peak %+.2f dBTP, deviation from target %+.2f dB.", track.source.audio_index, track.analysis.input_tp, track.gain.total_gain, track.measured_tp, excess, ) log_detail( "Audio %d: %d encoding attempt(s), target %+.2f dBTP, tolerance %.2f dB, " "warning limit %.2f dB.", track.source.audio_index, track.attempts, track.normalization.target_tp, track.normalization.tolerance, args.peak_warn_limit, ) # Create the candidate only directly before the mux. Errors in analysis or encoding # therefore leave no empty dummy files in the target directory. plan.candidate_output = create_candidate_output(plan.output) _mux_result, output_probe, timeline_checks = mux_with_timeline_control(tools, plan) problems = verify_output( tools, plan, output_probe, timeline_checks, full_decode=(args.verify == "full"), turbo=args.turbo ) if problems: keep = True for item in problems: log.error("Check: %s", item) raise StepFailed(f"The output has {len(problems)} problem(s).") commit_output(plan, args.force) if args.turbo: log.info( "Done. Turbo check passed; post-AAC true peak and final full-track and video hash checks were disabled. %s", plan.output, ) else: log.info("Done and verified. %s", plan.output) log_path = finalize_job_log( plan, job_handler, success=True, verified=not args.turbo, detail=("Turbo: post-AAC true peak and final full-track and video hash checks disabled" if args.turbo else ""), ) if log_path is not None: log_detail("Log: %s", log_path) except KeyboardInterrupt as exc: # A deliberate user abort is not a diagnostic error. Without # --keep-temp the working directory is removed at once. After a # hard process abort the next program start does this, based on # the owner marker. keep = bool(args.keep_temp) if workdir is not None and keep: mark_temp_keep(workdir, "--keep-temp / user abort") log.warning("Aborted by user; working directory kept: %s", workdir) if plan.candidate_output is not None and plan.candidate_output.exists(): if keep: log.warning("Unverified candidate file kept: %s", plan.candidate_output) else: plan.candidate_output.unlink(missing_ok=True) finalize_job_log(plan, job_handler, success=False, detail="Aborted by user") raise except BaseException as exc: keep = True # The attempt counts are the interesting data precisely in the failure case. # Without passing them on, exactly the case in which the peak control # loop does not converge would be lost. try: exc.peak_attempts = tuple(track.attempts for track in plan.tracks) except Exception: # noqa: BLE001 pass if workdir is not None: mark_temp_keep(workdir, f"Error: {exc}") log.error("Processing failed: %s", exc) if isinstance(exc, StepFailed) and exc.result is not None: log.debug("Command line %s", exc.result.commandline) log.debug("Complete subprocess diagnostics:\\n%s", exc.result.stderr) if workdir is not None: log.error("Working directory kept for diagnostics: %s", workdir) if plan.candidate_output is not None and plan.candidate_output.exists(): log.error("Unverified candidate file kept: %s", plan.candidate_output) failed_log = finalize_job_log(plan, job_handler, success=False, detail=str(exc)) if failed_log is not None: log.error("Error log: %s", failed_log) raise finally: if workdir is not None and not keep: shutil.rmtree(workdir, ignore_errors=True) return FileResult( source=path, output=plan.output, ok=True, summary=describe_conversion(plan, args), gains=describe_gains(plan), peak_attempts=tuple(track.attempts for track in plan.tracks), ) def handle_file(tools: Tools, path: Path, args: argparse.Namespace) -> FileResult: if not (args.inspect or args.dry_run): return process_file(tools, path, args) media = probe_media(tools, path, full_time_anchors=True) if not media.audio(): raise StepFailed("The file contains no audio track.") plan = build_plan(media, args) if args.analyze: for track in plan.tracks: prepare_audio_timeline(tools, path, track) if args.serial_audio or len(plan.tracks) <= 1: for track in plan.tracks: if analysis_required(track): analyse_track(tools, path, track, media.duration) else: track.gain = gain_without_analysis(track) else: inspect_workdir = create_managed_tempdir(prefix=f"{APP_NAME}-inspect-") try: spool = create_audio_spool(tools, path, plan.tracks, inspect_workdir, media.duration) analyse_tracks_spool( tools, spool, plan.tracks, media.duration, inspect_workdir, resolved_audio_workers(args, len(plan.tracks)), ) finally: shutil.rmtree(inspect_workdir, ignore_errors=True) if args.inspect: if args.json: print(json.dumps(inspect_json(plan, args.analyze), indent=2, ensure_ascii=False)) else: print(inspect_text(plan, args.analyze)) else: if args.json: print(json.dumps(dry_run_payload(tools, plan, args.analyze), indent=2, ensure_ascii=False)) else: print(dry_run_text(tools, plan, args.analyze)) return FileResult(source=path, output=plan.output, ok=True) def main(argv: list[str] | None = None) -> int: raw_argv = list(sys.argv[1:] if argv is None else argv) parser = build_parser() # argparse normally ends --help at once, and --version needed a special # path because of the mandatory INPUT argument. We intercept both # modes before normal parsing and show the real system status. if "--help" in raw_argv or "-h" in raw_argv: parser.print_help() ffmpeg_override, ffprobe_override, nero_override = _early_tool_overrides(raw_argv) print() print_dependency_report(ffmpeg_override, ffprobe_override, nero_override) return EXIT_OK if "--license" in raw_argv or "--licence" in raw_argv: sys.stdout.write(licence_text()) return EXIT_OK if "--version" in raw_argv: ffmpeg_override, ffprobe_override, nero_override = _early_tool_overrides(raw_argv) print(f"{DISPLAY_NAME} {VERSION}") print_dependency_report(ffmpeg_override, ffprobe_override, nero_override) return EXIT_OK args = parser.parse_args(raw_argv) # Any number of -v is allowed. The highest defined level is -vvvv. args.verbose = min(int(args.verbose or 0), 4) validate_args(parser, args) compact = not (args.verbose or args.inspect or args.dry_run or args.json) setup_logging(args.verbose, args.quiet, args.log, compact=compact) configure_progress(not args.no_progress and not args.quiet) cleanup_stale_tempdirs() processing = not (args.inspect or args.dry_run) try: tools = discover_tools(args, need_nero=processing) except ToolMissing as exc: log.error("%s", exc) return EXIT_TOOL_MISSING files_in = collect_inputs(args.inputs) if not files_in: log.error("No input file found.") return EXIT_USAGE log.info("Input: %d media file(s)%s", len(files_in), " from file or directory" if len(args.inputs) else "") global REPORTER, JOB_PROGRESS, LIVE_BLOCK, TIMING timing_path = Path(args.timing_log) if args.timing_log and not args.no_timing_log else None model = TimingModel(timing_path) TIMING = TimingLog(timing_path) if timing_path is not None else None if TIMING is not None: TIMING.start() items = prescan_durations(tools, files_in) known = [i for i in items if i.media] total_media = sum(i.media for i in known) if total_media: log_detail("Job: %d file(s), %s of media, estimated duration %s " "(model from %d earlier files).", len(items), _format_clock(total_media), _format_clock(sum(model.expected_seconds(i.media, i.tracks, bool(getattr(args, 'turbo', False))) for i in items)), model.samples) JOB_PROGRESS = JobProgress(items, model, enabled=compact and len(items) >= 1, turbo=bool(getattr(args, "turbo", False))) LIVE_BLOCK = LiveBlock(compact and PROGRESS_ENABLED and sys.stderr.isatty()) results: list[FileResult] = [] interrupted = False for position, path in enumerate(files_in, 1): item = items[position - 1] JOB_PROGRESS.start_file(position - 1) if TIMING is not None: TIMING.begin_file(position) file_started = time.monotonic() failure_attempts: tuple[int, ...] = () REPORTER = CompactReporter(position, len(files_in), path.name, enabled=compact) REPORTER.media_duration = item.media if not compact and len(files_in) > 1: if position > 1: sys.stderr.write("\n") width = len(str(len(files_in))) log_track_line("File (%*d/%d): %s", width, position, len(files_in), path) try: outcome = handle_file(tools, path, args) results.append(outcome) detail = outcome.summary if REPORTER.warnings: REPORTER.finish("Warning", detail, reason="; ".join(REPORTER.warnings)) else: REPORTER.finish("OK", detail) except (StepFailed, ToolMissing, argparse.ArgumentTypeError, OSError) as exc: compact_reason = getattr(exc, "compact_message", None) if not compact_reason: compact_reason = str(exc).split(". ")[0][:72] failure_attempts = tuple(getattr(exc, "peak_attempts", ()) or ()) REPORTER.finish("Failed", compact_reason) log.error("%s %s", path.name, exc) result = FileResult(path, None, False, [str(exc)]) step = getattr(exc, "result", None) if step is not None: log.debug("Command line %s", step.commandline) tail = [line for line in step.stderr.splitlines() if line.strip()] for line in tail[-20:]: log.error(" %s", line) if len(tail) > 20: log.error(" ... %d more lines, complete with --log", len(tail) - 20) results.append(result) if args.stop_on_error: break except KeyboardInterrupt: interrupted = True REPORTER.finish("Failed", "Aborted by user") log.warning("Aborted by user.") break finally: wall = time.monotonic() - file_started status = "Failed" if results and results[-1].source == path: status = "OK" if results[-1].ok else "Failed" if results[-1].ok and REPORTER is not None and REPORTER.warnings: status = "Warning" JOB_PROGRESS.finish_file(item, wall, status) if TIMING is not None: if item.media: attempts: tuple[int, ...] = () if results and results[-1].source == path: attempts = results[-1].peak_attempts if not attempts: attempts = tuple(failure_attempts) TIMING.file( item.media, wall, n=len(REPORTER.track_gains) if REPORTER else 0, ok=1 if status != "Failed" else 0, tr=1 if getattr(args, "turbo", False) else 0, sa=1 if getattr(args, "serial_audio", False) else 0, vf=getattr(args, "verify", "fast"), att=list(attempts), pr=sum(max(0, value - 1) for value in attempts), ) TIMING.end_file() REPORTER = None if JOB_PROGRESS is not None and LIVE_BLOCK is not None and JOB_PROGRESS.enabled: LIVE_BLOCK.close([JOB_PROGRESS.line(final=True)]) if TIMING is not None: TIMING.job_summary(len(results), sum(i.media for i in items if i.done and i.media), time.monotonic() - JOB_PROGRESS.started) failures = [r for r in results if not r.ok] if len(files_in) > 1 or failures: log.info( "Summary processed %d, successful %d, failed %d", len(results), len(results) - len(failures), len(failures), ) for item in failures: log.info(" %s %s", item.source, "; ".join(item.messages) or "error") return EXIT_INTERRUPTED if interrupted else (EXIT_FILE_FAILED if failures else EXIT_OK) if __name__ == "__main__": raise SystemExit(main())