Trio + PyQT5 in one program?

[ll2pakll]

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_())

[smurfix]

Well I have to admit that this wasn't terribly difficult (only moderately so …).

You might want to do something when Qt5TrioRunner.done is signalled, as it indicates that the Trio part has ended (or crashed; access r.result to find out).

class Qt5TrioRunner(QtCore.QObject):
    _sig = QtCore.pyqtSignal()
    _start = QtCore.pyqtSignal()
    done = QtCore.pyqtSignal()

    def __init__(self):
        super().__init__()

        from threading import Lock
        self._lock = Lock()
        self._fns = []
        self._result = None

        self._start.connect(self.start)
        self._sig.connect(self._handler)
        self._start.emit()

    def start(self):
        trio.lowlevel.start_guest_run(
            self.main,
            run_sync_soon_threadsafe=self._runner,
            done_callback=self._done,
        )

    async def main(self):
        """Trio main code. Override me."""
        await trio.sleep_forever()

    def _done(self, result):
        self.done.emit()
        self._result = result

    @property
    def result(self):
        return self._result.unwrap()

    def _handler(self):
        """Runs in the Qt main thread"""
        with self._lock:
            fns,self._fns = self._fns,[]
        for fn in fns:
            fn()

    def _runner(self, fn):
        with self._lock:
            self._fns.append(fn)
        self._sig.emit()

class MyRunner(Qt5TrioRunner):
    def __init__(self, ui):
        self.ui = ui

        # MUST BE LAST
        super().__init__()

    async def main(self):
        await self.ui.timer()

if __name__ == "__main__":
    import sys
    app = QtWidgets.QApplication(sys.argv)
    MainWindow = QtWidgets.QMainWindow()
    ui = PushMe(MainWindow)
    MainWindow.show()
    r = MyRunner(ui)
    sys.exit(app.exec_())

[smurfix]

@njsmith Should I park this code in the documentation or wherever?

[altendky]

This question came up a few places, sorry I didn't catch this one until now. They got running with https://qtrio.readthedocs.io/en/stable/ plus a little discussion and help.