lazymaster.py (6586B)
1 #!/usr/bin/env python3 2 import argparse 3 import json 4 import subprocess 5 import sys 6 from collections.abc import Sequence 7 8 DEFAULT_LUFS = -14.0 9 DEFAULT_LRA = 7.0 10 DEFAULT_TRUE_PEAK = -1.0 11 PCM_WAV_CODECS = {"pcm_s16le", "pcm_s24le", "pcm_s32le", "pcm_f32le", "pcm_f64le"} 12 13 14 def command_error(error: subprocess.CalledProcessError) -> str: 15 output = error.stderr or error.stdout or "" 16 return output.strip() 17 18 19 def run(command: Sequence[str], **kwargs: object) -> subprocess.CompletedProcess[str]: 20 return subprocess.run(command, check=True, text=True, **kwargs) 21 22 23 def audio_properties(input_file: str) -> tuple[str, str]: 24 probe = run( 25 [ 26 "ffprobe", 27 "-v", 28 "error", 29 "-select_streams", 30 "a:0", 31 "-show_entries", 32 "stream=codec_name,sample_rate", 33 "-of", 34 "json", 35 input_file, 36 ], 37 capture_output=True, 38 ) 39 try: 40 stream = json.loads(probe.stdout)["streams"][0] 41 sample_rate = stream["sample_rate"] 42 except (IndexError, KeyError, TypeError, json.JSONDecodeError) as error: 43 raise ValueError("ffprobe found no usable audio stream") from error 44 45 codec = stream.get("codec_name") 46 return (codec if codec in PCM_WAV_CODECS else "pcm_s24le"), sample_rate 47 48 49 def loudnorm_filter( 50 args: argparse.Namespace, measurements: dict[str, str] | None = None 51 ) -> str: 52 settings = f"I={args.lufs}:TP={args.true_peak}:LRA={args.lra}" 53 if measurements is None: 54 return f"loudnorm={settings}:print_format=json" 55 56 return ( 57 f"loudnorm={settings}:measured_I={measurements['input_i']}:" 58 f"measured_LRA={measurements['input_lra']}:" 59 f"measured_TP={measurements['input_tp']}:" 60 f"measured_thresh={measurements['input_thresh']}:" 61 f"offset={measurements['target_offset']}:linear=false" 62 ) 63 64 65 def parse_measurements(stderr: str) -> dict[str, str]: 66 end = stderr.rfind("}") 67 start = stderr.rfind("{", 0, end + 1) 68 if start < 0 or end < 0: 69 raise ValueError("ffmpeg did not print loudnorm measurements") 70 71 try: 72 measurements = json.loads(stderr[start : end + 1]) 73 required = ("input_i", "input_tp", "input_lra", "input_thresh", "target_offset") 74 if not all(key in measurements for key in required): 75 raise ValueError("ffmpeg returned incomplete loudnorm measurements") 76 except json.JSONDecodeError as error: 77 raise ValueError("could not parse loudnorm measurements") from error 78 79 return measurements 80 81 82 def measure(input_file: str, args: argparse.Namespace) -> dict[str, str]: 83 command = ["ffmpeg", "-hide_banner"] 84 if input_file != "-": 85 command.append("-nostdin") 86 command.extend(["-i", input_file, "-af", loudnorm_filter(args), "-f", "null", "-"]) 87 result = run(command, capture_output=True) 88 return parse_measurements(result.stderr) 89 90 91 def normalise( 92 input_file: str, 93 output_file: str, 94 measurements: dict[str, str], 95 args: argparse.Namespace, 96 ) -> None: 97 if input_file == "-": 98 raise ValueError( 99 "two-pass normalisation needs a seekable input file, not standard input" 100 ) 101 102 codec, sample_rate = audio_properties(input_file) 103 command = [ 104 "ffmpeg", 105 "-hide_banner", 106 "-nostdin", 107 "-i", 108 input_file, 109 "-map", 110 "0:a:0", 111 "-map_metadata", 112 "0", 113 "-af", 114 loudnorm_filter(args, measurements), 115 "-ar", 116 sample_rate, 117 "-c:a", 118 codec, 119 ] 120 if output_file == "-": 121 command.extend(["-f", "wav", "-"]) 122 else: 123 command.extend(["-y" if args.force else "-n", output_file]) 124 run(command) 125 126 127 def parse_args(argv: Sequence[str]) -> argparse.Namespace: 128 parser = argparse.ArgumentParser( 129 description="Measure or two-pass normalise an audio file using ffmpeg loudnorm." 130 ) 131 parser.add_argument( 132 "input", 133 metavar="INPUT", 134 help="input audio file, or - for standard input (analysis only)", 135 ) 136 parser.add_argument( 137 "output", 138 metavar="OUTPUT", 139 nargs="?", 140 help="normalised WAV file; use - for standard output", 141 ) 142 parser.add_argument( 143 "-o", 144 "--output", 145 dest="output_option", 146 metavar="OUTPUT", 147 help="normalised WAV file", 148 ) 149 parser.add_argument( 150 "-f", "--force", action="store_true", help="replace an existing output file" 151 ) 152 parser.add_argument( 153 "--lufs", 154 type=float, 155 default=DEFAULT_LUFS, 156 help=f"target integrated loudness (default: {DEFAULT_LUFS:g})", 157 ) 158 parser.add_argument( 159 "--lra", 160 type=float, 161 default=DEFAULT_LRA, 162 help=f"target loudness range (default: {DEFAULT_LRA:g})", 163 ) 164 parser.add_argument( 165 "--true-peak", 166 type=float, 167 default=DEFAULT_TRUE_PEAK, 168 help=f"maximum true peak in dBTP (default: {DEFAULT_TRUE_PEAK:g})", 169 ) 170 args = parser.parse_args(argv) 171 if args.output and args.output_option: 172 parser.error("OUTPUT and --output cannot be used together") 173 args.output = args.output_option or args.output 174 if args.input == args.output and args.output not in (None, "-"): 175 parser.error("input and output must be different files") 176 if args.force and args.output in (None, "-"): 177 parser.error("--force requires a file output") 178 return args 179 180 181 def main(argv: Sequence[str] | None = None) -> int: 182 args = parse_args(sys.argv[1:] if argv is None else argv) 183 try: 184 measurements = measure(args.input, args) 185 report = json.dumps(measurements, sort_keys=True) 186 print(report, file=sys.stderr if args.output == "-" else sys.stdout) 187 if args.output: 188 normalise(args.input, args.output, measurements, args) 189 except FileNotFoundError as error: 190 print( 191 f"lazymaster: required command not found: {error.filename}", file=sys.stderr 192 ) 193 return 127 194 except ValueError as error: 195 print(f"lazymaster: {error}", file=sys.stderr) 196 return 1 197 except subprocess.CalledProcessError as error: 198 print( 199 f"lazymaster: command failed with exit status {error.returncode}", 200 file=sys.stderr, 201 ) 202 if detail := command_error(error): 203 print(detail, file=sys.stderr) 204 return error.returncode or 1 205 except KeyboardInterrupt: 206 return 130 207 return 0 208 209 210 if __name__ == "__main__": 211 raise SystemExit(main())