Skip to content

Instantly share code, notes, and snippets.

@pedrovhb
Last active December 26, 2022 23:56
Show Gist options
  • Save pedrovhb/2a1ebaf65c0ceaf317bb88400db04c3d to your computer and use it in GitHub Desktop.
Save pedrovhb/2a1ebaf65c0ceaf317bb88400db04c3d to your computer and use it in GitHub Desktop.
A quick demo/proof of concept of running dearpygui with an asyncio object oriented interface.
from __future__ import annotations
import asyncio
from asyncio import Queue
from typing import NamedTuple, AsyncIterable, AsyncIterator, Any, Callable
import dearpygui.dearpygui as dpg
from loguru import logger
class ButtonEvent(NamedTuple):
app_data: dict
user_data: dict
sender: int
button_object: Button
class SliderEvent(NamedTuple):
app_data: int
user_data: dict
sender: int
slider_object: Slider
class Button(AsyncIterable[ButtonEvent]):
def __init__(self, window) -> None:
self._id = dpg.add_button(
label="Click Me",
callback=self._callback,
parent=window,
drag_callback=self._callback,
drop_callback=self._callback,
)
self._loop = asyncio.get_event_loop()
self._next_event: asyncio.Future[ButtonEvent] = self._loop.create_future()
async def __anext__(self) -> ButtonEvent:
return await self._next_event
def __aiter__(self) -> AsyncIterator[ButtonEvent]:
return self
async def clicked(self) -> ButtonEvent:
return await self._next_event
def _callback(self, sender: int, app_data: Any, user_data: Any) -> None:
event = ButtonEvent(app_data, user_data, sender, self)
# logger.debug(f"Button clicked: {event}")
self._next_event.set_result(event)
self._next_event = self._loop.create_future()
class Slider(AsyncIterable[ButtonEvent]):
def __init__(self, window) -> None:
self._id = dpg.add_slider_int(
label="Click Me",
callback=self._callback,
parent=window,
)
self._loop = asyncio.get_event_loop()
self._next_event: asyncio.Future[SliderEvent] = self._loop.create_future()
async def __anext__(self) -> SliderEvent:
return await self._next_event
def __aiter__(self) -> AsyncIterator[SliderEvent]:
return self
async def slode(self, predicate: Callable[[SliderEvent], bool] | None = None) -> SliderEvent:
if predicate is None:
return await self._next_event
async for event in self:
if predicate(event):
return event
def _callback(self, sender: int, app_data: int, user_data: dict[object, object]) -> None:
event = SliderEvent(app_data, user_data, sender, self)
self._next_event.set_result(event)
self._next_event = self._loop.create_future()
class DPGApp:
def __init__(
self,
title: str = "Custom Title",
width: int = 600,
height: int = 200,
) -> None:
dpg.create_context()
self._title = title
self._width = width
self._height = height
dpg.create_viewport(
title=self._title,
width=self._width,
height=self._height,
clear_color=[0, 0, 10, 255],
)
dpg.setup_dearpygui()
self._window = "_window"
with dpg.window(label=self._title) as self._window:
dpg.add_text("Hello, world")
self._button = self.add_button()
dpg.show_viewport()
self._running = True
self._loop = asyncio.get_event_loop()
self._initial_time = self._loop.time()
self._last_frame_time = self._initial_time
self._frames = 0
self._target_fps = 60
self._frame_time = 1 / self._target_fps
self._run()
def add_button(self) -> Button:
return Button(self._window)
def add_slider(self) -> Slider:
return Slider(self._window)
def _run(self):
if self._running:
self._loop.call_at(self._initial_time + self._frame_time * self._frames, self._run)
dpg.render_dearpygui_frame()
self._frames += 1
if self._frames % 100 == 0:
print(f"Avg. FPS: {self._frames / (self._loop.time() - self._initial_time)}")
print(f"Current FPS: {1 / (self._loop.time() - self._last_frame_time)}")
self._last_frame_time = self._loop.time()
def stop(self):
self._running = False
def __del__(self):
dpg.destroy_context()
async def main() -> None:
app = DPGApp()
button = app.add_button()
slider = app.add_slider()
i = 0
async def handle_slider(slider: Slider) -> None:
async for event in slider:
if event.app_data > 75:
print("Slider is above 75!")
elif event.app_data < 25:
print("Slider is below 25!")
asyncio.create_task(handle_slider(slider))
async for _ in button:
print(f"Button clicked - {i}")
i += 1
if i == 4:
break
logger.warning("Don't click the button anymore. You walk around, you find out.")
await button.clicked()
logger.warning("Now you dun goofed")
await slider.slode(lambda ev: ev.app_data > 75)
logger.error("BY A LOT")
logger.error("Self destructing in 3 seconds")
await asyncio.sleep(3)
app.stop()
if __name__ == "__main__":
asyncio.run(main())
async def main() -> None:
app = DPGApp()
button = app.add_button()
slider = app.add_slider()
async def handle_slider(slider: Slider) -> None:
async for event in slider:
if event.app_data > 75:
print("Slider is above 75!")
asyncio.create_task(handle_slider(slider)) # Continuously monitor the slider
# The below "blocks" execution (really only of the currently
# coroutine; the task above for instance will continue to
# run for every slider change) until the button is clicked 5 times
i = 0
async for _ in button:
print(f"Button clicked - {i}")
i += 1
if i == 4:
logger.info("Button clicked way too many times")
break
# Below, the loop is again "blocked" until the button is clicked once more
await button.clicked()
# Await for the slider value to be changed to a value above 75 *after* the button
# having been clicked for the 6th time above
await slider.slode(lambda ev: ev.app_data > 75)
# The program continues running and responding to input (e.g. button clicks, slider
# changes) as per the other running coroutines for 10 more seconds
logger.error("Self destructing in 10 seconds")
await asyncio.sleep(10)
app.stop()
class Slider(AsyncIterable[ButtonEvent]):
def __init__(self, window) -> None:
self._id = dpg.add_slider_int(
label="Click Me",
callback=self._callback,
parent=window,
)
self._loop = asyncio.get_event_loop()
self._next_event: asyncio.Future[SliderEvent] = self._loop.create_future()
async def __anext__(self) -> SliderEvent:
return await self._next_event
def __aiter__(self) -> AsyncIterator[SliderEvent]:
return self
async def slode(self, predicate: Callable[[SliderEvent], bool] | None = None) -> SliderEvent:
if predicate is None:
return await self._next_event
async for event in self:
if predicate(event):
return event
def _callback(self, sender: int, app_data: int, user_data: dict[object, object]) -> None:
event = SliderEvent(app_data, user_data, sender, self)
self._next_event.set_result(event)
self._next_event = self._loop.create_future()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment