annotate Syncopation models/PRS.py @ 0:76ce27beba95

Have uploaded the syncopation dataset and audio wavefies.

'''
Author: Chunyang Song
Institution: Centre for Digital Music, Queen Mary University of London

** Pressing's model **

Algorithm:

Only applicable to simple rhtyhms.

Generate hierarchical metrical structure for rhythms, they way as same as LHL model;


'''

def subdivide(sequence, segments_num):
	subSeq = []
	if len(sequence) % segments_num != 0:
		print 'Error: rhythm segment cannot be equally subdivided '
	else:
		n = len(sequence) / segments_num
		start , end = 0, n
		for i in range(segments_num):
			subSeq.append(sequence[start : end])
			start = end
			end = end + n
	
	return subSeq

# To check whether there is a need to continue subdividing and measuring
def checkContinue(sequence, division):
	isContinue = False
	if len(sequence) % division == 0:
		subs = subdivide (sequence, division)
		
		for s in subs:
			if 1 in s[1:]:	# If there are still onsets in-between the divisions
				isContinue = True
	else:
		print 'Error: the sequence cannot be equally subdivided!'
	return isContinue

def timeSpanTranscribe(sequence):
	l = len(sequence)
	transcribe = []

	if not (1 in sequence):  # Full rest
		transcribe = [0]
	else:
		divisor = 1
		while True:
			if l%divisor != 0:
				divisor = divisor + 1
			else:
				sampleStep = l/divisor # how many digits in each segment, divisor also represents the number of segments
				template = (([1] + [0]*(sampleStep-1) ) * divisor )

				sampled = []
				for i in range(l):
					sampled.append(sequence[i] and template[i])

				if sequence == sampled:
					break
				else:
					divisor = divisor + 1

		subs = subdivide(sequence, divisor)
		for s in subs:
			transcribe.append(s[0])

	return transcribe

# Identify the type of rhythm sequence : 0- null; 1-filled; 2-run; 3-upbeat; 5-syncopated
def syncType (sequence, followed_event):
	
	# Null is full rest or only one single on-beat note, therefore no 0 in sequence[1:]
	if not(1 in sequence[1:]) : 
		syncType = 0 

	else:
		ts = timeSpanTranscribe(sequence)

		# Filled is equally spaced onsets, therefore all 1s in the time span transcribe of the sequence
		if not(0 in ts ):
			syncType = 1  
		# Run is either starting with 1 and end with 0 in time span, or starting with 1 if next bar starts with 0
		elif ts[0] == 1 and ts[-1] == 0:
			syncType = 2		
		elif followed_event ==0 and ts[0] == 1:
			syncType = 2
		# Upbeat requires next bars starting with 1 and at least the last event in time span is 1
		elif followed_event == 1 and ts[-1] == 1: 
			syncType = 3
		# Syncopated start and end off-beat		
		elif sequence[0] == 0:
			syncType = 5
		else:
			print 'Error: un-recognizable syncopation type ', sequence
			syncType = None

	return syncType

def createHierarchy(rhythm, time_sig, bar):
	h = [] # A list of lists to record syncopation type(s) in each hierarchy level across bars
	s_bar = subdivide (rhythm, bar)
	
	if '4/4' in time_sig:
		 
		for i in range(bar):
			bar_level = s_bar[i]

			if i == bar -1:
				followed_event = s_bar[0][0]
			else:
				followed_event = s_bar[i+1][0]
			
			h. append ( [syncType(bar_level, followed_event)] )  # Indentify syncopation type at bar level 

		if checkContinue(rhythm, 2*bar):

			for i in range(bar):
				s_halfBar = subdivide (s_bar[i], 2)
				halfBar_h = []

				for j in range(2):
					halfBar_level = s_halfBar[j]

					if j == 1:
						followed_event = s_halfBar[0][0]
					else:
						followed_event = s_halfBar[j+1][0]

					halfBar_h.append (syncType (halfBar_level , followed_event))

				h.append(halfBar_h)

			if checkContinue(rhythm, 4*bar):
			
				for i in range(bar):
					s_quarter = subdivide (s_bar[i], 4)
					quarter_h = []

					for j in range(4):			
						quarter_level = s_quarter [j]

						if j == 3:
							followed_event = s_quarter[0][0]
						else:
							followed_event = s_quarter[j+1][0]

						quarter_h. append ( syncType (quarter_level , followed_event) )	# quarter note level

					h.append(quarter_h)

				if checkContinue( rhythm, 8*bar):
					
					for i in range(bar):
						s_eighth = subdivide (s_bar[i], 8)
						eighth_h = []

						for j in range(8):
							eighth_level = s_eighth [j]

							if j == 7:
								followed_event = eighth_level[0][0]
							else:
								followed_event = eighth_level[j+1][0]

							eighth_h.append (syncType (eighth_level, followed_event) )
					
						h.append(eighth_h)


	elif '3/4' in time_sig:	
		size_bar = len(s_bar)
		for i in range(size_bar):
			bar_level = s_bar[i]

			quarter_h = []
			eighth_h = []
			
			if i == size_bar -1:
				followed_event = s_bar[0][0]
			else:
				followed_event = s_bar[i+1][0]
			
			h. append ( [syncType(bar_level, followed_event)] )  # Indentify syncopation type at bar level 

			if checkContinue(bar_level, 3):
				s_quarter = subdivide (bar_level, 3)
				size_quarter = len(s_quarter)
				
				for j in range(size_quarter):
					quarter_level = s_quarter [j]

					if j == size_quarter -1:
						followed_event = s_quarter[0][0]
					else:
						followed_event = s_quarter[j+1][0]

					quarter_h. append ( syncType (quarter_level , followed_event) )	# quarter note level

					if checkContinue( quarter_level, 2):
						s_eighth = subdivide (quarter_level, 2)	# eighth note level
						size_eighth = len(s_eighth)

						for k in range(size_eighth):
							eighth_level = s_eighth [k]

							if k == size_eighth - 1:
								followed_event = eighth_level[0][0]
							else:
								followed_event = eighth_level[k+1][0]

							eighth_h.append (syncType (eighth_level, followed_event) )
						h.append(eighth_h)

				h.append(quarter_h)
		

	elif '6/8' in time_sig:		
		for i in range(bar):
			bar_level = s_bar[i]
			
			if i == bar -1:
				followed_event = s_bar[0][0]
			else:
				followed_event = s_bar[i+1][0]
			
			h. append ( [syncType(bar_level, followed_event)] )  # Indentify syncopation type at bar level 
			
		if checkContinue(rhythm, 2*bar):
			
			for i in range(bar):
				s_halfBar = subdivide (s_bar[i], 2)
				halfBar_h = []

				for j in range(2):
					halfBar_level = s_halfBar [j]
					
					if j == 1:
						followed_event = s_halfBar[0][0]
					else:
						followed_event = s_halfBar[j+1][0]

					halfBar_h. append ( syncType (halfBar_level , followed_event) )	

				h.append(halfBar_h)

			if checkContinue( rhythm, 6*bar):
				
				for i in range(bar):	
					s_eighth = subdivide (s_bar[i], 6)	# eighth note level
					eighth_h = []
					
					for j in range(6):
						eighth_level = s_eighth [j]

						if j == 5:
							followed_event = eighth_level[0][0]
						else:
							followed_event = eighth_level[j+1][0]

						eighth_h.append (syncType (eighth_level, followed_event) )
					
					h.append(eighth_h)

	else:
		print 'This time signature is not defined. Choose between 4/4, 3/4 or 6/8'

	return h


def pressing(rhythm, time_sig, category, bar):
	sync_oneLevel = []

	if 'poly' in category:
		return -1

	else:
		hierarchy = createHierarchy(rhythm, time_sig, bar)
		# print 'h', hierarchy

		if len(hierarchy) != 0:
			for h in hierarchy:
				sync_oneLevel.append (sum(h) / float(len(h)) )
			
			# Syncopation is the sum of averaged syncopation values of each level
			syncopation = sum (sync_oneLevel)
			
		else:
			syncopation = 0

		return syncopation 


# Retrieve the stimuli
# f = file('stimuli.txt')
f = file('stimuli_34only.txt')

#Calculate syncopation for each rhythm pattern
while True:
	line = f.readline().split(';')
	if len(line) == 1:
		 break
	else:
		sti_name = line[0]
		rhythmString = line[1].split()
		time_sig = line[2]
		category = line[3]
		bar = int([4])
		
		rhythm = map(int,rhythmString[0].split(','))

		print sti_name, pressing(rhythm, time_sig, category, bar)