Passing a function to a Worker (QObject) class in Python GUI application to prevent freezing/blocking

I am trying to find a way to successfully pass a function to a Worker class in Python using PyQT5. Instead of using the pre-defined run function (or Long-running task) in the sample Worker class code, I would like to be able to pass a custom function to the worker class. Below I’ve pasted the sample code I’m working with, followed by an adjustment I’ve tried.

from time import sleep
from PyQt5.QtCore import QObject, QThread, pyqtSignal,Qt
from PyQt5.QtWidgets import QApplication,QMainWindow,QLabel,QPushButton,QVBoxLayout,QWidget
import sys

# Step 1: Create a worker class
class Worker(QObject):
    finished = pyqtSignal()
    progress = pyqtSignal(int)
    
    def run(self):
        """Long-running task."""
        for i in range(5):
            sleep(1)
            self.progress.emit(i + 1)
        self.finished.emit()

class Window(QMainWindow):
    def __init__(self, parent=None):
        super().__init__(parent)
        self.clicksCount = 0
        self.setupUi()

    def setupUi(self):
        self.setWindowTitle("Freezing GUI")
        self.resize(300, 150)
        self.centralWidget = QWidget()
        self.setCentralWidget(self.centralWidget)
        # Create and connect widgets
        self.clicksLabel = QLabel("Counting: 0 clicks", self)
        self.clicksLabel.setAlignment(Qt.AlignHCenter | Qt.AlignVCenter)
        self.stepLabel = QLabel("Long-Running Step: 0")
        self.stepLabel.setAlignment(Qt.AlignHCenter | Qt.AlignVCenter)
        self.countBtn = QPushButton("Click me!", self)
        self.countBtn.clicked.connect(self.countClicks)
        self.longRunningBtn = QPushButton("Long-Running Task!", self)
        self.longRunningBtn.clicked.connect(self.runLongTask)
        # Set the layout
        layout = QVBoxLayout()
        layout.addWidget(self.clicksLabel)
        layout.addWidget(self.countBtn)
        layout.addStretch()
        layout.addWidget(self.stepLabel)
        layout.addWidget(self.longRunningBtn)
        self.centralWidget.setLayout(layout)

    def countClicks(self):
        self.clicksCount += 1
        self.clicksLabel.setText(f"Counting: {self.clicksCount} clicks")

    def reportProgress(self, n):
        self.stepLabel.setText(f"Long-Running Step: {n}")

    def runLongTask(self):
        # Step 2: Create a QThread object
        self.thread = QThread()
        # Step 3: Create a worker object
        self.worker = Worker()
        # Step 4: Move worker to the thread
        self.worker.moveToThread(self.thread)
        # Step 5: Connect signals and slots
        self.thread.started.connect(self.worker.run)
        self.worker.finished.connect(self.thread.quit)
        self.worker.finished.connect(self.worker.deleteLater)
        self.thread.finished.connect(self.thread.deleteLater)
        self.worker.progress.connect(self.reportProgress)
        # Step 6: Start the thread
        self.thread.start()

        # Final resets
        self.longRunningBtn.setEnabled(False)
        self.thread.finished.connect(
            lambda: self.longRunningBtn.setEnabled(True)
        )
        self.thread.finished.connect(
            lambda: self.stepLabel.setText("Long-Running Step: 0")
        )

app = QApplication(sys.argv)
win = Window()
win.show()
sys.exit(app.exec())

class Worker(QObject):
    finished = pyqtSignal(str)
    
    def __init__(self, *init_args, **init_kwargs):
        QObject.__init__(self, *init_args, **init_kwargs)
        self._return = None

    def run(self):
        """Long-running task."""
        self._return = self._target(*self._args, **self._kwargs)
        self.finished.emit(self._return)

Answer

As one of solutions, you can pass a function and an argument to the Worker’s __init__ method, like:

class Worker(QObject):
    finished = pyqtSignal()
    progress = pyqtSignal(int)
    result = pyqtSignal('QVariant')

    def __init__(self, function, args):
        super().__init__()
        self.function = function
        self.args = args

    def run(self):
        res = self.function(self.args)
        self.result.emit(res)
        self.finished.emit()

So you can pass a method and an argument when creating a worker and connect the method to handle the result:

        self.thread = QThread()
        self.worker = Worker(Proxy.GetInfo, code)
        self.worker.moveToThread(self.thread)
        self.thread.started.connect(self.worker.run)
        self.worker.finished.connect(self.thread.quit)
        self.worker.finished.connect(self.thread.deleteLater)
        self.worker.result.connect(self.receiveResult)
        self.thread.finished.connect(self.thread.deleteLater)
        self.thread.start()

Advertisement

Answer