Source code for direct.showbase.EventManager

"""Contains the EventManager class.  See :mod:`.EventManagerGlobal` for the
global eventMgr instance."""

__all__ = ['EventManager']


from .MessengerGlobal import *
from direct.directnotify.DirectNotifyGlobal import *
from direct.task.TaskManagerGlobal import taskMgr
from panda3d.core import PStatCollector, EventQueue, EventHandler
from panda3d.core import ConfigVariableBool


[docs]class EventManager: notify = None
[docs] def __init__(self, eventQueue = None): """ Create a C++ event queue and handler """ # Make a notify category for this class (unless there already is one) if EventManager.notify is None: EventManager.notify = directNotify.newCategory("EventManager") self.eventQueue = eventQueue self.eventHandler = None self._wantPstats = ConfigVariableBool('pstats-eventmanager', False)
[docs] def doEvents(self): """ Process all the events on the C++ event queue """ # use different methods for handling events with and without pstats tracking # for efficiency if self._wantPstats: processFunc = self.processEventPstats else: processFunc = self.processEvent isEmptyFunc = self.eventQueue.isQueueEmpty dequeueFunc = self.eventQueue.dequeueEvent while not isEmptyFunc(): processFunc(dequeueFunc())
[docs] def eventLoopTask(self, task): """ Process all the events on the C++ event queue """ self.doEvents() messenger.send("event-loop-done") return task.cont
[docs] def parseEventParameter(self, eventParameter): """ Extract the actual data from the eventParameter """ if eventParameter.isInt(): return eventParameter.getIntValue() elif eventParameter.isDouble(): return eventParameter.getDoubleValue() elif eventParameter.isString(): return eventParameter.getStringValue() elif eventParameter.isWstring(): return eventParameter.getWstringValue() elif eventParameter.isTypedRefCount(): return eventParameter.getTypedRefCountValue() elif eventParameter.isEmpty(): return None else: # Must be some user defined type, return the ptr # which will be downcast to that type. return eventParameter.getPtr()
[docs] def processEvent(self, event): """ Process a C++ event Duplicate any changes in processEventPstats """ # ************************************************************** # ******** Duplicate any changes in processEventPstats ********* # ************************************************************** # Get the event name eventName = event.name if eventName: paramList = [] for eventParameter in event.parameters: eventParameterData = self.parseEventParameter(eventParameter) paramList.append(eventParameterData) # Do not print the new frame debug, it is too noisy! if EventManager.notify.getDebug() and eventName != 'NewFrame': EventManager.notify.debug('received C++ event named: ' + eventName + ' parameters: ' + repr(paramList)) # ************************************************************** # ******** Duplicate any changes in processEventPstats ********* # ************************************************************** # Send the event, we used to send it with the event # name as a parameter, but now you can use extraArgs for that messenger.send(eventName, paramList) # Also send the event down into C++ land handler = self.eventHandler if handler: handler.dispatchEvent(event) else: # An unnamed event from C++ is probably a bad thing EventManager.notify.warning('unnamed event in processEvent')
[docs] def processEventPstats(self, event): """ Process a C++ event with pstats tracking Duplicate any changes in processEvent """ # ******************************************************** # ******** Duplicate any changes in processEvent ********* # ******************************************************** # Get the event name eventName = event.name if eventName: paramList = [] for eventParameter in event.parameters: eventParameterData = self.parseEventParameter(eventParameter) paramList.append(eventParameterData) # Do not print the new frame debug, it is too noisy! if EventManager.notify.getDebug() and eventName != 'NewFrame': EventManager.notify.debug('received C++ event named: ' + eventName + ' parameters: ' + repr(paramList)) # Send the event, we used to send it with the event # name as a parameter, but now you can use extraArgs for that # ******************************************************** # ******** Duplicate any changes in processEvent ********* # ******************************************************** name = eventName hyphen = name.find('-') if hyphen >= 0: name = name[0:hyphen] pstatCollector = PStatCollector('App:Show code:eventManager:' + name) pstatCollector.start() if self.eventHandler: cppPstatCollector = PStatCollector( 'App:Show code:eventManager:' + name + ':C++') messenger.send(eventName, paramList) # Also send the event down into C++ land handler = self.eventHandler if handler: cppPstatCollector.start() handler.dispatchEvent(event) cppPstatCollector.stop() pstatCollector.stop() else: # An unnamed event from C++ is probably a bad thing EventManager.notify.warning('unnamed event in processEvent')
[docs] def restart(self): if self.eventQueue is None: self.eventQueue = EventQueue.getGlobalEventQueue() if self.eventHandler is None: if self.eventQueue == EventQueue.getGlobalEventQueue(): # If we are using the global event queue, then we also # want to use the global event handler. self.eventHandler = EventHandler.getGlobalEventHandler() else: # Otherwise, we need our own event handler. self.eventHandler = EventHandler(self.eventQueue) taskMgr.add(self.eventLoopTask, 'eventManager')
[docs] def shutdown(self): taskMgr.remove('eventManager') # Flush the event queue. We do this after removing the task # since the task removal itself might also fire off an event. if self.eventQueue is not None: self.eventQueue.clear()