6
\$\begingroup\$

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')
```
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.