Issue
I was interested in trying to merge Python's tkinter
with asyncio
, and after having read this answer I was largely successful. For reference, you can recreate the mainloop
as follows:
import asyncio
import tkinter as tk
class AsyncTk(tk.Tk):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.is_running = True
async def async_loop(self):
"""Asynchronous equivalent of `Tk.mainloop`."""
while self.is_running:
self.update()
await asyncio.sleep(0)
def destroy(self):
super().destroy()
self.is_running = False
async def main():
root = AsyncTk()
await asyncio.gather(
root.async_loop(),
other_async_functions(),
)
However, this comment pointed out that in some cases, the GUI may freeze during an self.update()
call. One may instead use self.update_idletasks()
to prevent this, but since root.async_loop
is supposed to simulate the root.mainloop
, I fear never running some tasks may cause other problems.
I couldn't find the source code for how root.mainloop
works, though I did discover that replacing self.update()
with
self.tk.dooneevent(tk._tkinter.DONT_WAIT)
should produce more fine-grained concurrency by only doing one event instead of flushing all events (I'm assuming that's what it does, I couldn't find the documentation for this either but it seemed to work).
So my two questions are:
Is it fine to use just
self.update_idletasks()
only, and never run whatever elseself.update()
is supposed to run?How exactly does
root.mainloop()
work?
For some code that can be ran and experimented with:
"""Example integrating `tkinter`'s `mainloop` with `asyncio`."""
import asyncio
from random import randrange
from time import time
import tkinter as tk
class AsyncTk(tk.Tk):
"""
An asynchronous Tk class.
Use `await root.async_loop()` instead of `root.mainloop()`.
Schedule asynchronous tasks using `asyncio.create_task(...)`.
"""
is_running: bool
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.is_running = True
async def async_loop(self):
"""An asynchronous version of `root.mainloop()`."""
# For threaded calls.
self.tk.willdispatch()
# Run until `self.destroy` is called.
while self.is_running:
#self.update_idletasks() # NOTE: using `update_idletasks`
# prevents the window from freezing
# when you try to resize the window.
self.update()
#self.tk.dooneevent(tk._tkinter.DONT_WAIT)
await asyncio.sleep(0)
def destroy(self):
"""
Destroy this and all descendants widgets. This will
end the application of this Tcl interpreter.
"""
super().destroy()
# Mark the Tk as not running.
self.is_running = False
async def rotator(self, interval, d_per_tick):
"""
An example custom method for running code asynchronously
instead of using `tkinter.Tk.after`.
NOTE: Code that can use `tkinter.Tk.after` is likely
preferable, but this may not fit all use-cases and
may sometimes require more complicated code.
"""
canvas = tk.Canvas(self, height=600, width=600)
canvas.pack()
deg = 0
color = 'black'
arc = canvas.create_arc(
100,
100,
500,
500,
style=tk.CHORD,
start=0,
extent=deg,
fill=color,
)
while self.is_running:
deg, color = deg_color(deg, d_per_tick, color)
canvas.itemconfigure(arc, extent=deg, fill=color)
await asyncio.sleep(interval)
def deg_color(deg, d_per_tick, color):
"""Helper function for updating the degree and color."""
deg += d_per_tick
if 360 <= deg:
deg %= 360
color = f"#{randrange(256):02x}{randrange(256):02x}{randrange(256):02x}"
return deg, color
async def main():
root = AsyncTk()
await asyncio.gather(root.async_loop(), root.rotator(1/60, 2))
if __name__ == "__main__":
asyncio.run(main())
Solution
Although trying to rewrite the tkinter
loop seems troublesome, it seems rewriting the asyncio
loop is quite easy, given tkinter
's after
function. The main gist of it is this:
"""Example integrating `tkinter`'s `mainloop` with `asyncio`."""
import asyncio
import tkinter as tk
from typing import Any, Awaitable, TypeVar
T = TypeVar("T")
class AsyncTk(tk.Tk):
"""
A Tk class that can run asyncio awaitables alongside the tkinter application.
Use `root.run_with_mainloop(awaitable)` instead of `root.mainloop()` as a way to run
coroutines alongside it. It functions similarly to using `asyncio.run(awaitable)`.
Alternatively use `await root.async_loop()` if you need to run this in an asynchronous
context. Because this doesn't run `root.mainloop()` directly, it may not behave exactly
the same as using `root.run_with_mainloop(awaitable)`.
"""
is_running: bool
def __init__(self, /, *args: Any, **kwargs: Any) -> None:
super().__init__(*args, **kwargs)
self.is_running = True
def __advance_loop(self, loop: asyncio.AbstractEventLoop, timeout, /) -> None:
"""Helper method for advancing the asyncio event loop."""
# Stop soon i.e. only advance the event loop a little bit.
loop.call_soon(loop.stop)
loop.run_forever()
# If tkinter is still running, repeat this method.
if self.is_running:
self.after(timeout, self.__advance_loop, loop, timeout)
async def async_loop(self, /) -> None:
"""
An asynchronous variant of `root.mainloop()`.
Because this doesn't run `root.mainloop()` directly, it may not behave exactly
the same as using `root.run_with_mainloop(awaitable)`.
"""
# For threading.
self.tk.willdispatch()
# Run initial update.
self.update()
# Run until `self.destroy()` is called.
while self.is_running:
# Let other code run.
# Uses a non-zero sleep time because tkinter should be expected to be slow.
# This decreases the busy wait time.
await asyncio.sleep(tk._tkinter.getbusywaitinterval() / 10_000)
# Run one event.
self.tk.dooneevent(tk._tkinter.DONT_WAIT)
def run_with_mainloop(self, awaitable: Awaitable[T], /, *, timeout: float = 0.001) -> T:
"""
Run an awaitable alongside the tkinter application.
Similar to using `asyncio.run(awaitable)`.
Use `root.run_with_mainloop(awaitable, timeout=...)` to
customize the frequency the asyncio event loop is updated.
"""
if not isinstance(awaitable, Awaitable):
raise TypeError(f"awaitable must be an Awaitable, got {awaitable!r}")
elif not isinstance(timeout, (float, int)):
raise TypeError(f"timeout must be a float or integer, got {timeout!r}")
# Start a new event loop with the awaitable in it.
loop = asyncio.new_event_loop()
task = loop.create_task(awaitable)
# Use tkinter's `.after` to run the asyncio event loop.
self.after(0, self.__advance_loop, loop, max(1, int(timeout * 1000)))
# Run tkinter, which periodically checks
self.mainloop()
# After tkinter is done, wait until `asyncio` is done.
try:
return loop.run_until_complete(task)
finally:
loop.run_until_complete(loop.shutdown_asyncgens())
loop.close()
def destroy(self, /) -> None:
super().destroy()
self.is_running = False
The example application may be fixed up like this:
import asyncio
from random import randrange
import tkinter as tk
def deg_color(deg, d_per_tick, color):
"""Helper function for updating the degree and color."""
deg += d_per_tick
if 360 <= deg:
deg %= 360
color = f"#{randrange(256):02x}{randrange(256):02x}{randrange(256):02x}"
return deg, color
async def rotator(root, interval, d_per_tick):
"""
An example custom method for running code asynchronously
instead of using `tkinter.Tk.after`.
NOTE: Code that can use `tkinter.Tk.after` is likely
preferable, but this may not fit all use-cases and
may sometimes require more complicated code.
"""
canvas = tk.Canvas(root, height=600, width=600)
canvas.pack()
deg = 0
color = 'black'
arc = canvas.create_arc(
100,
100,
500,
500,
style=tk.CHORD,
start=0,
extent=deg,
fill=color,
)
while root.is_running:
deg, color = deg_color(deg, d_per_tick, color)
canvas.itemconfigure(arc, extent=deg, fill=color)
await asyncio.sleep(interval)
def main():
root = AsyncTk()
root.run_with_mainloop(rotator(root, 1/60, 2))
if __name__ == "__main__":
main()
Answered By - Simply Beautiful Art
0 comments:
Post a Comment
Note: Only a member of this blog may post a comment.