Front-end and Back-end Separation in PyQt5 with QML

Question

I want to add QML UI to an existed Python threading task, but I do not know how to do without changing the task too much in order to keep front- and back- end separated.

Let me explain the question with a minimized example. I want to add GUI with QML to control a process (or thread) start and stop as well as get and show the information from the process. For instance, the following is the process with some heavy works.

class Task(threading.Thread):
    def __init__(self):
        super().__init__()
        self.num = 0

    def run(self):
        for i in range(35):
            self.num = fib(i)
def fib(N):
    if N <= 1:
        return 1
    return fib(N - 1) + fib(N - 2)

Now, I try to create a QML file to get the self.num in Task after Task().start() without blocking.

import QtQuick 2.12
import QtQuick.Controls 2.12
ApplicationWindow {
    visible: true
    Column {
        Button {
            text: 'hello'
            onClicked: {
                backend.start_process()
            }
        }
        Text {
            text: backend.num
        }
    }
}

In order to keep front- and back-end separated, is there any way to connect the Task to QML UI without changing any script content in Task? According to this question, I have tried to create a Backend class handling any communication between front- and back-end and set the context property into QML (engine.rootContext().setContextProperty("backend", Backend())). For example,

class Backend(PyQt5.QtCore.QObject):
    changed = PyQt5.QtCore.pyqtSignal(int)

    def __init__(self):
        super().__init__()
        self._num = 0

    @PyQt5.QtCore.pyqtProperty(int, notify=changed)
    def num(self):
        return self._num

    @num.setter
    def num(self, val):
        if self._num == val:
            return
        self._num = val
        self.changed.emit(self._num)

    @PyQt5.QtCore.pyqtSlot()
    def start_process(self):
        t = Task()
        t.start()

However, I have no idea how to bidirectional binding the self.num and t.num in backend. Thus, I could not real-time update the t.num after start_process() has been called by the button in QML UI.

I want to minimize the modification of Task as it is the core functionality of an old program. Therefore, the Simple Example in PyQt5 website could not meet my requirement.

Furthermore, if I want to change Python threading.Thread to multiprocessing.Process, what should I do to get the variable in the process every time it updates in order to show on QML UI?

Answer 1

First of all, you do not need a setter like qproperty since according to your logic you should not modify the qproperty from QML. On the other hand you must pass the backend to your thread through args or kwargs and send the information invoking a slot:

import threading
import time
from PyQt5 import QtCore, QtGui, QtQml

class Task(threading.Thread):
    def run(self):
        self.backend = self._kwargs.get('backend')
        for i in range(35):
            num = fib(i)
            if self.backend is not None:
                QtCore.QMetaObject.invokeMethod(
                    self.backend, 
                    "set_num",
                    QtCore.Qt.QueuedConnection, 
                    QtCore.Q_ARG(int, num)
                )
                time.sleep(0.01)
def fib(N):
    if N <= 1:
        return 1
    return fib(N - 1) + fib(N - 2)

class Backend(QtCore.QObject):
    changed = QtCore.pyqtSignal(int)

    def __init__(self):
        super().__init__()
        self._num = 0

    @QtCore.pyqtProperty(int, notify=changed)
    def num(self):
        return self._num

    @QtCore.pyqtSlot(int)
    def set_num(self, val):
        if self._num == val:
            return
        self._num = val
        self.changed.emit(self._num)

    @QtCore.pyqtSlot()
    def start_process(self):
        Task(kwargs=dict(backend=self), daemon=True).start()

if __name__ == "__main__":
    import os
    import sys

    app = QtGui.QGuiApplication(sys.argv)
    backend = Backend()
    engine = QtQml.QQmlApplicationEngine()
    engine.rootContext().setContextProperty("backend", backend)
    qml_path = os.path.join(os.path.dirname(__file__), "main.qml")
    engine.load(QtCore.QUrl.fromLocalFile(qml_path))
    if not engine.rootObjects():
        sys.exit(-1)
    engine.quit.connect(app.quit)
    sys.exit(app.exec_())

---

Another way is to create a QEvent that has the information:

import threading
import time
from PyQt5 import QtCore, QtGui, QtQml

class InfoEvent(QtCore.QEvent):
    EventType = QtCore.QEvent.User

    def __init__(self, info):
        super(InfoEvent, self).__init__(InfoEvent.EventType)
        self._info = info

    @property
    def info(self):
        return self._info

class Task(threading.Thread):
    def run(self):
        self.backend = self._kwargs.get('backend')
        for i in range(35):
            num = fib(i)
            if self.backend is not None:
                event = InfoEvent(num)
                QtCore.QCoreApplication.sendEvent(self.backend, event)
                time.sleep(0.01)
def fib(N):
    if N <= 1:
        return 1
    return fib(N - 1) + fib(N - 2)

class Backend(QtCore.QObject):
    changed = QtCore.pyqtSignal(int)

    def __init__(self):
        super().__init__()
        self._num = 0

    def event(self, e):
        if e.type() == InfoEvent.EventType:
            self.set_num(e.info)
            return True
        return super(Backend, self).event(e)

    @QtCore.pyqtProperty(int, notify=changed)
    def num(self):
        return self._num

    @QtCore.pyqtSlot(int)
    def set_num(self, val):
        if self._num == val:
            return
        self._num = val
        self.changed.emit(self._num)

    @QtCore.pyqtSlot()
    def start_process(self):
        Task(kwargs=dict(backend=self), daemon=True).start()

if __name__ == "__main__":
    import os
    import sys

    app = QtGui.QGuiApplication(sys.argv)
    backend = Backend()
    engine = QtQml.QQmlApplicationEngine()
    engine.rootContext().setContextProperty("backend", backend)
    qml_path = os.path.join(os.path.dirname(__file__), "main.qml")
    engine.load(QtCore.QUrl.fromLocalFile(qml_path))
    if not engine.rootObjects():
        sys.exit(-1)
    engine.quit.connect(app.quit)
    sys.exit(app.exec_())

---

Another solution is that it no longer inherits from threading.Thread but from QThread that implements some signals that send information to the main thread.

from PyQt5 import QtCore, QtGui, QtQml

class Task(QtCore.QThread):
    valueChanged = QtCore.pyqtSignal(int)

    def run(self):
        for i in range(35):
            num = fib(i)
            self.valueChanged.emit(num)
            QtCore.QThread.msleep(1)

def fib(N):
    if N <= 1:
        return 1
    return fib(N - 1) + fib(N - 2)

class Backend(QtCore.QObject):
    changed = QtCore.pyqtSignal(int)

    def __init__(self):
        super().__init__()
        self._num = 0

    @QtCore.pyqtProperty(int, notify=changed)
    def num(self):
        return self._num

    @QtCore.pyqtSlot(int)
    def set_num(self, val):
        if self._num == val:
            return
        self._num = val
        self.changed.emit(self._num)

    @QtCore.pyqtSlot()
    def start_process(self):
        thread = Task(self)
        thread.valueChanged.connect(self.set_num)
        thread.start()

if __name__ == "__main__":
    import os
    import sys

    app = QtGui.QGuiApplication(sys.argv)
    backend = Backend()
    engine = QtQml.QQmlApplicationEngine()
    engine.rootContext().setContextProperty("backend", backend)
    qml_path = os.path.join(os.path.dirname(__file__), "main.qml")
    engine.load(QtCore.QUrl.fromLocalFile(qml_path))
    if not engine.rootObjects():
        sys.exit(-1)
    engine.quit.connect(app.quit)
    sys.exit(app.exec_())