configs/music-normalize/main.py

274 lines
8.0 KiB
Python
Raw Normal View History

2022-10-03 19:58:45 +02:00
#!/usr/bin/env python3
# multithreading
2022-10-04 18:16:56 +02:00
import multiprocessing
# audio format conversions
2022-10-03 19:58:45 +02:00
import ffmpy
# argument parsing
import argparse
2022-10-04 18:16:56 +02:00
# multiprocessing stuff
from multiprocessing import Pool
2022-10-03 19:58:45 +02:00
# executing some commands
import subprocess
2022-10-04 18:16:56 +02:00
# file/directory handling
import os
2022-10-03 19:58:45 +02:00
2022-10-04 18:16:56 +02:00
# most recent starttime for program
import time
2022-10-03 19:58:45 +02:00
# randomness
2022-10-04 18:16:56 +02:00
from random import randint
2022-10-03 19:58:45 +02:00
# typing hints
from typing import Any, Optional
2022-11-13 21:54:24 +01:00
import tempfile
# working with sound files
import soundfile
2022-10-03 19:58:45 +02:00
# loudness normalization
import pyloudnorm
# file copy
import shutil
"""
Normalize loudness of all music files in a given directory and its subdirectories.
2022-10-03 19:58:45 +02:00
"""
2022-10-04 18:16:56 +02:00
musicfile_extensions = (".flac", ".wav", ".mp3", ".m4a", ".aac", ".opus")
2022-10-03 19:58:45 +02:00
def loudnorm(inputfile: str, outputfile: str):
2022-10-04 18:16:56 +02:00
"""
Normalize audio to EBU R 128 standard using pyloudnorm
2022-10-04 18:16:56 +02:00
Parameters:
inputfile (str): Path to input file. Format must be supported by python-soundfile module
outputfile (str): Path to output file
"""
data, rate = soundfile.read(file=inputfile)
2022-10-04 18:16:56 +02:00
# measure loudness
meter = pyloudnorm.Meter(rate=rate)
loudness = meter.integrated_loudness(data=data)
# normalize audio
file_normalized = pyloudnorm.normalize.loudness(
2024-05-10 19:21:03 +02:00
data=data, input_loudness=loudness, target_loudness=-18.0
2022-10-03 19:58:45 +02:00
)
# write normalized audio to file
soundfile.write(file=outputfile, data=file_normalized, samplerate=rate)
def ffmpeg_to_wav(inputfile: str, outputfile: str):
"""
Convert a file into .wav for further processing
Parameters:
inputfile (str): Path to input file
outputfile (str): Path to output file
"""
# convert to wav in temporary directory
with tempfile.TemporaryDirectory() as tempdir:
# temporary input file
temp_input: str = os.path.join(
tempdir, os.path.splitext(os.path.basename(inputfile))[0] + ".wav"
)
# temporary output file
temp_output: str = os.path.join(
tempdir,
"normalized",
os.path.splitext(os.path.basename(inputfile))[0] + ".wav",
)
os.mkdir(os.path.join(tempdir, "normalized"))
# convert audio to wav
ff = ffmpy.FFmpeg(
inputs={inputfile: None}, outputs={temp_input: None}, global_options=("-y")
)
subprocess.run(ff.cmd, shell=True, capture_output=True)
# normalize loudness
loudnorm(inputfile=temp_input, outputfile=temp_output)
# convert audio back to lossy format
outputcmd = {
outputfile: "-c:a libopus" " " "-b:a 192k" " " "-compression_level 10"
}
ff = ffmpy.FFmpeg(
inputs={temp_output: None}, outputs=outputcmd, global_options=("-y")
)
subprocess.run(ff.cmd, shell=True, capture_output=True)
2022-10-03 19:58:45 +02:00
def ffmpeg_copy_metadata(inputfile: str, outputfile: str):
"""
Copy all metadata from the input file to the output file.
A temporary file is used in an intermediate step
Parameters:
inputfile (str): Path to input file
outputfile (str): Path to output file
"""
# store output file as temporary file. FFMPEG can't work on files in-place
with tempfile.NamedTemporaryFile() as temp_audio:
shutil.copyfile(outputfile, temp_audio.name)
2024-05-10 19:20:54 +02:00
# get input file extension
extension = os.path.splitext(os.path.basename(inputfile))[1]
inputcmd = {inputfile: None, temp_audio.name: None}
2024-05-10 19:20:54 +02:00
# NOTE opus maps metadata to the first audio stream. Other formats like flac, mp3 and m4a/aac by contrast map it to the input directly
if extension == ".opus":
outputcmd = {outputfile: "-map 1 -c copy -map_metadata 0:s"}
else:
outputcmd = {outputfile: "-map 1 -c copy -map_metadata 0"}
ff = ffmpy.FFmpeg(inputs=inputcmd, outputs=outputcmd, global_options=("-y"))
subprocess.run(ff.cmd, shell=True, capture_output=True)
def main(inputfile: str) -> Optional[list[Any]]:
2022-10-04 18:16:56 +02:00
"""
Main program loop
Parameters:
inputfile (str): Path to input file
Output:
dynamically normalised audio files (list)
2022-10-04 18:16:56 +02:00
"""
2022-10-04 18:16:56 +02:00
# set output folder to parent path + "normalized"
outputfolder = os.path.join(os.path.dirname(inputfile), "normalized")
2022-10-04 18:16:56 +02:00
# NOTE create output folder
# because multiple parallel processes are at work here,
# there might be conflicts with one trying to create the directory although it already exists
# this while loop makes sure the directory does exist
# the try/except block ensures the error is caught and (hopefully) doesn't happen again just after with random sleep
# there's very likely a better way to do this, idk
while not os.path.isdir(outputfolder):
try:
os.mkdir(outputfolder)
except:
time.sleep(randint(0, 4))
# output file path
infile_noextension: str = os.path.splitext(os.path.basename(inputfile))[0]
infile_extension: str = os.path.splitext(os.path.basename(inputfile))[1]
match infile_extension:
case ".flac" | ".wav":
print("Working on", inputfile)
outputfile: str = os.path.join(outputfolder, infile_noextension + ".flac")
# direct conversion start
loudnorm(inputfile=inputfile, outputfile=outputfile)
ffmpeg_copy_metadata(inputfile=inputfile, outputfile=outputfile)
print("Completed", inputfile)
case ".mp3" | ".m4a" | ".aac" | ".opus":
print("Working on", inputfile)
outputfile: str = os.path.join(outputfolder, infile_noextension + ".opus")
# conversion is started within the ffmpeg_to_wav function
ffmpeg_to_wav(inputfile=inputfile, outputfile=outputfile)
ffmpeg_copy_metadata(inputfile=inputfile, outputfile=outputfile)
print("Completed", inputfile)
case _:
print(
inputfile,
"does not use a known extension. This error shouldn't be happening actually",
)
return
2022-10-04 18:16:56 +02:00
if __name__ == "__main__":
"""
Handle arguments and other details for interactive usage
"""
# start time of program
starttime = time.time()
2022-10-04 18:16:56 +02:00
parser = argparse.ArgumentParser(description="")
# Input directory
parser.add_argument(
"-i", "--input-dir", required=True, type=str, help="Input source directory"
)
# number of cpus/threads to use, defaults to all available
parser.add_argument(
"-c",
"--cpu-count",
required=False,
type=int,
help="Number of cpu cores",
default=multiprocessing.cpu_count(),
)
# in case you wanted to rerun the conversion for everything
parser.add_argument(
"-r",
"--reset",
required=False,
action="store_true",
help="Rerun conversion for all files",
)
2022-10-04 18:16:56 +02:00
args = parser.parse_args()
srcfolder = args.input_dir
cpu = args.cpu_count
reset = args.reset
# file where last run timestamp is stored
2022-10-04 18:16:56 +02:00
timefile = os.path.join(srcfolder, "run.time")
2022-11-13 21:54:24 +01:00
# list of non-linear normalizations
nonlinear_all: Optional[list[Any]] = []
2022-11-13 21:54:24 +01:00
2022-10-04 18:16:56 +02:00
# get time of previous run
if reset:
timeprev = 0
elif os.path.isfile(timefile):
2022-10-04 18:16:56 +02:00
with open(timefile, "r") as file:
timeprev = file.read()
else:
timeprev = 0
musicfiles: list[str] = []
2022-10-04 18:16:56 +02:00
for root, dirs, files in os.walk(srcfolder):
# ignore the "normalized" subfolder
dirs[:] = [d for d in dirs if d not in ["normalized"]]
for file in files:
if file.endswith(musicfile_extensions):
filepath = os.path.join(root, file)
# only file newer than the last run are added
if os.path.getmtime(filepath) >= float(timeprev):
musicfiles.append(os.path.join(root, file))
with Pool(cpu) as p:
nonlinear_all: Optional[list[Any]] = p.map(main, musicfiles)
2022-10-04 18:16:56 +02:00
# write this run's time into file
with open(timefile, "w") as file:
file.write(str(starttime))