Background theme with QThread in PyQt

Question

I have a program that interacts with the radio that I use via gui, which I wrote in PyQt. Obviously, one of the main functions of the radio is to transmit data, but for this it is necessary to perform a recording cycle, which leads to a hang of gi. Since I never did thread processing, I tried to get rid of these hangs using QCoreApplication.processEvents(). The radio needs to sleep between transmissions, however, so gui still freezes depending on how long these sleepers last.

Is there an easy way to fix this with QThread? I was looking for tutorials on how to implement multithreading with PyQt, but most of them are related to setting up servers and are much more advanced than I need. To be honest, I don’t even need my thread to update anything during its operation, I just need to start it, transfer it in the background and stop it.

Answer 1

I created a small example that shows 3 different and simple ways to work with streams. I hope this helps you find the right approach to your problem.

 import sys import time from PyQt5.QtCore import (QCoreApplication, QObject, QRunnable, QThread, QThreadPool, pyqtSignal) # Subclassing QThread # http://qt-project.org/doc/latest/qthread.html class AThread(QThread): def run(self): count = 0 while count < 5: time.sleep(1) print("A Increasing") count += 1 # Subclassing QObject and using moveToThread # http://blog.qt.digia.com/blog/2007/07/05/qthreads-no-longer-abstract class SomeObject(QObject): finished = pyqtSignal() def long_running(self): count = 0 while count < 5: time.sleep(1) print("B Increasing") count += 1 self.finished.emit() # Using a QRunnable # http://qt-project.org/doc/latest/qthreadpool.html # Note that a QRunnable isn't a subclass of QObject and therefore does # not provide signals and slots. class Runnable(QRunnable): def run(self): count = 0 app = QCoreApplication.instance() while count < 5: print("C Increasing") time.sleep(1) count += 1 app.quit() def using_q_thread(): app = QCoreApplication([]) thread = AThread() thread.finished.connect(app.exit) thread.start() sys.exit(app.exec_()) def using_move_to_thread(): app = QCoreApplication([]) objThread = QThread() obj = SomeObject() obj.moveToThread(objThread) obj.finished.connect(objThread.quit) objThread.started.connect(obj.long_running) objThread.finished.connect(app.exit) objThread.start() sys.exit(app.exec_()) def using_q_runnable(): app = QCoreApplication([]) runnable = Runnable() QThreadPool.globalInstance().start(runnable) sys.exit(app.exec_()) if __name__ == "__main__": #using_q_thread() #using_move_to_thread() using_q_runnable() 

Answer 2

Take this answer updated for PyQt5, python 3.4

Use this as a template to start a worker who does not accept data and returns data because it is available for the form.

1 - The working class is reduced and placed in the file.py file for easy memorization and reuse of the software.

2 - The main.py file is the file that defines the GUI form class

3 - The thread object is not a subclass.

4 - Both the flow object and the work object belong to the Form object

5 - The steps of the procedure are in the comments.

  # worker.py from PyQt5.QtCore import QThread, QObject, pyqtSignal, pyqtSlot import time class Worker(QObject): finished = pyqtSignal() intReady = pyqtSignal(int) @pyqtSlot() def procCounter(self): # A slot takes no params for i in range(1, 100): time.sleep(1) self.intReady.emit(i) self.finished.emit() 

And the main file:

  #main.py from PyQt5.QtCore import QThread from PyQt5.QtWidgets import QApplication, QLabel, QWidget, QGridLayout import sys import worker class Form(QWidget): def __init__(self): super().__init__() self.label = QLabel("0") # 1 - create Worker and Thread inside the Form self.obj = worker.Worker() # no parent! self.thread = QThread() # no parent! # 2 - Connect Worker`s Signals to Form method slots to post data. self.obj.intReady.connect(self.onIntReady) # 3 - Move the Worker object to the Thread object self.obj.moveToThread(self.thread) # 4 - Connect Worker Signals to the Thread slots self.obj.finished.connect(self.thread.quit) # 5 - Connect Thread started signal to Worker operational slot method self.thread.started.connect(self.obj.procCounter) # * - Thread finished signal will close the app if you want! #self.thread.finished.connect(app.exit) # 6 - Start the thread self.thread.start() # 7 - Start the form self.initUI() def initUI(self): grid = QGridLayout() self.setLayout(grid) grid.addWidget(self.label,0,0) self.move(300, 150) self.setWindowTitle('thread test') self.show() def onIntReady(self, i): self.label.setText("{}".format(i)) #print(i) app = QApplication(sys.argv) form = Form() sys.exit(app.exec_()) 

Answer 3

A very good example from Matt, I fixed a typo, and pyqt4.8 is also common now, so I removed the dummy class as well and added an example for the dataReady signal

 # -*- coding: utf-8 -*- import sys from PyQt4 import QtCore, QtGui from PyQt4.QtCore import Qt # very testable class (hint: you can use mock.Mock for the signals) class Worker(QtCore.QObject): finished = QtCore.pyqtSignal() dataReady = QtCore.pyqtSignal(list, dict) @QtCore.pyqtSlot() def processA(self): print "Worker.processA()" self.finished.emit() @QtCore.pyqtSlot(str, list, list) def processB(self, foo, bar=None, baz=None): print "Worker.processB()" for thing in bar: # lots of processing... self.dataReady.emit(['dummy', 'data'], {'dummy': ['data']}) self.finished.emit() def onDataReady(aList, aDict): print 'onDataReady' print repr(aList) print repr(aDict) app = QtGui.QApplication(sys.argv) thread = QtCore.QThread() # no parent! obj = Worker() # no parent! obj.dataReady.connect(onDataReady) obj.moveToThread(thread) # if you want the thread to stop after the worker is done # you can always call thread.start() again later obj.finished.connect(thread.quit) # one way to do it is to start processing as soon as the thread starts # this is okay in some cases... but makes it harder to send data to # the worker object from the main gui thread. As you can see I'm calling # processA() which takes no arguments thread.started.connect(obj.processA) thread.finished.connect(app.exit) thread.start() # another way to do it, which is a bit fancier, allows you to talk back and # forth with the object in a thread safe way by communicating through signals # and slots (now that the thread is running I can start calling methods on # the worker object) QtCore.QMetaObject.invokeMethod(obj, 'processB', Qt.QueuedConnection, QtCore.Q_ARG(str, "Hello World!"), QtCore.Q_ARG(list, ["args", 0, 1]), QtCore.Q_ARG(list, [])) # that looks a bit scary, but its a totally ok thing to do in Qt, # we're simply using the system that Signals and Slots are built on top of, # the QMetaObject, to make it act like we safely emitted a signal for # the worker thread to pick up when its event loop resumes (so if its doing # a bunch of work you can call this method 10 times and it will just queue # up the calls. Note: PyQt > 4.6 will not allow you to pass in a None # instead of an empty list, it has stricter type checking app.exec_() 

Answer 4

According to the Qt developers, the subclassification of QThread is incorrect (see http://blog.qt.digia.com/blog/2010/06/17/youre-doing-it-wrong/ ). But this article is very difficult to understand (plus the title is a little lenient). I found a better blog post that gives a more detailed explanation of why you should use one streaming style over another: http://mayaposch.wordpress.com/2011/11/01/how-to-really-truly-use -qthreads-the-full-explanation /

In my opinion, you probably should never subclass a thread with the intention of overloading the run method. Although this really works, you basically get around how Qt wants you to work. In addition, you will miss things like events and the right stream-safe signals and slots. Plus, as you are likely to see in the aforementioned blog post, the “right” streaming method forces you to write more testable code.

Here are some examples of how to use QThreads in PyQt (I posted a separate answer below that uses QRunnable correctly and includes signals / slots, this answer is better if you have many asynchronous tasks that you need for load balancing).

 import sys from PyQt4 import QtCore from PyQt4 import QtGui from PyQt4.QtCore import Qt # very testable class (hint: you can use mock.Mock for the signals) class Worker(QtCore.QObject): finished = QtCore.pyqtSignal() dataReady = QtCore.pyqtSignal(list, dict) @QtCore.pyqtSlot() def processA(self): print "Worker.processA()" self.finished.emit() @QtCore.pyqtSlot(str, list, list) def processB(self, foo, bar=None, baz=None): print "Worker.processB()" for thing in bar: # lots of processing... self.dataReady.emit(['dummy', 'data'], {'dummy': ['data']}) self.finished.emit() class Thread(QtCore.QThread): """Need for PyQt4 <= 4.6 only""" def __init__(self, parent=None): QtCore.QThread.__init__(self, parent) # this class is solely needed for these two methods, there # appears to be a bug in PyQt 4.6 that requires you to # explicitly call run and start from the subclass in order # to get the thread to actually start an event loop def start(self): QtCore.QThread.start(self) def run(self): QtCore.QThread.run(self) app = QtGui.QApplication(sys.argv) thread = Thread() # no parent! obj = Worker() # no parent! obj.moveToThread(thread) # if you want the thread to stop after the worker is done # you can always call thread.start() again later obj.finished.connect(thread.quit) # one way to do it is to start processing as soon as the thread starts # this is okay in some cases... but makes it harder to send data to # the worker object from the main gui thread. As you can see I'm calling # processA() which takes no arguments thread.started.connect(obj.processA) thread.start() # another way to do it, which is a bit fancier, allows you to talk back and # forth with the object in a thread safe way by communicating through signals # and slots (now that the thread is running I can start calling methods on # the worker object) QtCore.QMetaObject.invokeMethod(obj, 'processB', Qt.QueuedConnection, QtCore.Q_ARG(str, "Hello World!"), QtCore.Q_ARG(list, ["args", 0, 1]), QtCore.Q_ARG(list, [])) # that looks a bit scary, but its a totally ok thing to do in Qt, # we're simply using the system that Signals and Slots are built on top of, # the QMetaObject, to make it act like we safely emitted a signal for # the worker thread to pick up when its event loop resumes (so if its doing # a bunch of work you can call this method 10 times and it will just queue # up the calls. Note: PyQt > 4.6 will not allow you to pass in a None # instead of an empty list, it has stricter type checking app.exec_() # Without this you may get weird QThread messages in the shell on exit app.deleteLater() 

Answer 5

PyQt has many options for obtaining asynchronous behavior. For things that require event handling (e.g. QtNetwork, etc.), you should use the QThread example, which I presented in my other answer to this thread. But for the vast majority of your flow needs, I think this solution is far superior to other methods.

The advantage of this is that QThreadPool schedules your QRunnable instances as tasks. This is similar to the task pattern used in Intel TBB. It is not as elegant as I like, but it is great for asynchronous behavior.

This allows you to use most of the Qt streaming power in Python using QRunnable and still use signals and slots. I use the same code in several applications, some of which are hundreds of asynchronous REST calls, some of which open files or directory directories, and the best part uses this method, the Qt task balances system resources for me.

 import time from PyQt4 import QtCore from PyQt4 import QtGui from PyQt4.QtCore import Qt def async(method, args, uid, readycb, errorcb=None): """ Asynchronously runs a task :param func method: the method to run in a thread :param object uid: a unique identifier for this task (used for verification) :param slot updatecb: the callback when data is receieved cb(uid, data) :param slot errorcb: the callback when there is an error cb(uid, errmsg) The uid option is useful when the calling code makes multiple async calls and the callbacks need some context about what was sent to the async method. For example, if you use this method to thread a long running database call and the user decides they want to cancel it and start a different one, the first one may complete before you have a chance to cancel the task. In that case, the "readycb" will be called with the cancelled task data. The uid can be used to differentiate those two calls (ie. using the sql query). :returns: Request instance """ request = Request(method, args, uid, readycb, errorcb) QtCore.QThreadPool.globalInstance().start(request) return request class Request(QtCore.QRunnable): """ A Qt object that represents an asynchronous task :param func method: the method to call :param list args: list of arguments to pass to method :param object uid: a unique identifier (used for verification) :param slot readycb: the callback used when data is receieved :param slot errorcb: the callback used when there is an error The uid param is sent to your error and update callbacks as the first argument. It there to verify the data you're returning After created it should be used by invoking: .. code-block:: python task = Request(...) QtCore.QThreadPool.globalInstance().start(task) """ INSTANCES = [] FINISHED = [] def __init__(self, method, args, uid, readycb, errorcb=None): super(Request, self).__init__() self.setAutoDelete(True) self.cancelled = False self.method = method self.args = args self.uid = uid self.dataReady = readycb self.dataError = errorcb Request.INSTANCES.append(self) # release all of the finished tasks Request.FINISHED = [] def run(self): """ Method automatically called by Qt when the runnable is ready to run. This will run in a separate thread. """ # this allows us to "cancel" queued tasks if needed, should be done # on shutdown to prevent the app from hanging if self.cancelled: self.cleanup() return # runs in a separate thread, for proper async signal/slot behavior # the object that emits the signals must be created in this thread. # Its not possible to run grabber.moveToThread(QThread.currentThread()) # so to get this QObject to properly exhibit asynchronous # signal and slot behavior it needs to live in the thread that # we're running in, creating the object from within this thread # is an easy way to do that. grabber = Requester() grabber.Loaded.connect(self.dataReady, Qt.QueuedConnection) if self.dataError is not None: grabber.Error.connect(self.dataError, Qt.QueuedConnection) try: result = self.method(*self.args) if self.cancelled: # cleanup happens in 'finally' statement return grabber.Loaded.emit(self.uid, result) except Exception as error: if self.cancelled: # cleanup happens in 'finally' statement return grabber.Error.emit(self.uid, unicode(error)) finally: # this will run even if one of the above return statements # is executed inside of the try/except statement see: # https://docs.python.org/2.7/tutorial/errors.html#defining-clean-up-actions self.cleanup(grabber) def cleanup(self, grabber=None): # remove references to any object or method for proper ref counting self.method = None self.args = None self.uid = None self.dataReady = None self.dataError = None if grabber is not None: grabber.deleteLater() # make sure this python obj gets cleaned up self.remove() def remove(self): try: Request.INSTANCES.remove(self) # when the next request is created, it will clean this one up # this will help us avoid this object being cleaned up # when it still being used Request.FINISHED.append(self) except ValueError: # there might be a race condition on shutdown, when shutdown() # is called while the thread is still running and the instance # has already been removed from the list return @staticmethod def shutdown(): for inst in Request.INSTANCES: inst.cancelled = True Request.INSTANCES = [] Request.FINISHED = [] class Requester(QtCore.QObject): """ A simple object designed to be used in a separate thread to allow for asynchronous data fetching """ # # Signals # Error = QtCore.pyqtSignal(object, unicode) """ Emitted if the fetch fails for any reason :param unicode uid: an id to identify this request :param unicode error: the error message """ Loaded = QtCore.pyqtSignal(object, object) """ Emitted whenever data comes back successfully :param unicode uid: an id to identify this request :param list data: the json list returned from the GET """ NetworkConnectionError = QtCore.pyqtSignal(unicode) """ Emitted when the task fails due to a network connection error :param unicode message: network connection error message """ def __init__(self, parent=None): super(Requester, self).__init__(parent) class ExampleObject(QtCore.QObject): def __init__(self, parent=None): super(ExampleObject, self).__init__(parent) self.uid = 0 self.request = None def ready_callback(self, uid, result): if uid != self.uid: return print "Data ready from %s: %s" % (uid, result) def error_callback(self, uid, error): if uid != self.uid: return print "Data error from %s: %s" % (uid, error) def fetch(self): if self.request is not None: # cancel any pending requests self.request.cancelled = True self.request = None self.uid += 1 self.request = async(slow_method, ["arg1", "arg2"], self.uid, self.ready_callback, self.error_callback) def slow_method(arg1, arg2): print "Starting slow method" time.sleep(1) return arg1 + arg2 if __name__ == "__main__": import sys app = QtGui.QApplication(sys.argv) obj = ExampleObject() dialog = QtGui.QDialog() layout = QtGui.QVBoxLayout(dialog) button = QtGui.QPushButton("Generate", dialog) progress = QtGui.QProgressBar(dialog) progress.setRange(0, 0) layout.addWidget(button) layout.addWidget(progress) button.clicked.connect(obj.fetch) dialog.show() app.exec_() app.deleteLater() # avoids some QThread messages in the shell on exit # cancel all running tasks avoid QThread/QTimer error messages # on exit Request.shutdown() 

When you exit the application, you want you to cancel all tasks or the application freezes before the completion of each scheduled task.

Answer 6

Based on the methods of the Worker objects mentioned in other answers, I decided to see if I could expand the solution to call more threads - in this case, the optimal number that the machine can start and deploy several employees with an indefinite completion time. To do this, I still need to subclass QThread, but only in order to assign a thread number and "redefine" the signals "finished" and "started" to include their thread number.

I focused a bit on the signals between the main gui, threads, and workers.

Similarly, other answers were aimed at not raising QThread, but I do not think this is a serious problem. However, my code also carefully destroys QThread objects.

However, I could not get the parent worker objects, so it seems advisable to send them the deleteLater () signal either when the stream function is finished or the GUI is destroyed. I had my own code to not do this.

Another improvement that I thought was necessary was to override the closeEvent GUI (QWidget) so that the threads were instructed to exit, and then the GUI would wait for all threads to complete. When I played with some other answers to this question, I had errors with a QThread error.

It may be useful to others. I definitely found this a useful exercise. Perhaps others will know the best way for the stream to report this.

 #!/usr/bin/env python3 #coding:utf-8 # Author: --<> # Purpose: To demonstrate creation of multiple threads and identify the receipt of thread results # Created: 19/12/15 import sys from PyQt4.QtCore import QThread, pyqtSlot, pyqtSignal from PyQt4.QtGui import QApplication, QLabel, QWidget, QGridLayout import sys import worker class Thread(QThread): #make new signals to be able to return an id for the thread startedx = pyqtSignal(int) finishedx = pyqtSignal(int) def __init__(self,i,parent=None): super().__init__(parent) self.idd = i self.started.connect(self.starttt) self.finished.connect(self.finisheddd) @pyqtSlot() def starttt(self): print('started signal from thread emitted') self.startedx.emit(self.idd) @pyqtSlot() def finisheddd(self): print('finished signal from thread emitted') self.finishedx.emit(self.idd) class Form(QWidget): def __init__(self): super().__init__() self.initUI() self.worker={} self.threadx={} self.i=0 i=0 #Establish the maximum number of threads the machine can optimally handle #Generally relates to the number of processors self.threadtest = QThread(self) self.idealthreadcount = self.threadtest.idealThreadCount() print("This machine can handle {} threads optimally".format(self.idealthreadcount)) while i <self.idealthreadcount: self.setupThread(i) i+=1 i=0 while i<self.idealthreadcount: self.startThread(i) i+=1 print("Main Gui running in thread {}.".format(self.thread())) def setupThread(self,i): self.worker[i]= worker.Worker(i) # no parent! #print("Worker object runningt in thread {} prior to movetothread".format(self.worker[i].thread()) ) self.threadx[i] = Thread(i,parent=self) # if parent isn't specified then need to be careful to destroy thread self.threadx[i].setObjectName("python thread{}"+str(i)) #print("Thread object runningt in thread {} prior to movetothread".format(self.threadx[i].thread()) ) self.threadx[i].startedx.connect(self.threadStarted) self.threadx[i].finishedx.connect(self.threadFinished) self.worker[i].finished.connect(self.workerFinished) self.worker[i].intReady.connect(self.workerResultReady) #The next line is optional, you may want to start the threads again without having to create all the code again. self.worker[i].finished.connect(self.threadx[i].quit) self.threadx[i].started.connect(self.worker[i].procCounter) self.destroyed.connect(self.threadx[i].deleteLater) self.destroyed.connect(self.worker[i].deleteLater) #This is the key code that actually get the worker code onto another processor or thread. self.worker[i].moveToThread(self.threadx[i]) def startThread(self,i): self.threadx[i].start() @pyqtSlot(int) def threadStarted(self,i): print('Thread {} started'.format(i)) print("Thread priority is {}".format(self.threadx[i].priority())) @pyqtSlot(int) def threadFinished(self,i): print('Thread {} finished'.format(i)) @pyqtSlot(int) def threadTerminated(self,i): print("Thread {} terminated".format(i)) @pyqtSlot(int,int) def workerResultReady(self,j,i): print('Worker {} result returned'.format(i)) if i ==0: self.label1.setText("{}".format(j)) if i ==1: self.label2.setText("{}".format(j)) if i ==2: self.label3.setText("{}".format(j)) if i ==3: self.label4.setText("{}".format(j)) #print('Thread {} has started'.format(self.threadx[i].currentThreadId())) @pyqtSlot(int) def workerFinished(self,i): print('Worker {} finished'.format(i)) def initUI(self): self.label1 = QLabel("0") self.label2= QLabel("0") self.label3= QLabel("0") self.label4 = QLabel("0") grid = QGridLayout(self) self.setLayout(grid) grid.addWidget(self.label1,0,0) grid.addWidget(self.label2,0,1) grid.addWidget(self.label3,0,2) grid.addWidget(self.label4,0,3) #Layout parents the self.labels self.move(300, 150) self.setGeometry(0,0,300,300) #self.size(300,300) self.setWindowTitle('thread test') self.show() def closeEvent(self, event): print('Closing') #this tells the threads to stop running i=0 while i <self.idealthreadcount: self.threadx[i].quit() i+=1 #this ensures window cannot be closed until the threads have finished. i=0 while i <self.idealthreadcount: self.threadx[i].wait() i+=1 event.accept() if __name__=='__main__': app = QApplication(sys.argv) form = Form() sys.exit(app.exec_()) 

And the working code below

 #!/usr/bin/env python3 #coding:utf-8 # Author: --<> # Purpose: Stack Overflow # Created: 19/12/15 import sys import unittest from PyQt4.QtCore import QThread, QObject, pyqtSignal, pyqtSlot import time import random class Worker(QObject): finished = pyqtSignal(int) intReady = pyqtSignal(int,int) def __init__(self, i=0): '''__init__ is called while the worker is still in the Gui thread. Do not put slow or CPU intensive code in the __init__ method''' super().__init__() self.idd = i @pyqtSlot() def procCounter(self): # This slot takes no params for j in range(1, 10): random_time = random.weibullvariate(1,2) time.sleep(random_time) self.intReady.emit(j,self.idd) print('Worker {0} in thread {1}'.format(self.idd, self.thread().idd)) self.finished.emit(self.idd) if __name__=='__main__': unittest.main()