diff --git a/media_shrinker.py b/media_shrinker.py index d72a8d5..02b2e0b 100644 --- a/media_shrinker.py +++ b/media_shrinker.py @@ -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 @@ -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( @@ -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, @@ -379,6 +386,7 @@ def build_audio_plan( output_dir=output_dir, ffmpeg_threads=ffmpeg_threads, segment=segment, + normalize=normalize, ) @@ -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 @@ -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", @@ -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) @@ -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 ] @@ -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, @@ -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) @@ -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( @@ -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 @@ -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, ) @@ -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" ) @@ -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 [ @@ -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, diff --git a/tests/test_media_shrinker.py b/tests/test_media_shrinker.py index 0ab2397..7383084 100644 --- a/tests/test_media_shrinker.py +++ b/tests/test_media_shrinker.py @@ -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: @@ -1215,6 +1309,7 @@ def test_parse_args_sets_conversion_options(self) -> None: "--exclude-dir-prefix", "split_", "--flac-all", + "--normalize", "--workers", "2", "--ffmpeg-threads", @@ -1241,6 +1336,7 @@ 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") @@ -1248,6 +1344,10 @@ def test_parse_args_sets_conversion_options(self) -> None: 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)