commit - 1089295496d18416b7f539d2ccce4714f733648c
commit + c015dc5373a588d0d9670a44686bd14afebac022
blob - 6d6a9468c3410678a717800e454412330895b33b
blob + 993424aaadc8779834309a4cf5a791f52e8c3f53
--- lazymaster/README.md
+++ lazymaster/README.md
# 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
#!/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())
