Simulation and clock¶

simulation ¶

The simulation engine.

SimulationResult `dataclass` ¶

SimulationResult(seed: int, start_time: float, end_time: float, events_processed: int, wall_time: float, metrics: dict[str, float] = dict(), details: dict[str, Any] = dict(), warnings: list[str] = list(), log: EventLog | None = None, series: dict[str, list[tuple[float, float]]] = dict())

Outcome of :meth:Simulation.run.

metrics is a flat name -> value mapping (e.g. "resource.teller.utilization") suitable for experiments and export; details holds the nested per-component summaries.

Simulation ¶

Simulation(seed: int | None = None, *, name: str = 'simulation', start: float = 0.0, epoch: datetime | None = None, time_unit: timedelta = timedelta(minutes=1), trace: bool = False, max_log_records: int | None = None, keep_entity_history: bool = True, keep_values: bool = True, record_series: bool = True, sampling: SamplingMode = 'native')

A discrete-event simulation: clock, event queue, randomness, components and metrics.

sim = Simulation(seed=42) server = sim.resource("server", capacity=1) def customer(sim): ... yield sim.request(server) ... yield 3.0 ... sim.release(server) _ = sim.process(customer(sim)) sim.run(until=10).metrics["resource.server.utilization"] 0.3

stream ¶

stream(name: str) -> RandomStream

A named, independent random stream derived from the simulation seed.

schedule ¶

schedule(event: Event | None = None, *, time: TimeLike | None = None, delay: DurationLike | None = None, priority: int | None = None, callback: EventCallback | None = None, event_type: str | None = None, payload: dict[str, Any] | None = None) -> Event

Schedule event at an absolute time or after delay.

Without an event, one is created from callback/event_type/payload.

call_at ¶

call_at(fn: Callable[[], Any], *, time: TimeLike | None = None, delay: DurationLike | None = None, priority: int = Priority.NORMAL, label: str = 'call') -> Event

Schedule a plain function call. Lighter than a full :class:Event and not logged.

reschedule ¶

reschedule(event: Event, *, time: TimeLike | None = None, delay: DurationLike | None = None, priority: int | None = None) -> Event

Move a pending (or cancelled) event to a new time.

cancel ¶

cancel(event: Event) -> bool

Cancel a pending event. Returns False if it was not pending.

peek ¶

peek() -> float

Time of the next pending event (inf if none).

step ¶

step() -> Event | None

Execute the next event. Returns it, or None if the queue is empty.

stop ¶

stop() -> None

Ask :meth:run to return after the current event.

save_checkpoint ¶

save_checkpoint(path: str | Path) -> Path

Save a checkpoint that :meth:load_checkpoint restores by deterministic replay.

See :mod:simulsi.core.checkpoint for what is stored and the requirements.

load_checkpoint `classmethod` ¶

load_checkpoint(path: str | Path, *, model: Any = None) -> Simulation

Rebuild a simulation saved with :meth:save_checkpoint, in exactly the saved state.

run ¶

run(until: TimeLike | None = None, *, max_events: int | None = None) -> SimulationResult

Process events in order and return a :class:SimulationResult.

With until, every event whose timestamp is <= until runs and the clock then stands exactly at until (even if the queue emptied earlier). Without it, the run lasts until no events remain. Runs can be resumed by calling :meth:run again with a later until.

on_finish ¶

on_finish(hook: Callable[[Simulation], None]) -> None

Register a function called whenever :meth:run returns (e.g. to set final metrics).

series ¶

series() -> dict[str, list[tuple[float, float]]]

Recorded step-function time series (empty when record_series=False).

reset_statistics ¶

reset_statistics() -> None

Discard statistics collected so far, typically at the end of a warm-up period.

warmup ¶

warmup(duration: DurationLike) -> None

Reset statistics once duration has elapsed (schedules :meth:reset_statistics).

process ¶

process(generator: ProcessGenerator, *, name: str | None = None, entity: Entity | None = None) -> Process

Start a generator as a process at the current time.

wait_until ¶

wait_until(predicate: Callable[[], bool], *, name: str = 'condition', on: Iterable[Any] | None = None) -> Signal

A signal that triggers (with the current time) once predicate() is true.

The predicate is checked now and then after every event until it holds, so yield sim.wait_until(lambda: stock.level < 20) works for any model state. Keep predicates cheap; each pending one costs a call per event.

With on=[resource, container, queue, ...] the predicate is only re-checked after an event that changed one of those components - much cheaper when many conditions are pending. The predicate must then depend only on those components (and on constants).

container ¶

container(name: str, capacity: float = math.inf, *, init: float = 0.0) -> Container

A level (stock, fuel, cash) that processes put to and get from.

request ¶

request(resource: Resource, *, priority: float = 0, entity: Entity | None = None, patience: DurationLike | None = None) -> Request

Ask for one unit of resource; yield the result to wait for it.

release ¶

release(target: Resource | Request) -> None

Release a request, or the most recent unit of resource held by the current process.

use ¶

use(resource: Resource, duration: DurationLike, *, priority: float = 0) -> Generator[Any, Any, Request]

Acquire resource, hold it for duration, release it: yield from sim.use(...).

The unit is released even if the process is interrupted while holding it.

entity ¶

entity(entity_type: str = 'entity', /, **attributes: Any) -> Entity

Create and register an entity with a reproducible id like "customer-7".

add_entity ¶

add_entity(entity: Entity) -> Entity

Register an entity. Auto-generated ids are replaced with per-simulation ones.

dispose ¶

dispose(entity: Entity, state: str = 'disposed') -> None

Mark an entity as having left the system; records its time in system.

snapshot ¶

snapshot(*, max_events: int = 50, max_entities: int = 500) -> dict[str, Any]

A JSON-serialisable view of the current state (for debugging and UIs).

clock ¶

Simulation clock.

Internally simulation time is always a float measured in abstract time units. A clock may optionally be anchored to the calendar via epoch (the datetime that corresponds to t = 0) and unit (the timedelta that one time unit represents). This keeps the hot path cheap while letting users schedule with timedelta / datetime values when that is natural.

Clock ¶

Clock(start: float = 0.0, *, epoch: datetime | None = None, unit: timedelta = timedelta(minutes=1))

Holds the current simulation time and converts user-facing time values.

duration ¶

duration(value: DurationLike) -> float

Convert a duration (number of time units or timedelta) to float time units.

to_time ¶

to_time(value: TimeLike) -> float

Convert an absolute time to float time units.

Numbers are taken as-is, timedelta values are offsets from t = 0 and datetime values require an epoch.

to_datetime ¶

to_datetime(time: float | None = None) -> datetime

Calendar time for time (default: now). Requires an epoch.

trace ¶

Structured simulation trace (opt-in event log).

EventLog ¶

EventLog(max_records: int | None = None)

Append-only list of :class:LogRecord objects.

max_records bounds memory for very long runs; once reached, further records are counted in dropped instead of stored.

export ¶

export(path: str | Path) -> Path

Write the log as JSON, CSV or Parquet, chosen by the file suffix.

checkpoint ¶

Checkpoints of running simulations, restored by deterministic replay.

A running simulation contains suspended Python generators, which cannot be serialised reliably. SimulSI therefore checkpoints how to get back rather than the raw memory: the model reference, resolved parameters, seed, simulation options, the number of events executed and the clock. Because a seeded SimulSI run is deterministic, rebuilding the model and replaying exactly that many events reproduces the state. A digest of all statistics (and the pending-event count) is stored too and checked after replay, so a model that is not deterministic, or that changed since the checkpoint, is detected instead of silently diverging.

Requirements and costs:

the simulation must come from :meth:Model.create / :meth:Model.simulate (so it can be rebuilt);
the model must be reproducible from its seed (no global randomness, wall clock, unordered iteration that affects scheduling);
restoring costs the time to re-run up to the checkpoint. Replay is usually much faster than the original interactive or segmented run, but it is not instant;
state changed from outside the event loop between runs (e.g. calling resource.set_capacity from your script between two run calls) is not recorded. Schedule such changes as events instead.

Finish hooks (sim.on_finish) are not re-run during replay; they run again at the end of the next run.

CheckpointError ¶

Bases: SimulsiError

A checkpoint could not be written, resolved or faithfully restored.

model_reference ¶

model_reference(model: Model) -> dict[str, Any]

How to find model again in another process (best effort).