Skip to content

feat: optional EBU R128 loudness normalization (--normalize) #185

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
seonghobae wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat: optional EBU R128 loudness normalization (--normalize) #185

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
65 changes: 60 additions & 5 deletions media_shrinker.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,7 @@
DEFAULT_TARGET_BYTES = 1_900_000_000
DEFAULT_MAX_SEGMENT_DURATION_SECONDS = 4 * 60 * 60
DEFAULT_SILENCE_NOISE = "-35dB"
LOUDNORM_FILTER = "loudnorm=I=-16:TP=-1.5:LRA=11"
DEFAULT_SILENCE_MIN_DURATION_SECONDS = 2.0
HARD_SPLIT_EPSILON_SECONDS = 0.001
DURATION_TOLERANCE_SECONDS = 2.0
Expand Down Expand Up @@ -321,8 +322,13 @@ def build_audio_plan(
prefer_flac: bool = False,
ffmpeg_threads: int | None = None,
segment: MediaSegment | None = None,
normalize: bool = False,
) -> ConversionPlan:
"""Build the preferred audio-only conversion plan for source_path."""
"""Build the preferred audio-only conversion plan for source_path.

When normalize is True the EBU R128 loudnorm audio filter is applied so
generated audio has consistent loudness; when False the args are unchanged.
"""

if probe.has_video:
raise MediaShrinkerError(
Expand Down Expand Up @@ -364,6 +370,7 @@ def build_audio_plan(
str(output_path),
]
)
args = _with_loudnorm(args, normalize)
args = _with_ffmpeg_threads(args, ffmpeg_threads)
return ConversionPlan(
strategy=strategy,
Expand All @@ -379,6 +386,7 @@ def build_audio_plan(
output_dir=output_dir,
ffmpeg_threads=ffmpeg_threads,
segment=segment,
normalize=normalize,
)


Expand All @@ -390,8 +398,13 @@ def build_opus_plan(
output_dir: Path,
ffmpeg_threads: int | None = None,
segment: MediaSegment | None = None,
normalize: bool = False,
) -> ConversionPlan:
"""Build a high-quality Opus plan that fits the target size."""
"""Build a high-quality Opus plan that fits the target size.

When normalize is True the EBU R128 loudnorm audio filter is applied so
generated audio has consistent loudness; when False the args are unchanged.
"""

duration_seconds = (
segment.duration_seconds if segment is not None else probe.duration_seconds
Expand Down Expand Up @@ -434,6 +447,7 @@ def build_opus_plan(
str(output_path),
]
)
args = _with_loudnorm(args, normalize)
args = _with_ffmpeg_threads(args, ffmpeg_threads)
return ConversionPlan(
strategy="opus-bitrate",
Expand Down Expand Up @@ -727,8 +741,13 @@ def convert_file(
protected_sources: Iterable[Path] = (),
resolved_protected_sources: frozenset[Path] | None = None,
original_size: int | None = None,
normalize: bool = False,
) -> list[ConversionResult]:
"""Convert one file and return generated segment results without deleting the source."""
"""Convert one file and return generated segment results without deleting the source.

When normalize is True generated audio is loudness-normalized with the EBU
R128 loudnorm filter; the default of False keeps prior byte-identical output.
"""

source = Path(source)
root = Path(root)
Expand Down Expand Up @@ -777,6 +796,7 @@ def convert_file(
overwrite=overwrite,
max_segment_duration_seconds=max_segment_duration_seconds,
protected_sources=resolved_sources,
normalize=normalize,
)
for segment in segments
]
Expand Down Expand Up @@ -874,8 +894,13 @@ def _execute_segment_conversion(
overwrite: bool,
max_segment_duration_seconds: float,
resolved_protected_sources: frozenset[Path],
normalize: bool = False,
) -> ConversionResult:
"""Execute the conversion plan(s) for a single segment."""
"""Execute the conversion plan(s) for a single segment.

When normalize is True both the flac and opus plans apply the EBU R128
loudnorm audio filter for consistent output loudness.
"""
plan = build_audio_plan(
rel_source,
probe,
Expand All @@ -884,6 +909,7 @@ def _execute_segment_conversion(
prefer_flac=prefer_flac,
ffmpeg_threads=ffmpeg_threads,
segment=segment,
normalize=normalize,
)

final_output = _resolve_collision(plan.output_path, overwrite=overwrite)
Expand All @@ -910,6 +936,7 @@ def _execute_segment_conversion(
output_dir=output_dir,
ffmpeg_threads=ffmpeg_threads,
segment=segment,
normalize=normalize,
)
final_output = _resolve_collision(opus_plan.output_path, overwrite=overwrite)
first_result = _execute_plan(
Expand Down Expand Up @@ -994,8 +1021,13 @@ def _convert_segment(
overwrite: bool,
max_segment_duration_seconds: float,
protected_sources: frozenset[Path] = frozenset(),
normalize: bool = False,
) -> ConversionResult:
"""Convert one media segment fitting the target size limit."""
"""Convert one media segment fitting the target size limit.

normalize is threaded through to the conversion plans so loudness
normalization can be applied when requested.
"""
# protected_sources is passed from convert_file where it is already fully resolved.
# _ensure_not_source_path explicitly checks against the current source independently.
resolved_protected_sources = protected_sources
Expand Down Expand Up @@ -1043,6 +1075,7 @@ def _convert_segment(
overwrite=overwrite,
max_segment_duration_seconds=max_segment_duration_seconds,
resolved_protected_sources=resolved_protected_sources,
normalize=normalize,
)


Expand Down Expand Up @@ -1217,6 +1250,15 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
action="store_true",
help="Prefer FLAC for every audio-only source, including lossy input",
)
parser.add_argument(
"--normalize",
action="store_true",
default=False,
help=(
"Apply EBU R128 loudness normalization (loudnorm=I=-16:TP=-1.5:LRA=11) "
"to generated audio. Off by default; output is byte-identical when omitted."
),
)
parser.add_argument(
"--workers", type=int, default=0, help="Parallel ffmpeg jobs. 0 = auto"
)
Expand Down Expand Up @@ -1286,6 +1328,7 @@ def process_candidate(candidate_tuple: tuple[Path, int]) -> list[ConversionResul
protected_sources=protected_sources,
resolved_protected_sources=resolved_candidates,
original_size=size,
normalize=args.normalize,
)
except Exception as exc: # noqa: BLE001 - batch processing records per-file failures.
return [
Expand Down Expand Up @@ -1484,6 +1527,18 @@ def _with_ffmpeg_threads(args: list[str], ffmpeg_threads: int | None) -> list[st
return [*args[:-1], "-threads", str(ffmpeg_threads), args[-1]]


def _with_loudnorm(args: list[str], normalize: bool) -> list[str]:
"""Insert the EBU R128 loudnorm audio filter before the output path when requested.

When normalize is False the argument list is returned unchanged so default
conversions remain byte-identical to prior behavior.
"""

if not normalize:
return args
return [*args[:-1], "-af", LOUDNORM_FILTER, args[-1]]


def _parse_probe_payload(
payload: dict[str, Any],
source_path: Path,
Expand Down
100 changes: 100 additions & 0 deletions tests/test_media_shrinker.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -916,6 +916,100 @@ def test_parse_silencedetect_intervals_pairs_long_silence_start_and_end(
)


class LoudnessNormalizationTests(unittest.TestCase):
_FLAC_PROBE = MediaProbe(
duration_seconds=3_600.0,
size_bytes=4_294_808_936,
audio_codec="pcm_s16le",
audio_bit_rate=1_411_200,
has_video=False,
format_name="wav",
)
_OPUS_PROBE = MediaProbe(
duration_seconds=10_000.0,
size_bytes=3_000_000_000,
audio_codec="aac",
audio_bit_rate=2_500_000,
has_video=False,
format_name="mov,mp4,m4a,3gp,3g2,mj2",
)

def _flac_plan(self, *, normalize: bool) -> ConversionPlan:
return build_audio_plan(
Path("meeting.wav"),
self._FLAC_PROBE,
target_bytes=1_900_000_000,
output_dir=Path("out"),
normalize=normalize,
)

def _opus_plan(self, *, normalize: bool) -> ConversionPlan:
return build_audio_plan(
Path("long.m4a"),
self._OPUS_PROBE,
target_bytes=1_900_000_000,
output_dir=Path("out"),
normalize=normalize,
)

def test_normalize_adds_loudnorm_filter_to_flac_plan(self) -> None:
plan = self._flac_plan(normalize=True)
self.assertEqual(plan.strategy, "flac-lossless")
self.assertIn("-af", plan.ffmpeg_args)
self.assertIn(media_shrinker.LOUDNORM_FILTER, plan.ffmpeg_args)
self.assertIn("loudnorm", media_shrinker.LOUDNORM_FILTER)
# The filter must precede the output path so ffmpeg applies it.
self.assertLess(
plan.ffmpeg_args.index(media_shrinker.LOUDNORM_FILTER),
len(plan.ffmpeg_args) - 1,
)

def test_normalize_adds_loudnorm_filter_to_opus_plan(self) -> None:
plan = self._opus_plan(normalize=True)
self.assertEqual(plan.strategy, "opus-bitrate")
self.assertIn("-af", plan.ffmpeg_args)
self.assertIn(media_shrinker.LOUDNORM_FILTER, plan.ffmpeg_args)

def test_default_off_flac_plan_is_byte_identical(self) -> None:
plan = self._flac_plan(normalize=False)
self.assertNotIn("-af", plan.ffmpeg_args)
self.assertNotIn(media_shrinker.LOUDNORM_FILTER, plan.ffmpeg_args)
# Default must equal the explicit normalize=False arg list unchanged.
self.assertEqual(
plan.ffmpeg_args,
build_audio_plan(
Path("meeting.wav"),
self._FLAC_PROBE,
target_bytes=1_900_000_000,
output_dir=Path("out"),
).ffmpeg_args,
)

def test_default_off_opus_plan_is_byte_identical(self) -> None:
plan = self._opus_plan(normalize=False)
self.assertNotIn("-af", plan.ffmpeg_args)
self.assertNotIn(media_shrinker.LOUDNORM_FILTER, plan.ffmpeg_args)
self.assertEqual(
plan.ffmpeg_args,
build_audio_plan(
Path("long.m4a"),
self._OPUS_PROBE,
target_bytes=1_900_000_000,
output_dir=Path("out"),
).ffmpeg_args,
)

def test_build_opus_plan_normalize_flag_adds_filter(self) -> None:
plan = media_shrinker.build_opus_plan(
Path("long.m4a"),
self._OPUS_PROBE,
target_bytes=1_900_000_000,
output_dir=Path("out"),
normalize=True,
)
self.assertIn(media_shrinker.LOUDNORM_FILTER, plan.ffmpeg_args)


class SilenceDetectionTests(unittest.TestCase):
@patch("media_shrinker.subprocess.run")
def test_detect_silence_intervals_success(self, mock_run: MagicMock) -> None:
Expand Down Expand Up @@ -1215,6 +1309,7 @@ def test_parse_args_sets_conversion_options(self) -> None:
"--exclude-dir-prefix",
"split_",
"--flac-all",
"--normalize",
"--workers",
"2",
"--ffmpeg-threads",
Expand All @@ -1241,13 +1336,18 @@ def test_parse_args_sets_conversion_options(self) -> None:
self.assertFalse(args.include_under_limit)
self.assertEqual(args.exclude_dir_prefix, ["split_"])
self.assertTrue(args.flac_all)
self.assertTrue(args.normalize)
self.assertEqual(args.workers, 2)
self.assertEqual(args.ffmpeg_threads, -1)
self.assertEqual(args.silence_noise, "-40dB")
self.assertEqual(args.silence_min_duration_seconds, 3.5)
self.assertTrue(args.execute)
self.assertTrue(args.overwrite)

def test_parse_args_normalize_defaults_false_and_opts_in(self) -> None:
self.assertFalse(media_shrinker.parse_args(["root"]).normalize)
self.assertTrue(media_shrinker.parse_args(["root", "--normalize"]).normalize)

def test_main_dry_run_lists_candidates(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
Expand Down
Loading