mi@0: from vampy import *
mi@0: from numpy import zeros,float64, float32, array
mi@0: import sys
mi@0: from scipy.signal import butter
mi@0: import numpy as np
mi@0: 
mi@0: ''' 
mi@0: Define a common base class where we define the methods common to plugins without 
mi@0: a fusion process involved. 
mi@0: The base class implements adaptive whitening and onset location backtracking.
mi@0: This also makes individual plugins easier to change / manage / overview.
mi@0: '''
mi@0: 
mi@0: class OnsetDetectBase(object):
mi@0: 	# WARNING: Apparently vampy doesn't handle errors in super classes (bases) as gracefully as they are handeled
mi@0: 	# in the single class scenario with no inheritnace. This is to be fixed later in vampy itself.
mi@0: 	# For now syntax errors, missing imports, etc... are likely to cause segfault, without printing a detailed python traceback.
mi@0: 	# However, the source of the error is still printed in debug mode, so at least we know which function to fix.
mi@0: 	
mi@0: 	def __init__(self,inputSampleRate):
mi@0: 		self.vampy_flags = vf_DEBUG
mi@0: 
mi@0: 		# basic common parameters
mi@0: 		self.preferredStepSecs = 0.01161
mi@0: 		# self.preferredStepSecs = 0.02322
mi@0: 		self.inputSampleRate = inputSampleRate
mi@0: 		self.stepSize = 0
mi@0: 		self.blockSize = 0
mi@0: 		self.channels = 0
mi@0: 		
mi@0: 		# user configurable parameters
mi@0: 		self.threshold = 50
mi@0: 		self.delta_threshold = 0.0
mi@0: 		self.backtracking_threshold = 1.9
mi@0: 		self.polyfitting_on = True
mi@0: 		self.medfilter_on = True
mi@0: 		self.LPfilter_on = True
mi@0: 		self.whitening_on = False
mi@0: 		self.simplePick = False
mi@0: 		
mi@0: 		# whitening
mi@0: 		self.whitenRelaxCoeff = 0.9997
mi@0: 		self.whitenFloor = 0.01
mi@0: 		self.magPeaks = None
mi@0: 		self.medianWin = 7
mi@0: 		self.aCoeffs = [1.0000, -0.5949, 0.2348]
mi@0: 		self.bCoeffs = [0.1600,  0.3200, 0.1600]
mi@0: 		self.cutoff = 0.34
mi@0: 		
mi@0: 		
mi@0: 	def initialise(self,channels,stepSize,blockSize):
mi@0: 		self.channels = channels
mi@0: 		self.stepSize = stepSize		
mi@0: 		self.blockSize = blockSize
mi@0: 		self.half_length = self.blockSize * 0.5 + 1.0
mi@0: 		self.magPeaks = zeros(self.half_length, dtype = float64)
mi@0: 		return True
mi@0: 		
mi@0: 	def reset(self):
mi@0: 		self.magPeaks = None
mi@0: 		return None
mi@0: 	
mi@0: 	def getMaker(self):
mi@0: 		return 'Mi Tian, Testing'
mi@0: 
mi@0: 	def getIdentifier(self):
mi@0: 		return 'vampy-base'		
mi@0: 		
mi@0: 	def getPreferredBlockSize(self):
mi@0: 		'''Preferred window size is twice the preferred step size'''
mi@0: 		# return 2048
mi@0: 		return int(self.getPreferredStepSize() * 2)
mi@0: 		
mi@0: 	def getPreferredStepSize(self):
mi@0: 		'''Preferred block size is set to 256 in the QM Vamp plugin in case SR is 22.5kHz'''
mi@0: 		step = int(self.inputSampleRate * self.preferredStepSecs + 0.0001)
mi@0: 		if step < 1 : return 1
mi@0: 		return step
mi@0: 		# return 1024
mi@0: 		
mi@0: 	def getMaxChannelCount(self):
mi@0: 		return 1
mi@0: 		
mi@0: 	def getInputDomain(self):
mi@0: 		return FrequencyDomain 
mi@0: 		
mi@0: 	def getParameterDescriptors(self):
mi@0: 		'''Define all common parameters of the plugins.'''
mi@0: 		
mi@0: 		threshold = ParameterDescriptor()
mi@0: 		threshold.identifier ='threshold'
mi@0: 		threshold.name ='Detection Sensitivity'
mi@0: 		threshold.description = 'Detection Sensitivity'
mi@0: 		threshold.unit = '%'
mi@0: 		threshold.minValue = 0
mi@0: 		threshold.maxValue = 100
mi@0: 		threshold.defaultValue = 50
mi@0: 		threshold.isQuantized = False
mi@0: 		
mi@0: 		delta_thd = ParameterDescriptor()
mi@0: 		delta_thd.identifier ='dthreshold'
mi@0: 		delta_thd.name ='Delta Threshold'
mi@0: 		delta_thd.description = 'Delta threshold used for adaptive theresholding using the median of the detection function'
mi@0: 		delta_thd.unit = ''
mi@0: 		delta_thd.minValue = -1.0
mi@0: 		delta_thd.maxValue = 1.0
mi@0: 		delta_thd.defaultValue = 0.0
mi@0: 		delta_thd.isQuantized = False
mi@0: 		
mi@0: 		# NOTE: GF: Not sure this should really be called a threshold. 'Tolerance' may be better.
mi@0: 		bt_thd = ParameterDescriptor()
mi@0: 		bt_thd.identifier ='bt-threshold'
mi@0: 		bt_thd.name ='Backtracking Threshold'
mi@0: 		bt_thd.description = 'Backtracking threshold used determine the stopping condition for backtracking the onset location'
mi@0: 		bt_thd.unit = ''
mi@0: 		bt_thd.minValue = -1.0
mi@0: 		bt_thd.maxValue = 3.0
mi@0: 		bt_thd.defaultValue = 1.9
mi@0: 		bt_thd.isQuantized = False
mi@0: 			
mi@0: 		cutoff = ParameterDescriptor()
mi@0: 		cutoff.identifier ='cut-off'
mi@0: 		cutoff.name ='cut off value'
mi@0: 		cutoff.description = 'low pass filter cut off value'
mi@0: 		cutoff.unit = ''
mi@0: 		cutoff.minValue = 0.1
mi@0: 		cutoff.maxValue = 0.6
mi@0: 		cutoff.defaultValue = 0.34
mi@0: 		cutoff.isQuantized = False
mi@0: 		
mi@0: 		med_thd = ParameterDescriptor()
mi@0: 		med_thd.identifier ='med-threshold'
mi@0: 		med_thd.name ='Median filter window'
mi@0: 		med_thd.description = 'Median filter window size'
mi@0: 		med_thd.unit = ''
mi@0: 		med_thd.minValue = 3.0
mi@0: 		med_thd.maxValue = 12.0
mi@0: 		med_thd.defaultValue = 7.0
mi@0: 		med_thd.isQuantized = True
mi@0: 		med_thd.quantizeStep = 1
mi@0: 		
mi@0: 		# save some typing by defining a descriptor type
mi@0: 		boolDescriptor = ParameterDescriptor()
mi@0: 		boolDescriptor.isQuantized = True
mi@0: 		boolDescriptor.minValue= 0
mi@0: 		boolDescriptor.maxValue= 1
mi@0: 		boolDescriptor.quantizeStep = 1
mi@0: 		
mi@0: 		polyfit = ParameterDescriptor(boolDescriptor)
mi@0: 		polyfit.identifier='polyfit'
mi@0: 		polyfit.name='polynomial fitting'
mi@0: 		polyfit.description='Use polynomial fitting to evaluate detection function peaks.'
mi@0: 		
mi@0: 		medfilt = ParameterDescriptor(boolDescriptor)
mi@0: 		medfilt.identifier='medfilt'
mi@0: 		medfilt.name='median filtering'
mi@0: 		medfilt.description='Use median filtering'
mi@0: 		
mi@0: 		filtfilt = ParameterDescriptor(boolDescriptor)
mi@0: 		filtfilt.identifier='filtfilt'
mi@0: 		filtfilt.name='low-pass filtering'
mi@0: 		filtfilt.description='Use zero-phase foward-backward low-pass filtering'
mi@0: 		
mi@0: 		whitening = ParameterDescriptor(boolDescriptor) 
mi@0: 		whitening.identifier='whitening'
mi@0: 		whitening.name='Adaptive whitening'
mi@0: 		whitening.description='Turn adaptive whitening on or off'
mi@0: 		whitening.defaultValue = False
mi@0: 		
mi@0: 		return ParameterList(threshold, delta_thd, bt_thd, cutoff, med_thd, whitening, polyfit, medfilt, filtfilt)
mi@0: 		
mi@0: 
mi@0: 	def setParameter(self,paramid,newval):	
mi@0: 		if paramid == 'threshold' :
mi@0: 			self.threshold = newval
mi@0: 			print >> sys.stderr, "sensitivity threshold: ", newval
mi@0: 		if paramid == 'dthreshold' :
mi@0: 			self.delta_threshold = newval
mi@0: 			print >> sys.stderr, "delta threshold: ", newval
mi@0: 		if paramid == 'bt-threshold' :
mi@0: 			self.backtracking_threshold = newval
mi@0: 			print >> sys.stderr, "backtracking threshold: ", newval
mi@0: 		if paramid == 'cut-off' :
mi@0: 			self.cutoff = newval
mi@0: 			self.bCoeffs, self.aCoeffs = butter(2, self.cutoff)
mi@0: 			print >> sys.stderr, "low pass filter cut off value: ", newval
mi@0: 		if paramid == 'med-threshold' :
mi@0: 			self.medianWin = newval
mi@0: 			print >> sys.stderr, "meidan filter windown: ", newval
mi@0: 		if paramid == 'medfilt' :
mi@0: 			self.medfilter_on = newval == 1.0
mi@0: 			print >> sys.stderr, "median filering: ", self.medfilter_on, newval
mi@0: 		if paramid == 'filtfilt' :
mi@0: 			self.LPfilter_on = newval == 1.0
mi@0: 			print >> sys.stderr, "foward-backward filering: ", self.LPfilter_on, newval		
mi@0: 		if paramid == 'polyfit' :
mi@0: 			self.polyfitting_on = newval == 1.0
mi@0: 			print >> sys.stderr, "polynomial fitting: ", self.polyfitting_on, newval
mi@0: 		if paramid == 'whitening' :
mi@0: 			self.whitening_on = newval == 1.0
mi@0: 			print >> sys.stderr, "whitening: ", self.whitening_on, newval
mi@0: 		
mi@0: 		return None
mi@0: 
mi@0: 		
mi@0: 	def getParameter(self,paramid):
mi@0: 		if paramid == 'whitening' :
mi@0: 			if self.whitening_on :
mi@0: 				return 1.0
mi@0: 			else :
mi@0: 				return 0.0
mi@0: 		if paramid == 'medfilt' :
mi@0: 			if self.medfilter_on :
mi@0: 				return 1.0
mi@0: 			else :
mi@0: 				return 0.0
mi@0: 		if paramid == 'filtfilt' :
mi@0: 			if self.LPfilter_on :
mi@0: 				return 1.0
mi@0: 			else :
mi@0: 				return 0.0		
mi@0: 		if paramid == 'polyfit' :
mi@0: 			if self.polyfitting_on :
mi@0: 				return 1.0
mi@0: 			else :
mi@0: 				return 0.0
mi@0: 		if paramid == 'threshold' :
mi@0: 			return self.threshold
mi@0: 		if paramid == 'dthreshold' :
mi@0: 			return self.delta_threshold
mi@0: 		if paramid == 'bt-threshold' :
mi@0: 			return self.backtracking_threshold
mi@0: 		# if paramid == 'tol-threshold' :
mi@0: 		if paramid == 'med-threshold' :
mi@0: 			return self.medianWin	
mi@0: 		if paramid == 'cut-off' :
mi@0: 			return self.cutoff	
mi@0: 		return 0.0
mi@0: 		
mi@0: 			
mi@0: 			
mi@0: 	def getGenericOutputDescriptors(self):
mi@0: 		'''Define 3 outputs ike in the QM plugin. First is the raw detecion function, second is the smoothed one,
mi@0: 		third is the actual note onset outputs. Note: in QM-Vamp the onsets are the first output.'''
mi@0: 		# We call this getGenericOutputDescriptors as we don't want the base to have real outputs. 
mi@0: 		# Identifiers shoudl be defined in the sub-classes therefore they are ommitted here.
mi@0: 
mi@0: 		DF_Descriptor = OutputDescriptor()
mi@0: 		DF_Descriptor.hasFixedBinCount=True
mi@0: 		DF_Descriptor.binCount=1
mi@0: 		DF_Descriptor.hasKnownExtents=False
mi@0: 		DF_Descriptor.isQuantized=False
mi@0: 		DF_Descriptor.sampleType = OneSamplePerStep
mi@0: 		DF_Descriptor.unit = ''		
mi@0: 		DF_Descriptor.name = 'Onset Detection Function'
mi@0: 		DF_Descriptor.description ='Onset Detection Function'
mi@0: 
mi@0: 		# NOTE: Just change what we need, all oter parameters are inherited from DF_Descriptor
mi@0: 		SDF_Descriptor = OutputDescriptor(DF_Descriptor)
mi@0: 		SDF_Descriptor.name = 'Smoothed Onset Detection Function'
mi@0: 		SDF_Descriptor.description ='Smoothed Onset Detection Function'
mi@0: 		SDF_Descriptor.sampleType = VariableSampleRate
mi@0: 		SDF_Descriptor.sampleRate = 1.0 / self.preferredStepSecs
mi@0: 		
mi@0: 		Onset_Descriptor = OutputDescriptor()
mi@0: 		Onset_Descriptor.name = 'Onsets'
mi@0: 		Onset_Descriptor.description ='Onsets using spectral difference'
mi@0: 		Onset_Descriptor.hasFixedBinCount=True
mi@0: 		Onset_Descriptor.binCount=0
mi@0: 		Onset_Descriptor.hasKnownExtents=False
mi@0: 		Onset_Descriptor.isQuantized=False
mi@0: 		Onset_Descriptor.sampleType = VariableSampleRate
mi@0: 		Onset_Descriptor.unit = ''
mi@0: 				
mi@0: 		return DF_Descriptor, SDF_Descriptor, Onset_Descriptor
mi@0: 
mi@0: 
mi@0: 	def backtrack(self, onset_index, smoothed_df):
mi@0: 		'''Backtrack the onsets to an earlier 'perceived' location from the actually detected peak...
mi@0: 		This is based on the rationale that the perceived onset tends to be a few frames before the detected peak.
mi@0: 		This tracks the position in the detection function back to where the peak is startng to build up.
mi@0: 		Notice the "out of the blue" parameter: 0.9. (Ideally, this should be tested, evaluated and reported...)'''
mi@0: 		prevDiff = 0.0
mi@0: 		while (onset_index > 1) :
mi@0: 			diff = smoothed_df[onset_index] - smoothed_df[onset_index-1]
mi@0: 			if diff < prevDiff * self.backtracking_threshold : break
mi@0: 			prevDiff = diff
mi@0: 			onset_index -= 1
mi@0: 		return onset_index
mi@0: 		
mi@0: 	def trackDF(self, onset1_index, df2):
mi@0: 		'''In the second round of detection, remove the known onsets from the DF by tracking from the peak given by the first round
mi@0: 		to a valley to deminish the recognised peaks on top of which to start new detection.'''	
mi@0: 		for idx in xrange(len(onset1_index)) :
mi@0: 			remove = True
mi@0: 			for i in xrange(onset1_index[idx], 1, -1) :
mi@0: 				if remove :
mi@0: 					if df2[i] >= df2[i-1] :	
mi@0: 						df2[i] == 0.0
mi@0: 					else:
mi@0: 						remove = False
mi@0: 		return df2	
mi@0: 		
mi@0: 	def whiten(self,magnitudeSpectrum):
mi@0: 		'''This function reproduces adaptive whitening as described in Dan Stowell's paper.'''
mi@0: 		if self.magPeaks is None :
mi@0: 			self.magPeaks = zeros(self.half_length, dtype = float32)
mi@0: 		m = array(magnitudeSpectrum, dtype=float32)
mi@0: 		idx = m < self.magPeaks
mi@0: 		# print " m", m[idx]
mi@0: 		
mi@0: 		m[idx] += (self.magPeaks[idx] - m[idx]) * self.whitenRelaxCoeff
mi@0: 		m[m < self.whitenFloor] = self.whitenFloor
mi@0: 		self.magPeaks = m
mi@0: 
mi@0: 		magnitudeSpectrum /= m
mi@0: 		
mi@0: 		return magnitudeSpectrum
mi@0: 	
mi@0: 	
mi@0: 
mi@0: