Parse *.afpk file and extract peaks, landmarks and hashes

parseafpk.py

import struct
import numpy as np

def afpk2peaks(peakfilename):
    PEAK_FMT = '<2i'
    PEAK_MAGIC = 'audfprintpeakV00'  # 16 chars, FWIW

    """ Read back a set of (time, bin) pairs written by peaks_save """
    peaks = []
    fmtsize = struct.calcsize(PEAK_FMT)
    with open(peakfilename, 'rb') as f:
        magic = f.read(len(PEAK_MAGIC))
        if magic != PEAK_MAGIC:
            raise IOError('%s is not a peak file (magic %s)'
                          % (peakfilename, magic))
        data = f.read(fmtsize)
        while data is not None and len(data) == fmtsize:
            peaks.append(struct.unpack(PEAK_FMT, data))
            data = f.read(fmtsize)
    return peaks

def landmarks2hashes(landmarks):
    """Convert a list of (time, bin1, bin2, dtime) landmarks
    into a list of (time, hash) pairs where the hash combines
    the three remaining values.
    """
    F1_BITS = 8
    DF_BITS = 6
    DT_BITS = 6

    B1_MASK = (1 << F1_BITS) - 1
    B1_SHIFT = DF_BITS + DT_BITS
    DF_MASK = (1 << DF_BITS) - 1
    DF_SHIFT = DT_BITS
    DT_MASK = (1 << DT_BITS) - 1

    landmarks = np.array(landmarks)
    hashes = np.zeros((landmarks.shape[0], 2), dtype=np.int32)
    hashes[:, 0] = landmarks[:, 0]
    hashes[:, 1] = (((landmarks[:, 1] & B1_MASK) << B1_SHIFT)
                    | (((landmarks[:, 2] - landmarks[:, 1]) & DF_MASK)
                       << DF_SHIFT)
                    | (landmarks[:, 3] & DT_MASK))
    return hashes

def peaks2landmarks(pklist):
    """ Take a list of local peaks in spectrogram
        and form them into pairs as landmarks.
        pklist is a column-sorted list of (col, bin) pairs as created
        by findpeaks().
        Return a list of (col, peak, peak2, col2-col) landmark descriptors.
    """
    maxpairsperpeak = 3
    targetdt = 63
    targetdf = 31
    mindt = 2
    # Form pairs of peaks into landmarks
    landmarks = []
    if len(pklist) > 0:
        # Find column of the final peak in the list
        scols = pklist[-1][0] + 1
        # Convert (col, bin) list into peaks_at[col] lists
        peaks_at = [[] for col in xrange(scols)]
        for (col, bin) in pklist:
            peaks_at[col].append(bin)

        # Build list of landmarks <starttime F1 endtime F2>
        for col in xrange(scols):
            for peak in peaks_at[col]:
                pairsthispeak = 0
                for col2 in xrange(col+mindt,
                                   min(scols, col+targetdt)):
                    if pairsthispeak < maxpairsperpeak:
                        for peak2 in peaks_at[col2]:
                            if abs(peak2-peak) < targetdf:
                                #and abs(peak2-peak) + abs(col2-col) > 2 ):
                                if pairsthispeak < maxpairsperpeak:
                                    # We have a pair!
                                    landmarks.append((col, peak,
                                                      peak2, col2-col))
                                    pairsthispeak += 1

    return landmarks
    
    
    
    
peakfilename = "./data/afpk/16SlippedDisc/01 Louise.afpk"

peaks = afpk2peaks(peakfilename)
lm = peaks2landmarks(peaks)
hashes = landmarks2hashes(lm)