Simulator Objects

Simulator objects structure observer behavior and manage state variables during scenario execution.

Observer Objects

Observer objects are base classes that implement the observer pattern to loosely couple an observable and observer.

Observer and Observable classes implement the observer pattern using the observer pattern for loose coupling of behavior between objects. They are primarily building-blocks for other classes within the library.

The Entity class is a base simulation component that maintains stateful properties such as a simulation clock (i.e., the time). As an Observable, an Entity object notifies any bound Observers when its time property changes. The Entity class has stub methods to manage simulation state including initialize (before start of a new execution), tick (compute the next state variables during a time step interval), and tock (commit the next state variables at the time step boundary). Typically, users will create a custom Entity subclass to specialize behavior for a particular simulation entity.

The Simulation class manages scenario time advancement for an execution consisting of member Entity objects. As an Observable, a Simulation object notifies any bound Observers when any of its properties change including time, mode, time_scale_factor, duration, or time_step. The Simulator class has methods to manage simulation execution including initialize (prepares a new execution), execute (runs a new execution), and terminate (prematurely end an execution). When intitialized, a Simulator will initialize all member Entity objects. When executed, a Simulator will sequentially compute the next time step boundary, call tick on all member Entity objects to compute the next state, wait until the appropriate wallclock time, and finally call tock on all member Entity objects to process the state change. Timing options during initialization and execution specify the starting and ending scenario time, starting wallclock time, scenario time step duration (i.e., time interval between state updates), time scale factor (i.e., number of scenario seconds per wallclock second).

class Observer

Bases: ABC

Abstract base class that can be notified of property changes from an associated Observable.

abstractmethod on_change(source: object, property_name: str, old_value: object, new_value: object) → None

Callback notifying of a change.

Parameters:

source (object) – object that triggered a property change
property_name (str) – name of the changed property
old_value (object) – old value of the named property
new_value (object) – new value of the named property

class Observable

Bases: object

Base class that can register (add/remove) and notify observers of property changes.

__init__(): Initializes a new observable.

add_observer(observer: Observer) → None

Adds an observer to this observable.

Parameters:: observer (Observer) – observer to be added

notify_observers(property_name: str, old_value: object, new_value: object) → None

Notifies observers of a property change.

Parameters:

property_name (str) – name of the changed property
old_value (object) – old value of the named property
new_value (object) – new value of the named property

remove_observer(observer: Observer) → Observer

Removes an observer from this observable.

Parameters:: observer (Observer) – obsever to be removed
Returns:: removed observer
Return type:: Observer

class MessageObserver

Bases: ABC

Abstract base class for message observers that can receive RabbitMQ messages.

abstractmethod on_message(ch, method, properties, body) → None

Callback for when a message is received.

Parameters:

ch – Channel object
method – Method frame
properties – Message properties
body – Message body

class MessageObservable

Bases: Observable

Observable that can notify observers of received messages.

__init__(): Initialize message observable

add_message_observer(observer: MessageObserver) → None

Add a message observer.

Parameters:: observer (MessageObserver) – The observer to add

notify_message_observers(ch, method, properties, body) → None

Notify all message observers about a received message.

Parameters:

ch – Channel object
method – Method frame
properties – Message properties
body – Message body

remove_message_observer(observer: MessageObserver) → None

Remove a message observer.

Parameters:: observer (MessageObserver) – The observer to remove

class PropertyChangeCallback(property_name: str, callback: Callable[[object, object], None])

Bases: Observer

Triggers a provided callback basedwhen a named property changes.

on_change(source: object, property_name: str, old_value: object, new_value: object) → None

Callback notifying of a change.

Parameters:

source (object) – object that triggered a property change
property_name (str) – name of the changed property
old_value (object) – old value of the named property
new_value (object) – new value of the named property

class ScenarioTimeIntervalCallback(callback: Callable[[object, datetime], None], time_inteval: timedelta, time_init: timedelta | None = None)

Bases: Observer

Triggers a provided callback at a fixed interval in scenario time.

on_change(source: object, property_name: str, old_value: object, new_value: object)

Callback notifying of a change.

Parameters:

source (object) – object that triggered a property change
property_name (str) – name of the changed property
old_value (object) – old value of the named property
new_value (object) – new value of the named property

class WallclockTimeIntervalCallback(simulator: Simulator, callback: Callable[[datetime], None], time_inteval: timedelta, time_init: timedelta = None)

Bases: Observer

Triggers a provided callback at a fixed interval in wallclock time.

on_change(source: object, property_name: str, old_value: object, new_value: object)

Callback notifying of a change.

Parameters:

source (object) – object that triggered a property change
property_name (str) – name of the changed property
old_value (object) – old value of the named property
new_value (object) – new value of the named property

class Entity(name: str = None)

Bases: Observable

A base entity that maintains its own clock (time) during scenario execution.

Notifies observers of changes to one observable property

time: current scenario time

name

The entity name (optional)

Type:: str

__init__(name: str = None)

Initializes a new entity.

Parameters:: name (str) – name of the entity (default: None)

get_time() → datetime

Retrieves the current scenario time.

Returns:: current scenario time
Return type:: datetime

initialize(init_time: datetime) → None

Initializes the entity at a designated initial scenario time.

Parameters:: init_time (datetime) – initial scenario time

tick(time_step: timedelta) → None

Computes the next state transition following an elapsed scenario duration (time step). If the entity hasn’t been initialized yet, the time_step will be stored but no time advancement will occur until the entity is initialized.

Parameters:: time_step (timedelta) – elapsed scenario duration

tock() → None: Commits the state transition pre-computed in tick and notifies observers of changes.

Scenario Objects

Scenario objects define states and methods for executing a simulation.

class Mode(value)

Bases: str, Enum

Enumeration of simulation modes.

The six simulation modes include

UNDEFINED: Simulation is in an undefined state that is not one of the other modes.
For example, the simulation reverts to UNDEFINED after adding a new entity and must be re-initialized before execution.
INITIALIZING: Simulation is in the process of initialization.
INITIALIZED: Simulation has finished initialization and is ready to execute.
EXECUTING: Simulation is in the process of execution.
TERMINATING: Simulation is in the process of termination.
TERMINATED: Simulation has finished termination and is ready for initialization.

class Simulator(wallclock_offset: timedelta = datetime.timedelta(0))

Bases: Observable

Object that manages simulation of entities in a scenario.

Notifies observers of changes to observable properties

time: current scenario time
mode: current execution mode
duration: scenario execution duration
time_step: scenario time step duration

__init__(wallclock_offset: timedelta = datetime.timedelta(0))

Initializes a new simulator.

Parameters:: wallclock_offset (timedelta) – difference between the system clock and trusted wallclock source (default: zero)

add_entity(entity: Entity) → None

Adds an entity the the simulation.

Parameters:: entity (Entity) – entity to be added

execute(init_time: datetime, duration: timedelta, time_step: timedelta, wallclock_epoch: datetime = None, time_scale_factor: float = 1) → None

Executes a simulation for a specified duration with uniform time steps. Requires that the simulator is in UNDEFINED, INITIALIZED, or TERMINATED mode.

Initializes the simulation (if not already in the INITIALIZED mode), waits for the specified wallclock epoch, and transitions to the EXECUTING mode. During execution, incrementally performs state transitions for each entity. At the end of the simulation, transitions to the TERMINATING and, finally, TERMINATED mode.

Parameters:

init_time (datetime) – initial scenario time
duration (timedelta) – scenario execution duration
time_step (timedelta) – scenario time step duration
wallclock_epoch (datetime) – wallclock time corresponding to the initial scenario time, None uses the current wallclock time (default: None)
time_scale_factor (float) – number of scenario seconds per wallclock second (default value: 1)

get_duration() → timedelta

Gets the scenario duration.

Returns:: current scenario duration
Return type:: timedelta

get_end_time() → datetime

Gets the scenario end time.

Returns:: final scenario time
Return type:: datetime

get_entities() → List[Entity]

Retrieves a list of all entities in the simulation.

Returns:: list of entities in the simulation
Return type:: List(Entity)

get_entities_by_name(name: str) → List[Entity]

Retrieves a list of entities by name.

Parameters:: name (str) – name of the entity
Returns:: list of entities with a matching name
Return type:: List(Entity)

get_entities_by_type(type: Type) → List[Entity]

Retrieves a list of entities by type (class).

Parameters:: type (Type) – type (class) of entity
Returns:: list of entities with a matching type
Return type:: List(Entity)

get_init_time() → datetime

Gets the initial scenario time.

Returns:: initial scenario time
Return type:: datetime

get_mode() → Mode

Gets the current simulation mode.

Returns:: current simulation mode
Return type:: Mode

get_simulation_epoch() → datetime

Gets the scenario epoch.

Returns:: current scenario epoch
Return type:: datetime

get_time() → datetime

Gets the current scenario time.

Returns:: current scenario time
Return type:: datetime

get_time_scale_factor() → float

Gets the time scale factor in scenario seconds per wall clock second (>1 is faster-than-real-time).

Returns:: current time scale factor
Return type:: float

get_time_step() → timedelta

Gets the scenario time step duration.

Returns:: time step duration
Return type:: timedelta

get_wallclock_epoch() → datetime

Gets the wallclock epoch.

Returns:: current wallclock epoch
Return type:: datetime

get_wallclock_time() → datetime

Gets the current wallclock time.

Returns:: current wallclock time
Return type:: datetime

get_wallclock_time_at_simulation_time(time: datetime) → datetime

Gets the wallclock time corresponding to the designated scenario time.

Parameters:: time (datetime) – scenario time
Returns:: wallclock time
Return type:: datetime

get_wallclock_time_step() → timedelta

Gets the wallclock time step duration.

Returns:: time step duration
Return type:: timedelta

initialize(init_time: datetime, wallclock_epoch: datetime = None, time_scale_factor: float = 1) → None

Initializes the simulation to an initial scenario time. Requires that the simulator is in UNDEFINED, INITIALIZED, or TERMINATED mode.

Transitions to the INITIALIZING mode, initializes all entities to the initial scenario time, sets the wallclock epoch (wallclock time corresponding with the initial scenario time), and finally transitions to the INITIALIZED mode.

Parameters:

init_time (datetime) – initial scenario time
wallclock_epoch (datetime) – wallclock time corresponding to the initial scenario time, None uses the current wallclock time (default: None)
time_scale_factor (float) – number of scenario seconds per wallclock second (default: 1)

pause() → None: Pauses the scenario execution. Requires that the simulator is in EXECUTING mode.

remove_entity(entity: Entity) → Entity

Removes an entity from the simulation.

Parameters:: entity (Entity) – entity to be removed
Returns:: removed entity
Return type:: Entity

resume() → None: Resumes the scenario execution. Requires that the simulator is in PAUSING or PAUSED mode.

set_duration(duration: timedelta) → None

Sets the scenario duration. Requires that the simulator is in EXECUTING mode.

Parameters:: duration (timedelta) – scenario duration

set_end_time(end_time: datetime) → None

Sets the scenario end time. Requires that the simulator is in EXECUTING mode.

Parameters:: end_time (datetime) – scenario end time

set_time_scale_factor(time_scale_factor: float, simulation_epoch: datetime = None) → None

Sets the time scale factor in scenario seconds per wallclock second (>1 is faster-than-real-time). Requires that the simulator is in EXECUTING mode.

Parameters:

time_scale_factor (float) – number of scenario seconds per wallclock second
simulation_epoch (datetime) – scenario time at which the time scale factor changes

set_time_step(time_step: timedelta) → None

Set the scenario time step duration. Requires that the simulator is in EXECUTING mode.

Parameters:: time_step (timedelta) – scenario time step duration

set_wallclock_offset(wallclock_offset: timedelta) → None

Set the wallclock offset (difference between system clock and trusted wallclock source). Requires that the simulator is in UNDEFINED, INITIALIZING, INITIALIZED, or TERMINATED mode.

Parameters:: wallclock_offset (timedelta) – difference between system clock and trusted wallclock source

terminate() → None: Terminates the scenario execution. Requires that the simulator is in EXECUTING mode.