Pass multiple models to QListView

Heads up! You've already completed this tutorial.

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.

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:


Over 10,000 developers have bought Create GUI Applications with Python & Qt!

To support developers in [[ countryRegion ]] I give a [[ localizedDiscount[couponCode] ]]% discount with the code [[ couponCode ]] — Enjoy!

For [[ activeDiscount.description ]] I'm giving a [[ activeDiscount.discount ]]% discount with the code [[ couponCode ]] — Enjoy!

Well done, you've finished this tutorial! Mark As Complete
[[ user.completed.length ]] completed [[ user.streak+1 ]] day streak