Skip to content

Instantly share code, notes, and snippets.

@danieljfarrell
Forked from ericfrederich/async_slot.py
Created March 20, 2019 12:44
Show Gist options
  • Save danieljfarrell/251adce3ceb63e93c5ada75a8e67e0af to your computer and use it in GitHub Desktop.
Save danieljfarrell/251adce3ceb63e93c5ada75a8e67e0af to your computer and use it in GitHub Desktop.
Example of using an asyncio coroutine as a Qt slot
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