IPython/utils/notification.py

# encoding: utf-8
"""
The IPython Core Notification Center.

See docs/source/development/notification_blueprint.txt for an overview of the
notification module.

Authors:

* Barry Wark
* Brian Granger
"""

#-----------------------------------------------------------------------------
#  Copyright (C) 2008-2009  The IPython Development Team
#
#  Distributed under the terms of the BSD License.  The full license is in
#  the file COPYING, distributed as part of this software.
#-----------------------------------------------------------------------------

#-----------------------------------------------------------------------------
# Code
#-----------------------------------------------------------------------------


class NotificationError(Exception):
    pass


class NotificationCenter(object):
    """Synchronous notification center.

    Examples
    --------
    Here is a simple example of how to use this::

        import IPython.util.notification as notification
        def callback(ntype, theSender, args={}):
            print ntype,theSender,args

        notification.sharedCenter.add_observer(callback, 'NOTIFICATION_TYPE', None)
        notification.sharedCenter.post_notification('NOTIFICATION_TYPE', object()) # doctest:+ELLIPSIS
        NOTIFICATION_TYPE ...
    """
    def __init__(self):
        super(NotificationCenter, self).__init__()
        self._init_observers()

    def _init_observers(self):
        """Initialize observer storage"""

        self.registered_types = set() #set of types that are observed
        self.registered_senders = set() #set of senders that are observed
        self.observers = {} #map (type,sender) => callback (callable)

    def post_notification(self, ntype, sender, *args, **kwargs):
        """Post notification to all registered observers.

        The registered callback will be called as::

            callback(ntype, sender, *args, **kwargs)

        Parameters
        ----------
        ntype : hashable
            The notification type.
        sender : hashable
            The object sending the notification.
        *args : tuple
            The positional arguments to be passed to the callback.
        **kwargs : dict
            The keyword argument to be passed to the callback.

        Notes
        -----
        * If no registered observers, performance is O(1).
        * Notificaiton order is undefined.
        * Notifications are posted synchronously.
        """

        if(ntype==None or sender==None):
            raise NotificationError(
                "Notification type and sender are required.")

        # If there are no registered observers for the type/sender pair
        if((ntype not in self.registered_types and
                None not in self.registered_types) or
            (sender not in self.registered_senders and
                None not in self.registered_senders)):
            return

        for o in self._observers_for_notification(ntype, sender):
            o(ntype, sender, *args, **kwargs)

    def _observers_for_notification(self, ntype, sender):
        """Find all registered observers that should recieve notification"""

        keys = (
                   (ntype,sender),
                   (ntype, None),
                   (None, sender),
                   (None,None)
               )

        obs = set()
        for k in keys:
            obs.update(self.observers.get(k, set()))

        return obs

    def add_observer(self, callback, ntype, sender):
        """Add an observer callback to this notification center.

        The given callback will be called upon posting of notifications of
        the given type/sender and will receive any additional arguments passed
        to post_notification.

        Parameters
        ----------
        callback : callable
            The callable that will be called by :meth:`post_notification`
            as ``callback(ntype, sender, *args, **kwargs)
        ntype : hashable
            The notification type. If None, all notifications from sender
            will be posted.
        sender : hashable
            The notification sender. If None, all notifications of ntype
            will be posted.
        """
        assert(callback != None)
        self.registered_types.add(ntype)
        self.registered_senders.add(sender)
        self.observers.setdefault((ntype,sender), set()).add(callback)

    def remove_all_observers(self):
        """Removes all observers from this notification center"""

        self._init_observers()



shared_center = NotificationCenter()