Skip to content

Instantly share code, notes, and snippets.

@aminechraibi
Created March 7, 2023 15:06
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save aminechraibi/d50d3e356e2d1a4a6b6965e06a9e7381 to your computer and use it in GitHub Desktop.
Save aminechraibi/d50d3e356e2d1a4a6b6965e06a9e7381 to your computer and use it in GitHub Desktop.
A simple example of a class you can use for running a background process in PyQt5
from PyQt5.QtCore import QObject, QRunnable, QThreadPool, pyqtSignal, pyqtSlot
class WorkerSignals(QObject):
finished = pyqtSignal()
result = pyqtSignal(object)
on_process = pyqtSignal(object)
class Worker(QRunnable):
def __init__(self, fn, *args, **kwargs):
super(Worker, self).__init__()
self.fn = fn
self.args = args
self.kwargs = kwargs
self.signals = WorkerSignals()
@pyqtSlot()
def run(self):
try:
result = self.fn(*self.args, **self.kwargs)
except Exception as e:
result = None
print(f"Exception in worker thread: {e}")
self.signals.result.emit(result)
self.signals.finished.emit()
class ThreadPool:
def __init__(self):
self.pool = QThreadPool.globalInstance()
def start(self, fn, callback=None, on_process=None, *args, **kwargs):
self.worker = Worker(fn, *args, **kwargs)
if callback:
self.worker.signals.result.connect(callback)
if on_process:
self.worker.signals.on_process.connect(on_process)
self.pool.start(self.worker)
if __name__ == '__main__':
"""
A simple example of using ThreadPool class
"""
from PyQt5.QtWidgets import QApplication, QMainWindow, QPushButton, QLabel, QVBoxLayout, QWidget, QProgressBar
from datetime import datetime
import time
app = QApplication([])
class MainWindow(QMainWindow):
def __init__(self):
super().__init__()
self.button = QPushButton("Start", self)
self.button.clicked.connect(self.start_thread)
self.label = QLabel("Waiting...", self)
self.progress = QProgressBar()
self.progress.setMaximum(5)
layout = QVBoxLayout()
layout.addWidget(self.button)
layout.addWidget(self.label)
layout.addWidget(self.progress)
widget = QWidget()
widget.setLayout(layout)
self.setCentralWidget(widget)
self.thread_pool = ThreadPool()
def start_thread(self):
self.button.setEnabled(False)
self.label.setText("Running...")
self.thread_pool.start(self.long_running_task, self.task_complete, self.task_on_process)
def task_on_process(self, result):
self.progress.setValue(result)
def long_running_task(self):
print("Starting long running task...")
for i in range(6):
self.thread_pool.worker.signals.on_process.emit(i)
time.sleep(3)
print("Long running task complete.")
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def task_complete(self, result):
self.label.setText(f"Result: {result}")
self.button.setEnabled(True)
window = MainWindow()
window.show()
app.exec_()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment