KaptureControlGui/KCG/base/callbacks.py

from PyQt4 import QtCore


class NoCallbackError(Exception):
    pass


class CallbackExistsError(Exception):
    pass


class CallbackHandler(QtCore.QObject):
    """
    Handler for custom callbacks.
    It can handle synchronous callbacks as well as async callbacks (using pyqtSignals)
    """

    callback_signal = QtCore.pyqtSignal(str, list, dict)

    def __init__(self):
        super(CallbackHandler, self).__init__()
        self._callbacks = {}
        self.callback_signal.connect(self.__async_callback_receiver)

    def callback(self, name, *args, **kwargs):
        """
        Call all registered callback method for name
        This passes all additional arguments and keyword arguments down to the callbacks
        NOTE: all callbacks therefore need to accept the same number of arguments
        :param name: the name for which the callbacks are to be called
        :param args: arguments to be passed to the callbacks
        :param kwargs: keyword arguments to be passed to the callbacks
        """
        if name in self._callbacks:
            for callback in self._callbacks[name]:
                callback(*args, **kwargs)
        else:
            raise NoCallbackError("Callback " + name + " not registered.")

    def __async_callback_receiver(self, name, *args):
        """
        Internal Method (Called when async callbacks are to be executed)
        """
        self.callback(str(name), *args[0], **args[1])

    def async_callback(self, name, *args, **kwargs):
        """
        Perform a async callback (same as callback but through pyqtSignal and therefore allowed in threads)
        :param name: the name for which the callbacks are to be called
        :param args: arguments to be passed to the callbacks
        :param kwargs: keyword arguments to be passed to the callbacks
        """
        if name in self._callbacks:
            self.callback_signal.emit(name, list(args), dict(kwargs))
        else:
            raise NoCallbackError("Callback " + name + " not registered.")

    def add_callback(self, name, callback):
        """
        Register a callback for name
        :param name: the name to register against
        :param callback: the callback to register
        """
        if name in self._callbacks:
            if callback in self._callbacks[name]:
                raise CallbackExistsError("Callback " + name + " already registered.")
            else:
                self._callbacks[name].append(callback)
        else:
            self._callbacks[name] = [callback, ]

    def delete_callback(self, name, callback):
        """
        Delete a callback from name
        if no callback for name is left the whole group is deleted
        :param name: the name to delete the callback from
        :param callback: the callback to delete
        """
        if name not in self._callbacks:
            raise NoCallbackError("Callback " + name + " not registered.")
        else:
            if callback in self._callbacks[name]:
                self._callbacks[name].remove(callback)
            else:
                raise NoCallbackError("Callback " + str(callback) + " not registered in " + name+".")
            if len(self._callbacks[name]) == 0:
                self.delete_callback_class(name)

    def delete_callback_class(self, name):
        """
        Delete a whole callback class
        :param name: the name of the class
        """
        if name in self._callbacks:
            del self._callbacks[name]
        else:
            raise NoCallbackError("Callback " + name + " not registered.")

callbacks = CallbackHandler()