Event systems

Implement a minimal publish/subscribe event bus so game objects communicate without holding direct references to each other.

When a player dies, several things need to happen: the lives counter decrements, a death animation plays, a sound effect fires, the camera shakes, and after a delay the game-over scene activates. If the Player class calls all of those directly, it needs a reference to the lives counter, the animation system, the sound manager, the camera, and the scene manager. That is five tight couplings. Add three more triggers and the class becomes impossible to test or reuse.

An event bus solves this. The Player publishes a PLAYER_DIED event and forgets about it. Each interested system has subscribed to that event and handles it independently. The Player and the lives counter never reference each other.

Pyodide (the in-browser Python runner) does not support pygame's display system. Read through this code carefully in the browser, then run it locally: pip install pygame followed by python eventbus.py.

Implementing the event bus

import pygame
import sys
from collections import defaultdict

# ── Event bus ────────────────────────────────────────────────────────────────

class EventBus:
    """Minimal publish/subscribe event bus."""

    def __init__(self):
        # Maps event_type (any hashable) → list of callables
        self._handlers = defaultdict(list)

    def subscribe(self, event_type, callback):
        """Register callback to be called when event_type is published."""
        self._handlers[event_type].append(callback)

    def publish(self, event_type, data=None):
        """Call every callback registered for event_type."""
        for cb in self._handlers[event_type]:
            cb(data)


# ── Event type constants ─────────────────────────────────────────────────────

PLAYER_DIED   = "player_died"
ENEMY_KILLED  = "enemy_killed"
SCORE_CHANGED = "score_changed"


# ── Game objects ─────────────────────────────────────────────────────────────

class Player:
    def __init__(self, bus):
        self.bus    = bus
        self.rect   = pygame.Rect(300, 220, 40, 40)
        self.health = 3

    def take_damage(self):
        self.health -= 1
        if self.health <= 0:
            self.bus.publish(PLAYER_DIED)

    def update(self, keys):
        if keys[pygame.K_LEFT]:  self.rect.x -= 4
        if keys[pygame.K_RIGHT]: self.rect.x += 4
        if keys[pygame.K_UP]:    self.rect.y -= 4
        if keys[pygame.K_DOWN]:  self.rect.y += 4

    def draw(self, screen):
        pygame.draw.rect(screen, (100, 210, 120), self.rect)


class HUD:
    """Listens for game events and updates display values."""

    def __init__(self, bus):
        self.score = 0
        self.font  = pygame.font.SysFont(None, 32)
        bus.subscribe(ENEMY_KILLED,  self._on_enemy_killed)
        bus.subscribe(PLAYER_DIED,   self._on_player_died)

    def _on_enemy_killed(self, data):
        points = data.get("points", 10) if data else 10
        self.score += points

    def _on_player_died(self, data):
        # Could flash the HUD, reduce lives, etc.
        pass

    def draw(self, screen):
        surf = self.font.render(
            f"Score: {self.score}   H = take damage", True, (255, 255, 255))
        screen.blit(surf, (10, 10))


# ── Main loop ────────────────────────────────────────────────────────────────

def main():
    pygame.init()
    screen = pygame.display.set_mode((640, 480))
    pygame.display.set_caption("Event bus demo")
    clock  = pygame.time.Clock()

    bus    = EventBus()
    player = Player(bus)
    hud    = HUD(bus)

    running = True
    while running:
        events = pygame.event.get()
        for event in events:
            if event.type == pygame.QUIT:
                running = False
            if event.type == pygame.KEYDOWN:
                if event.key == pygame.K_h:
                    player.take_damage()   # publishes PLAYER_DIED when health hits 0
                if event.key == pygame.K_k:
                    bus.publish(ENEMY_KILLED, {"points": 10})  # simulate a kill

        keys = pygame.key.get_pressed()
        player.update(keys)

        screen.fill((20, 20, 35))
        player.draw(screen)
        hud.draw(screen)
        pygame.display.flip()
        clock.tick(60)

    pygame.quit()
    sys.exit()


if __name__ == "__main__":
    main()

How it works

subscribe(event_type, callback) registers a callable against a key. The key is just a string constant — it can be anything hashable. Using module-level constants (PLAYER_DIED = "player_died") prevents typos and makes it easy to search for all usages.

publish(event_type, data) calls every registered callback with an optional data payload. Callbacks receive data as their single argument. Passing a dict lets you include context without changing every callback's signature.

The Player never imports HUD. When player.take_damage() calls self.bus.publish(PLAYER_DIED), it has no idea who is listening. You can add or remove listeners freely without touching Player.

One limitation: the bus as implemented here delivers events synchronously in the same frame. For deferred events — "play this animation then, on completion, trigger the next scene" — you would queue events and drain the queue at a controlled point in the update step. That is a straightforward extension of the same pattern.

Where to go next

Next: camera and viewport — moving the viewable window through a world that is larger than the screen.

Finished reading? Mark it complete to track your progress.

Implementing the event bus

How it works

Where to go next

On this page