wxWidgets/wxPython/wx/lib/evtmgr.py

#---------------------------------------------------------------------------
# Name:        wxPython.lib.evtmgr
# Purpose:     An easier, more "Pythonic" and more OO method of registering
#              handlers for wxWindows events using the Publish/Subscribe
#              pattern.
#
# Author:      Robb Shecter and Robin Dunn
#
# Created:     12-December-2002
# RCS-ID:      $Id$
# Copyright:   (c) 2003 by db-X Corporation
# Licence:     wxWindows license
#---------------------------------------------------------------------------
# 12/02/2003 - Jeff Grimmett (grimmtooth@softhome.net)
#
# o Updated for 2.5 compatability.
#

"""
A module that allows multiple handlers to respond to single wxWidgets
events.  This allows true NxN Observer/Observable connections: One
event can be received by multiple handlers, and one handler can
receive multiple events.

There are two ways to register event handlers.  The first way is
similar to standard wxPython handler registration::

    from wx.lib.evtmgr import eventManager
    eventManager.Register(handleEvents, EVT_BUTTON, win=frame, id=101)

There's also a new object-oriented way to register for events.  This
invocation is equivalent to the one above, but does not require the
programmer to declare or track control ids or parent containers::

    eventManager.Register(handleEvents, EVT_BUTTON, myButton)

This module is Python 2.1+ compatible.

"""
import  wx
import  pubsub # publish / subscribe library

#---------------------------------------------------------------------------


class EventManager:
    """
    This is the main class in the module, and is the only class that
    the application programmer needs to use.  There is a pre-created
    instance of this class called 'eventManager'.  It should not be
    necessary to create other instances.
    """
    def __init__(self):
        self.eventAdapterDict    = {}
        self.messageAdapterDict  = {}
        self.windowTopicLookup   = {}
        self.listenerTopicLookup = {}
        self.__publisher         = pubsub.Publisher()
        self.EMPTY_LIST          = []


    def Register(self, listener, event, source=None, win=None, id=None):
        """
        Registers a listener function (or any callable object) to
        receive events of type event coming from the source window.
        For example::
        
            eventManager.Register(self.OnButton, EVT_BUTTON, theButton)

        Alternatively, the specific window where the event is
        delivered, and/or the ID of the event source can be specified.
        For example::
        
            eventManager.Register(self.OnButton, EVT_BUTTON, win=self, id=ID_BUTTON)
            
        or::
        
            eventManager.Register(self.OnButton, EVT_BUTTON, theButton, self)
            
        """

        # 1. Check if the 'event' is actually one of the multi-
        #    event macros.
        if _macroInfo.isMultiEvent(event):
            raise 'Cannot register the macro, '+`event`+'.  Register instead the individual events.'

        # Support a more OO API.  This allows the GUI widget itself to
        # be specified, and the id to be retrieved from the system,
        # instead of kept track of explicitly by the programmer.
        # (Being used to doing GUI work with Java, this seems to me to be
        # the natural way of doing things.)
        if source is not None:
            id  = source.GetId()
            
        if win is None:
            # Some widgets do not function as their own windows.
            win = self._determineWindow(source)
            
        topic = (event, win, id)

        #  Create an adapter from the PS system back to wxEvents, and
        #  possibly one from wxEvents:
        if not self.__haveMessageAdapter(listener, topic):
            messageAdapter = MessageAdapter(eventHandler=listener, topicPattern=topic)
            try:
                self.messageAdapterDict[topic][listener] = messageAdapter
            except KeyError:
                self.messageAdapterDict[topic] = {}
                self.messageAdapterDict[topic][listener] = messageAdapter

            if not self.eventAdapterDict.has_key(topic):
                self.eventAdapterDict[topic] = EventAdapter(event, win, id)
        else:
            # Throwing away a duplicate request
            pass

        # For time efficiency when deregistering by window:
        try:
            self.windowTopicLookup[win].append(topic)
        except KeyError:
            self.windowTopicLookup[win] = []
            self.windowTopicLookup[win].append(topic)

        # For time efficiency when deregistering by listener:
        try:
            self.listenerTopicLookup[listener].append(topic)
        except KeyError:
            self.listenerTopicLookup[listener] = []
            self.listenerTopicLookup[listener].append(topic)

        # See if the source understands the listeningFor protocol.
        # This is a bit of a test I'm working on - it allows classes
        # to know when their events are being listened to.  I use
        # it to enable chaining events from contained windows only
        # when needed.
        if source is not None:
            try:
                # Let the source know that we're listening  for this
                # event.
                source.listeningFor(event)
            except AttributeError:
                pass

    # Some aliases for Register, just for kicks
    Bind = Register
    Subscribe = Register


    def DeregisterWindow(self, win):
        """
        Deregister all events coming from the given window.
        """
        win    = self._determineWindow(win)
        topics = self.__getTopics(win)

        if topics:
            for aTopic in topics:
                self.__deregisterTopic(aTopic)

            del self.windowTopicLookup[win]


    def DeregisterListener(self, listener):
        """
        Deregister all event notifications for the given listener.
        """
        try:
            topicList = self.listenerTopicLookup[listener]
        except KeyError:
            return

        for topic in topicList:
            topicDict = self.messageAdapterDict[topic]

            if topicDict.has_key(listener):
                topicDict[listener].Destroy()
                del topicDict[listener]

                if len(topicDict) == 0:
                    self.eventAdapterDict[topic].Destroy()
                    del self.eventAdapterDict[topic]
                    del self.messageAdapterDict[topic]

        del self.listenerTopicLookup[listener]


    def GetStats(self):
        """
        Return a dictionary with data about my state.
        """
        stats = {}
        stats['Adapters: Message'] = reduce(lambda x,y: x+y, [0] + map(len, self.messageAdapterDict.values()))
        stats['Adapters: Event']   = len(self.eventAdapterDict)
        stats['Topics: Total']     = len(self.__getTopics())
        stats['Topics: Dead']      = len(self.GetDeadTopics())
        return stats


    def DeregisterDeadTopics(self):
        """
        Deregister any entries relating to dead
        wxPython objects.  Not sure if this is an
        important issue; 1) My app code always de-registers
        listeners it doesn't need.  2) I don't think
        that lingering references to these dead objects
        is a problem.
        """
        for topic in self.GetDeadTopics():
            self.__deregisterTopic(topic)


    def GetDeadTopics(self):
        """
        Return a list of topics relating to dead wxPython
        objects.
        """
        return filter(self.__isDeadTopic, self.__getTopics())


    def __winString(self, aWin):
        """
        A string rep of a window for debugging
        """
        try:
            name = aWin.GetClassName()
            i    = id(aWin)
            return '%s #%d' % (name, i)
        except wx.PyDeadObjectError:
            return '(dead wx.Object)'


    def __topicString(self, aTopic):
        """
        A string rep of a topic for debugging
        """
        return '[%-26s %s]' % (aTopic[0].__name__, self.winString(aTopic[1]))


    def __listenerString(self, aListener):
        """
        A string rep of a listener for debugging
        """
        try:
            return aListener.im_class.__name__ + '.' + aListener.__name__
        except:
            return 'Function ' + aListener.__name__


    def __deregisterTopic(self, aTopic):
        try:
            messageAdapterList = self.messageAdapterDict[aTopic].values()
        except KeyError:
            # This topic isn't valid.  Probably because it was deleted
            # by listener.
            return

        for messageAdapter in messageAdapterList:
            messageAdapter.Destroy()

        self.eventAdapterDict[aTopic].Destroy()
        del self.messageAdapterDict[aTopic]
        del self.eventAdapterDict[aTopic]


    def __getTopics(self, win=None):
        if win is None:
            return self.messageAdapterDict.keys()

        if win is not None:
            try:
                return self.windowTopicLookup[win]
            except KeyError:
                return self.EMPTY_LIST


    def __isDeadWxObject(self, anObject):
        return isinstance(anObject, wx._wxPyDeadObject)


    def __isDeadTopic(self, aTopic):
        return self.__isDeadWxObject(aTopic[1])


    def __haveMessageAdapter(self, eventHandler, topicPattern):
        """
        Return True if there's already a message adapter
        with these specs.
        """
        try:
            return self.messageAdapterDict[topicPattern].has_key(eventHandler)
        except KeyError:
            return 0


    def _determineWindow(self, aComponent):
        """
        Return the window that corresponds to this component.
        A window is something that supports the Connect protocol.
        Most things registered with the event manager are a window,
        but there are apparently some exceptions.  If more are
        discovered, the implementation can be changed to a dictionary
        lookup along the lines of class : function-to-get-window.
        """
        if isinstance(aComponent, wx.MenuItem):
            return aComponent.GetMenu()
        else:
            return aComponent



#---------------------------------------------------------------------------
# From here down is implementaion and support classes, although you may
# find some of them useful in other contexts.
#---------------------------------------------------------------------------


class EventMacroInfo:
    """
    A class that provides information about event macros.
    """
    def __init__(self):
        self.lookupTable = {}


    def getEventTypes(self, eventMacro):
        """
        Return the list of event types that the given
        macro corresponds to.
        """
        try:
            return self.lookupTable[eventMacro]
        except KeyError:
            win = FakeWindow()
            try:
                eventMacro(win, None, None)
            except (TypeError, AssertionError):
                eventMacro(win, None)
            self.lookupTable[eventMacro] = win.eventTypes
            return win.eventTypes


    def eventIsA(self, event, macroList):
        """
        Return True if the event is one of the given
        macros.
        """
        eventType = event.GetEventType()
        for macro in macroList:
            if eventType in self.getEventTypes(macro):
                return 1
        return 0


    def macroIsA(self, macro, macroList):
        """
        Return True if the macro is in the macroList.
        The added value of this method is that it takes
        multi-events into account.  The macroList parameter
        will be coerced into a sequence if needed.
        """
        if callable(macroList):
            macroList = (macroList,)
        testList  = self.getEventTypes(macro)
        eventList = []
        for m in macroList:
            eventList.extend(self.getEventTypes(m))
        # Return True if every element in testList is in eventList
        for element in testList:
            if element not in eventList:
                return 0
        return 1


    def isMultiEvent(self, macro):
        """
        Return True if the given macro actually causes
        multiple events to be registered.
        """
        return len(self.getEventTypes(macro)) > 1


#---------------------------------------------------------------------------

class FakeWindow:
    """
    Used internally by the EventMacroInfo class.  The FakeWindow is
    the most important component of the macro-info utility: it
    implements the Connect() protocol of wxWindow, but instead of
    registering for events, it keeps track of what parameters were
    passed to it.
    """
    def __init__(self):
        self.eventTypes = []

    def Connect(self, id1, id2, eventType, handlerFunction):
        self.eventTypes.append(eventType)


#---------------------------------------------------------------------------

class EventAdapter:
    """
    A class that adapts incoming wxWindows events to
    Publish/Subscribe messages.

    In other words, this is the object that's seen by the
    wxWindows system.  Only one of these registers for any
    particular wxWindows event.  It then relays it into the
    PS system, which lets many listeners respond.
    """
    def __init__(self, func, win, id):
        """
        Instantiate a new adapter. Pre-compute my Publish/Subscribe
        topic, which is constant, and register with wxWindows.
        """
        self.publisher = pubsub.Publisher()
        self.topic     = ((func, win, id),)
        self.id        = id
        self.win       = win
        self.eventType = _macroInfo.getEventTypes(func)[0]

        # Register myself with the wxWindows event system
        try:
            func(win, id, self.handleEvent)
            self.callStyle = 3
        except (TypeError, AssertionError):
            func(win, self.handleEvent)
            self.callStyle = 2


    def disconnect(self):
        if self.callStyle == 3:
            return self.win.Disconnect(self.id, -1, self.eventType)
        else:
            return self.win.Disconnect(-1, -1, self.eventType)


    def handleEvent(self, event):
        """
        In response to a wxWindows event, send a PS message
        """
        self.publisher.sendMessage(topic=self.topic, data=event)


    def Destroy(self):
        try:
            if not self.disconnect():
                print 'disconnect failed'
        except wx.PyDeadObjectError:
            print 'disconnect failed: dead object'              ##????


#---------------------------------------------------------------------------

class MessageAdapter:
    """
    A class that adapts incoming Publish/Subscribe messages
    to wxWindows event calls.

    This class works opposite the EventAdapter, and
    retrieves the information an EventAdapter has sent in a message.
    Strictly speaking, this class is not required: Event listeners
    could pull the original wxEvent object out of the PS Message
    themselves.

    However, by pairing an instance of this class with each wxEvent
    handler, the handlers can use the standard API: they receive an
    event as a parameter.
    """
    def __init__(self, eventHandler, topicPattern):
        """
        Instantiate a new MessageAdapter that send wxEvents to the
        given eventHandler.
        """
        self.eventHandler = eventHandler
        pubsub.Publisher().subscribe(listener=self.deliverEvent, topic=(topicPattern,))

    def deliverEvent(self, message):
        event = message.data        # Extract the wxEvent
        self.eventHandler(event)    # Perform the call as wxWindows would

    def Destroy(self):
        pubsub.Publisher().unsubscribe(listener=self.deliverEvent)


#---------------------------------------------------------------------------
# Create globals

_macroInfo    = EventMacroInfo()

# For now a singleton is not enforced.  Should it be or can we trust
# the programmers?
eventManager  = EventManager()


#---------------------------------------------------------------------------
# simple test code


if __name__ == '__main__':
    app    = wx.PySimpleApp()
    frame  = wx.Frame(None, -1, 'Event Test', size=(300,300))
    button = wx.ToggleButton(frame, -1, 'Listen for Mouse Events')
    sizer  = wx.BoxSizer(wx.HORIZONTAL)
    sizer.Add(button, 0, 0 | wx.ALL, 10)
    frame.SetAutoLayout(1)
    frame.SetSizer(sizer)

    #
    # Demonstrate 1) register/deregister, 2) Multiple listeners receiving
    # one event, and 3) Multiple events going to one listener.
    #

    def printEvent(event):
        print 'Name:',event.GetClassName(),'Timestamp',event.GetTimestamp()

    def enableFrameEvents(event):
        # Turn the output of mouse events on and off
        if event.IsChecked():
            print '\nEnabling mouse events...'
            eventManager.Register(printEvent, wx.EVT_MOTION,    frame)
            eventManager.Register(printEvent, wx.EVT_LEFT_DOWN, frame)
        else:
            print '\nDisabling mouse events...'
            eventManager.DeregisterWindow(frame)

    # Send togglebutton events to both the on/off code as well
    # as the function that prints to stdout.
    eventManager.Register(printEvent,        wx.EVT_TOGGLEBUTTON, button)
    eventManager.Register(enableFrameEvents, wx.EVT_TOGGLEBUTTON, button)

    frame.CenterOnScreen()
    frame.Show(1)
    app.MainLoop()