My Pygame Evolution: Embracing Asyncio and Immutability for Scalable Design

Maintaining a blog is a lot of work, especially when it comes to finding new ideas. Over the weekend, I explored Pygame by building a simple game. I was very humbled by the experience throughout the build. Aha, this exploration turned out to be a worthy idea for a new article! Read on to relive the struggles involved while building a graphical game via Pygame together.

A cute illustration on the topic, by Microsoft Copilot

The Tic-Tac-Toe Catalyst: Unveiling Pygame’s Low-Level Nature

Over the weekend, in the EngineersMY chatroom, a friend shared his work in a recent game jam held in his workplace. He built the project with p5.js, piqued my interest. Eventually I decided to proceed with Pygame, and that decision led me to hours of exploration and perhaps the expense of a few hair follicles. Despite that, the lower-level library nudged me to build a custom architectural design that can potentially evolve into a proper framework.

Photo by Ryan Snaadt on Unsplash

Discussions about my friend’s project reminded me of a less than ideal job application that happened around late last year. The offered package was unattractive, and I was expected to work on graphical animation alongside full-stack development. My submitted video processing pipeline work received satisfactory feedback, though the hiring manager came back with another assignment. He requested a new task: to build a graphical application with p5.js, and it was also when I formally ended the application. That mention of p5.js was the first time in a formal setting.

Interestingly, I was still given a job offer despite all that, which I ultimately rejected.

Fast forward to the weekend, I consulted with my AI assistant and started my research about Processing, a project where p5.js was based on. Apparently there are ports for Python, though the official processing.py only works with Jython. All other alternatives either relied on the Java package or required special libraries. Consequently, I stumbled upon Pygame after a series of aimless searches.

Unlike Processing, Pygame felt like a lower level library for building graphical applications. Partially inspired by the discussion on the game jam, but mostly out of boredom, I decided to give it a try. The realization of how low-level Pygame is came when I was figuring out the specification and scope of my over-ambitious initial plan. At the time, my main loop was just barely running.

As a project to explore graphical application building, I eventually decided to limit the scope further to start small. Tic-tac-toe seemed like a good idea, and I should be able to build a reasonably functional version over a weekend. How wrong I was, as I am still doing some last minute changes as of writing this article…

Building a Reactive Core: Event-Driven Design in Pygame

Pygame is synchronous, and most of the objects and operations are not thread-safe. Due to the interactive nature of graphical applications, they often require a carefully planned strategy to achieve concurrency. AsyncIO is an obvious candidate for the task, though getting it to work with Pygame, requires some thought. Much of the implementation takes a cue from Javascript, especially in the event dispatching department. Immutability of certain components is also crucial, though mainly to keep me sane.

Let’s start with the main loop, essentially we want something like this

def run() -> None:
    pygame.init()

    continue_to_run = True
    clock = pygame.time.Clock()

    pygame.display.set_mode((300, 300))

    while continue_to_run:
        clock.tick(60)

        continue_to_run = event_process(pygame.event.get())

        pygame.display.flip()

    pygame.quit()

A potential problem I see with this arrangement is that, both handle_events and pygame.display.flip() are inevitably blocking. If any of them stalled for a prolonged period, we will miss the target frame rate. Ideally, we want something like this:

async def run():
    pygame.init()

    exit_event = asyncio.Event()

    pygame.display.set_mode((300, 300))

    asyncio.create_task(display_update())
    asyncio.create_task(event_process(exit_event))

    await exit_event.wait()

    pygame.quit()

Photo by Acton Crawford on Unsplash

A window popped up, and my event handling work seemed functional. Yet, sometimes when the application exits I get a SIGSEGV segmentation fault when exiting the program. After some research on the error, I found out that Pygame is not thread-safe, hence I should really avoid asyncio.to_thread. This means the following code, delegating the blocking display.flip() call to another thread, is invalid:

async def display_update():
    clock = pygame.time.Clock()

    while True:
        clock.tick(60)

        await asyncio.to_thread(pygame.display.flip)

With that discovery, I settled with the original main loop convention for a while, and only moved the event dispatching work to a coroutine.

async def run() -> None:
    pygame.init()

    clock = pygame.time.Clock()
    exit_event = asyncio.Event()

    pygame.display.set_mode((300, 300))

    while not exit_event.is_set():
        clock.tick(60)

        asyncio.create_task(event_process(pygame.event.get(), exit_event))

        pygame.display.flip()

    pygame.quit()

Unfortunately, due to the synchronous blocking nature of Pygame, the event_process tasks do not have a chance to run, though scheduled successfully. As my coding companion, the chatbot suggested adding asyncio.sleep(0) into the loop. According to the documentation:

Setting the delay to 0 provides an optimized path to allow other tasks to run. This can be used by long-running functions to avoid blocking the event loop for the full duration of the function call.

Adding the statement to the end of the loop allows the application to finally respond to events.

Previously, I mentioned I performed a last-minute update while writing this article. With this knowledge, we can apply the asyncio.sleep(0) fix to both display_update and event_process, enabling AsyncIO to run the scheduled tasks. In turn, this allows the refactoring of the huge main loop shown in the first snippet into the second snippet.

Photo by Scott Rodgerson on Unsplash

Before diving into the event dispatching part of the design, let’s talk about how I manage shared data. Certain components need to be shared across different parts of the application. Graphical applications are extremely interactive, hence I would need to respond to them concurrently. In this case, interactivity poses a new challenge — keeping track of changes of these shared objects.

Much of the design came from my experience with development work in JavaScript. Within a JavaScript application, it is possible to dispatch and subscribe to events almost everywhere, and changes can also be made to a page whenever we see fit. Keeping a reference while staying aware of these changes is a nightmare, and it is extremely unfriendly to my hair follicle.

Grouping them together into one dataclass, which serves as a registry is usually my first step. In this case, I prefer them to stay immutable, as shown below:

@dataclass(frozen=True)
class Application:
    screen: pygame.Surface = pygame.Surface((0, 0))
    elements: tuple[Element, ...] = tuple()
    state: GameState = GameState.INIT

The three objects consist of a pygame.Surface object for the main application window, an element tuple holding all the interactive widgets in the surface, and an Enum holding the current game state. Events are dispatched to be executed concurrently, so each execution is completely isolated from one another. Therefore, we need to synchronize the changes made to these shared objects to avoid race conditions.

Aiming to maintain the simplicity inspired by functional programs, I started by marking the dataclass as frozen, as shown in the snippet. Changes to the elements and state objects are deferred to two queues I introduced to the registry dataclass.

@dataclass(frozen=True)
class Application:
    ...

    state_update: asyncio.Queue[GameState] = asyncio.Queue()
    element_delta: asyncio.Queue[DeltaOperation] = asyncio.Queue()

Making the dataclass immutable would require a regeneration of application registry for every batch of events.

async def events_process(application: Application):
    while True:
        application = await application_refresh(application)

        await events_dispatch(
            application,
            pygame.event.get(),
        )

        await asyncio.sleep(0)

Whenever an event handler makes a change to the game state or manipulates the widgets on screen, the changes are pushed to the queue instead. Then the queue is consumed, to update the corresponding object:

from dataclasses import replace

async def application_refresh(application: Application) -> Application:
    with suppress(asyncio.queues.QueueEmpty):
        elements = application.elements

        while delta := application.element_delta.get_nowait():
            match delta:
                case ElementAdd():
                    elements += (delta.item,)

                case ElementUpdate():
                    elements = tuple(
                        delta.new if element == delta.item else element
                        for element in elements
                    )

                case ElementDelete():
                    elements = tuple(
                        element for element in elements if not element == delta.item
                    )

    with suppress(asyncio.queues.QueueEmpty):
        state = application.state

        while state_new := application.state_update.get_nowait():
            state = state_new

    return replace(
        application,
        elements=elements,
        state=state
    )

Wrapping the widget element with an ElementAdd, an ElementUpdate and an ElementDelete is just a way to provide metadata to indicate the manipulation type. On the other hand, the replace function, provided by the dataclasses module, is just a helpful shorthand in this case. Alternatively, we can recreate the Application object as follows:

    return Application(
        application.screen,
        elements,
        state,
        application.state_update,
        application.element_delta
    )

Moving on to the event dispatching part.

Due to my inexperience, I didn’t spent too much time drafting a design or a plan when I worked on the game. Still, as I went deeper I realized I really needed to start structuring the event handling code. Keeping that in mind, I continued with the build until I managed to get a bare minimum version running.

Drawing inspiration from JavaScript, I started my refactoring by revising the event handler registration. Directly inspired by the document.addEventListener method, my interpretation of it in this setting took this form:

async def add_event_listener(target: Eventable, kind: int, handler: Callable[..., None]) -> Eventable:
    event = Event(kind, handler)
    result = target

    match target:
        case Application():
            await target.event_delta.put(DeltaAdd(event))

        case Element():
            result = replace(target, events=target.events + (event,))

        case _:
            raise Exception("Unhandled event registration")

    return result

Besides the widgets in the application window, the application itself can also subscribe to an event. For that, I added a tuple of events as well as an event_delta queue to the Application dataclass. This function always returns either the application registry, or the recreated element. Publishing the recreated element to the element_delta queue is handled by the caller.

Unlike the display_update function, there is no clock.tick() function to regulate the timing of execution in the event_process function shown previously. After finishing scheduling the current batch, we want to immediately start fetching and processing a new batch of events as soon as possible. The current dispatching function is rather straightforward, as it just schedules the registered handler by cycling through the events tuple in the application registry as well as each of the elements.

async def events_dispatch(
    application: Application,
    events: Sequence[pygame.event.Event]
) -> None:
    for event in events:
        match event.type:
            case pygame.QUIT:
                application.exit_event.set()

            case pygame.MOUSEBUTTONDOWN:
                for element in application.elements:
                    for ev in element.events:
                        if not ev.kind == event.type:
                            continue

                        if element.position.collidepoint(mouse_x, mouse_y):
                            asyncio.create_task(
                                ev.handler(ev, element, application)
                            )

Basically these is all the code mainly for handling the orchestration work; it contains no game logic. Like how I was working on BigMeow, a structure emerges when I work more on the project. When I work on a personal project, I also tend to avoid writing object-oriented code as much as possible, and I am glad how well the current revision turns out.

Photo by BoliviaInteligente on Unsplash

Being a complete newbie to Pygame, Gemini proved helpful yet again. When I talked about the project with a friend, he jokingly asked if I vibe-coded it. However, the whole experience was not unlike how I ported the LLM word game to AsyncSSH, where Gemini assisted by helping me to navigate the documentation. Granted, the code example is usually good enough, but it helps to know how and why things are written in a certain way.

Lessons Learned and Horizons Gained: My Pygame Journey Continues

Refactoring work started right after I was more or less done with handling mouse click and triggered a change to game state. As the code gets cleaned up, it occurs to me that extracting the game logic into a separate module becomes trivial. To my surprise, after getting Gemini to review the code, I was told the separation would yield a mini framework for similar projects.

Not knowing where I could get feedback, I turned to Gemini to review my code, and we chatted about how this project could be extended. That follow-up discussion was a very enlightening session, apart from the revelation of an emerging framework. Presumably, the Application dataclass resembles an Entity-Component-System. This was definitely a concept worth adding to my learning backlog.

Photo by Glenn Carstens-Peters on Unsplash

If time permits, I would like to work on the separation and complete the game. Curiosity is definitely the biggest motivation behind the effort, rather than publishing the potentially bad framework. New system events like window.onload and window.onclose equivalents may also be implemented to better simulate my web development workflow. Follow the project on GitHub (inline link to project) if you are interested.

Unfortunately, I will not have as much free time for a while. On the flip-side, I no longer have to entertain needy and exploitative hiring managers too. With my incoming new job offer, I will adjust the publishing frequency for the coming weeks. I would also like to take this opportunity to thank everyone offering encouragement and help throughout this trying period. Spring does come, after winter.

To ensure this post is as clear and readable as possible, I leveraged Gemini as my editorial assistant. Gemini also provided complementary information to Pygame’s official documentation, helping to clarify complex concepts. However, please note that the voice and all code presented within this article are entirely my own. If you found this article helpful or insightful, consider following me on Medium and subscribing for more content like this.