# Comparing FFTs to deduplicate IVR recordings

I've got a Python script that traverses two file trees, checking all .wav files for duplication.

I'm still an undergrad, and have never worked with audio before. I'm not entirely sure this is going to be a valid comparison.

I lost the SO answer I was using as a guideline, but the general idea is to split the audio into chunks, FFT those chunks, then compare the results.

I also would appreciate any structure/style advice.

Note: All Olympus Recordings are valid, but I need to load them to have the ffts to compare.

import os
import os.path
import wave
import numpy
import struct
from shutil import copy
from subprocess import call, Popen, PIPE

oly_path = "/home/will/Desktop/soundfiles/Olympus Recordings"
titan_path = "/home/will/Desktop/soundfiles/TITAN Recordings"
wav_roots = [oly_path, titan_path]
destination_path = "/home/will/Desktop/soundfiles/output"
temp_wav_path = "/home/will/Desktop/soundfiles/temp.wav"

similarity_threshold = 95
validated_wavs = {}
bunk_wavs = {}

def get_fft_similarity(fft1, fft2):
hits = 0
len1 = len(fft1)
len2 = len(fft2)
limit = (len1 if len1<len2 else len2)

if limit == 0:
return 0

for i in xrange(limit):
if (fft1[i] == fft2[i]).all():
hits += 1

ratio = float(hits)/limit
return int(ratio*100)

def sort_wav(to_sort):
key = to_sort.filename
if "Olympus" in to_sort.source:
while key in validated_wavs:
to_sort.generate_new_name()
key = to_sort.filename

validated_wavs[key] = to_sort
return 1

if key in validated_wavs:
if get_fft_similarity(to_sort.fft, validated_wavs[key].fft) >= similarity_threshold:
bunk_wavs[key] = to_sort
return 0

while key in validated_wavs:
to_sort.generate_new_name()
key = to_sort.filename

for w in validated_wavs.values():
if get_fft_similarity(to_sort.fft, w.fft) >= similarity_threshold:
bunk_wavs[key] = to_sort
return 0

validated_wavs[key] = to_sort
return 1

def get_fft(fft_path):
return_list = []
frames_list = []
chunk_size = 36
start = 0
wav = wave.open(fft_path, 'r')
wav.close()

while start+chunk_size < len(frames):
data = struct.unpack(">fdddd", frames[start:start+chunk_size])
frames_list.append(data)
start += chunk_size
if len(frames_list) >= 1000:
return_list.append(numpy.fft.fft(frames_list))
frames_list = []

return return_list

def ensure_encoding(encoding_path):
command = ["soxi", "-e", encoding_path]
command_proc = Popen(command, stdin=PIPE, stdout=PIPE, stderr=PIPE)
encoding = command_proc.communicate()[0].rstrip()

if encoding != "Signed Integer PCM":
call(["sox", encoding_path, "-e", "signed", temp_wav_path])
else:
copy(encoding_path, temp_wav_path)

def validate(validate_path):
ensure_encoding(validate_path)
ivr_wav = IVR_Wav(validate_path)
return sort_wav(ivr_wav)

class IVR_Wav:
def __init__(self, source):
self.source = source
self.destination = destination_path + source.split("/")[-1]
self.size = os.path.getsize(source)
self.fft = get_fft(temp_wav_path)
self.filename = os.path.split(source)[-1]

def generate_new_name(self):
name = os.path.split(self.source)[-1].replace(".wav", "")
directory = os.path.split(self.source)[-2].split("/" if "/" in self.source else "\\")[-1]
self.filename = "".join([name, "_", directory, ".wav"])

self.destination = os.path.join(destination_path, self.filename)

def main():
valid = 0
total = 0
for wav_root in wav_roots:
for root, dirs, files in os.walk(wav_root):
for f in files:
if f.endswith("wav") and os.path.getsize(os.path.join(root, f)) > 16:
total += 1
temp_path = os.path.join(root, f)
valid += validate(temp_path)
print "\r", valid, "/", total, " valid::::::", len(validated_wavs),

main()


UPPERCASE for constants

As a convention, in Python constants are uppercase:

OLY_PATH = "/home/will/Desktop/soundfiles/Olympus Recordings"
TITAN_PATH = "/home/will/Desktop/soundfiles/TITAN Recordings"
WAV_ROOTS = [oly_path, titan_path]
DESTINATION_PATH = "/home/will/Desktop/soundfiles/output"
TEEMP_WAV_PATH = "/home/will/Desktop/soundfiles/temp.wav"


No need for temporary variables

    len1 = len(fft1)
len2 = len(fft2)


Are useless because you use them only once, thus leading to no efficiency gain and code clutter.

def get_fft_similarity(fft1, fft2):
hits = 0
limit = (len1 if len(fft1)<len(fft2) else len2)

if limit == 0:
return 0

for i in xrange(limit):
if (fft1[i] == fft2[i]).all():
hits += 1

ratio = float(hits)/limit
return int(ratio*100)


• float(hits)/limit you can remove float if you from __future__ import division

• Down below:

 for i in xrange(limit):
if (fft1[i] == fft2[i]).all():
hits += 1


You are using explicit indexing and a counter variable. Maybe zip is and sum will simplify your code.

Use more constants

You already use some, and that is nice, but things like:

• "Olympus"
• ".wav"
• "Signed Integer PCM"
• "soxi"
• ...

Should be in their own named constant.

Are you sure of your class?

Usually if a class has init and only one other function, maybe you can use just a function for simplicity: I am looking at class IVR_Wav

Peculiar double effect sort_wav

I noticed that sort_wav both modifies a global variable and returns 1 or 0. It is then called by validate. Maybe you could write a validator_helper to make the check and another function to update the dictionary.

• Thanks for all the feedback, just what I was looking for! I like the class, am planning on sorting the wavs, then iterating over validated_wavs.values to record the IVR_Wav.destination in a csv wherever the IVR_Wav.source appears. Is this better done during sort_wav()? Then store only the FFTs for comparison? – Will Jun 25 '15 at 17:33
• @Will I would do it the most straigthforward way possible, In/Output will take 95% of the time, so using two passes and two separate functions seems best to me. – Caridorc Jun 25 '15 at 17:34
• @Will I am so curious about ">fdddd", can you explain me what it means? (And then put it in a constant) – Caridorc Jun 25 '15 at 17:40
• ">fdddd" is apparently specifying little endian, a four byte-float, and four eight-byte doubles (all signed). I chose this format because it was a way to unpack the frames into 36-byte chunks. I ran getsizeof(frames[0]) to get 36 bytes per frame. Is this correct? – Will Jun 25 '15 at 17:45
• @Will It's exactly because I have no idea that I asked, so I can't confirm you that it is correct :) (By the way, if the matter is so complicated also add a comment near that string). – Caridorc Jun 25 '15 at 17:49