mod_signals.py

from weakref import WeakKeyDictionary, proxy
from types import FunctionType as function
from types import MethodType as instancedmethod

class MetaSignals(type):
    """
    register the list of signals in the class varable signals,
    including signals in superclasses.
    """
    def __init__(cls, name, bases, d):
        signals = d.get("signals", [])
        for superclass in cls.__bases__:
            signals.extend(getattr(superclass, 'signals', []))
        signals = dict([(x,None) for x in signals]).keys()
        d["signals"] = signals
        Signals.register(cls, signals)
        super(MetaSignals, cls).__init__(name, bases, d)


class Signals(object):
    _connections = WeakKeyDictionary()
    _supported = WeakKeyDictionary()

    def register(cls, sig_cls, signals):
        cls._supported[sig_cls] = signals
    register = classmethod(register)

    def connect(cls, obj, name, callback, user_arg=None):
        sig_cls = obj.__class__
        if not name in cls._supported.get(sig_cls, []):
            raise NameError, "No such signal %r for object %r" % \
                (name, obj)
        d = cls._connections.setdefault(obj, {})
        if type(callback) is instancedmethod:
            callback = (proxy(callback.__self__), callback.im_func.func_name)
        elif type(callback) is not function:
            raise TypeError('Help! I don't know what to do! Email the Urwid list with the output of this error:\nobj: %s\ntype: %s' % (str(callback), type(callback)))
        d.setdefault(name, []).append((callback, user_arg))
    connect = classmethod(connect)
        
    def disconnect(cls, obj, name, callback, user_arg=None):
        d = cls._connections.get(obj, {})
        if name not in d:
            return
        if (callback, user_arg) not in d[name]:
            return
        d[name].remove((callback, user_arg))
    disconnect = classmethod(disconnect)
 
    def emit(cls, obj, name, *args):
        result = False
        d = cls._connections.get(obj, {})
        for callback, user_arg in d.get(name, []):
            args_copy = args
            if user_arg is not None:
                args_copy = args + [user_arg]
            result |= bool(getattr(callback[0], callback[1])(*args_copy))
        return result
    emit = classmethod(emit)

_signals = Signals()
emit_signal = _signals.emit
register_signal = _signals.register
connect_signal = _signals.connect
disconnect_signal = _signals.disconnect