-
-
Save danieljfarrell/251adce3ceb63e93c5ada75a8e67e0af to your computer and use it in GitHub Desktop.
Example of using an asyncio coroutine as a Qt slot
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import math | |
import sys | |
import asyncio | |
from functools import partial, wraps | |
from PyQt5.QtWidgets import QApplication, QWidget, QPushButton, QGridLayout, QProgressBar, QErrorMessage | |
from PyQt5.QtCore import Qt | |
from quamash import QEventLoop | |
import traceback | |
def display_error(err): | |
app = QApplication.instance() | |
window = app.activeWindow() | |
dialog = QErrorMessage(window) | |
dialog.setWindowModality(Qt.WindowModal) | |
dialog.setWindowTitle("Error") | |
dialog.showMessage(err) | |
def slot_coroutine(async_func): | |
if not asyncio.iscoroutinefunction(async_func): | |
raise RuntimeError('Must be a coroutine!') | |
def log_error(future): | |
try: | |
future.result() | |
except Exception as err: | |
display_error(traceback.format_exc()) | |
@wraps(async_func) | |
def wrapper(self, *args): | |
loop = asyncio.get_event_loop() | |
future = loop.create_task(async_func(self, *args[:-1])) | |
future.add_done_callback(log_error) | |
return wrapper | |
class MyWidget(QWidget): | |
def __init__(self, parent=None): | |
super().__init__(parent) | |
layout = QGridLayout() | |
self.pairs = [] | |
for i in range(10): | |
b = QPushButton('Countdown: &%d' % i) # NB cannot use the clicked kwarg! | |
b.clicked.connect(partial(self.on_click_coroutine, i)) | |
p = QProgressBar(visible=False) | |
layout.addWidget(b, i, 0) | |
layout.addWidget(p, i, 1) | |
self.pairs.append((b, p)) | |
self.setLayout(layout) | |
@slot_coroutine | |
async def on_click_coroutine(self, i): | |
""" This is connected to the button's signal via button.clicked.connect(...). It is | |
a coroutine but the decorator wraps it, executes it as returns a function. | |
""" | |
if i == 2: | |
raise ValueError('Blah!! Two always fails.') | |
b, p = self.pairs[i] | |
await self.do_countdown(b, p, i) | |
@staticmethod | |
async def do_countdown(b, p, n): | |
b.setEnabled(False) | |
p.setVisible(True) | |
i = n * 100 | |
p.setMaximum(i) | |
while i >= 0: | |
await asyncio.sleep(0.01) | |
p.setValue(i) | |
b.setText('Countdown: %d' % math.ceil(i/100)) | |
i -= 1 | |
b.setText('Countdown: &%d' % n) | |
b.setEnabled(True) | |
p.setVisible(False) | |
def main(): | |
app = QApplication(sys.argv) | |
loop = QEventLoop(app) | |
asyncio.set_event_loop(loop) # NEW must set the event loop | |
w = MyWidget() | |
w.show() | |
with loop: | |
loop.run_forever() | |
if __name__ == '__main__': | |
main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment