EventBus¶

Publish-subscribe event system for loose coupling.

Location¶

src/pedre/events/base.py

Key Methods¶

subscribe(event_type: type[Event], callback: Callable) -> None

Subscribe to an event type.

Parameters:

event_type - Event class to listen for
callback - Function to call when event is published

Example:

def on_dialog_closed(event: DialogClosedEvent):
    print(f"Dialog with {event.npc_name} closed")

event_bus.subscribe(DialogClosedEvent, on_dialog_closed)

publish¶

publish(event: Event) -> None

Publish an event to all subscribers.

Parameters:

event - Event instance to publish

Example:

from pedre.plugins.events import DialogClosedEvent

event_bus.publish(DialogClosedEvent(
    npc_name="merchant",
    dialog_level=1
))

unsubscribe¶

unsubscribe(event_type: type[Event], callback: Callable) -> None

Unsubscribe from an event type.

Example:

event_bus.unsubscribe(DialogClosedEvent, on_dialog_closed)

Available Events¶

Event	Fields	Description
`DialogClosedEvent`	`npc_name`, `dialog_level`	Dialog window closed
`DialogOpenedEvent`	`npc_name`, `dialog_level`	Dialog window opened
`NPCInteractedEvent`	`npc_name`, `dialog_level`	Player interacted with NPC
`NPCMovementCompleteEvent`	`npc_name`	NPC reached destination
`NPCAppearCompleteEvent`	`npc_name`	NPC appear animation finished
`NPCDisappearCompleteEvent`	`npc_name`	NPC disappear animation finished
`InventoryClosedEvent`	`has_been_accessed`	Inventory screen closed
`ItemAcquiredEvent`	`item_id`	Item added to inventory
`ItemAcquisitionFailedEvent`	`item_id`, `reason`	Item could not be acquired
`ItemConsumedEvent`	`item_id`, `category`	Item consumed from inventory
`ObjectInteractedEvent`	`object_name`	Player interacted with object
`PortalEnteredEvent`	`portal_name`	Player entered a portal zone
`SceneStartEvent`	`scene_name`	Scene finished loading
`ScriptCompleteEvent`	`script_name`	Script finished executing

Creating Custom Events¶

from dataclasses import dataclass
from pedre.plugins.events import Event

@dataclass
class ItemCollectedEvent(Event):
    """Published when player collects an item."""
    item_name: str
    category: str
    position: tuple[float, float]

# Publish
event_bus.publish(ItemCollectedEvent(
    item_name="golden_key",
    category="keys",
    position=(320, 240)
))

# Subscribe
def on_item_collected(event: ItemCollectedEvent):
    print(f"Collected: {event.item_name}")

event_bus.subscribe(ItemCollectedEvent, on_item_collected)