r128.py

#!/usr/bin/env python
from __future__ import print_function

from glob import glob
import multiprocessing
import os
import signal
import subprocess
import sys


def get_loudness(file_location):
    # Run ffmpeg to obtain loudness stats
    command = [
        'ffmpeg', '-nostats', '-i', file_location,  '-filter_complex',
        'ebur128=peak=true', '-f', 'null', '-'
    ]
    try:
        output = subprocess.check_output(
            command, stderr=subprocess.STDOUT, stdin=subprocess.PIPE
        )
    except subprocess.CalledProcessError:
        return None

    # Split the output to obtain the summary
    # e.g.
    # [Parsed_ebur128_0 @ 0x7fb0a1d1a7e0] Summary:
    #
    #   Integrated loudness:
    #     I:         -16.1 LUFS
    #     Threshold: -26.7 LUFS
    #
    #   Loudness range:
    #     LRA:        14.5 LU
    #     Threshold: -36.7 LUFS
    #     LRA low:   -28.1 LUFS
    #     LRA high:  -13.5 LUFS
    #
    #   True peak:
    #     Peak:       -1.0 dBFS
    summary = output.split("Summary:")[-1]

    # Obtain integrated loudness stats
    integrated_loudness = (
        summary.split("Integrated loudness:")[1].split("Loudness range:")[0]
    )
    integrated_loudness_i = float(
        integrated_loudness.split("I:")[1].split("Threshold:")[0].split("LUFS")[0].strip()
    )
    integrated_loudness_threshold = float(
        integrated_loudness.split("Threshold:")[1].strip().split("LUFS")[0]
    )

    # Obtain loudness range stats
    loudness_range = summary.split("Loudness range:")[1].split("True peak:")[0]
    loudness_range_lra = float(
        loudness_range.split("LRA:")[1].split("Threshold:")[0].split("LU")[0].strip()
    )
    loudness_range_threshold = float(
        loudness_range.split("Threshold:")[1].split("LRA low:")[0].split("LUFS")[0].strip()
    )
    loudness_range_lra_low = float(
        loudness_range.split("LRA low:")[1].split("LRA high:")[0].split("LUFS")[0].strip()
    )
    loudness_range_lra_high = float(
        loudness_range.split("LRA high:")[1].split("LUFS")[0].strip()
    )

    # Obtain true peak stats
    true_peak = summary.split("True peak:")[1].strip()
    true_peak_peak = float(
        true_peak.split("Peak:")[1].split("dBFS")[0].strip()
    )

    return {
        'Integrated Loudness': {
            'I': integrated_loudness_i,
            'Threshold': integrated_loudness_threshold
        },
        'Loudness Range': {
            'LRA': loudness_range_lra,
            'Threshold': loudness_range_threshold,
            'LRA Low': loudness_range_lra_low,
            'LRA High': loudness_range_lra_high
        },
        'True Peak': {
            'Peak': true_peak_peak
        }
    }


def main():
    # Obtain arguments from user
    if len(sys.argv) < 2:
        print('Usage: {app} <filepath> [<filepath>, <filepath>, ...]'.format(
            app=__file__
        ))
        exit(1)
    raw_filepaths = sys.argv[1:]

    # Glob all filepaths provided and remove duplicates
    filepaths = set()
    for raw_filepath in raw_filepaths:
        filepaths.update(glob(raw_filepath))
    filepaths = sorted(filepaths)

    # After globbing, we may have an empty list of files, so exit
    if not filepaths:
        print('Warning: No files specified could be found')
        exit(1)

    def init_worker():
        signal.signal(signal.SIGINT, signal.SIG_IGN)

    try:
        pool = multiprocessing.Pool(
            min(multiprocessing.cpu_count(), len(filepaths)), init_worker
        )
        jobs = [pool.apply_async(get_loudness, [f]) for f in filepaths]
        loudnesses = [job.get(999) for job in jobs]
    except KeyboardInterrupt:
        pool.terminate()
        exit(2)

    # Print output
    max_filepath_length = max(len(os.path.basename(f)) for f in filepaths)
    max_length = max(max_filepath_length, 8)
    format_string_header = (
        '{filepath:^' + str(max_length + 4) + '}  '
        '{integrated:^10}  {lra:^10}  {true_peak:^10}'
    )
    format_string_stats = (
        '{filepath:<' + str(max_length + 4) + '}  '
        '{integrated:^10}  {lra:^10}  {true_peak:^10}'
    )

    print('')
    print(format_string_header.format(
        filepath='Filepath',
        integrated='Integrated',
        lra='LRA',
        true_peak='True Peak'
    ))
    for filepath, loudness in zip(filepaths, loudnesses):
        if not loudness:
            continue

        print(format_string_stats.format(
            filepath=os.path.basename(filepath),
            integrated=loudness['Integrated Loudness']['I'],
            lra=loudness['Loudness Range']['LRA'],
            true_peak=loudness['True Peak']['Peak']
        ))
    print('')

if __name__ == '__main__':
    main()