comparison dml-cla/python/chord_seq_key_relative.py @ 0:718306e29690 tip

commiting public release

+#!/usr/bin/python
+# Part of DML (Digital Music Laboratory)
+# Copyright 2014-2015 Daniel Wolff, City University
+ 
+# 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"
+
+# json testfile
+#
+#{ "module":"chord_seq_key_relative",
+#      "function":"aggregate",
+#      "arguments": [[
+#      {"keys": { "tag": "csv", "value":"D:\\mirg\\Chord_Analysis20141216\\Beethoven\\qm_vamp_key_standard.n3_50ac9\\1CD0000653_BD01_vamp_qm-vamp-plugins_qm-keydetector_key.csv"},
+#          "chords": { "tag": "csv", "value":"D:\\mirg\\Chord_Analysis20141216\\Beethoven\\chordino_simple.n3_1a812\\1CD0000653_BD01_vamp_nnls-chroma_chordino_simplechord.csv"},
+#            "trackuri": "Eins"},
+#      {"keys": { "tag": "csv", "value":"D:\\mirg\\Chord_Analysis20141216\\Beethoven\\qm_vamp_key_standard.n3_50ac9\\1CD0000653_BD01_vamp_qm-vamp-plugins_qm-keydetector_key.csv"},
+#          "chords": { "tag": "csv", "value":"D:\\mirg\\Chord_Analysis20141216\\Beethoven\\chordino_simple.n3_1a812\\1CD0000653_BD01_vamp_nnls-chroma_chordino_simplechord.csv"}}
+#      ]]
+#}
+
+# these for file reading etc
+import re
+import os
+import csv
+import numpy
+
+# spmf functions
+import chord_seq_spmf_helper as spmf
+
+from aggregate import *
+from csvutils import *
+
+# ---
+# roots
+# ---
+chord_roots = ["C","D","E","F","G","A","B"] 
+
+# create a dictionary for efficiency
+roots_dic = dict(zip(chord_roots, [0,2,4,5,7,9,11]))
+
+mode_lbls = ['major','minor']
+mode_dic = dict(zip(mode_lbls, range(0,2)))
+# ---
+# types
+# ---
+type_labels = ["", "6", "7", "m","m6", "m7", "maj7", "m7b5", "dim", "dim7", "aug"]
+type_dic = dict(zip(type_labels, range(0,len(type_labels))))
+
+base_labels = ["1","","2","b3","3","4","","5","","6","b7","7"]
+#base_dic = dict(zip(base_labels, range(0,len(base_labels))))
+
+# functions
+root_funs_maj = ['I','#I','II','#II','III','IV','#IV','V','#V','VI','#VI','VII']
+root_funs_min = ['I','#I','II','III','#III','IV','#IV','V','VI','#VI','VII','#VII']
+# dan's suggestion
+#root_funs_maj = ['I','#I','II','#II','(M)III','IV','#IV','V','#V','VI','#VI','(M)VII']
+#root_funs_min = ['I','#I','II','(m)III','#III','IV','#IV','V','VI','#VI','(m)VII','#VII']
+
+fun_dic_maj = dict(zip(range(0,len(root_funs_maj)),root_funs_maj))
+fun_dic_min = dict(zip(range(0,len(root_funs_min)),root_funs_min))
+# regex that separates roots and types, and gets chord base
+# this only accepts chords with a sharp (#) and no flats
+p = re.compile(r'(?P<root>[A-G,N](#|b)*)(?P<type>[a-z,0-9]*)(/(?P<base>[A-G](#|b)*))*')
+p2 = re.compile(r'(?P<key>[A-G](#|b)*)(\s/\s[A-G](#|b)*)*\s(?P<mode>[major|minor]+)')
+pclip = re.compile(r'(?P<clipid>[A-Z,0-9]+(\-|_)[A-Z,0-9]+((\-|_)[A-Z,0-9]+)*((\-|_)[A-Z,0-9]+)*)_(?P<type>vamp.*).(?P<ext>(csv|xml|txt|n3)+)')
+
+
+
+def chords_from_csv(filename):
+    # we assume CSV: time, chord_string
+    # return (time, chord_string)
+    return csv_map_rows(filename,2, lambda row:(float(row[0]),row[1]))
+
+def keys_from_csv(filename):
+    # we assume CSV: time, key_code, key_string
+    # return ( time, key_code, key_string)
+    return csv_map_rows(filename,3, lambda row:(float(row[0]),row[1],row[2]))
+
+# parsers for n3 / csv
+key_parser_table = { 'csv':keys_from_csv }
+chord_parser_table = { 'csv':chords_from_csv }
+
+# extracts relative chord sequences from inputs of chord / key data
+# input list of pairs with instances of features:
+#    (['chords'] chordino_simple.n3_1a812 , ['keys'] qm_vamp_key_standard.n3_50ac9,
+#     optional:  ['trackuri'] trackidentifier )
+# @note: in future we could add support for qm_key_tonic input
+#
+# opts : dictionary with opts["spm_algorithm"] =  SPADE, TKS or ClaSP algorithm?
+#                       and opts["spm_options"]  = "70%"
+# output:
+# 'sequences': seq, 'support': sup
+
+trackctr = 0
+
+def aggregate(inputs,opts={}):
+    print_status('In chord_seq_key_relative')
+    
+     
+    # SPADE, TKS or ClaSP algorithm?
+    algo = opts.get("spm_algorithm","CM-SPADE")
+    
+    # number of sequences
+    maxseqs = int(opts.get("spm_maxseqs",500)/2)
+    
+    # min. length of sequences
+    minlen = int(opts.get("spm_minlen",2))
+    
+    # min. length of sequences in seconds
+    maxtime = int(opts.get("spm_maxtime",1*60)/2)
+    
+    ignoreN = int(opts.get("spm_ignore_n",1))
+    
+    # min. length of sequences
+    minsup = int(opts.get("spm_minsupport",50))
+    
+    # we now safe the mode of each piece 
+    # to treat them separately
+    out_chords = [dict(), dict()];
+    # generate dict[trackuri] = [ (time,key,mode,fun,typ,bfun) ]
+    def accum(item):
+        global trackctr
+        # increase virtual identifier
+        trackctr += 1
+        
+        # get duration and normalised frequency for all tuning pitches (A3,A4,A5)
+        keys = decode_tagged(key_parser_table,item['keys'])
+        
+        # get most frequent key
+        key,mode = most_frequent_key(keys)
+                    
+        relchords = []      
+        for (time,chord) in decode_tagged(chord_parser_table,item['chords']):
+
+            # ignore chords that are 'N':
+            # a. the open pattern matching allows for arbitrary chords 
+            #   to appear inbetween those in a sequence
+            # b. the N chord potentially maps to any contents, so the
+            #   inclusion of N chord has limited (or no) use 
+            
+            # get chord function
+            (root,fun,typ, bfun) = chord2function(chord, key,mode)
+            
+            if not (ignoreN & (root == -1)): 
+                # translate into text
+                txt = fun2txt(fun,typ, bfun, mode)
+                # print 'Chord: ' + chord + ', function: ' + txt
+                
+                # add to chords of this clip
+                relchords.append((time,key,mode,fun,typ,bfun))
+
+        # save results into dict for this track
+        trackuri = item.get('trackuri',trackctr)  
+        out_chords[mode][trackuri] = relchords
+
+    # collate relative chord information per file 
+    st=for_each(inputs,accum)
+    # print_status('Finished accumulating')
+    
+    if trackctr < 2:
+        raise Exception("Need more than 1 track")
+    
+    seq = [[],[]]
+    sup = [[],[]]
+    
+    for mode in [0,1]:
+        # write to spmf file
+        spmffile = spmf.relchords2spmf(out_chords[mode])
+        #print_status('Wrote SPMF data ' + spmffile.name)
+       
+        
+        # run sequential pattern matching
+        if algo == "TKS":
+            algoopts = opts.get("spm_options","")
+            seqfile = spmf.spmf(spmffile.name,'TKS',[str(maxseqs), algoopts])
+        elif algo == "ClaSP":
+            algoopts = opts.get("spm_options",str(minsup) + "%")
+            seqfile = spmf.spmf(spmffile.name,'ClaSP',[algoopts, str(minlen)], timeout = maxtime)
+        elif algo == "SPADE":
+            algoopts = opts.get("spm_options",str(minsup) + "%")
+            seqfile = spmf.spmf(spmffile.name,'SPADE',[algoopts, str(minlen)], timeout = maxtime)
+        else:
+            print_status('Running CM-SPADE algo')
+            algoopts = opts.get("spm_options",str(minsup) + "%")
+            seqfile = spmf.spmf(spmffile.name,'CM-SPADE',[algoopts, str(minlen)], timeout = maxtime)
+             
+        #seqfile = spmf.spmf(spmffile.name,'BIDE+',['70%'])
+        #seqfile = "D:\mirg\Chord_Analysis20141216\Beethoven_60.txt"
+        
+        #print_status('SPADE finished in ' + seqfile)
+        # parse spmf output
+        seq[mode],sup[mode] = spmf.spmf2table(seqfile)
+        
+        #clean up
+        os.remove(spmffile.name)
+        os.remove(seqfile)
+    
+    # fold back sequences and support
+    # note that this results in the sequences being truncated together below
+    seq = [item for sublist in seq for item in sublist] 
+    sup = [item for sublist in sup for item in sublist] 
+    
+    # filter according to min. sequencelength and number of sequences
+    seq_out = []
+    sup_out = []
+    seq_count = 0
+    
+    # sort in descending support and pick up sequences of sufficient length
+    for i in numpy.argsort(sup)[::-1]:
+        if len(seq[i]) >= minlen:
+            seq_out.append(seq[i])
+            sup_out.append(sup[i])
+            seq_count += 1
+            
+        if seq_count >= maxseqs:
+            break
+
+    return { 'result': { 'sequences': seq_out, 'support': sup_out}, 
+             'stats' : st }
+
+
+# most simple note2num
+def note2num(notein = 'Cb'):
+    base = roots_dic[notein[0]]
+    if len(notein) > 1:
+        if notein[1] == 'b':
+            return (base - 1) % 12
+        elif notein[1] == '#':
+            return (base + 1) % 12
+        else: 
+            print "Error parsing chord " + notein
+            raise
+    else:
+        return base % 12
+
+
+# convert key to number
+def key2num(keyin = 'C major'):
+    # ---
+    # parse key string: separate root from rest
+    # ---
+    sepstring = p2.match(keyin)
+    if not sepstring:
+        print "Error parsing key " + keyin
+        raise
+    
+    # get relative position of chord and adapt for flats
+    key = sepstring.group('key')      
+    key = note2num(key)
+    
+    # ---
+    # parse mode. care for (unknown) string
+    # ---
+    mode = sepstring.group('mode')   
+
+    if mode:
+        mode = mode_dic[mode]
+    else:
+        mode = -1
+
+    return (key, mode)
+
+    
+
+# convert chord to relative function
+def chord2function(cin = 'B',key=3, mode=0):
+    # ---
+    # parse chord string: separate root from rest
+    # ---
+    sepstring = p.match(cin)
+    
+    # test for N code -> no chord detected
+    if sepstring.group('root') == 'N':
+        return (-1,-1,-1,-1)
+    
+    # get root and type otherwise 
+    root = note2num(sepstring.group('root'))
+    type = sepstring.group('type') 
+    
+    typ = type_dic[type]
+
+    # get relative position
+    fun = (root - key) % 12
+        
+    #--- do we have a base key?
+    # if yes return it relative to chord root
+    # ---
+    if sepstring.group('base'):
+        broot = note2num(sepstring.group('base'))
+        bfun = (broot - root) % 12
+    else:
+        # this standard gives 1 as a base key if not specified otherwise
+        bfun = 0
+        
+    
+    # ---
+    # todo: integrate bfun in final type list
+    # ---
+    
+    return (root,fun,typ,bfun)    
+
+# reads in any csv and returns a list of structure
+# time(float), data1, data2 ....data2
+def read_vamp_csv(filein = ''):
+    output = []
+    with open(filein, 'rb') as csvfile:
+        contents = csv.reader(csvfile, delimiter=',', quotechar='"')
+        for row in contents:
+            output.append([float(row[0])] + row[1:])
+    return output
+
+
+
+# histogram of the last entry in a list
+# returns the most frequently used key
+def histogram(keysin = []):
+    # build histogram 
+    histo = dict()
+    for row in keysin:
+        histo[row[-1]] = histo.get(row[-1], 0) + 1 
+
+    # return most frequent key
+    return (histo, max(histo.iterkeys(), key=(lambda key: histo[key])))
+
+def most_frequent_key(keys):   
+    # delete 'unknown' keys
+    keys = [(time,knum,key) for (time,knum,key) in keys if not key == '(unknown)']
+
+    # aggregate to one key 
+    (histo, skey) = histogram(keys)
+
+    # bet key number
+    (key,mode) = key2num(skey)    
+    return key,mode
+    
+     
+    
+def fun2txt(fun,typ, bfun,mode):
+    # now we can interpret this function 
+    # when given the mode of major or minor.
+    if (fun >= 0):
+        if (mode == 1):
+            pfun = fun_dic_min[fun]
+            md = '(m)'
+        elif (mode == 0):
+            pfun = fun_dic_maj[fun] 
+            md = '(M)'
+    else:
+        return 'N'
+
+    #if typ == 'm':
+    #    print 'Key: ' + skey + ', chord: ' + chord + ' function ' + str(fun) + ' type ' + typ + ' bfun ' + str(bfun)
+    type = type_labels[typ] if typ > 0 else ''
+    
+    blb = '/' + base_labels[bfun] if (bfun >= 0 and base_labels[bfun]) else ''
+    return md + pfun + type + blb
+
+def fun2num(fun,typ, bfun,mode):
+    # now we can interpret this function 
+    if not fun == -1:
+        return (mode+1)* 1000000 + (fun+1) * 10000 + (typ+1) * 100 + (bfun+1)
+    else: 
+        return 0
+
+        
+if __name__ == "__main__":
+    #chords2functions()
+    print "Creates a key-independent chord histogram. Usage: chord2function path_vamp_chords path_vamp_keys"
+    # sys.argv[1]
+    result = folder2histogram()
+    print "Please input a description for the chord function histogram"
+    c2j.data2json(result)