Processes, schedules and flow

process

Processes: generator functions that describe behaviour over simulated time.

A process is a Python generator. Each yield hands the engine something to wait for and the generator resumes with the result:

• a number or timedelta -> hold for that long (yield 4.5)
• sim.timeout(d, value) -> same, with a value returned from the yield
• sim.request(resource) -> resumes once a unit is granted
• queue.get() / put() -> resumes when an item arrives / space frees
• sim.signal() -> resumes when someone calls signal.succeed()
• sim.all_of(...) / any_of(...) -> combinations
• another :class:Process -> resumes when it finishes (with its return value)

Exceptions travel too: if the awaited thing fails, the exception is raised at the yield; an Interrupt is raised there when the process is interrupted.

Waitable

Waitable(sim: Simulation)

Something a process can wait for. Triggers once, with a value or an exception.

succeed

succeed(value: Any = None) -> Waitable

Trigger successfully. Waiters resume at the current simulation time.

fail

fail(exc: BaseException) -> Waitable

Trigger with an exception, which is raised in every waiting process.

add_callback

add_callback(cb: Callback) -> None

Call cb(self) once triggered (immediately-scheduled if already triggered).

Timeout

Timeout(sim: Simulation, delay: float, value: Any = None)

Bases: Waitable

Triggers after delay time units.

cancel

cancel() -> None

Withdraw the timeout if nobody is waiting on it any more.

Signal

Signal(sim: Simulation, name: str = 'signal')

Bases: Waitable

A one-shot flag that model code triggers explicitly with succeed/fail.

Condition

Condition(sim: Simulation, children: Iterable[Waitable])

Bases: Waitable

Base for :class:AllOf / :class:AnyOf. Value: {waitable: value} of triggered children.

Process

Process(sim: Simulation, generator: ProcessGenerator, *, name: str | None = None, entity: Entity | None = None)

Bases: Waitable

A running generator. Also a :class:Waitable that triggers when the generator ends.

target property

target: Waitable | None

What the process is currently waiting for.

interrupt

interrupt(cause: Any = None) -> None

Raise :class:~simulsi.errors.Interrupt inside the process at the current time.

A pending resource request the process was waiting on is withdrawn.

schedules

Time-varying inputs: shift (capacity) schedules and non-stationary arrivals.

• :func:capacity_schedule - set a resource's capacity from a table such as [(0, 2), (480, 4), (960, 1)], optionally repeating every period.
• :class:PiecewiseRate - an arrival rate that is constant within each interval of a table (e.g. hourly call volumes), optionally periodic.
• :func:arrivals - a source process for a Poisson process with a constant, piecewise-constant or arbitrary time-varying rate (NHPP).

Times in tables are measured from when the schedule or source is created (normally t = 0).

PiecewiseRate

PiecewiseRate(table: Sequence[tuple[float, float]], *, period: float | None = None)

A rate that is constant on each interval of table ((start_time, rate) rows).

rate = PiecewiseRate([(0, 2.0), (8, 6.0), (17, 1.0)], period=24) rate(9.5), rate(30.0), rate.max (6.0, 2.0, 6.0)

mean

mean() -> float

Average rate over one period (or the first times[-1] units without a period).

next_arrival

next_arrival(t: float, work: float) -> float

The time at which the integrated rate from t reaches work (inf if never).

capacity_schedule

capacity_schedule(sim: Simulation, resource: Resource, table: Sequence[tuple[float, int]], *, period: float | None = None, name: str | None = None) -> Process

Change resource capacity at the times in table ((time, capacity) rows).

With period the table repeats (e.g. period=1440 for a daily shift pattern in minutes). Lowering capacity never interrupts work in progress: units in use finish their job and the new limit applies to later grants.

arrivals

arrivals(sim: Simulation, rate: RateLike, spawn: Callable[[Simulation, int], Any], *, stream: str = 'arrivals', limit: int | None = None, until: float | None = None, max_rate: float | None = None, name: str = 'arrivals') -> Process

Start a source that calls spawn(sim, i) at each arrival of a Poisson process.

rate may be a number (homogeneous Poisson process), a :class:PiecewiseRate (exact, one draw per arrival) or any function of time rate(t) with an upper bound max_rate (generated by thinning). If spawn returns a generator it is started as a process. Stops after limit arrivals or at time until (measured like the rate table, from now).

flow

Flow helpers: batching, split/join and routing.

• :func:batch - collect size items from a queue (optionally with a timeout).
• :func:split / :func:join - fork work into parallel processes and wait for all of them.
• :class:Router - send entities down one of several paths with given probabilities, counting how many take each path.
• :func:shortest_queue - pick the resource with the fewest waiting.

Router

Router(sim: Simulation, name: str, routes: Mapping[str, float])

Probabilistic routing: router.choose() returns one of the route names.

Each route is counted in the metric route.<name>.<route>. Draws come from the stream route:<name>, so adding a router does not change the other random numbers in the model.

batch

batch(queue: Queue[T], size: int, *, timeout: float | None = None) -> Generator[Any, Any, list[T]]

Take size items from queue: items = yield from batch(q, 10).

With timeout, stops early once that much time has passed since the call and returns the (possibly shorter, possibly empty) batch collected so far.

split

split(sim: Simulation, generators: Iterable[ProcessGenerator]) -> list[Process]

Start each generator as a parallel process; pass the result to :func:join.

join

join(processes: Sequence[Process]) -> Generator[Any, Any, list[Any]]

Wait for all processes; returns their return values in order.

results = yield from join(split(sim, [cut(sim), paint(sim)])). If a process fails, its exception is raised here.

shortest_queue

shortest_queue(resources: Sequence[Resource]) -> Resource

The resource with the fewest waiting requests; ties go to the most free units, then to the earliest in resources.

trace

Trace-driven input: replay recorded arrivals and values instead of sampling.

• :func:load_trace - read a CSV into a list of records (numbers converted).
• :func:trace_arrivals - start a source that creates an arrival at each recorded time, passing the record to spawn.
• :class:TraceReplay - hand out recorded values (e.g. service times) in order, as a drop-in for distribution.sample(stream).

Trace-driven runs are useful for validating a model against history: with the real arrivals and service times, the model should reproduce the real waits. Every replication of a trace-driven input is identical, so vary something else (or use sampled inputs) for confidence intervals.

TraceExhausted

Bases: SimulsiError

A :class:TraceReplay ran out of values.

TraceReplay

TraceReplay(values: Sequence[Any], *, cycle: bool = False)

Recorded values handed out in order.

service = TraceReplay(times) then service.sample() (or service.sample(stream), so it can stand in for a distribution) gives the next value. Raises :class:TraceExhausted at the end unless cycle=True. Create it inside the model's build function so every run starts from the first value.

load_trace

load_trace(path: str | Path) -> list[dict[str, Any]]

Read a CSV file with a header row; numeric cells become int/float.

trace_arrivals

trace_arrivals(sim: Simulation, records: Iterable[float | Mapping[str, Any]], spawn: Callable[[Simulation, Any], Any], *, time: str = 'time', name: str = 'trace') -> Process

Create one arrival per record at its recorded time (measured from now).

records are numbers (the times) or mappings with a time field. spawn(sim, record) is called at each arrival; if it returns a generator, that is started as a process. Times must be non-decreasing and non-negative.

disruption

Failure and disruption primitives.

• :class:FailureProcess - a resource alternates between up (for a random time-to-failure) and down (until recovered). Recovery is described by a :class:RecoveryProcess: a random repair time, optionally needing a unit of a repair-crew resource first (so repairs can queue for technicians).
• :class:ScheduledDisruption - a deterministic window during which something is changed and later restored (capacity cut, demand spike, ...).
• :func:capacity_reduction - the most common scheduled disruption.

All randomness comes from named streams, so disruptions are reproducible and do not disturb other streams (common random numbers stay intact).

RecoveryProcess dataclass

RecoveryProcess(time_to_repair: DistributionLike, crew: Resource | None = None)

How a failed resource comes back: repair time, optionally needing a crew unit.

FailureProcess

FailureProcess(sim: Simulation, resource: Resource, time_to_failure: DistributionLike, recovery: RecoveryProcess | DistributionLike, *, units: int | None = None, interrupt: bool = False, name: str | None = None, start: bool = True)

Random failures of a resource.

Each cycle: wait time_to_failure (measured from the end of the last repair), take units units down (default: all), recover via recovery, bring them back. With interrupt=True jobs holding the failed units are interrupted (see :meth:Resource.fail).

Metrics: failure.<name>.count (counter) and failure.<name>.downtime (tally of outage durations).

ScheduledDisruption

ScheduledDisruption(sim: Simulation, *, at: float, duration: float | None, apply: Callable[[], Any], revert: Callable[[], Any] | None = None, name: str = 'disruption')

Apply a change at at and revert it duration later.

demand spike: double the arrival rate between t=100 and t=160

ScheduledDisruption(sim, at=100, duration=60,

apply=lambda: state.update(rate=2 * base), revert=lambda: state.update(rate=base))

capacity_reduction

capacity_reduction(sim: Simulation, resource: Resource, *, at: float, duration: float, capacity: int, name: str | None = None) -> ScheduledDisruption

Temporarily set resource capacity to capacity during [at, at + duration).