commit c015dc5373a588d0d9670a44686bd14afebac022 from: mtmn date: Sun Aug 9 01:21:25 2026 UTC lazymaster: keep audio format, tune loudness commit - 1089295496d18416b7f539d2ccce4714f733648c commit + c015dc5373a588d0d9670a44686bd14afebac022 blob - 6d6a9468c3410678a717800e454412330895b33b blob + 993424aaadc8779834309a4cf5a791f52e8c3f53 --- lazymaster/README.md +++ lazymaster/README.md @@ -1,16 +1,9 @@ # lazymaster -2-pass audio loudness normalization using `ffmpeg` and [loudnorm](https://ffmpeg.org/ffmpeg-filters.html#loudnorm) filter. -It normalizes audio to **-13 LUFS**, **-1 dBTP**, and **LRA 8**. +Two-pass audio loudness normalisation with `ffmpeg` loudnorm. -## Usage - -**Analysis:** -```bash -./lazymaster.py input.wav +```sh +./lazymaster.py input.wav [output.wav] ``` -**Two-pass normalization:** -```bash -./lazymaster input.wav output.wav -``` +Defaults: -14 LUFS, -1 dBTP, LRA 7. Run `./lazymaster.py --help` for options. blob - 71cc39fa891e339279bc0f5b312de7473e266887 blob + ad048a5532f848a57d5dc376c014d61e19410b64 --- lazymaster/lazymaster.py +++ lazymaster/lazymaster.py @@ -1,69 +1,211 @@ #!/usr/bin/env python3 +import argparse +import json import subprocess import sys -import json -from typing import cast +from collections.abc import Sequence +DEFAULT_LUFS = -14.0 +DEFAULT_LRA = 7.0 +DEFAULT_TRUE_PEAK = -1.0 +PCM_WAV_CODECS = {"pcm_s16le", "pcm_s24le", "pcm_s32le", "pcm_f32le", "pcm_f64le"} -def main(): - if len(sys.argv) < 2: - print("lazymaster input.wav [output.wav]") - sys.exit(1) - input_file = sys.argv[1] +def command_error(error: subprocess.CalledProcessError) -> str: + output = error.stderr or error.stdout or "" + return output.strip() - # Measure loudness - cmd_measure = [ + +def run(command: Sequence[str], **kwargs: object) -> subprocess.CompletedProcess[str]: + return subprocess.run(command, check=True, text=True, **kwargs) + + +def audio_properties(input_file: str) -> tuple[str, str]: + probe = run( + [ + "ffprobe", + "-v", + "error", + "-select_streams", + "a:0", + "-show_entries", + "stream=codec_name,sample_rate", + "-of", + "json", + input_file, + ], + capture_output=True, + ) + try: + stream = json.loads(probe.stdout)["streams"][0] + sample_rate = stream["sample_rate"] + except (IndexError, KeyError, TypeError, json.JSONDecodeError) as error: + raise ValueError("ffprobe found no usable audio stream") from error + + codec = stream.get("codec_name") + return (codec if codec in PCM_WAV_CODECS else "pcm_s24le"), sample_rate + + +def loudnorm_filter( + args: argparse.Namespace, measurements: dict[str, str] | None = None +) -> str: + settings = f"I={args.lufs}:TP={args.true_peak}:LRA={args.lra}" + if measurements is None: + return f"loudnorm={settings}:print_format=json" + + return ( + f"loudnorm={settings}:measured_I={measurements['input_i']}:" + f"measured_LRA={measurements['input_lra']}:" + f"measured_TP={measurements['input_tp']}:" + f"measured_thresh={measurements['input_thresh']}:" + f"offset={measurements['target_offset']}:linear=false" + ) + + +def parse_measurements(stderr: str) -> dict[str, str]: + end = stderr.rfind("}") + start = stderr.rfind("{", 0, end + 1) + if start < 0 or end < 0: + raise ValueError("ffmpeg did not print loudnorm measurements") + + try: + measurements = json.loads(stderr[start : end + 1]) + required = ("input_i", "input_tp", "input_lra", "input_thresh", "target_offset") + if not all(key in measurements for key in required): + raise ValueError("ffmpeg returned incomplete loudnorm measurements") + except json.JSONDecodeError as error: + raise ValueError("could not parse loudnorm measurements") from error + + return measurements + + +def measure(input_file: str, args: argparse.Namespace) -> dict[str, str]: + command = ["ffmpeg", "-hide_banner"] + if input_file != "-": + command.append("-nostdin") + command.extend(["-i", input_file, "-af", loudnorm_filter(args), "-f", "null", "-"]) + result = run(command, capture_output=True) + return parse_measurements(result.stderr) + + +def normalise( + input_file: str, + output_file: str, + measurements: dict[str, str], + args: argparse.Namespace, +) -> None: + if input_file == "-": + raise ValueError( + "two-pass normalisation needs a seekable input file, not standard input" + ) + + codec, sample_rate = audio_properties(input_file) + command = [ "ffmpeg", + "-hide_banner", + "-nostdin", "-i", input_file, + "-map", + "0:a:0", + "-map_metadata", + "0", "-af", - "loudnorm=I=-13:TP=-1:LRA=8:print_format=json", - "-f", - "null", - "-", + loudnorm_filter(args, measurements), + "-ar", + sample_rate, + "-c:a", + codec, ] + if output_file == "-": + command.extend(["-f", "wav", "-"]) + else: + command.extend(["-y" if args.force else "-n", output_file]) + run(command) - result = subprocess.run(cmd_measure, capture_output=True, text=True) - # Parse output values +def parse_args(argv: Sequence[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Measure or two-pass normalise an audio file using ffmpeg loudnorm." + ) + parser.add_argument( + "input", + metavar="INPUT", + help="input audio file, or - for standard input (analysis only)", + ) + parser.add_argument( + "output", + metavar="OUTPUT", + nargs="?", + help="normalised WAV file; use - for standard output", + ) + parser.add_argument( + "-o", + "--output", + dest="output_option", + metavar="OUTPUT", + help="normalised WAV file", + ) + parser.add_argument( + "-f", "--force", action="store_true", help="replace an existing output file" + ) + parser.add_argument( + "--lufs", + type=float, + default=DEFAULT_LUFS, + help=f"target integrated loudness (default: {DEFAULT_LUFS:g})", + ) + parser.add_argument( + "--lra", + type=float, + default=DEFAULT_LRA, + help=f"target loudness range (default: {DEFAULT_LRA:g})", + ) + parser.add_argument( + "--true-peak", + type=float, + default=DEFAULT_TRUE_PEAK, + help=f"maximum true peak in dBTP (default: {DEFAULT_TRUE_PEAK:g})", + ) + args = parser.parse_args(argv) + if args.output and args.output_option: + parser.error("OUTPUT and --output cannot be used together") + args.output = args.output_option or args.output + if args.input == args.output and args.output not in (None, "-"): + parser.error("input and output must be different files") + if args.force and args.output in (None, "-"): + parser.error("--force requires a file output") + return args + + +def main(argv: Sequence[str] | None = None) -> int: + args = parse_args(sys.argv[1:] if argv is None else argv) try: - stderr_output = result.stderr - json_start = stderr_output.find("{") - json_end = stderr_output.rfind("}") + 1 - if json_start == -1 or json_end == -1: - raise ValueError("json output is missing") + measurements = measure(args.input, args) + report = json.dumps(measurements, sort_keys=True) + print(report, file=sys.stderr if args.output == "-" else sys.stdout) + if args.output: + normalise(args.input, args.output, measurements, args) + except FileNotFoundError as error: + print( + f"lazymaster: required command not found: {error.filename}", file=sys.stderr + ) + return 127 + except ValueError as error: + print(f"lazymaster: {error}", file=sys.stderr) + return 1 + except subprocess.CalledProcessError as error: + print( + f"lazymaster: command failed with exit status {error.returncode}", + file=sys.stderr, + ) + if detail := command_error(error): + print(detail, file=sys.stderr) + return error.returncode or 1 + except KeyboardInterrupt: + return 130 + return 0 - stats = cast(dict[str, str], json.loads(stderr_output[json_start:json_end])) - measured_i = stats["input_i"] - measured_tp = stats["input_tp"] - measured_lra = stats["input_lra"] - measured_thresh = stats["input_thresh"] - offset = stats["target_offset"] - except (ValueError, json.JSONDecodeError, KeyError) as e: - print(f"error parsing loudness stats from ffmpeg output {e}") - sys.exit(1) - - print( - f"I={measured_i}, TP={measured_tp}, LRA={measured_lra}, Thresh={measured_thresh}, Offset={offset}" - ) - - # Normalize loudness - if len(sys.argv) > 2: - output_file = sys.argv[2] - cmd_normalize = [ - "ffmpeg", - "-i", - input_file, - "-af", - f"loudnorm=I=-13:TP=-1:LRA=8:measured_I={measured_i}:measured_LRA={measured_lra}:measured_TP={measured_tp}:measured_thresh={measured_thresh}:offset={offset}:linear=true", - "-y", - output_file, - ] - - _ = subprocess.run(cmd_normalize, check=True) - - if __name__ == "__main__": - main() + raise SystemExit(main())