⚠️ Beta State

PyBevy is in an early and experimental stage. The API is incomplete, subject to breaking changes without notice, and you should expect bugs. Many features are still under development.

Events and Observers

Decouple systems with events, observers, and lifecycle hooks.

Introduction

In a complex game, systems need to communicate without being tightly coupled. For example, a collision system detects a hit, and a scoring system reacts to it.

pybevy provides two mechanisms:

  • Events (Event): Data objects that carry information about something that happened.
  • Observers (On[T]): Systems that react to events when they are triggered.
  • Lifecycle hooks (OnAdd, OnRemove, OnDespawn): React to ECS changes.
from pybevy.prelude import *
from pybevy.app import ScheduleRunnerPlugin

Defining Events

Events are classes that inherit from Event. They carry data about what happened. Unlike components and resources, events do NOT need a decorator.

class ScorePoint(Event):
    def __init__(self, points: int, reason: str):
        self.points = points
        self.reason = reason
 
class GameOver(Event):
    pass

A Resource to Track Score

@resource
class Score(Resource):
    def __init__(self):
        self.value = 0

Triggering Events

Any system can trigger events using commands.trigger(). The event is queued and delivered to all matching observers during the command flush.

@component
class Enemy(Component):
    pass
 
def setup(commands: Commands):
    commands.insert_resource(Score())
    # Spawn some enemies
    commands.spawn(Enemy())
    commands.spawn(Enemy())
    commands.spawn(Enemy())
 
def kill_enemies(
    commands: Commands,
    query: Query[Entity, With[Enemy]],
):
    for entity in query:
        commands.despawn(entity)
        # Trigger an event that any observer can react to
        commands.trigger(ScorePoint(points=100, reason="enemy defeated"))

Observing Events

Observers are registered with app.add_observer(). When the matching event is triggered, the observer runs. The On[T] parameter gives access to the event data.

def on_score_point(trigger: On[ScorePoint], score: ResMut[Score]):
    event = trigger.event()
    score.value += event.points
    print(f"+{event.points} points ({event.reason})! Total: {score.value}")

Lifecycle Observers

You can also observe when components are added, removed, or entities are despawned. This is useful for cleanup logic or initialization.

def on_enemy_despawned(trigger: On[OnDespawn, Enemy]):
    entity = trigger.entity()
    print(f"Enemy entity {entity} was despawned")

Entity-Specific Observers

You can attach an observer to a specific entity using .observe() on EntityCommands. This observer only fires for events targeting that particular entity.

class TakeDamage(Event):
    def __init__(self, amount: int):
        self.amount = amount
 
def on_take_damage(trigger: On[TakeDamage]):
    event = trigger.event()
    entity = trigger.entity()
    print(f"Entity {entity} took {event.amount} damage")
 
def setup_player(commands: Commands):
    # This observer only fires for events on this specific entity
    commands.spawn(
        Name("Player"),
    ).observe(on_take_damage)

Running the App

We register global observers with add_observer() at the app level.

@entrypoint
def main(app: App) -> App:
    return (
        app
        .add_plugins(ScheduleRunnerPlugin.run_once())
        .add_observer(on_score_point)
        .add_observer(on_enemy_despawned)
        .add_systems(Startup, (setup, setup_player))
        .add_systems(Update, kill_enemies)
    )
 
if __name__ == "__main__":
    main().run()

When you run this, the kill_enemies system despawns all enemies and triggers ScorePoint events. The on_score_point observer reacts to each one, updating the score. The on_enemy_despawned lifecycle observer fires for each despawn.

This decoupling means the scoring system doesn't need to know how enemies are killed — it just reacts to the event.