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.