All Samples(1) | Call(0) | Derive(0) | Import(1)
src/m/t/mt-midi-sequencer-HEAD/src/core/audio/EventManager.py mt-midi-sequencer(Download)
import EventWorker #import OutputAdapter import core.audio.Sequence as Sequence import core.audio.Output as Output import time import threading import Queue import core.Constants as Constants import sys import pygame import pygame.midi from pygame.locals import * ''' Singleton Implementation Meta-Info: If ever someone creates a module named EventManagerInstance, there will be problems! ''' def getInstance(): if ('EventManagerInstance' not in sys.modules): raise Exception('You must start EventManager first in order to get a running instance.') return sys.modules['EventManagerInstance'] ''' - Class: EventManager - Description: This class is the worker's manager. Here we can register new objects with callback functions while the worker is busy working ''' class EventManager(threading.Thread): def __init__(self, **kwargs): kwargs.setdefault('logging', True) self.__logging = kwargs.get('logging') self.__log('initializing EventManager') # thread safe queue for music data self.playDataQueue=Queue.Queue(maxsize=Constants.EventManager_playDataQueue_maxsize) # mandatory for threading threading.Thread.__init__(self) # setup worker self.worker=EventWorker.EventWorker(manager=self, logging=self.__logging) self.worker.setDaemon(True) # pygame midi output self.midi_out = kwargs.get('midi_out') if self.midi_out==None: raise Exception('EventManager must be instantiated with an associated pygame midi Output!') # setup output self.output=Output.Output(manager=self, midi_out=self.midi_out) self.output.setDaemon(True) self.__callbacks={} # dictionary for callback methods <obj>:<callback> self.__sequenceCounter = -1 sys.modules['EventManagerInstance']=self self.start() ''' register an object ''' def register(self, object, method): self.__log('registering callback method for ' + str(object.id)) self.__callbacks[object]=method # tell worker to get new registered method: self.worker.refreshConfig.set() ''' unregister an object ''' def unregister(self, object): self.__log('unregistering callback method for ' + str(object.id)) del self.__callbacks[object] # tell worker to get new registered method: self.worker.refreshConfig.set() ''' for EventWorker to fetch it ''' def getCallbacks(self): return self.__callbacks ''' set speed in bpm = beats per minute ''' def setBPM(self,bpm): self.output.setBPM(bpm) def getBPM(self): return self.output.getBPM() def setVolume(self, volume): self.output.setVolume(volume) def getVolume(self): return self.output.getVolume() ''' tunnel for log messages ''' def __log(self, msg): if(self.__logging): print 'EventManager:\t' + msg ''' get next Sequence ID from internal counter this method should be used each time a sequence is instantiated so there are no duplicates ''' def getNextSequenceID(self): self.__sequenceCounter = self.__sequenceCounter + 1 print self.__sequenceCounter return self.__sequenceCounter ''' make new sequence ''' def createSequence(self): seq = Sequence.Sequence(id=self.getNextSequenceID(),logging=Constants.LOGGING_sequences, manager=self) self.register(seq, seq.getMidiData) return seq ''' mainloop ''' def run(self): # start EventWorker-Thread self.__log('starting EventWorker-Thread') self.worker.start() # start Output-Thread self.__log('starting Output-Thread') self.output.start()