i'm attaching a demo of what it takes to write your own merging buffer in
python. it took me 4 hours (moderately experienced programmer new to
python). there are some tricky things to look out for, and i had the
advantage of a built-in threadsafe PriorityQueue. i have to set my own
internal latency to around 10ms in order to keep the missed timestamps
around 1-3%, and i lose whatever fancy driver/hardware optimizations apple
might have in place (did they get "active midi timing" from emagic?). if
portmidi took care of this, large classes of applications might be able to
avoid concurrency altogether, and timing-sensitive applications would be
feasible in slower dynamic/interpreted environments (and those that don't
expose threading). these issues definitely seem like perfect things for
portmidi to save users from.
-e
import pypm
import threading
import time
import bisect
import Queue
def allNotesOff(m):
for i in range(16):
for j in range(2**7-1):
m.Write([[[0x90 + i,j,0],0]])
def mean(x):
return sum(x)/len(x)
def std(x):
m=mean(x)
return mean(map(lambda y:abs(y-m),x))
class midiBuffer(threading.Thread):
def __init__(self):
self.stop=False
self.incoming=Queue.PriorityQueue() #threadsafe
self.outgoing=[]
threading.Thread.__init__(self)
def run(self):
latency=1 #latency is in ms, 0 means ignore timestamps
dev = 0
MidiOut=pypm.Output(dev, latency)
myLatency=10
t=0
class report():pass
rpt=report()
rpt.sleeps=0
rpt.latencies=[]
try:
while not self.stop:
try:
while True:
s,e=self.incoming.get_nowait()
if s<=t:
MidiOut.Write(e)
else:
#since PriorityQueue has no peek method, transfer events to a peekable format
self.outgoing.insert(bisect.bisect(map(lambda x:x[0],self.outgoing),s),[s,e])
except Queue.Empty: pass
while pypm.Time()<=t:
rpt.sleeps+=1
time.sleep(.0003) #prevent proc hog
t=pypm.Time()
while len(self.outgoing)>0 and t>=self.outgoing[0][0]-myLatency:
rpt.latencies.append(self.outgoing[0][0]-t)
MidiOut.Write(self.outgoing.pop(0)[1])
finally:
print "sleeps: " + str(rpt.sleeps)
misses=len(filter(lambda x:x<0,rpt.latencies))
print "latencies (ms, negative=late): mean=" + str(mean(rpt.latencies)) + " std=" + str(std(rpt.latencies)) + " misses=" + str(misses) + " of " + str(len(rpt.latencies)) + " (" + str(100*misses/len(rpt.latencies)) +"%)"
allNotesOff(MidiOut)
del MidiOut
def playChord(buff,notes,chan,vel,dur,onset):
ChordList = []
for i in range(len(notes)):
ChordList.append([[0x90 + chan,notes[i],vel], onset])
buff.incoming.put((onset,ChordList))
ChordList = []
offset=onset+dur*1000
for i in range(len(notes)):
ChordList.append([[0x90 + chan,notes[i],0],offset])
buff.incoming.put((offset,ChordList))
def test():
pypm.Initialize()
b=midiBuffer()
b.start()
tempo=120
dur=60.0/tempo/2
notes=[60, 63, 66, 70]
n=0
while n<100:
t=pypm.Time()
playChord(b,[notes[0]-12],1,127,dur,t)
playChord(b,notes,0,60,dur,t)
n+=1
while pypm.Time()<t+2*dur*1000: pass
b.stop=True
b.join()
pypm.Terminate()
test()
_______________________________________________
media_api mailing list
[email protected]
http://lists.create.ucsb.edu/mailman/listinfo/media_api
