Last active
May 15, 2024 14:55
-
-
Save thegamecracks/564bd55af973827f8a05b48f197d5c09 to your computer and use it in GitHub Desktop.
Running tkinter and asyncio event loops in separate threads
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import asyncio | |
import concurrent.futures | |
import threading | |
from tkinter import Event, Tk | |
from tkinter.ttk import Button | |
from typing import Self | |
class EventThread(threading.Thread): | |
"""Runs an asyncio event loop in a separate thread. | |
Starting and stopping can be done with :meth:`start()` and :meth:`stop()`, | |
but for reliability, you should use the class in a context manager instead:: | |
>>> with EventThread() as event_thread: | |
... loop = event_thread.loop | |
... future = asyncio.run_coroutine_threadsafe(my_task(), loop) | |
... result = future.result() | |
>>> # Event loop will be cleaned up before exiting | |
""" | |
def __init__(self, *args, **kwargs) -> None: | |
super().__init__(*args, **kwargs) | |
self.loop_fut = concurrent.futures.Future() | |
self.stop_fut = concurrent.futures.Future() | |
self.finished_fut = concurrent.futures.Future() | |
def __enter__(self) -> Self: | |
self.start() | |
return self | |
def __exit__(self, exc_type, exc_val, tb) -> None: | |
self.stop() | |
self.join() | |
@property | |
def loop(self) -> asyncio.AbstractEventLoop: | |
return self.loop_fut.result() | |
def run(self) -> None: | |
try: | |
asyncio.run(self._run_forever()) | |
finally: | |
self.finished_fut.set_result(None) | |
def stop(self) -> None: | |
try: | |
self.stop_fut.set_result(None) | |
except concurrent.futures.InvalidStateError: | |
pass | |
async def _run_forever(self) -> None: | |
self.loop_fut.set_result(asyncio.get_running_loop()) | |
await asyncio.wrap_future(self.stop_fut) | |
class TkApp(Tk): | |
def __init__(self, event_thread: EventThread) -> None: | |
super().__init__() | |
self.event_thread = event_thread | |
self._connect_lifetime(event_thread) | |
self.bind("<<Destroy>>", self._on_destroy) | |
def destroy(self) -> None: | |
# asyncio tasks that make any tkinter calls, including event_generate(), | |
# require the tkinter mainloop to still be running. As such, we need to | |
# make sure the event loop closes before tkinter does. | |
self.event_thread.stop() | |
def _connect_lifetime(self, event_thread: EventThread) -> None: | |
event_callback = lambda fut: self.event_generate("<<Destroy>>") | |
event_thread.finished_fut.add_done_callback(event_callback) | |
def _on_destroy(self, event: Event) -> None: | |
return super().destroy() | |
# Example usage: | |
async def some_task(): | |
print("Running a task...") | |
try: | |
await asyncio.sleep(5) | |
except asyncio.CancelledError: | |
print("Task was cancelled!") | |
raise | |
else: | |
print("Task was completed!") | |
def start_task(app: Tk, event_thread: EventThread): | |
coro = some_task() | |
fut = asyncio.run_coroutine_threadsafe(coro, event_thread.loop) | |
fut.add_done_callback(lambda fut: app.event_generate("<<TaskFinished>>")) | |
# Upon completion, callback will run in the event loop's thread. | |
# Whatever you do should be thread-safe, such as pushing the result | |
# to a queue and generating an event. | |
# | |
# Sidenote, if you want to push to an asyncio queue from tkinter you must | |
# do that thread-safely as well, otherwise waiting tasks won't wake up: | |
# | |
# >>> asyncio.run_coroutine_threadsafe(queue.put("Hello world!"), event_thread.loop) | |
# # or: | |
# >>> event_thread.loop.call_soon_threadsafe(queue.put_nowait, "Hello world!") | |
def on_task_finished(event: Event): | |
print("Tkinter received a TaskFinished event") | |
def main(): | |
with EventThread() as event_thread: | |
app = TkApp(event_thread) | |
app.geometry("320x240") | |
app.grid_columnconfigure(0, weight=1) | |
app.grid_rowconfigure(0, weight=1) | |
command = lambda: start_task(app, event_thread) | |
button = Button(app, text="Start a task", command=command) | |
button.grid(sticky="nesw", padx=20, pady=20) | |
app.bind("<<TaskFinished>>", on_task_finished) | |
try: | |
app.mainloop() | |
except BaseException: | |
# This is probably a KeyboardInterrupt. If we exit now, the context | |
# manager will stop our tasks, but it may trigger callbacks | |
# that perform tkinter calls. To make sure they don't fail, | |
# allow tkinter to run for just a bit longer. | |
app.destroy() | |
app.mainloop() | |
raise | |
if __name__ == "__main__": | |
try: | |
main() | |
except KeyboardInterrupt: | |
print("KeyboardInterrupt caught") |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Running a task... | |
Running a task... | |
Task was completed! | |
Tkinter received a TaskFinished event | |
Task was cancelled! | |
Tkinter received a TaskFinished event |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Running a task... | |
Running a task... | |
Running a task... | |
Task was cancelled! | |
Task was cancelled! | |
Task was cancelled! | |
Tkinter received a TaskFinished event | |
Tkinter received a TaskFinished event | |
Tkinter received a TaskFinished event | |
KeyboardInterrupt caught |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
License