Simple integration of async and tkinter in Python

Question

I've been a fan of coroutines and asynchronous programming for a while, and I recently took a trip down memory lane to when I was using Python's tkinter module for a GUI. I wanted to combine the two worlds, one of smooth coroutines and one of callbacks.

My hope is to find a better solution for the basic event loop, implemented in Window._coro_step. The current way of calling again in 1 ms seems too hackish, but the tkinter mainloop has no way to be run in non-blocking steps. (Please take a look at my formatting and style too. Sorry for the lack of comments.)

Here's my unfinished but working code:

import tkinter
from collections import deque
from contextlib import closing
from types import coroutine
class Window:
 def __init__(self, title, min_size, start_size, screens=None, gvars=None):
 if screens is None:
 screens = {}
 if gvars is None:
 gvars = {}
 self.title = str(title)
 self.min_size = tuple(min_size)
 self.start_size = tuple(start_size)
 self.screens = dict(screens)
 self.gvars = dict(gvars)
 self.current = None
 self._events = deque()
 self._window = None
 self._canvas = None
 self._coro = None
 self._wait = None
 def run(self, start_screen, start_title=None):
 self._window = window = tkinter.Tk()
 window.minsize(*self.min_size)
 window.rowconfigure(1, weight=1)
 window.columnconfigure(1, weight=1)
 window.title(start_title if start_title is None else self.title)
 self._canvas = canvas = tkinter.Canvas(window, highlightthickness=0)
 canvas.rowconfigure(1, weight=1)
 canvas.columnconfigure(1, weight=1)
 canvas.grid(row = 1, column=1, sticky='nsew')
 window.geometry('x'.join(map(str, self.start_size)))
 Widget.canvas = canvas
 self.switch(start_screen)
 window.after(1, self._coro_step)
 with closing(self):
 window.mainloop()
 def switch(self, screen_name):
 # Switch the current coroutine to 'screen_name'
 canvas = self._canvas
 if self.current is not None:
 self._coro.close()
 for event_name in self.current.event_names:
 canvas.unbind(event_name)
 if screen_name == '':
 current = self.current
 else:
 self.current = current = self.screens[screen_name]
 self._coro, self._wait = current.setup(self, canvas, self.gvars)
 for event_name in current.event_names:
 canvas.bind(event_name, self._events.append)
 def destroy(self):
 self._window.destroy()
 self._coro.close()
 self._coro = None
 def close(self):
 # Idempotent
 try: self._window.destroy()
 except: pass
 if self._coro is not None:
 try: self._coro.close()
 except: pass
 self._coro = None
 async def aevents(self, *, forever=False, ty='get_events'):
 coro = getattr(self, ty)
 multiple = ty.endswith('events')
 if forever and multiple:
 while True:
 for event in await coro():
 yield event
 elif forever:
 while True:
 yield await coro()
 elif multiple:
 for event in await coro():
 yield event
 else:
 yield await coro()
 def _coro_step(self):
 if self._coro is None:
 return
 # Mini event loop that doesn't block
 while True:
 if (
 type(self._wait) is not tuple
 or self._wait[0] not in {
 'get_event', 'get_events',
 'poll_event', 'poll_events',
 }
 ):
 exc = ValueError('invalid yield type')
 self._coro.throw(exc)
 raise exc # In case the coroutine yielded another value
 allow_empty, allow_multiple = self._wait[0].split('_')
 allow_empty = (allow_empty == 'poll') # 'get' or 'poll
 allow_multiple = (allow_multiple == 'events') # 'event' or 'events
 if allow_multiple:
 send = []
 try:
 while True:
 send.append(self._events.popleft())
 except IndexError:
 if not allow_empty and not send:
 # Reschedule for a later time
 self._window.after(1, self._coro_step)
 return
 else:
 send = None
 try:
 send = self._events.popleft()
 except IndexError:
 if not allow_empty:
 # Reschedule for a later time
 self._window.after(1, self._coro_step)
 return
 try:
 self._wait = self._coro.send(send)
 except StopIteration as e:
 result = e.value
 if result in {'return', None}:
 self.destroy()
 return
 else:
 self.switch(result)
 else:
 if allow_multiple and allow_empty:
 # Reschedule for a later time
 self._window.after(1, self._coro_step)
 return
 # Wrappers around decorated coroutines
 async def get_event(self):
 return await self._get_event()
 async def get_events(self):
 return await self._get_events()
 async def poll_event(self):
 return await self._poll_event()
 async def poll_events(self):
 return await self._poll_events()
 # Raw yielding coroutines
 @staticmethod
 @coroutine
 def _get_event():
 return (yield ('get_event',))
 @staticmethod
 @coroutine
 def _get_events():
 return (yield ('get_events',))
 @staticmethod
 @coroutine
 def _poll_event():
 return (yield ('poll_event',))
 @staticmethod
 @coroutine
 def _poll_events():
 return (yield ('poll_events',))
class Screen:
 event_names = [
 '<ButtonPress>', '<ButtonRelease>',
 '<KeyPress>', '<KeyRelease>',
 '<Configure>', '<Deactivate>', '<Destroy>',
 '<Enter>', '<Motion>', '<Leave>',
 '<Expose>', '<FocusIn>', '<FocusOut>',
 '<MouseWheel>', '<Visibility>',
 ]
 def __init__(self, name='', screen=None):
 if screen is None:
 screen = lambda window, canvas, gvars: iter([])
 self.name = name
 self._func_screen = screen
 self._gen_screen = None
 def __repr__(self):
 return f'<{type(self).__name__} name={self.name}>'
 def setup(self, window, canvas, gvars):
 self._gen_screen = gen_screen = self._func_screen(window, canvas, gvars)
 return gen_screen, gen_screen.send(None)
class Widget:
 canvas = None
 def __init__(self):
 self.canvas = type(self).canvas
 self.id = None
 def __repr__(self):
 return f'<{type(self).__name__} at {hex(id(self))}>'
 def draw(self):
 ...
 def check(self, event):
 ...
 def cget(self, item):
 self.canvas.itemcget(self.id, item)
 def config(self, **kwargs):
 self.canvas.itemconfig(self.id, **kwargs)
class RectWidget(Widget):
 def __init__(self, x1, y1, x2, y2, colour='#000000'):
 super().__init__()
 self.x1 = x1
 self.y1 = y1
 self.x2 = x2
 self.y2 = y2
 self.colour = colour
 def draw(self):
 if self.id is not None:
 self.canvas.delete(self.id)
 self.id = self.canvas.create_rectangle(
 self.x1, self.y1, self.x2, self.y2,
 fill=self.colour, width=0,
 )
 return self.id
 def check(self, event):
 super().check(event)
 def touches(self, x=None, y=None):
 if x is None:
 x = self.canvas.winfo_pointerx() - self.canvas.winfo_rootx()
 if y is None:
 y = self.canvas.winfo_pointery() - self.canvas.winfo_rooty()
 x1, x2 = sorted([self.x1, self.x2])
 y1, y2 = sorted([self.y1, self.y2])
 return x1 <= x <= x2 and y1 <= y <= y2
class TextWidget(Widget):
 def __init__(self, x, y, text='', colour='#000000', font='Helvetica', size=50):
 super().__init__()
 self.x = x
 self.y = y
 self.text = text
 self.colour = colour
 self.font = font
 self.size = size
 def draw(self):
 if self.id is not None:
 self.canvas.delete(self.id)
 self.id = self.canvas.create_text(
 self.x, self.y,
 text=self.text, fill=self.colour,
 font=(self.font, str(self.size)),
 )
 return self.id
 def check(self, event):
 super().check(event)
async def main_screen(window, canvas, gvars):
 canvas.delete('all')
 canvas.configure(bg='#7777FF')
 width, height = canvas.winfo_width(), canvas.winfo_height()
 scale = min(width/320, height/220)
 pressed = False
 buttons = {
 (-60, -25, 60, 5): {'return': 'main'},
 (-60, 15, 60, 45): {'return': 'main'},
 (-60, 55, 60, 85): {'return': 'return'},
 }
 text = {
 (width/2, scale*40, 'Async Window', 30): {},
 (width/2, height/2 - scale*10, 'MAIN', 18): {},
 (width/2, height/2 + scale*30, 'OTHER', 18): {},
 (width/2, height/2 + scale*70, 'QUIT', 18): {},
 }
 for info in buttons:
 x1, y1, x2, y2 = info
 x1, x2 = map(lambda i: width/2 + scale*i, (x1, x2))
 y1, y2 = map(lambda i: height/2 + scale*i, (y1, y2))
 rect_widget = RectWidget(x1, y1, x2, y2, '')
 rect_widget.colour = '#EEEEEE' if rect_widget.touches() else 'white'
 rect_widget.draw()
 buttons[info]['widget'] = rect_widget
 for info in text:
 x, y, t, size = info
 text_widget = TextWidget(x, y, t, '#CCCCCC', size=round(size*scale))
 text_widget.draw()
 text[info]['result'] = text_widget
 async for event in window.aevents(forever=True):
 e_type = str(event.type)
 if e_type == 'Destroy':
 return
 elif e_type == 'Configure':
 return ''
 elif e_type == 'Motion':
 x, y = event.x, event.y
 for pos, info in buttons.items():
 button = info['widget']
 if button.touches(x, y):
 button.config(fill='#DDDDDD' if pressed else '#EEEEEE')
 else:
 button.config(fill='white')
 if pressed == button:
 pressed = True
 elif e_type == 'ButtonPress':
 x, y = event.x, event.y
 pressed = True
 for pos, info in buttons.items():
 button = info['widget']
 if button.touches(x, y):
 button.config(fill='#DDDDDD')
 pressed = button
 else:
 button.config(fill='white')
 elif e_type == 'ButtonRelease':
 x, y = event.x, event.y
 for pos, info in buttons.items():
 button = info['widget']
 if button.touches(x, y):
 if pressed == button:
 return info['return']
 else:
 button.config(fill='#EEEEEE')
 else:
 button.config(fill='white')
 pressed = False
main = Screen('main', main_screen)
window = Window('Async Window', (320, 220), (350, 250), {'main', main})
if __name__ == '__main__':
 window.run('main')
```

Answer 1

I've been working on this for a while now, and from that hard work is tkio, a curio inspired asynchronous library for Python's tkinter. Here are some notable changes from my previous design.

1. I removed the idea of screens / windows and added in tasks / loop (from curio and asyncio). The idea about screens was me thinking about video games and their different menus and such. It wouldn't apply to other things like a Sudoku solver or bootleg MS Paint as they would only have one "screen". Tasks are something well known and when combined with synchronizing can replicate the screen layout (one task runs at a time).

2. The loop is run from tkinter's wait_window method. Not many people use this but it is basically mainloop but one that ends when the specified widget is destroyed. With this in mind, mainloop could just be wait_window(root) where root is the top level Tk instance.

3. The event and after callbacks run the loop. This means that there needs to be a difference between sending in a coroutine to run versus a suspend point. Luckily there are asynchronous generators which can suspend in between yield points. The loop can be run by sending in a coroutine using cycle = loop.asend(coro) on the loop and event / after callbacks can run the cycle using cycle.send("EVENT_WAKE") or something similar.

Everything else is from curio: traps, cancellation, and timeout just to name a few. I've also used many different resources to create this and it has been a great experience. I hope that this will be of help to someone and upgrade tkinter to the async world.