Connecting Asyncio and Tkinter event loops?

To run Tkinter with asyncio providing the event loop, you may want to take advantage of Tcl_SetNotifier , which is Tcl’s API to use an external event loop.

I have been using this to run multiple GUI frameworks (tkinter, GTK, and PyQt) at the same time. Below is a minimal example script with tkinter and GTK.

Advantages of this approach are

  • No threads are needed;
  • It does not use polling, i.e. there is no sleep anywhere.

The example script uses select from the Python standard library instead of async , but I think the same could be accomplished using async, provided there is some way to call a “setup” function just before select, and a “check” function just after select .

Note that the script makes use of Tcl’s public API functions Tcl_QueueEvent, Tcl_SetServiceMode, Tcl_ServiceAll, and Tcl_SetNotifier; and the Tcl constants TCL_SERVICE_ALL and TCL_SERVICE_NONE. These are currently not exposed through _tkinter, but it would be trivial to add them (I added these to my local copy of _tkinter).

import os
import select
import sys
import time

import tkinter
from tkinter import _tkinter

import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk, Gdk, GLib


# --- Create an event source for tkinter

class TkinterEventHandler:

    def __init__(self, callback):
        self.callback = callback
        self.ready_events = 0

    def process_event(self, flags):
        if not (flags & _tkinter.FILE_EVENTS):
            return False

        events = self.ready_events
        self.ready_events = 0
        if events != 0:
            self.callback(_tkinter.READABLE)

        return True

    def __call__(self, events):
        if self.ready_events == 0:
            _tkinter.queue_event(self.process_event)
        self.ready_events = events


class TkinterEventSource:

    def __init__(self):
        self.handlers = {}
        self.timer_abstime = 0

    def set_timer(self, interval_usec=None):
        mode = _tkinter.set_service_mode(_tkinter.SERVICE_ALL)
        if interval_usec is None:
            self.timer_abstime = 0
        else:
            now = time.perf_counter_ns() // 1000
            self.timer_abstime = now + interval_usec

    def wait_for_event(self, timeout):
        return 0

    def init_notifier(self):
        return

    def create_file_handler(self, fd, mask, callback):
        event_handler = TkinterEventHandler(callback)
        if mask & _tkinter.READABLE:
            self.handlers[fd] = event_handler

    def delete_file_handler(self, fd):
        try:
            del self.handlers[fd]
        except KeyError:
            pass

    def finalize_notifier(self, clientData):
        return

    def alert_notifier(self, clientData):
        return

    def service_mode_hook(self, mode):
        return
    
    def setup(self):
        timer_abstime = self.timer_abstime
        if timer_abstime > 0:
            now = time.perf_counter_ns() // 1000
            timeout_usec = timer_abstime - now
            if timeout_usec < 0:
                timeout = 0
            else:
                timeout = timeout_usec / 1000000
        else:
            timeout = None
        fds = list(self.handlers.keys())
        return timeout, fds

    def check(self, ready_events):
        timer_abstime = self.timer_abstime
        result = 0;
        now = time.perf_counter_ns() // 1000
        if timer_abstime < now:
            self.timer_abstime = 0
        ready_rlist, ready_wlist, ready_xlist = ready_events
        for fd in ready_rlist:
            handler = self.handlers.get(fd)
            if handler is None:
                continue
            handler(_tkinter.READABLE)
        _tkinter.service_all()

# --- Tell Tcl to use TkinterEventSource as its notifier

tkinter_event_source = TkinterEventSource()
_tkinter.set_notifier(tkinter_event_source.set_timer,
                      tkinter_event_source.wait_for_event,
                      tkinter_event_source.create_file_handler,
                      tkinter_event_source.delete_file_handler,
                      tkinter_event_source.init_notifier,
                      tkinter_event_source.finalize_notifier,
                      tkinter_event_source.alert_notifier,
                      tkinter_event_source.service_mode_hook
                     )


# --- Create an event source for GTK

class GTKEventSource:

    def setup(self):
        context = GLib.MainContext.default()
        if not context.acquire():
            raise RuntimeError("Failed to acquire the context")
        ready, priority = context.prepare()
        timeout_msec, self.fds = context.query(priority)
        fds = [fd.fd for fd in self.fds]
        if timeout_msec == -1:
            timeout = None
        else:
            timeout = timeout_msec / 1000
        context.release()
        return timeout, fds

    def check(self, ready_events):
        context = GLib.MainContext.default()
        if not context.acquire():
            raise RuntimeError("Failed to acquire the context")
        ready_rlist, ready_wlist, ready_xlist = ready_events
        ready_rlist = set(ready_rlist)
        for fd in self.fds:
            if fd.fd in ready_rlist:
                fd.revents |= GLib.IO_IN
        some_ready = context.check(GLib.MAXINT, self.fds)
        context.dispatch()
        context.release()

gtk_event_source = GTKEventSource()


# --- Make one example window for tkinter

class TkinterWindow:

    def __init__(self):
        self.window = tkinter.Tk()
        self.window.title("Tkinter")

        self.button1 = tkinter.Button(self.window,
                                      text="Click me",
                                      command=self.clicked,
                                      font='Helvetica 20',
                                      width=20,
                                      background='pale green')
        self.button1.pack()

        pipe_read_fd, self.pipe_write_fd = os.pipe()
        self.window.tk.createfilehandler(pipe_read_fd,
                                         tkinter.READABLE,
                                         self.read_from_pipe)
        self.button2 = tkinter.Button(self.window,
                                      text="Trigger the file descriptor",
                                      command=self.write_to_pipe,
                                      font='Helvetica 20',
                                      width=20,
                                      background='lightblue')
        self.button2.pack()

        self.button3 = tkinter.Button(self.window,
                                      text="Start the timer (1 sec)",
                                      command=self.update_timer,
                                      font='Helvetica 20',
                                      width=20,
                                      background='steelblue')
        self.button3.pack()

    def clicked(self):
        print("Tkinter button clicked")

    def write_to_pipe(self):
        message = "Tkinter file descriptor triggered"
        n = os.write(self.pipe_write_fd, message.encode())

    def read_from_pipe(self, fd, mask):
        message = os.read(fd, 1024)
        print(message.decode())

    def update_timer(self, counter=0):
        print("Tkinter timer update %d" % counter)
        self.window.after(1000, self.update_timer, counter+1)

tkinter_window = TkinterWindow()

# --- Make one example window for GTK

settings = Gtk.Settings.get_default()
settings.props.gtk_theme_name = "Adwaita"

class GTKWindow(Gtk.ApplicationWindow):

    def __init__(self, **kargs):
        super().__init__(**kargs, title='GTK')

        self.grid = Gtk.Grid()
        self.add(self.grid)

        button1 = self.create_button(1, text="Click me", color="red")
        button1.connect("clicked", self.clicked)
        self.grid.attach(button1, 0, 0, 1, 1)

        pipe_read_fd, self.pipe_write_fd = os.pipe()
        GLib.io_add_watch(pipe_read_fd, GLib.IOCondition.IN, self.read_from_pipe)
        button2 = self.create_button(2, text="Trigger the file descriptor", color='orchid')
        button2.connect("clicked", self.write_to_pipe)
        self.grid.attach(button2, 0, 1, 1, 1)

        button3 = self.create_button(3, text="Start the timer (2 sec)", color='salmon')
        button3.connect("clicked", self.start_timer)
        self.grid.attach(button3, 0, 2, 1, 1)

        self.show_all()

    def create_button(self, number, text, color):
        button = Gtk.Button.new_with_label(text)
        key = "custom-button%d" % number
        button.get_style_context().add_class(key)
        css_provider = Gtk.CssProvider()
        css_provider.load_from_data(b"""
            .%s {
                font-family: Helvetica;
                font-size: 20pt;
                background: %s;
                color: black;
            }
        """ % (key.encode(),  color.encode()))
        # Apply the CSS to the display
        style_context = button.get_style_context()
        style_context.add_provider(css_provider, Gtk.STYLE_PROVIDER_PRIORITY_USER)
        style_context.add_class(key) # Add the CSS class to the button
        return button

    def clicked(self, button):
        print("GTK button clicked")

    def write_to_pipe(self, button):
        message = "GTK file descriptor triggered"
        n = os.write(self.pipe_write_fd, message.encode())

    def read_from_pipe(self, fd, condition):
        if condition & GLib.IOCondition.IN:
            message = os.read(fd, 1024)
            print(message.decode())
            return True

    def update_timer(self):
        print("GTK timer update %d" % self.counter)
        self.counter += 1
        return True

    def start_timer(self, button):
        self.counter = 0
        GLib.timeout_add(2000, self.update_timer)


glib_window = GTKWindow()


# --- run the event loop with the two GUI frameworks as event sources

sources = [tkinter_event_source, gtk_event_source ]


while True:
    rlist = []
    wlist = []
    xlist = []
    shortest_timeout = None
    for source in sources:
        timeout, fds = source.setup()
        if timeout is not None:
            if shortest_timeout is None:
                shortest_timeout = timeout
            else:
                if timeout < shortest_timeout:
                    shortest_timeout = timeout
        rlist.extend(fds)
    ready_events = select.select(rlist, wlist, xlist, shortest_timeout)
    for source in sources:
        source.check(ready_events)
1 Like