honney/movie_sync
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add two-pass EBU R128 loudness normalization for audio extraction

Browse Source
This commit is contained in:
Honney OC committed 2026-09-13 20:37:40 +00:00
1 parent d038882824
commit 154aeebb01
2 files changed
+108 -17
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

extract.py
+102 -16
View File
@@ -11,6 +11,7 @@ import subprocess
import os
import sys
import shutil
import json
# ─── Input / config (config.toml is base config, fallback to hardcoded) ───
CONFIG_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.toml")
@@ -20,6 +21,10 @@ _cfg_start = 0.0
_cfg_out_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "extracted")
_cfg_target_fps = "24000/1001"
_cfg_eps = 0.02
_cfg_normalize = True
_cfg_target_lufs = -14
_cfg_target_tp = -1.0
_cfg_target_lra = 11
try:
try:
import tomllib # py3.11+
@@ -41,6 +46,24 @@ try:
_cfg_eps = float(_ex["target_fps_eps"])
except Exception:
pass
_au = _cfg.get("audio", {})
if _au.get("normalize") is not None:
_cfg_normalize = bool(_au["normalize"])
if _au.get("target_lufs") is not None:
try:
_cfg_target_lufs = float(_au["target_lufs"])
except Exception:
pass
if _au.get("target_tp") is not None:
try:
_cfg_target_tp = float(_au["target_tp"])
except Exception:
pass
if _au.get("target_lra") is not None:
try:
_cfg_target_lra = float(_au["target_lra"])
except Exception:
pass
if _cfg_inputs:
print(f"[config] loaded {CONFIG_PATH}: duration={_cfg_duration}s start={_cfg_start}s target_fps={_cfg_target_fps}")
except Exception as _e:
@@ -84,6 +107,10 @@ try:
except Exception:
TARGET_FPS_FLOAT = 24000 / 1001
FPS_EPS = _cfg_eps # tolerance for float comparison
NORMALIZE = _cfg_normalize
TARGET_LUFS = _cfg_target_lufs
TARGET_TP = _cfg_target_tp
TARGET_LRA = _cfg_target_lra
def _parse_rate(s: str) -> float | None:
s = s.strip()
@@ -123,6 +150,61 @@ def probe_fps(path: str) -> float | None:
def fps_matches(a: float | None, b: float, eps: float = FPS_EPS) -> bool:
return a is not None and abs(a - b) < eps
def _parse_loudnorm_json(stderr: str) -> dict | None:
"""Extract JSON block from loudnorm stderr output."""
try:
start = stderr.rindex("{")
end = stderr.rindex("}") + 1
return json.loads(stderr[start:end])
except (ValueError, json.JSONDecodeError):
return None
def extract_audio_loudnorm(input_file: str, duration: int, start_sec: float,
out_path: str, stem: str) -> None:
"""Two-pass EBU R128 loudnorm extraction: measure then apply."""
ss_args = ["-ss", str(start_sec)] if start_sec and start_sec != 0 else []
base = ["ffmpeg", "-y", *ss_args, "-i", input_file, "-t", str(duration), "-vn"]
measure_af = f"loudnorm=I={TARGET_LUFS}:TP={TARGET_TP}:LRA={TARGET_LRA}:print_format=json"
# ─── Pass 1: measure ───
print(f"[loudness] {stem}: pass 1/2 measuring…")
cmd_measure = base + ["-af", measure_af, "-f", "null", "-"]
r = subprocess.run(cmd_measure, capture_output=True, text=True)
if r.returncode != 0:
print(f"[loudness] {stem}: pass 1 failed ({r.stderr.strip()[-80:]}), falling back to simple extract", file=sys.stderr)
cmd_fallback = base + [*AUDIO_CODEC, out_path]
r2 = subprocess.run(cmd_fallback, capture_output=True, text=True)
if r2.returncode != 0:
sys.exit(f"ffmpeg (audio) failed:\n{r2.stderr}")
return
measured = _parse_loudnorm_json(r.stderr)
if measured is None:
print(f"[loudness] {stem}: could not parse measurement, falling back to simple extract", file=sys.stderr)
cmd_fallback = base + [*AUDIO_CODEC, out_path]
r2 = subprocess.run(cmd_fallback, capture_output=True, text=True)
if r2.returncode != 0:
sys.exit(f"ffmpeg (audio) failed:\n{r2.stderr}")
return
mi = measured.get("input_i", "-24")
mtp = measured.get("input_tp", "-2")
mlra = measured.get("input_lra", "7")
mthresh = measured.get("input_thresh", "-34")
offset = measured.get("target_offset", "0")
print(f"[loudness] {stem}: measured I={mi} LUFS, TP={mtp} dBTP, LRA={mlra}, offset={offset} dB")
# ─── Pass 2: apply with measured values ───
print(f"[loudness] {stem}: pass 2/2 applying loudnorm → {TARGET_LUFS} LUFS…")
apply_af = (f"loudnorm=I={TARGET_LUFS}:TP={TARGET_TP}:LRA={TARGET_LRA}"
f":measured_I={mi}:measured_TP={mtp}:measured_LRA={mlra}"
f":measured_thresh={mthresh}:offset={offset}:linear=true")
cmd_apply = base + ["-af", apply_af, *AUDIO_CODEC, out_path]
r = subprocess.run(cmd_apply, capture_output=True, text=True)
if r.returncode != 0:
sys.exit(f"ffmpeg (audio) pass 2 failed:\n{r.stderr}")
print(f"[loudness] {stem}: normalized to {TARGET_LUFS} LUFS ✔")
# ─── Check ffmpeg ───
if shutil.which(f"ffmpeg") is None:
sys.exit(f"ERROR: 'ffmpeg' not found on PATH. Install it and try again.")
@@ -194,22 +276,26 @@ for INPUT_FILE in INPUT_FILES:
sys.exit(f"ffmpeg (video) failed:\n{r.stderr}")
print(f"[video: {stem}] done ✔")
# ─── Extract audio-only (first 90s, no video, CPU PCM re-encode) ───
cmd_audio = [
"ffmpeg",
"-y",
*ss_args,
"-i", INPUT_FILE,
"-t", str(DURATION),
"-vn", # no video
*AUDIO_CODEC, # pcm_f32le 48kHz mono, CPU (no GPU equiv.)
audio_out
]
print(f"[audio: {stem}] running ffmpeg…")
r = subprocess.run(cmd_audio, capture_output=True, text=True)
if r.returncode != 0:
sys.exit(f"ffmpeg (audio) failed:\n{r.stderr}")
print(f"[audio: {stem}] done ✔")
# ─── Extract audio-only (duration s, no video, CPU PCM re-encode) ───
if NORMALIZE:
print(f"[audio: {stem}] extracting with two-pass loudnorm…")
extract_audio_loudnorm(INPUT_FILE, DURATION, START_SEC, audio_out, stem)
else:
cmd_audio = [
"ffmpeg",
"-y",
*ss_args,
"-i", INPUT_FILE,
"-t", str(DURATION),
"-vn", # no video
*AUDIO_CODEC, # pcm_f32le 48kHz mono, CPU (no GPU equiv.)
audio_out
]
print(f"[audio: {stem}] running ffmpeg (no normalization)…")
r = subprocess.run(cmd_audio, capture_output=True, text=True)
if r.returncode != 0:
sys.exit(f"ffmpeg (audio) failed:\n{r.stderr}")
print(f"[audio: {stem}] done ✔")
print(f"-" * 50)
print(f"All done. Files are in:", OUT_DIR)