Async like pattern in pyqt? Or cleaner background call pattern?
I'm trying to write a short(one file pyqt) program which is responsive(so dependencies outside python/lxml/qt, especially ones I can't just stick in the file have some downsides for this use case but I might still be willing to try them). I'm trying to perform possibly lengthy(and cancelable) operations on a worker thread(actually the background operation has a lock around it to prevent multiple operations at once(since the library it uses can only be used one call at a time) and timeouts so spawning multiple threads would be fine also).
As far as I can figure out the "basic" way to do this with qt is. (note code is not tested so it may be wrong)
class MainWindow(QWidget):
#self.worker moved to background thread
def initUI(self):
...
self.cmd_button.clicked.connect(self.send)
...
@pyqtslot()
def send(self):
...
...#get cmd from gui
QtCore.QTimer.singleShot(0, lambda : self.worker(cmd))
@pyqtslot(str)
def end_send(self, result):
...
...# set some gui to display result
...
class WorkerObject(QObject):
def send_cmd(self, cmd):
... get result of cmd
QtCore.QTimer.singleShot(0, lambda: self.main_window.end_send())
(Am I using QTimer right(it runs on different thread right)?)
I'd really prefer to have something simpler and more abstracted along the lines of c#'s async. (note I haven't used asyncio so I might be getting some things wrong)
class MainWindow(QWidget):
...
@asyncio.coroutine
def send(self):
...
...#get cmd from gui
result = yield from self.worker(cmd)
#set gui textbox to result
class WorkerObject(QObject):
@asyncio.coroutine
def send_cmd(self, cmd):
... get result of cmd
yield from loop.run_in_executor(None, self.model.send_command, cmd)
I heard that python 3 had similar features and there was a back port but does it work properly with qt?
If anyone knows of another saner pattern. that too would be useful/an acceptable answer.
Answer
The short answer to your question ("is there a way to use an asyncio-like pattern in PyQt?") is yes, but it's pretty complicated and arguably not worth it for a small program. Here's some prototype code that allows you to use an asynchronous pattern like you described:
import types
import weakref
from functools import partial
from PyQt4 import QtGui
from PyQt4 import QtCore
from PyQt4.QtCore import QThread, QTimer
## The following code is borrowed from here:
# http://stackoverflow.com/questions/24689800/async-like-pattern-in-pyqt-or-cleaner-background-call-pattern
# It provides a child->parent thread-communication mechanism.
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
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
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())
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_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)
## End borrowed code
## Begin Coroutine-framework code
class AsyncTask(QtCore.QObject):
""" Object used to manage asynchronous tasks.
This object should wrap any function that you want
to call asynchronously. It will launch the function
in a new thread, and register a listener so that
`on_finished` is called when the thread is complete.
"""
def __init__(self, func, *args, **kwargs):
super(AsyncTask, self).__init__()
self.result = None # Used for the result of the thread.
self.func = func
self.args = args
self.kwargs = kwargs
self.finished = False
self.finished_cb_ran = False
self.finished_callback = None
self.objThread = RunThreadCallback(self, self.func, self.on_finished,
*self.args, **self.kwargs)
self.objThread.start()
def customEvent(self, event):
event.callback()
def on_finished(self, result):
""" Called when the threaded operation is complete.
Saves the result of the thread, and
executes finished_callback with the result if one
exists. Also closes/cleans up the thread.
"""
self.finished = True
self.result = result
if self.finished_callback:
self.finished_ran = True
func = partial(self.finished_callback, result)
QTimer.singleShot(0, func)
self.objThread.quit()
self.objThread.wait()
class RunThreadCallback(QtCore.QThread):
""" Runs a function in a thread, and alerts the parent when done.
Uses a custom QEvent to alert the main thread of completion.
"""
def __init__(self, parent, func, on_finish, *args, **kwargs):
super(RunThreadCallback, self).__init__(parent)
self.on_finished = on_finish
self.func = func
self.args = args
self.kwargs = kwargs
def run(self):
try:
result = self.func(*self.args, **self.kwargs)
except Exception as e:
print "e is %s" % e
result = e
finally:
CallbackEvent.post_to(self.parent(), self.on_finished, result)
def coroutine(func):
""" Coroutine decorator, meant for use with AsyncTask.
This decorator must be used on any function that uses
the `yield AsyncTask(...)` pattern. It shouldn't be used
in any other case.
The decorator will yield AsyncTask objects from the
decorated generator function, and register itself to
be called when the task is complete. It will also
excplicitly call itself if the task is already
complete when it yields it.
"""
def wrapper(*args, **kwargs):
def execute(gen, input=None):
if isinstance(gen, types.GeneratorType):
if not input:
obj = next(gen)
else:
try:
obj = gen.send(input)
except StopIteration as e:
result = getattr(e, "value", None)
return result
if isinstance(obj, AsyncTask):
# Tell the thread to call `execute` when its done
# using the current generator object.
func = partial(execute, gen)
obj.finished_callback = func
if obj.finished and not obj.finished_cb_ran:
obj.on_finished(obj.result)
else:
raise Exception("Using yield is only supported with AsyncTasks.")
else:
print("result is %s" % result)
return result
result = func(*args, **kwargs)
execute(result)
return wrapper
## End coroutine-framework code
If you put the above code into a module (say qtasync.py) you can import it into a script and use it like so to get asyncio-like behavior:
import sys
import time
from qtasync import AsyncTask, coroutine
from PyQt4 import QtGui
from PyQt4.QtCore import QThread
class MainWindow(QtGui.QMainWindow):
def __init__(self):
super(MainWindow, self).__init__()
self.initUI()
def initUI(self):
self.cmd_button = QtGui.QPushButton("Push", self)
self.cmd_button.clicked.connect(self.send_evt)
self.statusBar()
self.show()
def worker(self, inval):
print "in worker, received '%s'" % inval
time.sleep(2)
return "%s worked" % inval
@coroutine
def send_evt(self, arg):
out = AsyncTask(self.worker, "test string")
out2 = AsyncTask(self.worker, "another test string")
QThread.sleep(3)
print("kicked off async task, waiting for it to be done")
val = yield out
val2 = yield out2
print ("out is %s" % val)
print ("out2 is %s" % val2)
out = yield AsyncTask(self.worker, "Some other string")
print ("out is %s" % out)
if __name__ == "__main__":
app = QtGui.QApplication(sys.argv)
m = MainWindow()
sys.exit(app.exec_())
Output (when the button is pushed):
in worker, received 'test string'
in worker, received 'another test string'
kicked off async task, waiting for it to be done
out is test string worked
out2 is another test string worked
in worker, received 'Some other string'
out is Some other string worked
As you can see, worker gets run asynchronously in a thread whenever it gets called via the AsyncTask class, but its return value can be yielded directly from send_evt, without needing to use callbacks.
The code uses the coroutine-supporting features (generator_object.send) of Python generators, and a recipe I found on ActiveState that provides a child->main thread communication mechanism, to implement some very basic coroutines. The coroutines are quite limited: You can't return anything from them, and you can't chain coroutine calls together. It's probably possible to implement both of those things, but also probably not worth the effort, unless you really need them. I haven't done much negative testing with this either, so exceptions in workers and elsewhere may not be handled properly. What it does do well, though, is allow you to call methods in separate threads via the AsyncTask class, and then yield a result from the thread when one is ready, without blocking the Qt event loop. Normally this kind of thing would be done with callbacks, which can be difficult to follow and is generally less readable than having all the code in a single function.
You're welcome to use this approach if the limitations I mentioned are acceptable to you, but this is really just a proof-of-concept; you would need to do a whole bunch of testing before you think about put it into production anywhere.
As you mentioned, Python 3.3 and 3.4 makes asynchronous programming easier with the introduction of yield from and asyncio, respectively. I think yield from would actually be quite useful here to allow chaining coroutines (meaning have one coroutine call another and get a result from it). asyncio has no PyQt4 event-loop integration, so it's usefulness is pretty limited.
Another option would be to drop the coroutine piece of this altogether and just use the callback-based inter-thread communication mechanism directly:
import sys
import time
from qtasync import CallbackEvent # No need for the coroutine stuff
from PyQt4 import QtGui
from PyQt4.QtCore import QThread
class MyThread(QThread):
""" Runs a function in a thread, and alerts the parent when done.
Uses a custom QEvent to alert the main thread of completion.
"""
def __init__(self, parent, func, on_finish, *args, **kwargs):
super(MyThread, self).__init__(parent)
self.on_finished = on_finish
self.func = func
self.args = args
self.kwargs = kwargs
self.start()
def run(self):
try:
result = self.func(*self.args, **self.kwargs)
except Exception as e:
print "e is %s" % e
result = e
finally:
CallbackEvent.post_to(self.parent(), self.on_finished, result)
class MainWindow(QtGui.QMainWindow):
def __init__(self):
super(MainWindow, self).__init__()
self.initUI()
def initUI(self):
self.cmd_button = QtGui.QPushButton("Push", self)
self.cmd_button.clicked.connect(self.send)
self.statusBar()
self.show()
def customEvent(self, event):
event.callback()
def worker(self, inval):
print("in worker, received '%s'" % inval)
time.sleep(2)
return "%s worked" % inval
def end_send(self, cmd):
print("send returned '%s'" % cmd)
def send(self, arg):
t = MyThread(self, self.worker, self.end_send, "some val")
print("Kicked off thread")
if __name__ == "__main__":
app = QtGui.QApplication(sys.argv)
m = MainWindow()
sys.exit(app.exec_())
Output:
Kicked off thread
in worker, received 'some val'
send returned 'some val worked'
This could get a bit unwieldy if you're dealing with a long callback chain, but it doesn't rely on the more unproven coroutine code.