PyQt5中前后端分离与QML
Front-end and Back-end Separation in PyQt5 with QML
我想将 QML UI 添加到现有的 Python 线程任务，但我不知道如何在不改变任务太多的情况下保持前端和后端分开了。
让我用一个最小化的例子来解释这个问题。我想用 QML 添加 GUI 来控制进程（或线程）的启动和停止以及获取和显示来自进程的信息。例如，下面是一些繁重工作的过程。
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)
现在，我尝试创建一个 QML 文件以在 Task().start() 之后无阻塞地获取 Task 中的 self.num。
import QtQuick 2.12
import QtQuick.Controls 2.12
ApplicationWindow {
visible: true
Column {
Button {
text: 'hello'
onClicked: {
backend.start_process()
}
}
Text {
text: backend.num
}
}
}
为了保持前后端分离，有没有办法在不改变Task中任何脚本内容的情况下将Task连接到QMLUI？根据 ，我尝试创建一个 Backend class 来处理前端和后端之间的任何通信，并将上下文 属性 设置为 QML（engine.rootContext().setContextProperty("backend", Backend())).例如，
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()
但是，我不知道如何双向绑定 backend 中的self.num 和t.num。因此，在 QML UI.
中的按钮调用 start_process() 后，我无法实时更新 t.num
我想尽量减少对 Task 的修改，因为它是旧程序的核心功能。因此，Simple Example in PyQt5 website无法满足我的要求。
另外，如果我想把Pythonthreading.Thread改成multiprocessing.Process，我应该怎么做才能在每次更新的过程中获取变量，以便在QML上显示UI?
首先，您不需要像 qproperty 这样的 setter，因为根据您的逻辑，您不应该修改 QML 中的 qproperty。另一方面，您必须通过 args 或 kwargs 将后端传递给您的线程，并发送调用插槽的信息：
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_())
另一种方法是创建一个包含以下信息的 QEvent：
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_())
另一种解决方案是它不再继承自threading.Thread，而是继承自QThread，它实现了一些向主线程发送信息的信号。
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_())
我想将 QML UI 添加到现有的 Python 线程任务，但我不知道如何在不改变任务太多的情况下保持前端和后端分开了。
让我用一个最小化的例子来解释这个问题。我想用 QML 添加 GUI 来控制进程（或线程）的启动和停止以及获取和显示来自进程的信息。例如，下面是一些繁重工作的过程。
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)
现在，我尝试创建一个 QML 文件以在 Task().start() 之后无阻塞地获取 Task 中的 self.num。
import QtQuick 2.12
import QtQuick.Controls 2.12
ApplicationWindow {
visible: true
Column {
Button {
text: 'hello'
onClicked: {
backend.start_process()
}
}
Text {
text: backend.num
}
}
}
为了保持前后端分离，有没有办法在不改变Task中任何脚本内容的情况下将Task连接到QMLUI？根据 Backend class 来处理前端和后端之间的任何通信，并将上下文 属性 设置为 QML（engine.rootContext().setContextProperty("backend", Backend())).例如，
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()
但是，我不知道如何双向绑定 backend 中的self.num 和t.num。因此，在 QML UI.
start_process() 后，我无法实时更新 t.num
我想尽量减少对 Task 的修改，因为它是旧程序的核心功能。因此，Simple Example in PyQt5 website无法满足我的要求。
另外，如果我想把Pythonthreading.Thread改成multiprocessing.Process，我应该怎么做才能在每次更新的过程中获取变量，以便在QML上显示UI?
首先，您不需要像 qproperty 这样的 setter，因为根据您的逻辑，您不应该修改 QML 中的 qproperty。另一方面，您必须通过 args 或 kwargs 将后端传递给您的线程，并发送调用插槽的信息：
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_())
另一种方法是创建一个包含以下信息的 QEvent：
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_())
另一种解决方案是它不再继承自threading.Thread，而是继承自QThread，它实现了一些向主线程发送信息的信号。
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_())