Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
phaseanalyze: analyze a set of audio files to find optimum phase configuration
#!/usr/bin/env python3
# -*- coding: UTF-8 -*-
Analyze a set of audio files to find optimum phase configuration.
import argparse
import itertools
import os
import subprocess
import tabulate
TMPDIR = "/tmp"
class TooLongError(Exception):
"""Raise when input files are too long to analyse."""
def __init__(self, audiofile, duration):
self.message = "File {} is {} seconds long!".format(audiofile, duration)
class IdentificationFailed(Exception):
"""Raise when sndfile-info fails."""
def __init__(self, audiofile):
self.message = "Audio identification of {} failed!".format(audiofile)
class MixAnalyzeFailed(Exception):
"""Raise when sox mix call fails."""
def __init__(self, cmd):
self.message = "Call to sox failed!\n {}".format("".join(cmd))
def readaudiofile(audiofile):
"""Return channels, frames, samplerate"""
print("analysing file {0}...".format(audiofile))
info =["sndfile-info", audiofile], stdout=subprocess.PIPE)
channels = None
frames = None
rate = None
for infoline in info.stdout.decode().split("\n"):
if infoline.startswith("Channels"):
channels = int(infoline.split(":")[-1].strip())
if infoline.startswith("Frames"):
frames = int(infoline.split(":")[-1].strip())
if infoline.startswith("Sample Rate"):
rate = int(infoline.split(":")[-1].strip())
if None in (channels, frames, rate):
raise IdentificationFailed(audiofile)
return channels, frames, rate
def splitmultichannel(audiofile, channels):
"""Split into mono audio files."""
audiobasename = audiofile.rpartition("/")[2]
monofiles = []
for ch in range(channels):
monofile = "{}/{}_{}".format(TMPDIR, ch + 1, audiobasename)
cmd = ["sox", audiofile, monofile, "remix", str(ch + 1)]
return monofiles
def processpermutation(audiofiles, permutation):
"""Mix files with sox and measure RMS level."""
cmd = ["sox", "-m"]
for phase, audiofile in zip(permutation, audiofiles):
if phase:
cmd.extend([phase, str(audiofile)])
cmd.extend(["-n", "stats"])
info =, stderr=subprocess.PIPE)
for infoline in info.stderr.decode().split("\n"):
if infoline.startswith("RMS lev dB"):
return float(infoline.split("dB")[-1].strip())
raise MixAnalyzeFailed(cmd)
def execute(args):
audiofiles = []
tmpmonofiles = []
for audiofile in args.audiofiles:
channels, frames, rate = readaudiofile(audiofile)
if (duration := int(frames / rate)) > MAXIMUM_DURATION:
raise TooLongError(audiofile, duration)
if channels > 1:
tmpmonofiles.extend(splitmultichannel(audiofile, channels))
results = {}
for permutation in itertools.product(["", "-v -1"], repeat=len(audiofiles)):
print(".", end="", flush=True)
inversepermutation = tuple("" if x else "-v -1" for x in permutation)
if inversepermutation not in results:
results[permutation] = processpermutation(audiofiles, permutation)
if len("".join(permutation)) < len("".join(inversepermutation)):
results[permutation] = results[inversepermutation]
del results[inversepermutation]
maxkey = max(results, key=lambda key: results[key])
optimization = [["Channel", "Phase"]]
optimization.extend(zip(audiofiles, ["ø" if x else "" for x in maxkey]))
print("\n\n", tabulate.tabulate(optimization, headers="firstrow"))
print("\nUnchanged RMS {} dB".format(results[tuple("" for i in range(len(audiofiles)))]))
print("Optimized RMS {} dB".format(results[maxkey]))
print("\n{} phase permutations calculated.".format(len(results)))
# delete mono audio files
for tmf in tmpmonofiles:
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("audiofiles", metavar="FILE", nargs="+", help="audio files to analyze")
args = parser.parse_args()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment