python - 在单独的线程中运行 asyncio 循环，信号来自和到循环

Question

我正在尝试制作一个在后台与多个 BLE 设备进行通信的 UI。
为此，我实现了一个运行 asyncio.loop 的单独线程。这是必要的，因为我使用 bleak 0.9.1 连接到设备。
使用信号和槽从 UI 线程获取数据到工作线程工作正常。但是，它在另一个方向上不起作用。据我所知，这是因为线程忙于运行循环并且永远不会停止这样做。因此，它无法处理来自 UI 线程的输入。
下面是显示问题的示例代码。
有没有办法在运行 asyncio 循环时处理线程中的输入槽？

import sys
from PyQt5.QtWidgets import QWidget, QPushButton, QApplication, QVBoxLayout
from PyQt5.QtCore import QThread, QObject, pyqtSignal, pyqtSlot
import asyncio

class Test_Thread(QObject):
    signal_back = pyqtSignal(int)

    def __init__(self,
                 loop: asyncio.AbstractEventLoop,
                 parent=None):
        super(Test_Thread, self).__init__(parent)
        self.text = "Task1 not configured"
        self.loop = loop
        self.counter = 0

    @pyqtSlot(str)
    def set_text_slot(self, txt):
        self.text = txt

    async def do_stuff1(self):
        while True:
            print(self.text)
            await asyncio.sleep(2.0)

    async def do_stuff2(self):
        while True:
            self.counter += 1
            self.signal_back.emit(self.counter)
            await asyncio.sleep(1.0)

    def work(self):
        #run the event loop
        try:
            asyncio.ensure_future(self.do_stuff1(), loop=self.loop)
            asyncio.ensure_future(self.do_stuff2(), loop=self.loop)
            self.loop.run_forever()
        finally:
            print("Disconnect...")


class Window(QWidget):

    set_text_signal = pyqtSignal(str)

    def __init__(self, parent=None):
        super(Window, self).__init__()
        self.initUi()
        self.startThread()

    def initUi(self):
        layout = QVBoxLayout()
        self.button = QPushButton('User input')
        self.button.clicked.connect(self.sendtotask)
        layout.addWidget(self.button)
        self.setLayout(layout)
        self.show()

    def startThread(self):
        loop = asyncio.get_event_loop()
        self.asyciothread = Test_Thread(loop)
        self.thread = QThread()
        self.asyciothread.moveToThread(self.thread)

        self.set_text_signal.connect(self.asyciothread.set_text_slot)
        self.asyciothread.signal_back.connect(self.receivefromthread)
        self.thread.started.connect(self.asyciothread.work)

        self.thread.start()

    @pyqtSlot(int)
    def receivefromthread(self, number):
        print(str(number))

    def sendtotask(self):
        self.set_text_signal.emit("Task: Configured")


if __name__ == "__main__":
    app = QApplication(sys.argv)
    ui = Window()
    ui.show()
    sys.exit(app.exec_())

Answer 1

没有必要使用线程来将 asyncio 与 Qt 一起使用，因为有像 asyncqt 这样的库。和 qasync 启用它:

import asyncio
import sys

from PyQt5.QtWidgets import QWidget, QPushButton, QApplication, QVBoxLayout
from PyQt5.QtCore import QObject, pyqtSignal, pyqtSlot

from asyncqt import QEventLoop
# from qasync import QEventLoop


class Worker(QObject):
    signal_back = pyqtSignal(int)

    def __init__(self, loop: asyncio.AbstractEventLoop, parent=None):
        super(Worker, self).__init__(parent)
        self.text = "Task1 not configured"
        self.loop = loop
        self.counter = 0

    @pyqtSlot(str)
    def set_text_slot(self, txt):
        self.text = txt

    async def do_stuff1(self):
        while True:
            print(self.text)
            await asyncio.sleep(2.0)

    async def do_stuff2(self):
        while True:
            self.counter += 1
            self.signal_back.emit(self.counter)
            await asyncio.sleep(1.0)

    def work(self):
        asyncio.ensure_future(self.do_stuff1(), loop=self.loop)
        asyncio.ensure_future(self.do_stuff2(), loop=self.loop)


class Window(QWidget):
    set_text_signal = pyqtSignal(str)

    def __init__(self, parent=None):
        super(Window, self).__init__()
        self.initUi()
        self.start_task()

    def initUi(self):
        layout = QVBoxLayout(self)
        self.button = QPushButton("User input")
        self.button.clicked.connect(self.sendtotask)
        layout.addWidget(self.button)

    def start_task(self):
        loop = asyncio.get_event_loop()
        self.worker = Worker(loop)
        self.set_text_signal.connect(self.worker.set_text_slot)
        self.worker.signal_back.connect(self.receive_from_worker)
        self.worker.work()

    @pyqtSlot(int)
    def receive_from_worker(self, number):
        print(str(number))

    def sendtotask(self):
        self.set_text_signal.emit("Task: Configured")


if __name__ == "__main__":
    app = QApplication(sys.argv)
    loop = QEventLoop(app)
    asyncio.set_event_loop(loop)
    ui = Window()
    ui.show()
    with loop:
        loop.run_forever()