Shield your eyes!

gistfile1.py

# Test with:
#
# $ curl -o sample1.wav https://api.twilio.com/2010-04-01/Accounts/AC44d08277b616939d314d0c81832c389a/Recordings/REb9c469219996e51b1b9aee996b3ff48309
# $ curl -o sample2.wav https://api.twilio.com/2010-04-01/Accounts/AC44d08277b616939d314d0c81832c389a/Recordings/REf7ba239cedd653ce75390c925ed51aad
# $ python compare_audio.py sample1.wav sample2.wav

import wave
import sys
import numpy as np

import struct
import itertools

file1 = sys.argv[1]
file2 = sys.argv[2]

def dB(a,base=1.0):
    return 10.0*np.log10(a/base)

def get_mags(results, x, y):
    highest_mag = 0
    highest_freq = -1
    for freq, mag in results.iteritems():
        normalized_mag = np.log(np.abs(mag)+1)
        if freq > x and freq < y and normalized_mag > highest_mag:
            highest_mag = normalized_mag
            highest_freq = freq

    return highest_freq

def calc(filename):
    print '[processing %s]' % filename
    wf = wave.open(filename, 'r')
    frames = wf.getnframes()
    data = wf.readframes(frames)
    wf.close()

    # read data
    data = struct.unpack('{n}h'.format(n=frames), data)
    data = np.array(data)

    result_mags = []
    result_freqs = []
    chunk_size = 8000
    amountPossible = len(data)/chunk_size

    signatures = []
    for i in xrange(amountPossible):
        fft = np.fft.fft(data[i*chunk_size:i*chunk_size+chunk_size])
        freqs = np.fft.fftfreq(len(fft))
        result_mags.append(fft)
        result_freqs.append(freqs)
        idx=np.argmax(np.abs(fft)**2)
        freq=freqs[idx]
        freq_in_hertz=abs(freq*chunk_size)

        freqs = map(lambda x: abs(x * chunk_size), freqs)
        results = dict(itertools.izip(freqs, fft))
        low = get_mags(results, 0, 40)
        lowmid = get_mags(results, 40, 80)
        mid = get_mags(results, 80, 120)
        highmid = get_mags(results, 120, 180)
        high = get_mags(results, 180, 300)
        line = '%f %f %f %f %f' % (low, lowmid, mid, highmid, high)
        #print line

        FUZ_FACTOR = 2
        signature = (high-(high%FUZ_FACTOR)) * 100000000 + (highmid-(highmid%FUZ_FACTOR)) * 100000 + (mid-(mid%FUZ_FACTOR)) * 100 + (lowmid-(lowmid%FUZ_FACTOR))

        print 'signature: %s @ %d' % (signature, i)
        signatures.append(signature)

    return signatures

def find_offset(sig1, sig2):
    # find first matching hash in sig2, then return offset
    for i, sig in enumerate(sig1):
        if sig in sig2:
            j = sig2.index(sig)
            print 'found %s at offset (sig1: %d, sig2: %d)' % (sig, i, j)
            return (i, j)

    print 'offset not found'
    return None

def compare_audio(sig1, sig2):
    offsets = find_offset(sig1, sig2)
    if offsets is not None:
        (sig1_start, sig2_start) = offsets
        matches = 0
        for s1, s2 in zip(sig1[sig1_start:], sig2[sig2_start:]):
            if s1 == s2:
                matches += 1

        print 'found %d matches, (%d/%d ~= %f%%)' % (matches, matches, len(sig1), (1.0*matches/len(sig1)*100))
        return (1.0*matches/len(sig1))*100
    return 0

s1_signatures = calc(file1)
s2_signatures = calc(file2)

print 'Comparing %s to %s:' % (file1, file2)
comparison = compare_audio(s1_signatures, s2_signatures)

if comparison > 20:
    # pretty close
    sys.exit(0)
else:
    sys.exit(1)