Event handling module for Eventure.
This module provides the core Event and EventBus classes for implementing a robust event system with type-safe event handling and wildcard subscriptions.
@dataclass
class Event()Represents a single game event that occurred at a specific tick.
Events are immutable records of state changes in the game. Each event:
- Is tied to a specific tick number
- Has a UTC timestamp for real-world time reference
- Contains a type identifier for different kinds of events
- Includes arbitrary data specific to the event type
- Has a unique event_id in the format tick-typeHash-sequence
- May reference a parent event that caused this event (for cascade tracking)
Arguments:
tick- Game tick when the event occurredtimestamp- UTC timestamp when the event occurredtype- Event type from the EventType enumdata- Dictionary containing event-specific dataid- Optional explicit event ID (generated if not provided)parent_id- Optional ID of the parent event that caused this one
UTC timestamp
Will be set in post_init
Reference to parent event that caused this one
def to_json() -> strConvert event to JSON string for storage or transmission.
@classmethod
def from_json(cls, json_str: str) -> "Event"Create event from JSON string for loading or receiving.
Event logging module for Eventure.
This module provides the EventLog class for managing and storing events in the game.
class EventLog()Manages the sequence of game events and provides replay capability.
The EventLog is the core of the game's state management system:
- Maintains ordered sequence of all events
- Tracks current tick number
- Provides methods to add events and advance time
- Handles saving and loading of event history
- Supports tracking cascades of related events
The event log can be saved to disk and loaded later to:
- Restore a game in progress
- Review game history
- Debug game state issues
- Analyze gameplay patterns
- Trace causality chains between events
@property
def current_tick() -> intCurrent game tick number.
Ticks are the fundamental unit of game time. Each tick can contain zero or more events that modify the game state.
def advance_tick() -> NoneAdvance to next tick.
This should be called once per game update cycle. Multiple events can occur within a single tick, but they will always be processed in the order they were added.
def add_event(type: str,
data: Dict[str, Any],
parent_event: Optional[Event] = None) -> EventAdd a new event at the current tick.
Arguments:
type- Event type as a stringdata- Dictionary containing event-specific dataparent_event- Optional parent event that caused this event (for cascade tracking)
Returns:
The newly created and added Event
Notes:
Events are immutable once created. To modify game state, create a new event rather than trying to modify existing ones.
def get_events_at_tick(tick: int) -> List[Event]Get all events that occurred at a specific tick.
This is useful for:
- Debugging what happened at a specific point in time
- Processing all state changes for a given tick
- Analyzing game history
def get_event_by_id(event_id: str) -> Optional[Event]Get an event by its unique ID.
Arguments:
event_id- The unique ID of the event to find
Returns:
The event with the given ID, or None if not found
def get_event_cascade(event_id: str) -> List[Event]Get the cascade of events starting from the specified event ID.
This returns the event with the given ID and all events that have it as an ancestor in their parent chain.
Arguments:
event_id- The ID of the root event in the cascade
Returns:
A list of events in the cascade, ordered by tick and sequence
def create_query()Create an EventQuery instance for this event log.
Returns:
An EventQuery instance that can be used to visualize and analyze this event log.
def save_to_file(filename: str) -> NoneSave event log to file.
The entire game state can be reconstructed from this file. Each event is stored as a separate line of JSON for easy parsing and appending.
@classmethod
def load_from_file(cls, filename: str) -> "EventLog"Load event log from file.
Creates a new EventLog instance and populates it with events from the saved file. The current tick is set to the highest tick found in the loaded events.
Event bus module for Eventure.
This module provides the EventBus class for publishing events and subscribing to them.
class EventBus()Central event bus for publishing events and subscribing to them.
The EventBus decouples event producers from event consumers, allowing components to communicate without direct references to each other.
Features:
- Subscribe to specific event types
- Publish events to all interested subscribers
- Automatic event creation with current tick and timestamp
- Support for event cascade tracking through parent-child relationships
def __init__(event_log: EventLog)Initialize the event bus.
Arguments:
event_log- Reference to an EventLog for event creation and tick information
def subscribe(event_type: str, handler: Callable[[Event],
None]) -> Callable[[], None]Subscribe a handler to a specific event type.
Arguments:
event_type- The type of event to subscribe to as a string. Supports three types of wildcards:- Global wildcard "*" to receive all events regardless of type
- Prefix wildcard "prefix.*" to receive all events with the given prefix
- Suffix wildcard "*.suffix" to receive all events with the given suffix
handler- Function to call when an event of this type is published
Returns:
A function that can be called to unsubscribe the handler
Examples:
```python
# Subscribe to a specific event type
bus.subscribe("player.move", on_player_move)
# Subscribe to all player events
bus.subscribe("player.*", on_any_player_event)
# Subscribe to all error events
bus.subscribe("*.error", on_any_error_event)
# Subscribe to all events
bus.subscribe("*", on_any_event)
```
def publish(event_type: str,
data: Dict[str, Any],
parent_event: Optional[Event] = None) -> EventPublish an event to all subscribers.
Arguments:
event_type- The type of event to publish as a stringdata- Dictionary containing event-specific dataparent_event- Optional parent event that caused this event (for cascade tracking)
Returns:
The created event
Notes:
This method adds the event to the event log. It also dispatches the event to all subscribers.
def dispatch(event: Event) -> NoneDispatch the event to all interested subscribers.
Arguments:
event- The event to dispatch
Notes:
This method supports three types of wildcard subscriptions:
- Global wildcard "*" which will receive all events regardless of type
- Prefix wildcard "prefix.*" which will receive all events with the given prefix
- Suffix wildcard "*.suffix" which will receive all events with the given suffix
The event is dispatched to handlers in this order:
- Exact type match subscribers
- Prefix wildcard subscribers
- Suffix wildcard subscribers
- Global wildcard subscribers
Event query and visualization module for Eventure.
This module provides tools for querying and visualizing event logs, including event cascade relationships and parent-child event tracking.
class EventQuery()Provides query and visualization capabilities for event logs.
This class offers methods to analyze and display event relationships, helping with debugging and understanding complex event cascades.
def __init__(event_log: EventLog)Initialize with an event log to query.
Arguments:
event_log- The event log to query and visualize
def print_event_cascade(file: TextIO = sys.stdout,
show_data: bool = True) -> NonePrint events organized by tick with clear cascade relationships. Optimized for showing parent-child relationships within the same tick.
This method provides a visual representation of the event log, showing how events relate to each other across ticks and within the same tick. It's especially useful for debugging complex event sequences and understanding cause-effect relationships between events.
Arguments:
file- File-like object to print to (defaults to stdout).show_data- Whether to show event data (defaults to True).
def get_events_by_type(event_type: str) -> List[Event]Get all events matching a specific type.
Arguments:
event_type- Event type to filter by
Returns:
List of matching events
def get_events_by_data(key: str, value: Any) -> List[Event]Get all events with matching data key-value pair.
Arguments:
key- Key in event data to matchvalue- Value to match against
Returns:
List of matching events
def get_child_events(parent_event: Event) -> List[Event]Get all events that are direct children of the given event.
Arguments:
parent_event- Parent event to find children for
Returns:
List of child events
def get_cascade_events(root_event: Event) -> List[Event]Get all events in the cascade starting from the given root event. This includes the root event, its children, their children, etc.
Arguments:
root_event- Root event to find cascade for
Returns:
List of events in the cascade (including the root)
def print_event_details(event: Event,
file: TextIO = sys.stdout,
show_data: bool = True) -> NonePrint details of a single event.
Arguments:
event- Event to print details forfile- File-like object to print toshow_data- Whether to show event data
def print_single_cascade(root_event: Event,
file: TextIO = sys.stdout,
show_data: bool = True) -> NonePrint a single event cascade starting from the root event.
Arguments:
root_event- Root event to start cascade fromfile- File-like object to print toshow_data- Whether to show event data
def count_events_by_type() -> Dict[str, int]Count events by type.
Returns:
Dictionary mapping event types to counts
def get_events_at_tick(tick: int) -> List[Event]Get all events that occurred at a specific tick.
Arguments:
tick- Tick number to filter by
Returns:
List of events at the specified tick
def get_root_events(tick: Optional[int] = None) -> List[Event]Get all root events, optionally filtered by tick.
Root events are those with no parent or whose parent is in a previous tick.
Arguments:
tick- Optional tick to filter by
Returns:
List of root events