Coroutines in Pyglet

example.py

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()

tasks.py

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()

There's something experimental similar in https://github.com/pyglet/pyglet/blob/7df9ee869242f482579f1aa0ed5c61e39c4a444f/pyglet/experimental/net.py#L154C18-L154C42 now.

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.