Core Module

The core module contains the fundamental building blocks for discrete event simulation.

Simulation

The main simulation engine with event scheduling and hierarchical composition.

class simcraft.Simulation[source]

Bases: object

Core discrete event simulation engine.

The Simulation class provides the fundamental infrastructure for discrete event simulation including event scheduling, time advancement, hierarchical model composition, and execution control.

Features

Efficient event scheduling with priority support
Hierarchical model composition (parent-child sandboxes)
Multiple execution modes (run until, run for, step)
Warmup period handling
Real-time execution support
Event tracing and debugging

Examples

Basic simulation:

>>> class BankSimulation(Simulation):
...     def __init__(self):
...         super().__init__()
...         self.customers_served = 0
...
...     def on_init(self):
...         self.schedule(self.customer_arrival, delay=0)
...
...     def customer_arrival(self):
...         self.customers_served += 1
...         self.schedule(self.customer_arrival,
...                       delay=self.rng.exponential(5.0))
...
>>> sim = BankSimulation()
>>> sim.run(until=100)
>>> print(f"Served {sim.customers_served} customers")

Hierarchical composition:

>>> class Server(Simulation):
...     def __init__(self, parent: Simulation):
...         super().__init__(parent=parent)
...
>>> class System(Simulation):
...     def __init__(self):
...         super().__init__()
...         self.server1 = Server(parent=self)
...         self.server2 = Server(parent=self)

__init__(parent=None, config=None, name='')[source]

Initialize simulation.

Parameters:

parent (Optional[Simulation]) – Parent simulation for hierarchical composition
config (Optional[SimulationConfig]) – Simulation configuration
name (str) – Optional name for the simulation

Return type:

None

property clock: Clock: Get the simulation clock.

property now: float: Get current simulation time.

property rng: RandomGenerator: Get random number generator.

property name: str: Get simulation name.

property state: SimulationState: Get current simulation state.

property is_running: bool: Check if simulation is currently running.

property is_warmed_up: bool: Check if warmup period has completed.

property parent: Simulation | None: Get parent simulation.

property root: Simulation: Get root simulation.

property events_pending: int: Get number of pending events.

property events_processed: int: Get number of processed events.

schedule(action, delay=0.0, at=None, args=(), kwargs=None, tag='', priority=0)[source]

Schedule an event for future execution.

Parameters:

action (Callable) – Function to execute
delay (float) – Time delay from current time (ignored if ‘at’ is specified)
at (Optional[float]) – Absolute simulation time for execution
args (tuple) – Positional arguments for the action
kwargs (dict) – Keyword arguments for the action
tag (str) – Optional tag for event identification
priority (int) – Priority level (higher = executed first at same time)

Returns:

The scheduled event (can be used for cancellation)

Return type:

Event

Examples

>>> sim.schedule(process_arrival, delay=5.0)
>>> sim.schedule(process_arrival, at=sim.now + 5.0)
>>> sim.schedule(process_arrival, delay=0, priority=10)  # High priority

cancel_event(event)[source]

Cancel a scheduled event.

Parameters:: event (Event) – The event to cancel
Returns:: True if event was successfully cancelled
Return type:: bool

cancel_events_by_tag(tag)[source]

Cancel all events with a specific tag.

Parameters:: tag (str) – Tag to match
Returns:: Number of events cancelled
Return type:: int

run(until=None, for_duration=None, events=None)[source]

Run the simulation.

Parameters:

until (Optional[float]) – Run until this simulation time
for_duration (Optional[float]) – Run for this duration from current time
events (Optional[int]) – Run for this many events

Returns:

Self for method chaining

Return type:

Simulation

Examples

>>> sim.run(until=100)  # Run until time 100
>>> sim.run(for_duration=50)  # Run for 50 time units
>>> sim.run(events=1000)  # Run 1000 events
>>> sim.run()  # Run until no more events

step()[source]

Execute a single event.

Returns:: True if an event was processed, False if no events remain
Return type:: bool

warmup(duration=None)[source]

Run warmup period.

Parameters:: duration (Optional[float]) – Warmup duration (uses config value if not specified)
Returns:: Self for method chaining
Return type:: Simulation

reset()[source]

Reset simulation to initial state.

Returns:: Self for method chaining
Return type:: Simulation

on_init()[source]

Hook called during initialization.

Override this method to set up initial events and state.

Return type:: None

on_end()[source]

Hook called when simulation ends.

Override this method for cleanup and result collection.

Return type:: None

on_warmup_end()[source]

Hook called when warmup period ends.

Override this method to reset statistics after warmup.

Return type:: None

add_monitor(name, monitor)[source]

Register a monitor/statistics collector.

Parameters:

name (str) – Monitor name for retrieval
monitor (Any) – Monitor instance

Return type:

None

get_monitor(name)[source]

Get a registered monitor.

Parameters:: name (str) – Monitor name
Returns:: Monitor instance or None
Return type:: Any

get_trace()[source]

Get event execution trace.

Returns:: List of (time, action_name, tag) tuples
Return type:: List[Tuple[float, str, str]]

class simcraft.core.simulation.SimulationConfig[source]

Bases: object

Configuration for simulation execution.

time_unit

Base time unit for the simulation

Type:: TimeUnit

seed

Random seed for reproducibility

Type:: Optional[int]

warmup_duration

Duration of warmup period

Type:: float

max_events

Maximum number of events to process (0 = unlimited)

Type:: int

real_time_factor

Factor for real-time execution (0 = as fast as possible)

Type:: float

log_level

Logging verbosity level

Type:: int

collect_trace

Whether to collect event execution trace

Type:: bool

time_unit: TimeUnit = 'hours'

seed: int | None = None

warmup_duration: float = 0.0

max_events: int = 0

real_time_factor: float = 0.0

log_level: int = 30

collect_trace: bool = False

__init__(time_unit=TimeUnit.HOURS, seed=None, warmup_duration=0.0, max_events=0, real_time_factor=0.0, log_level=30, collect_trace=False)

Parameters:

time_unit (TimeUnit)
seed (int | None)
warmup_duration (float)
max_events (int)
real_time_factor (float)
log_level (int)
collect_trace (bool)

Return type:

None

class simcraft.core.simulation.SimulationState[source]

Bases: Enum

Simulation lifecycle states.

CREATED = 1

INITIALIZED = 2

RUNNING = 3

PAUSED = 4

COMPLETED = 5

TERMINATED = 6

Event

Event management and scheduling.

class simcraft.core.event.Event[source]

Bases: object

A discrete event scheduled for future execution.

Events are ordered by their scheduled time, with ties broken by their sequence index to ensure deterministic execution order.

scheduled_time

The simulation time at which this event should be executed

Type:: float

action

The function to call when the event is executed

Type:: Callable

args

Positional arguments to pass to the action

Type:: tuple

kwargs

Keyword arguments to pass to the action

Type:: dict

owner

The simulation that scheduled this event

Type:: Simulation

index

Unique sequence number for tie-breaking

Type:: int

tag

Optional tag for event identification and filtering

Type:: str

priority

Priority level (higher = executed first at same time)

Type:: int

cancelled

Whether this event has been cancelled

Type:: bool

Examples

>>> def handle_arrival(customer_id: int):
...     print(f"Customer {customer_id} arrived")
...
>>> event = Event(
...     scheduled_time=10.0,
...     action=handle_arrival,
...     args=(1,),
...     index=0
... )
>>> event.invoke()
Customer 1 arrived

scheduled_time: float

action: Callable[..., Any]

args: Tuple[Any, ...]

kwargs: dict

owner: 'Simulation' | None = None

index: int = 0

tag: str = ''

priority: int = 0

cancelled: bool = False

invoke()[source]

Execute the event action.

Returns:: The return value of the action, if any
Return type:: Any

Notes

Cancelled events will not execute and return None.

cancel()[source]

Cancel this event.

The event will remain in the event list but will not execute when its scheduled time is reached.

Return type:: None

__init__(scheduled_time, action, args=<factory>, kwargs=<factory>, owner=None, index=0, tag='', priority=0, cancelled=False)

Parameters:

scheduled_time (float)
action (Callable[..., Any])
args (Tuple[Any, ...])
kwargs (dict)
owner (Optional['Simulation'])
index (int)
tag (str)
priority (int)
cancelled (bool)

Return type:

None

class simcraft.core.event.ConditionalEvent[source]

Bases: object

An event that executes only when a condition is met.

Useful for events that depend on system state.

condition

Function that returns True when event should execute

Type:: Callable[[], bool]

action

The function to call when condition is met

Type:: Callable

check_interval

How often to check the condition

Type:: float

max_attempts

Maximum number of condition checks (0 = unlimited)

Type:: int

condition: Callable[[], bool]

action: Callable[[...], Any]

args: Tuple[Any, ...]

kwargs: dict

check_interval: float = 0.1

max_attempts: int = 0

check_and_execute()[source]

Check condition and execute if met.

Returns:: True if condition was met and action executed
Return type:: bool

__init__(condition, action, args=<factory>, kwargs=<factory>, check_interval=0.1, max_attempts=0, _attempt_count=0)

Parameters:

condition (Callable[[], bool])
action (Callable[[...], Any])
args (Tuple[Any, ...])
kwargs (dict)
check_interval (float)
max_attempts (int)
_attempt_count (int)

Return type:

None

class simcraft.core.event.EventList[source]

Bases: object

Efficient container for managing scheduled events.

Uses sortedcontainers.SortedList for O(log n) insertion and O(1) minimum extraction.

Examples

>>> events = EventList()
>>> events.add(Event(scheduled_time=5.0, action=lambda: None, index=0))
>>> events.add(Event(scheduled_time=3.0, action=lambda: None, index=1))
>>> next_event = events.pop_next()
>>> print(next_event.scheduled_time)
3.0

__init__()[source]

Initialize empty event list.

Return type:: None

add(event)[source]

Add an event to the list.

Parameters:: event (Event) – The event to schedule
Return type:: None

pop_next()[source]

Remove and return the next event.

Returns:: The next event to execute, or None if list is empty
Return type:: Optional[Event]

peek_next()[source]

Return the next event without removing it.

Returns:: The next event, or None if list is empty
Return type:: Optional[Event]

remove(event)[source]

Remove a specific event from the list.

Parameters:: event (Event) – The event to remove
Returns:: True if event was found and removed
Return type:: bool

clear()[source]

Remove all events from the list.

Return type:: None

Entity

Simulation entities with lifecycle tracking.

class simcraft.Entity[source]

Bases: object

Base class for all simulation entities.

Entities are the objects that flow through and interact within the simulation. They maintain their own state, track timing information, and can carry arbitrary attributes.

id

Custom identifier for the entity

Type:: Optional[str]

state

Current lifecycle state

Type:: EntityState

created_at

Simulation time when entity was created

Type:: float

attributes

Custom attributes dictionary

Type:: Dict[str, Any]

Examples

>>> class Customer(Entity):
...     def __init__(self, priority: int = 0):
...         super().__init__()
...         self.priority = priority
...
>>> customer = Customer(priority=1)
>>> customer.set_attribute("vip", True)
>>> print(customer.index, customer.get_attribute("vip"))

id: str | None = None

state: EntityState = 1

created_at: float = 0.0

attributes: Dict[str, Any]

property index: int: Get unique sequence index for this entity.

set_attribute(key, value)[source]

Set a custom attribute.

Parameters:

key (str) – Attribute name
value (Any) – Attribute value

Return type:

None

get_attribute(key, default=None)[source]

Get a custom attribute.

Parameters:

key (str) – Attribute name
default (Any) – Value to return if attribute not found

Returns:

Attribute value or default

Return type:

Any

has_attribute(key)[source]

Check if entity has an attribute.

Parameters:: key (str) – Attribute name
Returns:: True if attribute exists
Return type:: bool

activate()[source]

Transition entity to ACTIVE state.

Return type:: None

wait()[source]

Transition entity to WAITING state.

Return type:: None

enter_service()[source]

Transition entity to IN_SERVICE state.

Return type:: None

block()[source]

Transition entity to BLOCKED state.

Return type:: None

complete()[source]

Transition entity to COMPLETED state.

Return type:: None

dispose()[source]

Transition entity to DISPOSED state.

Return type:: None

property is_active: bool: Check if entity is in an active processing state.

property is_waiting: bool: Check if entity is waiting.

property is_completed: bool: Check if entity has completed processing.

property is_disposed: bool: Check if entity has been disposed.

__init__(id=None, state=EntityState.CREATED, created_at=0.0, attributes=<factory>, _counter=<factory>, _index=0)

Parameters:

id (str | None)
state (EntityState)
created_at (float)
attributes (Dict[str, Any])
_counter (count)
_index (int)

Return type:

None

class simcraft.core.entity.EntityState[source]

Bases: Enum

Standard entity lifecycle states.

CREATED = 1

ACTIVE = 2

WAITING = 3

IN_SERVICE = 4

BLOCKED = 5

COMPLETED = 6

DISPOSED = 7

class simcraft.core.entity.EntityFactory[source]

Bases: object

Factory for creating entities with consistent configuration.

Examples

>>> factory = EntityFactory(Customer, priority=1)
>>> customer1 = factory.create()
>>> customer2 = factory.create(vip=True)

__init__(entity_class, **default_kwargs)[source]

Initialize factory.

Parameters:

entity_class (Type[Entity]) – Class of entities to create
**default_kwargs – Default keyword arguments for entity creation

Return type:

None

create(**kwargs)[source]

Create a new entity.

Parameters:: **kwargs (Any) – Override default arguments
Returns:: New entity instance
Return type:: Entity

property created_count: int: Get number of entities created by this factory.

class simcraft.core.entity.EntityPool[source]

Bases: object

Pool for reusing entity instances.

Useful for high-frequency entity creation to reduce garbage collection overhead.

Examples

>>> pool = EntityPool(Customer, initial_size=100)
>>> customer = pool.acquire()
>>> # ... use customer ...
>>> pool.release(customer)

__init__(entity_class, initial_size=0, **default_kwargs)[source]

Initialize pool.

Parameters:

entity_class (Type[Entity]) – Class of entities to pool
initial_size (int) – Number of entities to pre-create
**default_kwargs – Default arguments for entity creation

Return type:

None

acquire(**kwargs)[source]

Get an entity from the pool.

Parameters:: **kwargs (Any) – Override default arguments for new entities
Returns:: Available entity (either reused or newly created)
Return type:: Entity

release(entity)[source]

Return an entity to the pool.

Parameters:: entity (Entity) – Entity to return
Return type:: None

property available_count: int: Get number of available entities.

property in_use_count: int: Get number of entities currently in use.

property total_count: int: Get total number of entities in pool.

Clock

Simulation time management.

class simcraft.core.clock.Clock[source]

Bases: object

Simulation clock for centralized time management.

The clock tracks the current simulation time, warmup status, and provides utilities for time conversion and formatting.

now

Current simulation time in the base time unit (hours by default)

Type:: float

time_unit

The base time unit for the simulation

Type:: TimeUnit

start_datetime

Optional real-world start time for datetime mapping

Type:: Optional[datetime]

warmup_end

Time at which warmup period ends (statistics start being collected)

Type:: float

is_warmed_up

Whether the simulation has completed its warmup period

Type:: bool

Examples

>>> clock = Clock()
>>> clock.advance(1.5)  # Advance 1.5 hours
>>> print(clock.now)
1.5
>>> print(clock.now_in(TimeUnit.MINUTES))
90.0

now: float = 0.0

time_unit: TimeUnit = 'hours'

start_datetime: datetime | None = None

warmup_end: float = 0.0

property is_warmed_up: bool: Check if simulation has passed warmup period.

property elapsed: float: Get total elapsed simulation time.

property elapsed_since_warmup: float: Get elapsed time since warmup ended.

advance(delta)[source]

Advance the clock by a time delta.

Parameters:: delta (float) – Time to advance in base time units
Return type:: None

advance_to(time)[source]

Advance the clock to a specific time.

Parameters:: time (float) – Target time in base time units
Raises:: ValueError – If target time is before current time
Return type:: None

now_in(unit)[source]

Get current time in specified units.

Parameters:: unit (TimeUnit) – Target time unit
Returns:: Current time converted to specified unit
Return type:: float

convert(value, from_unit, to_unit)[source]

Convert time value between units.

Parameters:

value (float) – Time value to convert
from_unit (TimeUnit) – Source time unit
to_unit (TimeUnit) – Target time unit

Returns:

Converted time value

Return type:

float

property datetime_now: datetime | None

Get current simulation time as real-world datetime.

Returns None if start_datetime is not set.

set_warmup(duration)[source]

Set warmup period duration from current time.

Parameters:: duration (float) – Warmup duration in base time units
Return type:: None

reset()[source]

Reset clock to initial state.

Return type:: None

__init__(now=0.0, time_unit=TimeUnit.HOURS, start_datetime=None, warmup_end=0.0, _initial_time=0.0)

Parameters:

now (float)
time_unit (TimeUnit)
start_datetime (datetime | None)
warmup_end (float)
_initial_time (float)

Return type:

None

class simcraft.core.clock.TimeUnit[source]

Bases: Enum

Enumeration of supported time units.

SECONDS = 'seconds'

MINUTES = 'minutes'

HOURS = 'hours'

DAYS = 'days'

to_hours(value)[source]

Convert value in this unit to hours.

Parameters:: value (float)
Return type:: float

from_hours(hours)[source]

Convert hours to this time unit.

Parameters:: hours (float)
Return type:: float