123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899 |
- 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()
|