Background thread with QThread in PyQt
I created a little example that shows 3 different and simple ways of dealing with threads. I hope it will help you find the right approach to your problem.
import sysimport timefrom PyQt5.QtCore import (QCoreApplication, QObject, QRunnable, QThread, QThreadPool, pyqtSignal)# Subclassing QThread# http://qt-project.org/doc/latest/qthread.htmlclass 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-abstractclass 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()
Take this answer updated for PyQt5, python 3.4
Use this as a pattern to start a worker that does not take data and return data as they are available to the form.
1 - Worker class is made smaller and put in its own file worker.py for easy memorization and independent software reuse.
2 - The main.py file is the file that defines the GUI Form class
3 - The thread object is not subclassed.
4 - Both thread object and the worker object belong to the Form object
5 - Steps of the procedure are within the comments.
# worker.pyfrom PyQt5.QtCore import QThread, QObject, pyqtSignal, pyqtSlotimport timeclass 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 is:
# 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_())
Very nice example from Matt, I fixed the typo and also pyqt4.8 is common now so I removed the dummy class as well and added an example for the dataReady signal
# -*- coding: utf-8 -*-import sysfrom PyQt4 import QtCore, QtGuifrom 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 laterobj.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 argumentsthread.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 checkingapp.exec_()