Thomas_Parker | 2020-10-01 12:27:43 UTC | #1
I have 3 worker manager classes each containing their own threadpools and worker queues.
I am using the worker manager 6 example. However when setting progress.setModel I can only pass one worker manager.
How do I get both worker managers to be in one QListView?
Thank You
martin | 2020-10-02 19:19:53 UTC | #2
Hi @Thomas_Parker welcome to the forum.
PyQt6 Crash Course — a new tutorial in your Inbox every day
Beginner-focused crash course explaining the basics with hands-on examples.
You can't actually set more than one model on a view, as a view can only pull data from one source at a time. However, you can already achieve what you want with the current manager implementation.
If you look at the definition of the WorkerManager you'll see it has two variables _workers and _state. These are defined as class variables (defined on the class definition rather than in the __init__ -- which means they are shared among all instances of the class.
python
class WorkerManager(QAbstractListModel):
"""
Manager to handle our worker queues and state.
Also functions as a Qt data model for a view
displaying progress for each worker.
"""
_workers = {}
_state = {}
That's a bit weird to be honest, and I should probably change it. But it does mean that you can get what you want simply by defining two WorkerManager instances and passing one of them as the model.
Since they share their internal state, either will report the full state for both.
Here's the MainWindow definition to achieve this --
python
class MainWindow(QMainWindow):
def __init__(self):
super().__init__()
self.workers1 = WorkerManager()
self.workers2 = WorkerManager()
self.workers1.status.connect(self.statusBar().showMessage)
self.workers2.status.connect(self.statusBar().showMessage)
layout = QVBoxLayout()
self.progress = QListView()
self.progress.setModel(self.workers1)
delegate = ProgressBarDelegate()
self.progress.setItemDelegate(delegate)
layout.addWidget(self.progress)
self.text = QPlainTextEdit()
self.text.setReadOnly(True)
start1 = QPushButton("Start a worker 1")
start1.pressed.connect(self.start_worker1)
start2 = QPushButton("Start a worker 2")
start2.pressed.connect(self.start_worker2)
clear = QPushButton("Clear")
clear.pressed.connect(self.workers1.cleanup)
clear.pressed.connect(self.workers2.cleanup)
layout.addWidget(self.text)
layout.addWidget(start1)
layout.addWidget(start2)
layout.addWidget(clear)
w = QWidget()
w.setLayout(layout)
self.setCentralWidget(w)
self.show()
# tag::startWorker[]
def start_worker1(self):
x = random.randint(0, 1000)
y = random.randint(0, 1000)
w = Worker(x, y)
w.signals.result.connect(self.display_result)
w.signals.error.connect(self.display_result)
self.workers1.enqueue(w)
def start_worker2(self):
x = random.randint(0, 1000)
y = random.randint(0, 1000)
w = Worker(x, y)
w.signals.result.connect(self.display_result)
w.signals.error.connect(self.display_result)
self.workers2.enqueue(w)
# end::startWorker[]
def display_result(self, job_id, data):
self.text.appendPlainText("WORKER %s: %s" % (job_id, data))
If you click either the start1/start2 buttons you'll see the workers appear in the list.
But yeah, as I say, that's a bit weird. So what we should do instead is separate the model from the manager, and share this separate model explicitly with both managers. Below is a full working example to do that.
The key changes are in the definition of the WorkerModel which holes everything relating to _state. The signals/etc. are modified to connect to the model directly. On the MainWindow we create a model instance, and then pass it into each WorkerManager so the state is shared between them -- jobs for both are shown in the progress list.
python
import random
import subprocess
import sys
import time
import traceback
import uuid
from PyQt5.QtCore import (
QAbstractListModel, QObject, QRect, QRunnable, Qt, QThreadPool, QTimer,
pyqtSignal, pyqtSlot,
)
from PyQt5.QtGui import QBrush, QColor, QPen
from PyQt5.QtWidgets import (
QApplication, QListView, QMainWindow, QPlainTextEdit, QProgressBar,
QPushButton, QStyledItemDelegate, QVBoxLayout, QWidget,
)
STATUS_WAITING = "waiting"
STATUS_RUNNING = "running"
STATUS_ERROR = "error"
STATUS_COMPLETE = "complete"
STATUS_COLORS = {
STATUS_RUNNING: "#33a02c",
STATUS_ERROR: "#e31a1c",
STATUS_COMPLETE: "#b2df8a",
}
DEFAULT_STATE = {"progress": 0, "status": STATUS_WAITING}
class WorkerSignals(QObject):
"""
Defines the signals available from a running worker thread.
Supported signals are:
finished
No data
error
`tuple` (exctype, value, traceback.format_exc() )
result
`object` data returned from processing, anything
progress
`int` indicating % progress
"""
error = pyqtSignal(str, str)
result = pyqtSignal(str, object) # We can send anything back.
finished = pyqtSignal(str)
progress = pyqtSignal(str, int)
status = pyqtSignal(str, str)
class Worker(QRunnable):
"""
Worker thread
Inherits from QRunnable to handle worker thread setup, signals and wrap-up.
:param args: Arguments to pass for the worker
:param kwargs: Keywords to pass for the worker
"""
def __init__(self, *args, **kwargs):
super().__init__()
# Store constructor arguments (re-used for processing).
self.signals = WorkerSignals()
# Give this job a unique ID.
self.job_id = str(uuid.uuid4())
# The arguments for the worker
self.args = args
self.kwargs = kwargs
self.signals.status.emit(self.job_id, STATUS_WAITING)
@pyqtSlot()
def run(self):
"""
Initialize the runner function with passed args, kwargs.
"""
self.signals.status.emit(self.job_id, STATUS_RUNNING)
x, y = self.args
try:
value = random.randint(0, 100) * x
delay = random.random() / 10
result = []
for n in range(100):
# Generate some numbers.
value = value / y
y -= 1
# The following will sometimes throw a division by zero error.
result.append(value)
# Pass out the current progress.
self.signals.progress.emit(self.job_id, n + 1)
time.sleep(delay)
except Exception as e:
print(e)
# We swallow the error and continue.
self.signals.error.emit(self.job_id, str(e))
self.signals.status.emit(self.job_id, STATUS_ERROR)
else:
self.signals.result.emit(self.job_id, result)
self.signals.status.emit(self.job_id, STATUS_COMPLETE)
self.signals.finished.emit(self.job_id)
class WorkerModel(QAbstractListModel):
"""
Model holding the worker states (used by the managers)
to track progress.
"""
_state = {}
def add(self, identifier, data):
self._state[identifier] = data
self.layoutChanged.emit()
# Model interface
def data(self, index, role):
if role == Qt.DisplayRole:
# See below for the data structure.
job_ids = list(self._state.keys())
job_id = job_ids[index.row()]
return job_id, self._state[job_id]
def rowCount(self, index):
return len(self._state)
def receive_status(self, job_id, status):
self._state[job_id]["status"] = status
self.layoutChanged.emit()
def receive_progress(self, job_id, progress):
self._state[job_id]["progress"] = progress
self.layoutChanged.emit()
def cleanup(self):
"""
Remove any complete/failed workers from worker_state.
"""
for job_id, s in list(self._state.items()):
if s["status"] in (STATUS_COMPLETE, STATUS_ERROR):
del self._state[job_id]
self.layoutChanged.emit()
class WorkerManager(QObject):
"""
Manager to handle our worker queues and state.
"""
_workers = {}
status = pyqtSignal(str)
def __init__(self, model):
super().__init__()
# Store the model we'll use for tracking state.
self.model = model
# Create a threadpool for our workers.
self.threadpool = QThreadPool()
# self.threadpool.setMaxThreadCount(1)
self.max_threads = self.threadpool.maxThreadCount()
print("Multithreading with maximum %d threads" % self.max_threads)
self.status_timer = QTimer()
self.status_timer.setInterval(100)
self.status_timer.timeout.connect(self.notify_status)
self.status_timer.start()
def notify_status(self):
n_workers = len(self._workers)
running = min(n_workers, self.max_threads)
waiting = max(0, n_workers - self.max_threads)
self.status.emit(
"{} running, {} waiting, {} threads".format(
running, waiting, self.max_threads
)
)
def enqueue(self, worker):
"""
Enqueue a worker to run (at some point) by passing it to the QThreadPool.
"""
worker.signals.error.connect(self.receive_error)
worker.signals.status.connect(self.model.receive_status)
worker.signals.progress.connect(self.model.receive_progress)
worker.signals.finished.connect(self.done)
self.threadpool.start(worker)
self._workers[worker.job_id] = worker
# Set default status to waiting, 0 progress.
self.model.add(worker.job_id, DEFAULT_STATE.copy())
def receive_error(self, job_id, message):
print(job_id, message)
def done(self, job_id):
"""
Task/worker complete. Remove it from the active workers
dictionary. We leave it in worker_state, as this is used to
to display past/complete workers too.
"""
del self._workers[job_id]
class ProgressBarDelegate(QStyledItemDelegate):
def paint(self, painter, option, index):
# data is our status dict, containing progress, id, status
job_id, data = index.model().data(index, Qt.DisplayRole)
if data["progress"] > 0:
color = QColor(STATUS_COLORS[data["status"]])
brush = QBrush()
brush.setColor(color)
brush.setStyle(Qt.SolidPattern)
width = option.rect.width() * data["progress"] / 100
rect = QRect(option.rect) # Copy of the rect, so we can modify.
rect.setWidth(width)
painter.fillRect(rect, brush)
pen = QPen()
pen.setColor(Qt.black)
painter.drawText(option.rect, Qt.AlignLeft, job_id)
class MainWindow(QMainWindow):
def __init__(self):
super().__init__()
self.model = WorkerModel()
self.workers1 = WorkerManager(self.model)
self.workers2 = WorkerManager(self.model)
self.workers1.status.connect(self.statusBar().showMessage)
self.workers2.status.connect(self.statusBar().showMessage)
layout = QVBoxLayout()
self.progress = QListView()
self.progress.setModel(self.model)
delegate = ProgressBarDelegate()
self.progress.setItemDelegate(delegate)
layout.addWidget(self.progress)
self.text = QPlainTextEdit()
self.text.setReadOnly(True)
start1 = QPushButton("Start a worker 1")
start1.pressed.connect(self.start_worker1)
start2 = QPushButton("Start a worker 2")
start2.pressed.connect(self.start_worker2)
clear = QPushButton("Clear")
# We clear the model.
clear.pressed.connect(self.model.cleanup)
layout.addWidget(self.text)
layout.addWidget(start1)
layout.addWidget(start2)
layout.addWidget(clear)
w = QWidget()
w.setLayout(layout)
self.setCentralWidget(w)
self.show()
def start_worker1(self):
x = random.randint(0, 1000)
y = random.randint(0, 1000)
w = Worker(x, y)
w.signals.result.connect(self.display_result)
w.signals.error.connect(self.display_result)
self.workers1.enqueue(w)
def start_worker2(self):
x = random.randint(0, 1000)
y = random.randint(0, 1000)
w = Worker(x, y)
w.signals.result.connect(self.display_result)
w.signals.error.connect(self.display_result)
self.workers2.enqueue(w)
def display_result(self, job_id, data):
self.text.appendPlainText("WORKER %s: %s" % (job_id, data))
app = QApplication(sys.argv)
window = MainWindow()
app.exec_()
Hope that makes sense? Let me know if anything needs clarifying.
Thomas_Parker | 2020-10-05 08:39:07 UTC | #3
Awesome, can't thank you enough for taking the time to answer my questions complete with example code. thank you very much! :smiley:
The complete guide to packaging Python GUI applications with PyInstaller.
[[ discount.discount_pc ]]% OFF for
the next [[ discount.duration ]]
[[discount.description ]]
with the code [[ discount.coupon_code ]]
Purchasing Power Parity
Developers in [[ country ]] get [[ discount.discount_pc ]]% OFF on all books & courses with code [[ discount.coupon_code ]]
[[ user.completed.length ]] completed
[[ user.streak+1 ]] day streak
