How can I use trio in conjunction with PyQT5? The thing is that my program has interface written using PyQT5, and now I need to run eventloop trio, to work with the network, because I use trio WebSocket to connect to the server. I read that I should use trio.lowlevel.start_guest_run for this purpose. But in the documentation it says that in addition to the trio function I must pass run_sync_soon_threadsafe and done_callback as arguments. The documentation gives an example with asyncio and says that I have to define similar functions for my event_loop, in my case for PyQT. Unfortunately my knowledge is not enough to do it myself. I wrote a very simple application using PyQT and put in the body of the class an asynchronous function that if it works correctly should change the inscription on the timer every second. In addition you can enter text in the input box and by pressing the button this text will be displayed at the bottom. I did not run the asynchronous function, as that is my question. At best, I expect the answer to my question to be a modified program in which the asynchronous function and the PyQT5 components run in the same thread using trio. Thank you for your reply.
# -*- coding: utf-8 -*-
from PyQt5 import QtCore, QtGui, QtWidgets
import trio
class Ui_MainWindow(object):
def setupUi(self, MainWindow):
MainWindow.setObjectName("MainWindow")
MainWindow.resize(804, 595)
self.centralwidget = QtWidgets.QWidget(MainWindow)
self.centralwidget.setObjectName("centralwidget")
self.verticalLayoutWidget = QtWidgets.QWidget(self.centralwidget)
self.verticalLayoutWidget.setGeometry(QtCore.QRect(10, 10, 781, 591))
self.verticalLayoutWidget.setObjectName("verticalLayoutWidget")
self.verticalLayout = QtWidgets.QVBoxLayout(self.verticalLayoutWidget)
self.verticalLayout.setContentsMargins(0, 0, 0, 0)
self.verticalLayout.setObjectName("verticalLayout")
spacerItem = QtWidgets.QSpacerItem(20, 40, QtWidgets.QSizePolicy.Minimum, QtWidgets.QSizePolicy.Expanding)
self.verticalLayout.addItem(spacerItem)
self.lineEdit = QtWidgets.QLineEdit(self.verticalLayoutWidget)
font = QtGui.QFont()
font.setPointSize(20)
self.lineEdit.setFont(font)
self.lineEdit.setObjectName("lineEdit")
self.verticalLayout.addWidget(self.lineEdit)
self.pushButton = QtWidgets.QPushButton(self.verticalLayoutWidget)
font = QtGui.QFont()
font.setPointSize(20)
self.pushButton.setFont(font)
self.pushButton.setObjectName("pushButton")
self.verticalLayout.addWidget(self.pushButton)
self.label = QtWidgets.QLabel(self.verticalLayoutWidget)
font = QtGui.QFont()
font.setPointSize(20)
self.label.setFont(font)
self.label.setAlignment(QtCore.Qt.AlignCenter)
self.label.setObjectName("label")
self.verticalLayout.addWidget(self.label)
spacerItem1 = QtWidgets.QSpacerItem(20, 40, QtWidgets.QSizePolicy.Minimum, QtWidgets.QSizePolicy.Expanding)
self.verticalLayout.addItem(spacerItem1)
self.label_2 = QtWidgets.QLabel(self.verticalLayoutWidget)
font = QtGui.QFont()
font.setPointSize(20)
self.label_2.setFont(font)
self.label_2.setAlignment(QtCore.Qt.AlignCenter)
self.label_2.setObjectName("label_2")
self.verticalLayout.addWidget(self.label_2)
spacerItem2 = QtWidgets.QSpacerItem(20, 40, QtWidgets.QSizePolicy.Minimum, QtWidgets.QSizePolicy.Expanding)
self.verticalLayout.addItem(spacerItem2)
MainWindow.setCentralWidget(self.centralwidget)
self.retranslateUi(MainWindow)
QtCore.QMetaObject.connectSlotsByName(MainWindow)
def retranslateUi(self, MainWindow):
_translate = QtCore.QCoreApplication.translate
MainWindow.setWindowTitle(_translate("MainWindow", "MainWindow"))
self.pushButton.setText(_translate("MainWindow", "Pushme"))
self.label.setText(_translate("MainWindow", ""))
self.label_2.setText(_translate("MainWindow", "Time"))
class PushMe(Ui_MainWindow):
def __init__(self, MainWindow):
super(PushMe, self).__init__()
self.setupUi(MainWindow)
self.PushMe = MainWindow
self.lineEdit.setPlaceholderText("type something")
self.connect_button()
# self.timer()
def connect_button(self):
self.pushButton.clicked.connect(self.set_text)
def set_text(self):
self.label.setText(self.lineEdit.text())
async def timer(self):
a = 1
while True:
self.label_2.setText(str(a))
a += 1
await trio.sleep(1)
if __name__ == "__main__":
import sys
app = QtWidgets.QApplication(sys.argv)
MainWindow = QtWidgets.QMainWindow()
ui = PushMe(MainWindow)
MainWindow.show()
sys.exit(app.exec_())
Advertisement
Answer
I have to admit that I haven’t been keeping up on maintenance on QTrio, but it does mix Qt with Trio. https://qtrio.readthedocs.io/en/stable/ Depending what route you want to go you could either use it as a whole library, or you could just pick out the pieces you want.
trio.lowlevel.start_guest_run(
self.trio_main,
async_fn,
args,
run_sync_soon_threadsafe=self.run_sync_soon_threadsafe,
done_callback=self.trio_done,
clock=self.clock, # type: ignore[arg-type]
instruments=self.instruments,
)
def run_sync_soon_threadsafe(self, fn: typing.Callable[[], object]) -> None:
"""Helper for the Trio guest to execute a sync function in the Qt host
thread when called from the Trio guest thread. This call will not block waiting
for completion of ``fn`` nor will it return the result of calling ``fn``.
Args:
fn: A no parameter callable.
"""
import qtrio.qt
event = qtrio.qt.ReenterEvent(fn=fn)
self.application.postEvent(self.reenter, event)