annotate dml-cla/python/tuning_stats.py @ 0:718306e29690 tip

commiting public release

# Part of DML (Digital Music Laboratory)
# Copyright 2014-2015 Daniel Wolff, City University; Steven Hargreaves; Samer Abdallah, University of London
 
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
# 
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
# 
# You should have received a copy of the GNU General Public
# License along with this library; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA

# -*- coding: utf-8 -*-
__author__='wolffd, hargreavess, abdallahs'

# this script derives standard statistics for tuning frequency, 
# in particular:
# average
# standard deviation
# histogram

from rdflib import RDF, RDFS
from csvutils import *
from aggregate import *
from n3Parser import get_rdf_graph_from_n3 
import numpy

def transcription_from_csv(filename):
    # we assume CSV: time, duration, pitch, velocity, note_name 
    # return (time, duration, pitch, note_name)
    return csv_map_rows(filename,5, lambda row:(float(row[0]),float(row[1]),float(row[2]),row[4]))

def transcription_from_n3(filename):
    graph=get_rdf_graph_from_n3(filename)
    notes = [ ( event_time_prop(graph, ev, tl_ns.beginsAt), 
                event_time_prop(graph, ev, tl_ns.duration),
                graph.value(ev,af_ns.feature).split(' ')[0],
                graph.value(ev,RDFS.label) )
              for ev in subject((RDF.type, af_ns.Note)) ]

parser_table = { 'n3':transcription_from_n3, 
                 'csv':transcription_from_csv }

offset = { 'D':7, 'E':5, 'A':0 }

def fold_pitch(freq,name): 
    # semis =(4+offset[name[0]]-int(name[1]))
    # print_status("folding by %d" % semis)
    return freq * 2**(4+offset[name[0]]/12.0-int(name[1]))
def tuning_note(n):
    return n[1] in ['3','4','5'] and n[0] in ['A','E','D']


def per_file(inputs):
    means = []
    hists = []
    hist_edges =[]

    def accum(item):
        # get duration and normalised frequency for all tuning pitches (A3,A4,A5)
        a_notes = [ (note[1],fold_pitch(note[2],note[3])) 
                    for note in decode_tagged(parser_table,item) 
                    if tuning_note(note[3]) ]

        if len(a_notes)==0:
            print_status("No notes for "+str(item))
        else:
            # get frequency and duration columns
            freq = numpy_column(a_notes,1)
            dur = numpy_column(a_notes,0)
            # get mean values per clip now,
            # then statistics over clips later
            avg, std = weighted_stats(freq, weights = dur)
            (counts, edges) = histogram(freq, 100, 390, 490, weights=dur)
            
            means.append(avg) 
            hists.append(counts)
            if len(hist_edges) == 0: 
                hist_edges.extend(edges)
            
    st=for_each(inputs,accum)

    avg, std = stats(numpy.array(means,dtype=float))

    # !!! does this make any sense?
    hist_mean, hist_std = stats(numpy.array(hists,dtype=float))

    return { 'result': { 'mean': avg, 'std-dev': std, 
                         'hist': continuous_hist(hist_edges,hist_mean) }, 
             'stats' : st }


def aggregate(inputs):
    notes = [] # will contain all notes in all inputs
    def accum(item):
        # get duration and normalised frequency for all tuning pitches (A3,A4,A5)
        # and collect them all in notes
        notes.extend( [ (note[1],fold_pitch(note[2],note[3])) 
                        for note in decode_tagged(parser_table,item) 
                        if tuning_note(note[3]) ] )
    
    # execute accumulation for each accum                    
    stats=for_each(inputs,accum)
            
    # get frequency and duration columns
    dur = numpy_column(notes,0)
    freq = numpy_column(notes,1)

    # get basic statistics
    avg, std = weighted_stats(freq, weights=dur)

    # get histogram weighted by duration
    counts, edges = histogram(freq, 100, 390, 490, weights=dur)
        
    return { 'result': { 'mean': avg, 'std_dev': std, 
                         'hist': continuous_hist(edges,counts) },
             'stats' : stats }

# convert one column, specified by datapos, to numpy array
def numpy_column(data,datapos):
    return numpy.array([ row[datapos] for row in data ], dtype=float)

#calculates the histogram
# nbins: number of bins
# lb: lower bound
# ub: upper bound
def histogram(colu, nbins, lb, ub, weights = []):
    counts,edges = numpy.histogram(colu, bins=nbins, range=[lb, ub], weights=weights)
    counts = counts / numpy.sum(counts)
    
    return (counts.tolist(), edges.tolist())

# calculates unweighted statistics for the histograms
def stats(counts):
    avg = numpy.average(counts, axis = 0).tolist()
    std = numpy.std(counts, axis =0)
    return (avg,std)
    
#calculates weighted statistics for  numerical input
def weighted_stats(colu, weights = []):
    avg = numpy.average(colu, axis = 0 ,weights = weights)
    #weighted standard deviation
    std = numpy.sqrt(numpy.average((colu-avg)**2, axis = 0, weights=weights))
    #std = numpy.std(colu, weights = weights).tolist()
    #med = numpy.median(colu, weights = weights).tolist()
    # could use https://pypi.python.org/pypi/wquantiles for weighted median
    return (avg,std)