⚠️ 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.ecs import Event, OnDespawn
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}")

Full System Parameters in Observers

Observers support the same system parameters as regular systems: Commands, Query, Res, ResMut, Assets, AssetServer, View, MessageWriter, and MessageReader. This makes observers powerful enough to respond to events by modifying the world — not just reading data.

Here's an observer that uses Commands and Query to react to an event by spawning a new entity and modifying existing ones:

@component
class ScoreBonus(Component):
    pass
 
def on_score_with_bonus(
    trigger: On[ScorePoint],
    commands: Commands,
    score: ResMut[Score],
    query: Query[Entity, With[ScoreBonus]],
):
    event = trigger.event()
    # Use the score resource
    score.value += event.points
    # Use a query to check for bonus entities
    bonus_count = len(query)
    if bonus_count > 0:
        score.value += event.points  # Double points if any bonus exists
    # Use commands to spawn a celebration entity
    commands.spawn(Name(f"celebration_{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.

Timers and Time

Core Features

States and Run Conditions

Core Features