Skip to content

Instantly share code, notes, and snippets.

@happy-monk
Created October 22, 2021 13:23
Show Gist options
  • Save happy-monk/103fcb66025a359a85fca64c218054f6 to your computer and use it in GitHub Desktop.
Save happy-monk/103fcb66025a359a85fca64c218054f6 to your computer and use it in GitHub Desktop.
Coroutines in Pyglet
import pyglet
from pyglet.graphics import Batch
import tasks
window = pyglet.window.Window()
batch = Batch()
@window.event
def on_draw():
window.clear()
batch.draw()
async def run():
img = pyglet.resource.image('sprite.png')
sprite = pyglet.sprite.Sprite(
img,
batch=batch,
)
while 1:
await move_around_square(sprite, 5, 100, 100, 300, 300)
async def move_around_square(sprite, t, x1, y1, x2, y2):
await move_sprite(sprite, t/4, x1, y1, x2, y1)
await move_sprite(sprite, t/4, x2, y1, x2, y2)
await move_sprite(sprite, t/4, x2, y2, x1, y2)
await move_sprite(sprite, t/4, x1, y2, x1, y1)
async def move_sprite(sprite, t, x1, y1, x2, y2):
phase = 0
finished = False
while not finished:
dt = await tasks.tick()
phase += dt / t
if phase >= 1.0:
phase = 1.0
finished = True
x = x1 * (1.0 - phase) + x2 * phase
y = y1 * (1.0 - phase) + y2 * phase
sprite.update(x=x, y=y)
tasks.TaskLoop().schedule_updates()
tasks.start_task(run())
if __name__ == '__main__':
pyglet.app.run()
from inspect import isawaitable
from typing import *
import contextvars
import pyglet.clock
T = TypeVar('T')
class TaskLoop:
current = contextvars.ContextVar['TaskLoop']('current')
def __init__(self) -> None:
self.dt = 0.0
self.tasks: Set['Task'] = set()
def schedule_updates(self):
self.current.set(self)
pyglet.clock.schedule(self._update)
def _update(self, dt: float) -> None:
self.dt = dt
self.current.set(self)
self._run_tasks()
def _run_tasks(self) -> None:
for task in list(self.tasks):
task.step()
if task.done:
self.tasks.discard(task)
def start_task(self, coro: 'Union[Task[T], Awaitable[T]]') -> 'Task[T]':
if isinstance(coro, Task):
return coro
assert isawaitable(coro)
task = Task(coro)
self.tasks.add(task)
return task
class Task(Generic[T]):
done = False
result: T = None
exception: Exception = None
def __init__(self, coro: Awaitable[T]) -> None:
self.coro = coro
self._iter = self.coro.__await__()
def step(self) -> None:
if self.done:
return
try:
next(self._iter)
except StopIteration as e:
self.result = e.value
self.done = True
except Exception as e:
self.exception = e
self.done = True
# TODO
import traceback
traceback.print_exc()
async def wait(self) -> None:
while not self.done:
await Yield
class YieldType:
def __await__(self):
yield
Yield = YieldType()
async def tick() -> float:
await Yield
return TaskLoop.current.get().dt
async def sleep(dt: float) -> None:
while 1:
dt -= await tick()
if dt <= 0.0:
return
def start_task(coro: Union[Task[T], Awaitable[T]]) -> Task[T]:
return TaskLoop.current.get().start_task(coro)
async def wait(coros: Iterable[Union[Task[Any], Awaitable[object]]]) -> None:
tasks = list(map(TaskLoop.current.get().start_task, coros))
while 1:
if all(t.done for t in tasks):
return
await tick()
@matanox
Copy link

matanox commented Dec 15, 2023

@happy-monk
Copy link
Author

happy-monk commented Jun 28, 2024

There's something experimental similar in ... now.

This is not the same. They are trying to use asyncio in separate for networking. I am trying to use coroutines inside pyglet event loop as a replacement for the state machines.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment