From 70a9b238106a5e858b49615fb5310c2c39e0c072 Mon Sep 17 00:00:00 2001 From: jenstandstad Date: Fri, 7 Aug 2026 10:47:09 +0200 Subject: [PATCH] Improved loudnorm --- gnommo/cli.py | 43 +++++++++++++++++++++++- gnommo/preprocessor.py | 75 +++++++++++++++++++++++++++++++++++++++++- 2 files changed, 116 insertions(+), 2 deletions(-) diff --git a/gnommo/cli.py b/gnommo/cli.py index 339e82b..be22252 100644 --- a/gnommo/cli.py +++ b/gnommo/cli.py @@ -2612,7 +2612,12 @@ def cmd_preprocess( """ from concurrent.futures import ThreadPoolExecutor, as_completed from .parser import parse_project_config, parse_videos - from .preprocessor import preprocess_video, RES_CONFIGS + from .preprocessor import ( + preprocess_video, + RES_CONFIGS, + measure_loudnorm_stats, + _resolve_auto_channel, + ) from .models import VideoSource as _VideoSource from .cache import set_active_project @@ -2829,6 +2834,40 @@ def cmd_preprocess( print(f" Filters: {len(segment_source.filter)} step(s)") return 0 + # --- Shared loudness reference --- + # Probe the FIRST narration take once and reuse its loudness profile for every + # take (linear loudnorm). Per-segment dynamic loudnorm otherwise re-measures each + # take independently, so a pausier/quieter one gets boosted louder than the next + # (s1-9 ended up louder than s10-39 even though the raw takes matched). Computed + # from the stable first source so incremental reprocessing of any single take + # still lands on the same reference. Falls back to per-segment loudnorm if probing + # fails. NOTE: changing the first take means re-running preprocess --force on the + # narration so every take re-normalises against the new reference. + shared_loudnorm_stats = None + _an_cfg = next( + (f for f in talkinghead_filter + if isinstance(f, dict) and f.get("type") == "audio_normalize"), + None, + ) + if _an_cfg is not None and source_files and not dry_run: + _ref_source = source_files[0] + _ref_id = _ref_source.stem + if using_compressed and _ref_id.endswith("_compressed"): + _ref_id = _ref_id[: -len("_compressed")] + _ref_channel = existing_narration.get(_ref_id, {}).get("use_audio_channels", "auto") + if _ref_channel == "auto": + _ref_channel = _resolve_auto_channel(_ref_source) + shared_loudnorm_stats = measure_loudnorm_stats( + _ref_source, _an_cfg, _ref_channel, verbose + ) + if shared_loudnorm_stats: + print( + f" Loudness reference from {_ref_id} " + f"(I={shared_loudnorm_stats['measured_I']} LUFS) — applied to all takes" + ) + else: + print(" Loudness reference probe failed — using per-segment loudnorm") + # --- Process segments --- successfully_processed: list[tuple[str, _VideoSource]] = [] @@ -2849,6 +2888,7 @@ def cmd_preprocess( force=force, custom_gnommo_scratch=gnommo_scratch, res=res, + shared_loudnorm_stats=shared_loudnorm_stats, ) return task @@ -2884,6 +2924,7 @@ def cmd_preprocess( force, gnommo_scratch, res=res, + shared_loudnorm_stats=shared_loudnorm_stats, ) output_path = ( cache_narration_dir or narration_dir diff --git a/gnommo/preprocessor.py b/gnommo/preprocessor.py index 74829b4..5d5f13f 100644 --- a/gnommo/preprocessor.py +++ b/gnommo/preprocessor.py @@ -769,6 +769,7 @@ def preprocess_video( force: bool = False, custom_gnommo_scratch: Optional[Path] = None, res: str = "full", + shared_loudnorm_stats: Optional[dict] = None, ) -> Path: """ Apply preprocessing filters to a video source. @@ -921,6 +922,7 @@ def preprocess_video( take=None, use_audio_channels=channel, skip_loudnorm=video_source.defer_loudnorm, + shared_loudnorm_stats=shared_loudnorm_stats, ) current_input = step_output batch_num += 1 @@ -2292,6 +2294,62 @@ def apply_transcribe( return output_path +def measure_loudnorm_stats( + input_path: Path, + config: dict[str, Any], + use_audio_channels: str = "both", + verbose: bool = False, +) -> Optional[dict]: + """Run loudnorm's analysis (first) pass on one file and return its measured values. + + Used to derive ONE shared loudness reference from a single narration take. Feeding + those measured values into `loudnorm ... linear=true` on EVERY take makes them all + receive the same gain, so per-segment loudnorm can't drift their levels apart (the + s1-9-louder-than-s10-39 bug). The channel mapping is applied so the measurement + matches the channel the render will actually use. Returns None on any failure — the + caller then falls back to ordinary per-segment loudnorm. + """ + import json as _json + import re as _re + import subprocess + + cfg = parse_audio_normalize_config(config) + pan = "" + if use_audio_channels == "left": + pan = "pan=stereo|c0=c0|c1=c0," + elif use_audio_channels == "right": + pan = "pan=stereo|c0=c1|c1=c1," + af = ( + f"{pan}loudnorm=I={cfg.target_lufs:.1f}:LRA={cfg.target_lra:.1f}" + f":TP={cfg.target_tp:.1f}:print_format=json" + ) + cmd = [ + "ffmpeg", "-hide_banner", "-nostats", + "-i", str(input_path), "-af", af, "-f", "null", "-", + ] + try: + proc = subprocess.run(cmd, capture_output=True, text=True) + except Exception: + return None + # loudnorm prints a JSON object near the end of stderr. + match = _re.search(r'\{[^{}]*"input_i"[^{}]*\}', proc.stderr, _re.DOTALL) + if not match: + return None + try: + data = _json.loads(match.group(0)) + stats = { + "measured_I": data["input_i"], + "measured_TP": data["input_tp"], + "measured_LRA": data["input_lra"], + "measured_thresh": data["input_thresh"], + } + except Exception: + return None + if verbose: + print(f" Loudness reference: {stats}") + return stats + + def apply_audio_normalize( input_path: Path, output_path: Path, @@ -2300,6 +2358,7 @@ def apply_audio_normalize( take: float = None, use_audio_channels: str = "both", skip_loudnorm: bool = False, + shared_loudnorm_stats: Optional[dict] = None, ) -> None: """ Apply audio normalization: denoise, compress, and loudness normalize. @@ -2439,11 +2498,25 @@ def apply_audio_normalize( # 8. Loudness normalization (loudnorm - EBU R128) # Skip if skip_loudnorm=True (for segments that will be concatenated) if cfg.normalize and not skip_loudnorm: - audio_filters.append( + loudnorm = ( f"loudnorm=I={cfg.target_lufs:.1f}" f":LRA={cfg.target_lra:.1f}" f":TP={cfg.target_tp:.1f}" ) + # With a shared reference (from measure_loudnorm_stats on the first narration + # take), use linear mode so EVERY take gets the SAME gain — otherwise loudnorm's + # default dynamic pass re-measures each take independently and a pausier/quieter + # segment gets boosted louder than the next. + if shared_loudnorm_stats: + s = shared_loudnorm_stats + loudnorm += ( + f":measured_I={s['measured_I']}" + f":measured_TP={s['measured_TP']}" + f":measured_LRA={s['measured_LRA']}" + f":measured_thresh={s['measured_thresh']}" + f":linear=true" + ) + audio_filters.append(loudnorm) if not audio_filters: # No filters enabled, just copy