comparison utils/SegProperties.py @ 0:26838b1f560f

initial commit of a segmenter project

+#!/usr/bin/env python
+# encoding: utf-8
+"""
+SegProperties.py
+
+Created by mi tian on 2015-04-02.
+Copyright (c) 2015 __MyCompanyName__. All rights reserved.
+"""
+
+import sys
+import os
+
+class FeatureGMM(object):
+	'''Represent segment candidates using single GMMs and compute pairwise distances.'''
+	def getGaussianParams(self, length, featureRate, timeWindow):
+
+		win_len = round(timeWindow * featureRate)
+		win_len = win_len + (win_len % 2) - 1
+
+		# a 50% overlap between windows
+		stepsize = ceil(win_len * 0.5)
+		num_win = int(floor( (length) / stepsize))
+		gaussian_rate = featureRate / stepsize
+
+		return stepsize, num_win, win_len, gaussian_rate
+
+	def GaussianDistance(self, feature, featureRate, timeWindow):
+
+		stepsize, num_win, win_len, gr = self.getGaussianParams(feature.shape[0], featureRate, timeWindow)
+		print 'stepsize, num_win, feature', stepsize, num_win, feature.shape, featureRate, timeWindow
+		gaussian_list = []
+		gaussian_timestamps = []
+		tsi = 0
+
+		# f = open('/Users/mitian/Documents/experiments/features.txt','w')
+		# print 'divergence computing..'
+		for num in xrange(num_win): 
+			# print num, num * stepsize , (num * stepsize) + win_len
+			gf=GaussianFeature(feature[int(num * stepsize) : int((num * stepsize) + win_len), :],2)
+			# f.write("\n%s" %str(gf))
+			gaussian_list.append(gf)
+			tsi = int(floor( num * stepsize + 1)) 
+			gaussian_timestamps.append(self.timestamp[tsi])
+
+		# f.close()
+
+		# print 'gaussian_list', len(gaussian_list), len(gaussian_timestamps)
+		dm = np.zeros((len(gaussian_list), len(gaussian_list)))
+
+		for v1, v2 in combinations(gaussian_list, 2): 
+			i, j = gaussian_list.index(v1), gaussian_list.index(v2)
+			dm[i, j] = v1.distance(v2)
+			dm[j, i] = v2.distance(v1)
+			# print 'dm[i,j]',dm[i,j]
+		# sio.savemat("/Users/mitian/Documents/experiments/dm-from-segmenter.mat",{"dm":dm})
+		return dm, gaussian_timestamps
+
+	def getGMMs(self, feature, segment_boundaries):
+		'''Return GMMs for located segments'''
+		gmm_list = []
+		gmm_list.append(GmmDistance(feature[: segment_boundaries[0], :], components = 1))
+		for i in xrange(1, len(segment_boundaries)):
+			gmm_list.append(GmmDistance(feature[segment_boundaries[i-1] : segment_boundaries[i], :], components = 1))
+		return gmm_list
+
+
+class FusedPeakSelection(object):
+	'''Peak selection from fusion of individual results.'''  
+	def getFusedPeaks(self, combined_thresh, individual_thresh, individual_tol, combined_tol, w1=None, w2=None, w3=None, w4=None):
+		'''Return a list a peak position and the corresponding confidence.'''
+		confidence_array = np.zeros_like(w1)
+		conf1 = np.zeros_like(w1)
+		len_arr = len(w1)
+		
+		# keep peaks retrieved by single feature if its confidence is above individual_thresh
+		w1_keep = np.where(w1>=individual_thresh)[0]
+		w2_keep = np.where(w2>=individual_thresh)[0]
+		w3_keep = np.where(w3>=individual_thresh)[0]
+		w4_keep = np.where(w4>=individual_thresh)[0]
+		confidence_array[w1_keep] += w1[w1_keep]
+		confidence_array[w2_keep] += w2[w2_keep]
+		confidence_array[w3_keep] += w3[w3_keep]
+		confidence_array[w4_keep] += w4[w4_keep]
+		
+		confidence_array[confidence_array>1] = 1
+		
+		# deal with peaks picked individual features with high confidence first
+		i=0
+		while i < len_arr:
+			if confidence_array[i] > 0:
+				temp = [confidence_array[i]]
+				pos = [i]
+				i += 1
+		
+				# start searching neighborhood for local maximum
+				while (i+individual_tol < len_arr and np.max(confidence_array[i:i+individual_tol]) > 0):
+					temp += [confidence_array[i+delta] for delta in xrange(individual_tol) if confidence_array[i+delta]>0]
+					pos += [i+delta for delta in xrange(individual_tol) if confidence_array[i+delta]>0]
+					i += individual_tol
+					
+				if len(temp) == 1:
+					conf1[pos[0]] = temp[0]
+				else:
+					# p = int(np.rint(np.sum(np.multiply(pos,temp))/ np.sum(temp)))
+					# conf1[p] = 1
+					p = int(np.mean(pos))
+					conf1[p] = np.mean(temp)
+			else:	
+				i += 1
+		conf1[conf1>1] = 1
+		
+		# Process peaks with low confidence but located by multiple features in the same neighborhood
+		# conf2 = copy(conf1)
+		conf2 = np.zeros_like(conf1)
+		weight1, weight2, weight3, weight4 = copy(w1), copy(w2), copy(w3), copy(w4)
+		weight1[weight1>individual_thresh] = 0.0
+		weight2[weight2>individual_thresh] = 0.0
+		weight3[weight3>individual_thresh] = 0.0
+		weight4[weight4>individual_thresh] = 0.0
+		combined = weight1 + weight2 + weight3 + weight4
+		combined = (combined - np.min(combined)) / (np.max(combined) - np.min(combined))
+		if combined[0]>0.3: combined[0] = 0.8
+		
+		i = 0	
+		while i < len_arr:
+			if combined[i] > 0:
+				temp = [combined[i]]
+				pos = [i]
+				i += 1
+				
+				# start searching neighborhood for local maximum
+				while (i+combined_tol < len_arr and np.max(combined[i:i+combined_tol]) > 0):
+					temp += [combined[i+delta] for delta in xrange(combined_tol) if combined[i+delta]>0]
+					pos += [i+delta for delta in xrange(combined_tol) if combined[i+delta]>0]
+					i += combined_tol
+					
+				if len(temp) == 1:
+					conf2[pos[0]] += temp[0]
+				else:
+					p = int(np.rint(np.sum(np.multiply(pos,temp))/ np.sum(temp)))
+					conf2[p] += np.sum(np.multiply(pos,temp)) / p 
+			else:	
+				i += 1
+		
+		conf2[conf2<combined_thresh] = 0
+		conf2[conf2>1] = 1
+		
+		combined_conf = conf1 + conf2	
+		combined_conf[combined_conf>1] = 1
+		conf = np.zeros_like(combined_conf)
+		# Combine selections from the obove two steps	
+		i=0
+		while i < len_arr:
+			if combined_conf[i] > 0.3:
+				temp = [combined_conf[i]]
+				pos = [i]
+				i += 1
+		
+				# start searching neighborhood for local maximum
+				while (i+individual_tol < len_arr and np.max(combined_conf[i:i+individual_tol]) > 0.5):
+					temp += [combined_conf[i+delta] for delta in xrange(individual_tol) if combined_conf[i+delta]>0.5]
+					pos += [i+delta for delta in xrange(individual_tol) if combined_conf[i+delta]>0.5]
+					i += individual_tol
+					
+				if len(temp) == 1:
+					conf[pos[0]] = combined_conf[pos[0]]
+				elif (np.max(temp)== 1 and np.sort(temp)[-2] < combined_thresh):
+					p = pos[np.argmax(temp)]
+					conf[p] = np.max(temp)
+				else:
+					p = int(np.rint(np.sum(np.multiply(pos,temp))/ np.sum(temp)))
+					conf[p] = np.mean(np.multiply(pos,temp)) / p 
+			else:	
+				i += 1
+
+		peaks = list(np.where(conf>combined_thresh)[0])
+		return peaks, conf1, conf2, conf
+		
+	def getPeakWeights(self, sdf, peak_list):
+		'''Compute peak confidence.
+		Return: array with confidence values at peak positions and zeros otherwise'''
+		mask = np.zeros_like(sdf)
+		mask[peak_list] = 1.0
+		return sdf * mask
+		
+	def selectPeak(self, peak_candidates, featureset, winlen=5):
+		dist_list = []
+		feature_types = len(featureset)
+		gt_dist, hm_dist, tb_dist, tp_dist = [], [], [], []
+		
+		for idx, x in enumerate(peak_candidates):
+			prev_features = tuple([featureset[i][x-winlen:x, :] for i in xrange(feature_types)])
+			post_features = tuple([featureset[i][x:x+winlen, :] for i in xrange(feature_types)])
+			gt_dist.append(np.sum(pairwise_distances(prev_features[0], post_features[0])))
+			hm_dist.append(np.sum(pairwise_distances(prev_features[1], post_features[1])))
+			tb_dist.append(np.sum(pairwise_distances(prev_features[2], post_features[2])))
+			tp_dist.append(np.sum(pairwise_distances(prev_features[3], post_features[3])))
+		
+		return peak_candidates[np.argmax(gt_dist)], peak_candidates[np.argmax(hm_dist)], peak_candidates[np.argmax(tb_dist)], peak_candidates[np.argmax(tp_dist)]
+		
+	def getPeakFeatures(self, peak_candidates, featureset, winlen):
+		'''
+		args: winlen: length of feature window before and after an investigated peak 
+			 featureset: A list of audio features for measuring the dissimilarity.
+		
+		return: peak_features
+				A list of tuples of features for windows before and after each peak.
+		'''
+		prev_features = []
+		post_features = []
+		feature_types = len(featureset)
+
+		# print peak_candidates[-1], winlen, featureset[0].shape
+		# if peak_candidates[-1] + winlen > featureset[0].shape[0]:
+		#	peak_candidates = peak_candidates[:-1]
+		# for x in peak_candidates:
+		#	prev_features.append(tuple([featureset[i][x-winlen:x, :] for i in xrange(feature_types)]))
+		#	post_features.append(tuple([featureset[i][x:x+winlen, :] for i in xrange(feature_types)]))
+		prev_features.append(tuple([featureset[i][:peak_candidates[0], :] for i in xrange(feature_types)]))
+		post_features.append(tuple([featureset[i][peak_candidates[0]:peak_candidates[1], :] for i in xrange(feature_types)]))
+		for idx in xrange(1, len(peak_candidates)-1):
+			prev_features.append(tuple([featureset[i][peak_candidates[idx-1]:peak_candidates[idx], :] for i in xrange(feature_types)]))
+			post_features.append(tuple([featureset[i][peak_candidates[idx]:peak_candidates[idx+1], :] for i in xrange(feature_types)]))
+		prev_features.append(tuple([featureset[i][peak_candidates[-2]:peak_candidates[-1], :] for i in xrange(feature_types)]))
+		post_features.append(tuple([featureset[i][peak_candidates[-1]:, :] for i in xrange(feature_types)]))
+		return prev_features, post_features
+	
+	def segStats(self, feature_array, boundary_list):
+		'''Return some basic stats of features associated with two boundaries.'''
+		feature_stats = []
+		for i in xrange(1, len(boundary_list)):
+			feature_stats.append(np.std(feature_array[boundary_list[i-1]:boundary_list[i]], axis=0))
+		return feature_stats
+			
+	def segmentDev(self, prev_features, post_features, metric='kl'):
+		'''Deviations are measured for each given feature type. 
+		peak_candidates: peaks from the 1st round detection
+		peak_features: Features for measuring the dissimilarity for parts before and after each peak.
+					dtype: tuple. 
+		'''
+		dev_list = []
+		n_peaks = len(prev_features)
+		n_features = len(prev_features[0])
+		# print 'n_peaks, n_features', n_peaks, n_features
+		if metric == 'kl':
+			for x in xrange(n_peaks):
+				f1, f2 = prev_features[x], post_features[x]
+				dev_list.append(tuple([GmmDistance(f1[i], components=1).skl_distance_full(GmmDistance(f2[i], components=1)) for i in xrange(n_features)]))
+		elif metric == 'euclidean':
+			for x in xrange(n_peaks):
+				f1, f2 = prev_features[x], post_features[x]
+				dev_list.append(tuple([pairwise_distances(f1[i], f2[i]) for i in xrange(n_features)]))
+		return dev_list
+	
+def main():
+	pass
+
+
+if __name__ == '__main__':
+	main()
+