Over the last weekend, I spent some time mucking around with Pygame. Despite facing some challenges along the way, I ended up with a working prototype. There was neither design nor specification drafted for the project. Nonetheless, mid-development, a reusable design emerged, and we discussed that in last week’s article. Following that discovery, we will explore the implementation this week and answer the question ‚ “Can we make a framework out of it?”.
Cute illustration from Copilot
The Unexpected Framework: From Pygame Hacks to a Reusable Design
Understanding Awaitables: Coroutines, Tasks, and Futures in Python
As a brief recap to the last article, we know how Pygame is designed to be primitive, and lacks inherent structure. Building a Pygame application would therefore require a developer to begin by writing the main application loop. On the other hand, this also presents an opportunity to design a bespoke structure for graphical application development. By refactoring our work from last week, we will witness how a framework emerges, while adhering the original design goal to stay immutable and reliable.
Adapting the synchronous main application loop to AsyncIO took quite some effort. Despite the complexity involved, it was worth it because it broke the tight coupling of display update and event dispatching loop. Previously, events were only fetched once per display update cycle. If the application was written in a pure event driven manner, delegating work into multiple events would introduce latency.
Now that both event dispatching and display update run independently, ensuring synchronization of application states between the two components is also crucial. Immutability was therefore introduced such that application data can only be changed through publishing changes to relevant queues.
With all the work laid out, this opens up an interesting potential, allowing the orchestration code for event dispatching and display updates to be split from the game logic. Application setup work, such as event handler registrations and the initialization of application-specific data, is the primary area that differs from one application to another.
Sounds like a plan, let’s split it out, and complete the tic-tac-toe game we started.
Photo by Daniel McCullough on Unsplash
Building the Core: Immutable State and Asynchronous Orchestration
It shouldn’t be difficult, right? Given the design goal mentioned earlier, the splitting work was indeed not too difficult. On the other hand, keeping track of the declared dataclasses is another story. Regardless, we will uncover an architectural pattern that forms the framework core through some refactoring work. Some minor revisions will be made to the state management, as well as the main application loop. All this work yields a reusable framework for simple graphical applications.
Previously we were creating multiple queues for different fields in our Application dataclass. Having multiple queues serving similar purpose to update values is fine, though I personally find it wasteful. Why not just consolidate them?
And I did exactly that.
Photo by USGS on Unsplash
Firstly, we start by introducing an Enum to represent the field we intend to update
class ApplicationDataField(Enum):
STATE = auto()
ELEMENTS = auto()
EVENTS = auto()
Secondly, we update the DeltaOperation to always require the field we intend to update
@dataclass
class DeltaOperation(ABC):
field: ApplicationDataField
item: Any
Lastly, we need to update the application_refresh function, which is the only place where the Application object is updated throughout the framework.
async def application_refresh(application: Application) -> Application:
result = application
with suppress(asyncio.queues.QueueEmpty):
while delta := application.delta_data.get_nowait():
field = application_get_field(application, delta.field)
match delta:
case DeltaAdd():
result = replace(
result,
**{field: getattr(result, field) + (delta.item,)},
)
case DeltaUpdate():
result = replace(
result,
**{
field: tuple(
delta.new if item == delta.item else item
for item in getattr(result, field)
)
},
)
case DeltaDelete():
result = replace(
result,
**{
field: tuple(
item
for item in getattr(result, field)
if not item == delta.item
)
},
)
return result
Actual implementation is a little bit longer, but I truncated for brevity. Logically, the current version is rather similar to the previous version, though it only consumes from one single queue (application.delta_data) to update the states in an Application object.
After the minor refactoring to the state management, we shift our attention to the main application loop.
async def main_loop(
application_setup: Awaitable[Application],
) -> None:
pygame.init()
application = await application_setup
pygame.event.post(pygame.event.Event(SystemEvent.INIT.value))
tasks = []
tasks.append(
asyncio.create_task(
display_update(
application.delta_screen,
application.clock,
)
)
)
tasks.append(
asyncio.create_task(events_process(application))
)
await application.exit_event.wait()
pygame.quit()
It is almost similar to the original run function in our original code, with just 2 small changes. Remember I mentioned about wanting a window.onload like system events? Here we are emitting an INIT event before we start the display update and event dispatching loops. Additionally, setup code is passed in as anAwaitable object, that is only scheduled to run after pygame.init().
Understanding Awaitables: Coroutines, Tasks, and Futures in Python
That’s practically it, though extracting them out to another module, is a pain due to how these functions reference numerous dataclasses and other helpers. Alongside these changes, I also added some helpers to deal with in-game state management. Spiritually, the design of these helper functions loosely follows Redux, though the implementation is a lot simpler in my implementation. We shall see them in action when we go through the game implementation next.
Framework in Action: Bringing Tic-Tac-Toe to Life
Framework is now done, what’s next?
What’s a better way to test the framework, than building a game for real? Now that we delegate the orchestration work to another module, we can now focus purely on the game logic. Throughout the discussion, we can observe how the framework design enables building of a simple game with just dispatching events and implementing the handlers. Game state management as mentioned earlier will be discussed as well.
Firstly, let’s set-up the game:
async def setup() -> Application:
application = Application(screen=pygame.display.set_mode((300, 300)))
application = await add_event_listeners(
application,
(
(SystemEvent.INIT.value, handle_init),
(CustomEvent.RESET.value, handle_reset),
(CustomEvent.MAP_UPDATE.value, handle_update),
(CustomEvent.JUDGE.value, handle_judge),
),
)
return application
A new helper add_event_listeners was written for adding multiple event listeners, though they work similarly as the add_event_listener we discussed last week. In case you are wondering at this point, yes, this is all we needed to implement our tic-tac-toe game. Since we split out the orchestration code to another module, our run function now becomes:
from core import main_loop
async def run() -> None:
await main_loop(setup())
Remember main_loop() takes an Awaitable as argument? This is the reason why setup() was written as a coroutine function, and passed into main_loop function without the await keyword.
Before we start the display update and event dispatching cycles, let’s initialize the game by drawing relevant elements to the application screen object:
async def handle_init(
_event: pygame.event.Event, target: Application, logger: BoundLogger, **detail: Any
):
for row, col in product(range(3), range(3)):
element = await box_draw(target, row, col, 100)
element = await add_event_listener(
element,
pygame.MOUSEBUTTONDOWN,
handle_box_click,
)
await target.delta_data.put(DeltaAdd(ApplicationDataField.ELEMENTS, element))
await screen_update(target, element) # type: ignore
pygame.event.post(pygame.event.Event(CustomEvent.RESET.value))
During the initialization phase, the handler draws the 3 x 3 tiles to the screen via box_draw, and then registers a mouse click event to each of them. Each tile is then sent to 2 queues, one to populate the application.elements tuple (via target.delta_data.put), and another to update the display (via screen_update). Once done, the handler invokes the RESET event to start a new game.
async def handle_reset(
_event: pygame.event.Event, target: Application, logger: BoundLogger, **detail: Any
):
await state_merge(target, "board", winner=None, state=GameState.RING, board=None)
for element in target.elements.values():
delta = await box_redraw(target, (255, 255, 255), element, Symbol.EMPTY) # type: ignore
await target.delta_data.put(
DeltaUpdate(ApplicationDataField.ELEMENTS, element, delta)
)
await screen_update(target, delta)
As the game is resetting, it first reinitializes the game state via state_merge.The helper sends a DeltaOperation to update application.state to
{
"board": {
"winner": None,
"state": GameState.END,
"board": None
}
}
Elements in the game board also get redrawn (via box_redraw) to ensure they do not contain a ring or cross symbol. As we did earlier, all of them are then sent to the queues for application and display update.
Our game is now accepting input. And once a click is registered, the registered event handler is called.
async def handle_box_click(
_event: pygame.event.Event,
target: Box,
application: Application,
logger: BoundLogger,
**detail: Any,
) -> None:
element = None
match state_get(application, "board").get("state"):
case GameState.END:
await logger.ainfo("END")
pygame.event.post(pygame.event.Event(CustomEvent.RESET.value))
case GameState.TIE:
await logger.ainfo("TIE")
pygame.event.post(pygame.event.Event(CustomEvent.RESET.value))
case GameState.RING if target.value == Symbol.EMPTY:
element = await box_update(
application,
target,
Symbol.RING,
)
await state_merge(application, "board", state=GameState.CROSS)
case GameState.CROSS if target.value == Symbol.EMPTY:
element = await box_update(
application,
target,
Symbol.CROSS,
)
await state_merge(application, "board", state=GameState.RING)
case _:
await logger.aerror("Invalid click")
if element:
pygame.event.post(pygame.event.Event(CustomEvent.MAP_UPDATE.value))
Click events are handled differently, depending on the state of the board. If there is no valid moves available, in the events of an ended or tie game, then we log a message, and emit a RESET event. Otherwise, we update the clicked element to the symbol corresponding to the current state of the board, followed by a board state change for the next symbol. Lastly, we trigger a MAP_UPDATE event to update the current mapping of the board for judging.
@dataclass
class BoardMap:
mapping: dict[tuple[int, int], Symbol]
async def handle_update(
_event: pygame.event.Event, target: Application, logger: BoundLogger, **detail: Any
):
await state_merge(
target,
"board",
map=BoardMap(
{
(item.column, item.row): item.value
for item in target.elements.values()
if isinstance(item, Box)
}
),
)
pygame.event.post(pygame.event.Event(CustomEvent.JUDGE.value))
BoardMap is just a dataclass keeping the current state of the board, what symbol is placed at each tile. As soon as it is updated, a JUDGE event is triggered to evaluate the game.
WIN_LINES = [
# Rows
{(0, 0), (0, 1), (0, 2)},
{(1, 0), (1, 1), (1, 2)},
{(2, 0), (2, 1), (2, 2)},
# Columns
{(0, 0), (1, 0), (2, 0)},
{(0, 1), (1, 1), (2, 1)},
{(0, 2), (1, 2), (2, 2)},
# Diagonals
{(0, 0), (1, 1), (2, 2)},
{(0, 2), (1, 1), (2, 0)},
]
async def handle_judge(
_event: pygame.event.Event, target: Application, logger: BoundLogger, **detail: Any
):
if not (bmap := state_get(target, "board").get("map")):
return
tiles, winner = (), None
for symbol in (Symbol.RING, Symbol.CROSS):
places = {coor for coor, item in bmap.mapping.items() if symbol == item}
for line in WIN_LINES:
if line.issubset(places):
winner = symbol
tiles = line
break
if winner:
break
if winner:
await logger.ainfo("Winner is found", winner=winner)
await state_merge(target, "board", winner=winner, state=GameState.END)
for element in target.elements.values():
if (element.column, element.row) in tiles: # type: ignore
delta = await box_redraw(target, (255, 255, 0), element, winner) # type: ignore
await target.delta_data.put(
DeltaUpdate(ApplicationDataField.ELEMENTS, element, delta)
)
await screen_update(
target,
delta, # type: ignore
)
elif (
len([symbol for _, symbol in bmap.mapping.items() if symbol == Symbol.EMPTY])
== 0
):
await logger.ainfo("End with a tie")
await state_merge(target, "board", state=GameState.TIE)
Not the most robust code I would proudly print on a t-shirt and parade with it. Essentially, it judges the game to find the winning line and symbol. Redrawing the winning tiles will take place whenever applicable. Besides that, the relevant state of the board is also queued for update (either an END or TIE) whenever needed.
By utilizing events and state_merge strategically, the simple game of tic-tac-toe is implemented. All the event dispatching work and display update are completely handled by the framework itself. All we needed to do, when changes occur, is to submit the change to the relevant queue. While there are certainly potential improvements to be made, as a weekend project I deem it good enough.
Beyond Tic-Tac-Toe: Lessons Learned and What’s Next
Can we build more things with it?
Short answer to the question is yes, but I don’t see much point in doing so for most people. Instead, we should reflect on what we learned so far. What is the purpose of the project? What have we achieved so far? What do we do with this piece of code?
Extracting the orchestration code into a framework is a totally unexpected outcome. Yet, I am happy with the turns of events leading to what we have now. With some polish, the resulting framework should be able to be applied for other simple graphical applications. Personally, I am quite curious on seeing how the framework scales with more elements on screen.
Photo by Dan Freeman on Unsplash
As an experimental project, it certainly is quite rough on the edges. Much thought and polish are still needed in the framework API design for consistency. There are also some experimental code snippets that are not covered in the article as they are irrelevant to the game featured this week. Some of the most interesting helpers I can’t wait to try out are the functions inspired by setTimeout and element.dispatchEvent().
Perhaps a re-implementation of the code rain animation a-la the Matrix?
Code Rain’s Revelation: Embracing Existence Before Perfection
Now, back to the original question asked earlier, at the beginning of this section. The framework is proven capable of building a simple game, as shown in this article. Despite that, do go for a proper game engine like Godot for serious projects. Treat the work presented here as a personal passion project, though I am open for feedback and collaboration. Seriously, this exploration makes me appreciate how game engines do all the heavy-lifting work.
Finally, thanks for reading this far. If you are interested in checking out the project, it is still hosted on GitHub at the same repository. Though I cannot commit to a fixed publication schedule due to my new job, I shall try writing again soon.
A quick note on how this article came together. While Gemini helped with the drafting and refining of the language, all the core ideas, the technical insights, and every line of code you’ve seen are entirely mine. This project truly reflects my personal journey and passion for game development. If you enjoyed this deep dive and want to follow more of my explorations into coding and game development, consider subscribing or following me on Medium. Your support means a lot!
Top comments (0)