运行 单独线程中的异步循环,信号来自和循环
Running an asyncio loop in a separate thread, Signals from, and to loop
我正在尝试制作一个 UI,它在后台与多个 BLE 设备进行通信。
为此,我实现了一个运行 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_())
没有必要使用线程来将 asyncio 与 Qt 一起使用,因为有像 asyncqt
and 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()
我正在尝试制作一个 UI,它在后台与多个 BLE 设备进行通信。 为此,我实现了一个运行 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_())
没有必要使用线程来将 asyncio 与 Qt 一起使用,因为有像 asyncqt
and 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()