我正在尝试制作一个在后台与多个 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_())