Event#
- class bamengine.core.event.Event[source]#
Bases:
ABCBase class for all events (systems) in the BAM-ECS architecture.
An Event encapsulates economic logic that operates on roles and mutates simulation state in-place. Events are executed by the Pipeline in the exact order specified.
- Class Attributes:
name (
str) – Event name in snake_case, automatically derived from class name.
Design Guidelines
Inherit from Event and implement execute() method
Use name class variable for unique identification
Events receive full Simulation instance for maximum flexibility
Use self.get_logger() for event-specific logging
Examples
Define an event using traditional syntax:
>>> from dataclasses import dataclass >>> from bamengine.core import Event >>> from bamengine.simulation import Simulation >>> >>> @dataclass(slots=True) ... class MyCustomEvent(Event): ... def execute(self, sim: Simulation) -> None: ... logger = self.get_logger() ... logger.info("Executing custom logic") ... # Mutate simulation state here >>> # Event is auto-registered as 'my_custom_event'
Define an event using the @event decorator (simplified):
>>> from bamengine import event >>> >>> @event ... class AnotherEvent: ... def execute(self, sim): ... pass # Implementation here >>> # Event is auto-registered as 'another_event'
Execute an event:
>>> import bamengine as bam >>> sim = bam.Simulation.init(n_firms=100, seed=42) >>> event = sim.get_event("firms_adjust_price") >>> event.execute(sim)
Per-event logging configuration:
>>> logger = event.get_logger() >>> logger.info("Starting execution") >>> if logger.isEnabledFor(logging.DEBUG): ... logger.debug("Expensive debug info: %s", compute_stats())
Notes
Events are registered automatically via __init_subclass__ hook. The order of event execution is critical and must be explicitly defined in the pipeline configuration (no automatic dependency sorting).
See also
- name = ''#
- classmethod __init_subclass__(name='', **kwargs)[source]#
Auto-register Event subclasses in the global registry.
This hook is called automatically when a class inherits from Event. It handles event registration and automatic name conversion.
- Parameters:
name (
str, optional) – Custom name for the event. If not provided, uses the class name converted to snake_case.**kwargs (
Any) – Additional keyword arguments passed to parent __init_subclass__.
Notes
This method is called twice when using @dataclass(slots=True): once during class definition and once when dataclass creates the final class. The name preservation logic handles this correctly.
Examples
Normal usage (automatic snake_case conversion):
>>> @dataclass(slots=True) ... class FirmsAdjustPrice(Event): ... def execute(self, sim): ... pass >>> # Registered as 'firms_adjust_price'
Custom name:
>>> @dataclass(slots=True) ... class MyEvent(Event, name="custom_event_name"): ... def execute(self, sim): ... pass >>> # Registered as 'custom_event_name'
- get_logger()[source]#
Get logger for this event with per-event log level applied.
- Returns:
Logger instance with event-specific configuration.
- Return type:
logging.BamLogger
Examples
Use logger in event execute method:
>>> class MyEvent(Event): ... def execute(self, sim): ... logger = self.get_logger() ... logger.info("Starting execution") ... # Expensive computation ... if logger.isEnabledFor(logging.DEBUG): ... logger.debug("Details: %s", expensive_stats())
Configure per-event log levels:
>>> # In config YAML or kwargs >>> config = { ... "logging": { ... "events": { ... "firms_adjust_price": "DEBUG", ... "labor_market_round": "WARNING", ... } ... } ... }
Notes
Logger name format: ‘bamengine.events.{event_name}’ Per-event log levels can be configured via config/defaults.yml or kwargs. Use isEnabledFor() to avoid expensive computations when logging is disabled.
- abstractmethod execute(sim)[source]#
Execute the event’s logic.
Mutates simulation state in-place. This method must be implemented by all Event subclasses.
- Parameters:
sim (
Simulation) – The simulation instance containing all state and configuration.- Returns:
All mutations are in-place.
- Return type:
Examples
Implement execute in a custom event:
>>> from bamengine import event, ops >>> >>> @event ... class CustomPricingEvent: ... def execute(self, sim): ... prod = sim.get_role("Producer") ... # Apply 10% markup to all prices ... new_prices = ops.multiply(prod.price, 1.1) ... ops.assign(prod.price, new_prices)
Access configuration and RNG:
>>> @event ... class StochasticEvent: ... def execute(self, sim): ... shock = sim.config.h_rho ... random_values = sim.rng.uniform(0, shock, size=sim.n_firms) ... # Use random_values in calculations
Notes
The execute method receives full Simulation access, including: - All roles: sim.get_role(“RoleName”) or sim.prod, sim.wrk, etc. - Configuration: sim.config - RNG: sim.rng - Economy state: sim.ec
- __repr__()[source]#
Provide informative repr showing event name.
- Returns:
String representation in format “EventClassName(name=’event_name’)”.
- Return type:
Examples
>>> from bamengine.events import FirmsAdjustPrice >>> event = FirmsAdjustPrice() >>> repr(event) "FirmsAdjustPrice(name='firms_adjust_price')"
- __init__()#