#! /usr/bin/env python import sys import os import string import copy as C from numpy import * def fullpath_minus_extension(fn) : parts = fn.split('.') return string.join(parts[0:len(parts)-1],'.') def load_beats(fn) : fid = open(fn,'r') beats=list() for l in fid : if l.count(',')>0 : #comma-delimited beats.append(map(float,l.split(','))) else : curr_beats = l.split() #space delimited if len(curr_beats)==1 or len(curr_beats)==2 : #just a time or, with 2 columns, in the format from sonic visualizer, ignore column 2 beats.append(float(curr_beats[0])) else : beats.append(map(float,curr_beats)) return array(beats) def davies_global_score(global_hist) : global_hist/=sum(global_hist) global_hist[where(global_hist==0.0)]=1 global_entropy = -sum(global_hist * log(global_hist)) global_score= 100.0 * (1-exp(global_entropy)/len(global_hist)) return global_score def davies_get_beat_error(tgt_beats,pred_beats) : f_tgt_beats = tgt_beats.copy() f_pred_beats = pred_beats.copy() error=list() # deal first annotation for k in range(len(f_tgt_beats)) : if k==0 : # would be 1 in matlabe actmin = 3 pre_interval = f_tgt_beats[k]-3; else : pre_interval = 0.5*(f_tgt_beats[k]-f_tgt_beats[k-1]) actmin = f_tgt_beats[k] - pre_interval # if k==length(annots), # actmax = max(annots(k),beats(end)); # post_interval = max(annots(end),beats(end)) - annots(k); # deal with last annotation... if k==len(f_tgt_beats)-1 : # might be len, not len-1??? actmax = max(f_tgt_beats[k],f_pred_beats[len(f_pred_beats)-1]) post_interval = max(f_tgt_beats[len(f_tgt_beats)-1],f_pred_beats[len(f_pred_beats)-1]) - f_tgt_beats[k] else : post_interval = 0.5*(f_tgt_beats[k+1]-f_tgt_beats[k]) actmax = f_tgt_beats[k] + post_interval # don't know how to do this in one line... :( a1 = f_pred_beats[where(f_pred_beats>=actmin)].copy() a1 = a1[where(a10 : for i in range(len(a1)) : newerror = a1[i] - f_tgt_beats[k] if (newerror<0) : error.append(0.5*newerror/pre_interval) else : error.append(0.5*newerror/post_interval) error = array(error) return error def evaluate_davies(tgt_beats,pred_beats,hist_bins) : if len(tgt_beats)==0 or len(pred_beats)==0 : print 'Tgt or pred empty in beat_dixon' d=dict() d['score']=0. d['hist']=ones(hist_bins) return d bins = arange(-.5,0.5+1./hist_bins,1./(hist_bins-1)) tgt_beats = tgt_beats[where(tgt_beats>3)] # remove any tgt_beats before 3 seconds in the file pred_beats = pred_beats[where(pred_beats>3)] # remove any pred_beats before 3 seconds in the annots = tgt_beats.copy() beats = pred_beats.copy() # beats compared to annotations fwd_error = davies_get_beat_error(annots,beats) fwd_error = fwd_error[where(fwd_error>=-0.5)] fwd_error = fwd_error[where(fwd_error<=0.5)] # get histogram bin heights (fwd_outbins,tmpbins) = histogram(fwd_error, bins=bins) fwd_binvals = zeros(len(fwd_outbins),float) fwd_binvals += fwd_outbins # calculate entropy fwd_binvals/=sum(fwd_binvals) fwd_binvals[where(fwd_binvals==0.0)]=1 fwd_entropy = -sum(fwd_binvals * log(fwd_binvals)) annots = pred_beats.copy() beats = tgt_beats.copy() # annotations compared to beats bwd_error = davies_get_beat_error(annots,beats) bwd_error = bwd_error[where(bwd_error>=-0.5)] bwd_error = bwd_error[where(bwd_error<=0.5)] # get histogram bin heights (bwd_outbins,tmpbins) = histogram(bwd_error, bins=bins) bwd_binvals = zeros(len(bwd_outbins),float) bwd_binvals += bwd_outbins # calculate entropy bwd_binvals/=sum(bwd_binvals) bwd_binvals[where(bwd_binvals==0.0)]=1 bwd_entropy = -sum(bwd_binvals * log(bwd_binvals)) if fwd_entropy > bwd_entropy : max_entropy = fwd_entropy hist = fwd_outbins # store un-normalised bin values to dictionary output else : max_entropy = bwd_entropy hist = bwd_outbins # now calculate linearised entropy measure score= 100.0 * (1-exp(max_entropy)/len(hist)) d=dict() d['score']=score d['hist']=hist return d def evaluate_dixon(tgt_beats,pred_beats,typ='hard') : #beat_dixon: #for each note in the reorted times, match closest real time. If within tolerance, correct. # n #---------------- #n + Fpos + Fneg #His window is 70ms on either side of beat #typ is used to set for easy or hard where easy allows for 50% out of phase; can be 'easy' or 'hard' #tgt_beats and pred_beats should be in seconds. This code (below) is written for msec tgt_beats*=1000.0 pred_beats*=1000.0 if len(tgt_beats)==0 and len(pred_beats)==0 : d=dict() d['hits']=array([],int) d['fpos']=array([],int) d['fneg']=array([],int) d['score']=0 elif len(tgt_beats)==0 : #print 'No target beats thus all predicted beats are false-positives' d=dict() d['hits']=array([],int) d['fpos']=C.deepcopy(pred_beats) d['fneg']=array([],int) d['score']=0 return d elif len(pred_beats)==0 : #print 'No predicted beats thus all target beats are false-negatives' d=dict() d['hits']=array([],int) d['fpos']=array([],int) d['fneg']=C.deepcopy(tgt_beats) d['score']=0 return d offset=70 pred_idx=1 fpos=array([]) fneg=array([]) pred_hitidxs=array([],int) tgt_hitidxs=array([],int) #calculate hits for i in range(len(tgt_beats)) : b=tgt_beats[i] #compute hit phits=where( (pred_beats>=b-offset) & (pred_beats<=b+offset) )[0] if len(phits)>0 : ptimes=pred_beats[phits]-b hitidx=argmin(abs(ptimes)) pred_hitidxs=append(pred_hitidxs,phits[hitidx]) #indexes into our predbeats which are hits tgt_hitidxs=append(tgt_hitidxs,i) #indexes into our targets which are accounted for (hit) #calculate hits for easy version (50% out) if typ=='easy' : easypred_hitidxs=array([],int) easytgt_hitidxs=array([],int) easytgt_beats=tgt_beats + gradient(tgt_beats)/2.0 for i in range(len(easytgt_beats)) : b=easytgt_beats[i] #compute hit phits=where( (pred_beats>=b-offset) & (pred_beats<=b+offset) )[0] if len(phits)>0 : ptimes=pred_beats[phits]-b hitidx=argmin(abs(ptimes)) easypred_hitidxs=append(easypred_hitidxs,phits[hitidx]) #indexes into our predbeats which are hits easytgt_hitidxs=append(easytgt_hitidxs,i) #indexes into our targets which are accounted for (hit) if len(easypred_hitidxs)>len(pred_hitidxs) : pred_hitidxs=easypred_hitidxs tgt_hitidxs=easytgt_hitidxs #now compute fpos, fneg wholly based on hits fnegidxs=setdiff1d(arange(len(tgt_beats)),tgt_hitidxs) #difference hits from targets (tgt) gives fneg fposidxs=setdiff1d(arange(len(pred_beats)),pred_hitidxs) #difference hits from predictions gives fpos hits=pred_beats[pred_hitidxs] #plot the predictions for hits (not the target beat) fneg=tgt_beats[fnegidxs] fpos=pred_beats[fposidxs] #score=len(tgt_beats)./(len(tgt_beats)+len(fpos)+len(fneg)) score=len(hits)/float(len(hits)+len(fpos)+len(fneg)) d=dict() d['hits']=C.deepcopy(hits) d['fpos']=C.deepcopy(fpos) d['fneg']=C.deepcopy(fneg) d['score']=score return d def die_with_usage() : print 'apm_evaluate.py ' print 'Flags:' print ' -eval : is eval method to use; choices are "davies" and "dixon";' print ' PENDING if flag is not present, both "dixon" and "davies" are used' print ' -davies_hist_bins : is the number of histogram bins. Default is 41' print ' -verbose : will yield verbose printing' print 'Args:' print ' : is the model to use (e.g. "apm" or "eck")' print ' : are the files to evaluate. This can include a single file,]' print ' a list of files or a directory in which all valid files are evaluated' print ' In general .mp3 or .wav filenames should be used. The trailing suffix is removed and used as a filename stub' print ' For example, if is "mysong.mp3" or "mysong.wav" then "mysong..beats" and "mysong.true.beats"' print ' will be used as prediction and target files' sys.exit(0) if __name__=='__main__' : if len(sys.argv)<2 : die_with_usage() evaluators=list(('dixon','davies')) evaltype='davies' model='' verbose=False davies_hist_bins=41 while len(sys.argv)>1 : if sys.argv[1]=='-eval': evaltype=sys.argv[2] sys.argv.pop(1) elif sys.argv[1]=='-verbose': verbose=True elif sys.argv[1]=='-davies_hist_bins': davies_hist_bins=int(sys.argv[2]) print 'Set davies_hist_bins to',davies_hist_bins sys.argv.pop(1) else : break sys.argv.pop(1) if len(sys.argv)>2 : model=sys.argv[1] sys.argv.pop(1) else : die_with_usage() if len(sys.argv)<2 : die_with_usage files = sys.argv[1:] if evaluators.count(evaltype)==0 : print 'Evaluation method',evaltype,'does not exist. Use one of:' print ' ',str(evaluators).replace('[','').replace(']','') sys.exit(0) print 'Evaluating using evaluating method "%s" on predictions from model "%s"' % (evaltype,model) print 'Evaluating files: ' if os.path.isdir(files[0]) : print 'Evaluating directory',files[0] dir=files[0] currdir = os.getcwd() os.chdir(dir) files = list() for root,dirs,fls in os.walk('.') : for f in fls: if f.lower().endswith('.mp3') or f.lower().endswith('.wav') : true_beats ='%s.%s' % (fullpath_minus_extension(os.path.join(root,f)),'true.beats') pred_beats ='%s.%s' % (fullpath_minus_extension(os.path.join(root,f)),('%s.beats' % model)) if os.path.exists(true_beats) and os.path.exists(pred_beats): f_full = os.path.abspath(os.path.join(root,f)) files.append(f_full) os.chdir(currdir) results=list() for f in files : stub = fullpath_minus_extension(f) tgt_fn = '%s.true.beats' % stub pred_fn = '%s.%s.beats' % (stub,model) tgt_beats = load_beats(tgt_fn) pred_beats = load_beats(pred_fn) # print tgt_beats # print pred_beats #print len(tgt_beats), 'target beats',tgt_beats #print len(pred_beats), 'pred beats ',pred_beats if evaltype=='dixon' : result_dict=evaluate_dixon(tgt_beats,pred_beats) results.append(result_dict) elif evaltype=='davies' : result_dict = evaluate_davies(tgt_beats=tgt_beats,pred_beats=pred_beats,hist_bins=davies_hist_bins) results.append(result_dict) if verbose : print '%2.2f %s' % (result_dict['score'],f) #now do some analysis of overall performance #all types should have a field 'acc' in the dictionary meanscore=0 for r in results : meanscore+=r.get('score',0) meanscore/=float(len(results)) print 'Mean score is', meanscore if evaltype=='all' or evaltype=='davies' : davies_hist=zeros(davies_hist_bins) for r in results : davies_hist+=r['hist'] global_score= davies_global_score(davies_hist) print 'Global score is', global_score