Welcome, guest | Sign In | My Account | Store | Cart

A mechanism for communication from any thread to the main thread.

It uses the same custom Event, but adds a classmethod convenience for taking a callback and params, wrapping it into an event and posting it to the receiver. Also adds in support for weak method references to the callback, in case the source object gets deleted before its callback is actually called.

Merges forks of both: http://code.activestate.com/recipes/81253/#c5 http://code.activestate.com/recipes/578299-pyqt-pyside-thread-safe-global-queue-main-loop-int/

Python, 337 lines
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
"""
ActiveState:
http://code.activestate.com/recipes/578634-pyqt-pyside-thread-safe-callbacks-main-loop-integr/

ref and proxy classes are based on:
    http://code.activestate.com/recipes/81253/#c5

Modified proxy to support a quiet concept for callbacks that can 
simply pass if they were not valid, instead of raising an exception.

Callback event dispatch based on:
    http://code.activestate.com/recipes/578299-pyqt-pyside-thread-safe-global-queue-main-loop-int/

Modified to simplify the process, by removing the Queue and threaded dispatcher,
and just using a more developed Event object that posts directly to the event loop.

"""

import weakref
import types

from functools import partial
from multiprocessing.pool import ThreadPool

__all__ = [
    "ref",
    "proxy",
    "CallbackEvent",
    "CallbackThreadPool",
]

class ref(object):
    """
    A weak method implementation
    """
    def __init__(self, method):
        try:

            if method.im_self is not None:
                # bound method
                self._obj = weakref.ref(method.im_self)

            else:
                # unbound method
                self._obj = None

            self._func = method.im_func
            self._class = method.im_class

        except AttributeError:
            # not a method
            self._obj = None
            self._func = method
            self._class = None
 
    def __call__(self):
        """
        Return a new bound-method like the original, or the
        original function if refers just to a function or unbound
        method.
        Returns None if the original object doesn't exist
        """
        if self.is_dead():
            return None

        if self._obj is not None:
            # we have an instance: return a bound method
            return types.MethodType(self._func, self._obj(), self._class)
        
        else:
            # we don't have an instance: return just the function
            return self._func
  
    def is_dead(self):
        """
        Returns True if the referenced callable was a bound method and
        the instance no longer exists. Otherwise, return False.
        """
        return self._obj is not None and self._obj() is None
 
    def __eq__(self, other):
        try:
            return type(self) is type(other) and self() == other()
        except:
            return False
  
    def __ne__(self, other):
        return not self == other
 
 
#
# The modified proxy class, adding a quiet option
#
class proxy(ref):
    """
    Exactly like ref, but calling it will cause the referent method to
    be called with the same arguments. If the referent's object no longer lives,
    ReferenceError is raised.

    If quiet is True, then a ReferenceError is not raise and the callback 
    silently fails if it is no longer valid. 
    """
 
    def __init__(self, method, quiet=False):
        super(proxy, self).__init__(method)
        self._quiet = quiet
 
    def __call__(self, *args, **kwargs):
        func = ref.__call__(self)

        if func is None:
            if self._quiet:
                return
            else:
                raise ReferenceError('object is dead')

        else:
            return func(*args, **kwargs)
  
    def __eq__(self, other):
        try:
            func1 = ref.__call__(self)
            func2 = ref.__call__(other)
            return type(self) == type(other) and func1 == func2

        except:
            return False


#
# PyQt4 / PySide Thread-Safe Callback Dispatch
#
try:
    from PySide import QtGui, QtCore
except ImportError:
    from PyQt4 import QtGui, QtCore


class _Invoker(QtCore.QObject):

    def customEvent(self, e):
        e.callback()


class CallbackEvent(QtCore.QEvent):
    """
    A custom QEvent that contains a callback reference

    Also provides class methods for conveniently executing 
    arbitrary callback, to be dispatched to the event loop.
    """
    EVENT_TYPE = QtCore.QEvent.Type(QtCore.QEvent.registerEventType())

    __invoker = _Invoker()

    def __init__(self, func, *args, **kwargs):
        super(CallbackEvent, self).__init__(self.EVENT_TYPE)
        self.func = func
        self.args = args
        self.kwargs = kwargs

    def callback(self):
        """
        Convenience method to run the callable. 

        Equivalent to:  
            self.func(*self.args, **self.kwargs)
        """
        self.func(*self.args, **self.kwargs)

    @classmethod
    def post(cls, func, *args, **kwargs):
        """
        Post a callable to run in the main thread
        """
        cls.post_to(cls.__invoker, func, *args, **kwargs)

    @classmethod
    def post_to(cls, receiver, func, *args, **kwargs):
        """
        Post a callable to be delivered to a specific
        receiver as a CallbackEvent. 

        It is the responsibility of this receiver to 
        handle the event and choose to call the callback.
        """
        # We can create a weak proxy reference to the
        # callback so that if the object associated with
        # a bound method is deleted, it won't call a dead method
        if not isinstance(func, proxy):
            reference = proxy(func, quiet=True)
        else:
            reference = func

        event = cls(reference, *args, **kwargs)

        # post the event to the given receiver
        QtGui.QApplication.postEvent(receiver, event)


class CallbackThreadPool(ThreadPool):
    """
    Simple wrapper around ThreadPool to wrap callbacks in a weakref 
    that get posted as CallbackEvents in the main thread.
    """
    def apply_async(self, fn, args=None, kwargs=None, callback=None):
        proxyCbk = self._async_helper(callback)
        args = args or tuple
        kwargs = kwargs or {}
        return super(CallbackThreadPool, self).apply_async(fn, args, kwargs, proxyCbk)

    def map_async(self, fn, iterable, chunk=None, callback=None):
        proxyCbk = self._async_helper(callback)
        return super(CallbackThreadPool, self).map_async(fn, iterable, chunk, callback)

    @staticmethod
    def _async_helper(callback):
        if callback:
            proxyCbk = partial(CallbackEvent.post, proxy(callback, quiet=True))
        else:
            proxyCbk = None

        return proxyCbk



if __name__ == "__main__":
    #
    # Usage Example
    #

    import time
    from threading import Event 


    class Gui(QtGui.QDialog):

        def __init__(self):
            super(Gui, self).__init__()
            self.resize(250, 300)

            self._list = QtGui.QListWidget(self)

            self._button1 = QtGui.QPushButton("Test CallbackEvent", self)     
            self._button2 = QtGui.QPushButton("Test CallbackPool", self)     
            
            layout = QtGui.QVBoxLayout(self)
            layout.setSpacing(2)
            layout.addWidget(self._button1)

            line = QtGui.QFrame(self)
            line.setFrameStyle(line.HLine)
            layout.addSpacing(6)
            layout.addWidget(line)
            layout.addSpacing(6)

            layout.addWidget(self._list)
            layout.addWidget(self._button2)

            self._pool = CallbackThreadPool(4)

            self._button1.clicked.connect(self.runCallbackEvents)
            self._button2.clicked.connect(self.runCallbackPool)

        def customEvent(self, event):
            print "Running callback from Gui.customEvent()"
            event.callback()

        def runCallbackEvents(self):
            self._list.clear()
            self._list.addItem("Printing results to console...")

            thread = Worker(self)
            thread.start()

        def runCallbackPool(self):
            self._list.clear()

            def action(a, b):
                time.sleep(1)
                return a+b

            def callback(result):
                self._list.addItem("Result: %s" % result)

            for i in xrange(20):
                self._pool.apply_async(action, (i, i), callback=callback)


    # Just some random worker
    class Worker(QtCore.QThread):

        def __init__(self, parent=None):
            super(Worker, self).__init__(parent)
            self.__quitting = Event()

        def run(self):

            for i in xrange(5):

                msg = "#{0}: Hi".format(i)

                # Instruct the say() method to run in the main thread,
                # being delivered to the parent
                CallbackEvent.post_to(self.parent(), self.say, msg)
                self.msleep(200)

                # Run the same callback, but without a specific receiver. 
                # This will just be run directly in an anonymous receiver.
                CallbackEvent.post(self.say, msg)
                self.msleep(200)

                # And lets just have one happen from this worker thread too
                self.say(msg)
                self.msleep(200)

                if self.__quitting.is_set():
                    break

            print "All done!"

        def stop(self):
            self.__quitting.set()
            self.wait()

        def say(self, word):
            mainThread = QtGui.qApp.thread()
            isMainThread = mainThread == self.currentThread()

            print word, "from", "main" if isMainThread else "worker", "thread"


    app = QtGui.QApplication([])
    gui = Gui()
    gui.show()
    gui.raise_()
    app.exec_()

I couldn't see a real benefit to having the original example use a threaded dispatcher, when all it was doing was taking it out of a queue, and posting it as an event. I thought a simpler approach would be to just have the same custom event, and a function that takes the callback parameters, and posts it directly to the event loop.

The postEvent() is not a blocking call, so it doesn't really need to be part of a thread / queue solution. I do think this example could be expanded with the weakMethod solution, for properly wrapping and handling a callback to an object that might have gotten cleaned up.

There is also a small CallbackThreadPool class that acts as a ThreadPool helper, wrapping callbacks in a weakref and posting them as CallbackEvents to the main thread.