Utilities API

The builder functions are convenience wrappers that wire together an Environment, a ShopFloor, a Router, and a release policy into a ready-to-run system, so you can spin up a complete simulation in a single call. The distribution helpers sample processing times, with RunningStats providing an online (Welford) accumulator for mean and variance. The log sinks write the environment's log events (and, at DEBUG, its domain events) as plain text, JSON lines, SQLite rows or an in-memory history.

Builders

build_immediate_release_system

build_immediate_release_system(
    *,
    env: Environment,
    scenario: Scenario | None = None,
    prefix: str = "",
    default_metrics: bool = True,
    priority_policies: Callable[..., float] | None = None,
    collect_workload: bool = False,
    collect_time_series: bool = False,
    retain_job_history: bool = False,
) -> BuiltSystem[None]

Build an immediate release (push) system with no workload control.

Creates a simple push system where jobs enter the shopfloor immediately upon arrival without any release control.

Parameters:

Name	Type	Description	Default
`env`	`Environment`	The simulation environment.	required
`scenario`	`Scenario \| None`	Environment description (shop type, machine count, arrival process, service-time, due-date rule). Defaults to a fresh `Scenario()`: a 6-machine pure job shop at rho=0.90.	`None`
`prefix`	`str`	Prefix of every entity id the builder creates: servers `wc-<i>`, `shopfloor`, `router` and `psp` (if any) become `f"{prefix}wc-<i>"` and so on. Use distinct prefixes to build several systems in one environment.	`''`
`priority_policies`	`Callable[..., float] \| None`	Optional callable used to assign job priorities at servers.	`None`
`default_metrics`	`bool`	Attach the shop floor's default EMA collector (True by default).	`True`
`collect_workload`	`bool`	If True, attach a `CurrentWorkloadCollector` (listed in `env.collectors`).	`False`
`collect_time_series`	`bool`	If True, attach a `ServerTimeSeries` (queue length and utilization) to each server (listed in `env.collectors`).	`False`
`retain_job_history`	`bool`	If True, servers retain completed job references.	`False`

Returns:

Type	Description
`BuiltSystem[None]`	`BuiltSystem[None]` with `psp=None` (push system; no release control)
`BuiltSystem[None]`	and `policy=None` (no policy object). Unpacks as
`BuiltSystem[None]`	`psp, servers, shop_floor, router, policy`.

Example

env = Environment() _, servers, shop_floor, router, _ = build_immediate_release_system(env=env) env.run(until=1000) print(f"Jobs completed: {len(shop_floor.jobs_done)}")

build_focus_system

build_focus_system(
    *,
    env: Environment,
    scenario: Scenario | None = None,
    prefix: str = "",
    default_metrics: bool = True,
    focus_weights: tuple[
        float, float, float, float, float
    ] = (0.25, 0.25, 0.25, 0.25, 0.0),
    collect_workload: bool = False,
) -> BuiltSystem[None]

Build an immediate-release (push) system that dispatches with FOCUS.

Jobs enter the shopfloor on arrival (no release control); queue ordering at every server uses the FOCUS self-establishing rule (Kasper, Land, Teunter 2023, Omega 114, 102726) via FocusPriorityRule. Use this to study FOCUS as a standalone dispatching rule, independent of DRACO.

Parameters:

Name	Type	Description	Default
`env`	`Environment`	The simulation environment.	required
`scenario`	`Scenario \| None`	Environment description (shop type, machine count, arrival process, service-time, due-date rule). Defaults to a fresh `Scenario()`: a 6-machine pure job shop at rho=0.90.	`None`
`prefix`	`str`	Prefix of every entity id the builder creates: servers `wc-<i>`, `shopfloor`, `router` and `psp` (if any) become `f"{prefix}wc-<i>"` and so on. Use distinct prefixes to build several systems in one environment.	`''`
`focus_weights`	`tuple[float, float, float, float, float]`	FOCUS mechanism weights `(w1, w2, w3, w4, w5)` for (pi, omega, psi, gamma, beta); must each be in `[0, 1]` and sum to 1. Defaults to beta-dormant `(0.25, 0.25, 0.25, 0.25, 0.0)`.	`(0.25, 0.25, 0.25, 0.25, 0.0)`
`default_metrics`	`bool`	Attach the shop floor's default EMA collector (True by default).	`True`
`collect_workload`	`bool`	If True, attach a `CurrentWorkloadCollector` (listed in `env.collectors`).	`False`

Returns:

Type	Description
`BuiltSystem[None]`	`BuiltSystem[None]` with `psp=None` (push system; no PSP) and
`BuiltSystem[None]`	`policy=None`. The FOCUS dispatching rule remains reachable via
`BuiltSystem[None]`	`router.priority_policies`. Unpacks as
`BuiltSystem[None]`	`psp, servers, shop_floor, router, policy`.

Example

env = Environment() _, servers, shop_floor, router, _ = build_focus_system(env=env) env.run(until=1000)

References

Kasper, A., Land, M., Teunter, R. (2023). Towards system state dispatching in high-variety manufacturing. Omega, 114, 102726. https://doi.org/10.1016/j.omega.2022.102726

build_lumscor_system

build_lumscor_system(
    *,
    env: Environment,
    scenario: Scenario | None = None,
    prefix: str = "",
    default_metrics: bool = True,
    check_timeout: float,
    wl_norm_level: float,
    allowance_factor: int,
    collect_workload: bool = False,
) -> BuiltSystem[LumsCor]

Build a LumsCor (load-based) pull system with workload control.

Creates a pull system using LUMS-COR (Land's Upper limit for Make-Span with CORrected workload) release policy. Jobs are held in a Pre-Shop Pool and released only when server workloads stay below configured norms.

Uses CorrectedWIPStrategy which discounts downstream workload by position, and includes starvation avoidance triggers for idle servers.

Parameters:

Name	Type	Description	Default
`env`	`Environment`	The simulation environment.	required
`scenario`	`Scenario \| None`	Environment description (shop type, machine count, arrival process, service-time, due-date rule). Defaults to a fresh `Scenario()`: a 6-machine pure job shop at rho=0.90.	`None`
`prefix`	`str`	Prefix of every entity id the builder creates: servers `wc-<i>`, `shopfloor`, `router` and `psp` (if any) become `f"{prefix}wc-<i>"` and so on. Use distinct prefixes to build several systems in one environment.	`''`
`check_timeout`	`float`	Time between pool release checks.	required
`wl_norm_level`	`float`	Workload norm threshold for each server. Jobs are released only if adding them keeps corrected WIP at or below this level.	required
`allowance_factor`	`int`	Buffer time per server for due date calculation. Higher values result in earlier (more conservative) releases.	required
`default_metrics`	`bool`	Attach the shop floor's default EMA collector (True by default).	`True`
`collect_workload`	`bool`	If True, attach a `CurrentWorkloadCollector` (listed in `env.collectors`).	`False`

Returns:

Type	Description
`BuiltSystem[LumsCor]`	`BuiltSystem[LumsCor]` whose `policy` is the wired `LumsCor`
`BuiltSystem[LumsCor]`	instance (inspect `wl_norm`, retune mid-run). Unpacks as
`BuiltSystem[LumsCor]`	`psp, servers, shop_floor, router, policy`.

Example

env = Environment() psp, servers, shop_floor, router, policy = build_lumscor_system( ... env=env, check_timeout=10.0, wl_norm_level=5.0, allowance_factor=2 ... ) env.run(until=1000)

References

Land, M.J. (2006). Parameters and sensitivity in workload control. International Journal of Production Economics, 104(2), 625-638. https://doi.org/10.1016/j.ijpe.2005.03.001

build_slar_system

build_slar_system(
    *,
    env: Environment,
    scenario: Scenario | None = None,
    prefix: str = "",
    default_metrics: bool = True,
    allowance_factor: float,
    collect_workload: bool = False,
) -> BuiltSystem[Slar]

Build a SLAR (Superfluous Load Avoidance Release) pull system.

Creates a pull system using SLAR release policy based on planned slack times (PST). Jobs are released from the Pre-Shop Pool when servers risk starvation or when urgent jobs need insertion.

Release triggers

Starvation avoidance: When queue is empty or has one job, release the job with earliest planned start time.
Urgent job insertion: When all queued jobs are non-urgent, insert the most urgent job with shortest processing time.

Parameters:

Name	Type	Description	Default
`env`	`Environment`	The simulation environment.	required
`allowance_factor`	`float`	Slack allowance per operation (parameter 'k' in paper). Higher values provide more buffer time per server.	required
`scenario`	`Scenario \| None`	Environment description (shop type, machine count, arrival process, service-time, due-date rule). Defaults to a fresh `Scenario()`: a 6-machine pure job shop at rho=0.90.	`None`
`prefix`	`str`	Prefix of every entity id the builder creates: servers `wc-<i>`, `shopfloor`, `router` and `psp` (if any) become `f"{prefix}wc-<i>"` and so on. Use distinct prefixes to build several systems in one environment.	`''`
`default_metrics`	`bool`	Attach the shop floor's default EMA collector (True by default).	`True`
`collect_workload`	`bool`	If True, attach a `CurrentWorkloadCollector` (listed in `env.collectors`).	`False`

Returns:

Type	Description
`BuiltSystem[Slar]`	`BuiltSystem[Slar]` whose `policy` is the wired `Slar` instance.
`BuiltSystem[Slar]`	Unpacks as `psp, servers, shop_floor, router, policy`.

Example

env = Environment() psp, servers, shop_floor, router, policy = build_slar_system( ... env=env, allowance_factor=3.0 ... ) env.run(until=1000)

References

Land, M.J. & Gaalman, G.J.C. (1998). The performance of workload control concepts in job shops: Improving the release method. International Journal of Production Economics, 56-57, 347-364. https://doi.org/10.1016/S0925-5273(98)00052-8

build_slar_limit_system

build_slar_limit_system(
    *,
    env: Environment,
    scenario: Scenario | None = None,
    prefix: str = "",
    default_metrics: bool = True,
    allowance_factor: float,
    wl_norm_level: float,
    collect_workload: bool = False,
) -> BuiltSystem[SlarLimit]

Build a SLAR-Limit (load-bounded SLAR) pull system.

Creates a pull system using SLAR-Limit release policy. SLAR-Limit extends classic SLAR by adding a workload-norm limit to the urgent insertion branch: urgent PSP candidates are iterated in ascending SPT order and the first whose corrected workload contribution PT/(i+1) fits all server norms is released. If none fits, the policy falls back to the standard SLAR postponed-starvation branch.

Uses CorrectedWIPStrategy on the shopfloor (required by the norm check).

Parameters:

Name	Type	Description	Default
`env`	`Environment`	The simulation environment.	required
`allowance_factor`	`float`	Slack allowance per operation (parameter 'k' in the SLAR paper).	required
`wl_norm_level`	`float`	Workload norm threshold applied uniformly to every server. An urgent PSP candidate is released only if adding its corrected contribution keeps every server in its routing at or below this level.	required
`scenario`	`Scenario \| None`	Environment description (shop type, machine count, arrival process, service-time, due-date rule). Defaults to a fresh `Scenario()`: a 6-machine pure job shop at rho=0.90.	`None`
`prefix`	`str`	Prefix of every entity id the builder creates: servers `wc-<i>`, `shopfloor`, `router` and `psp` (if any) become `f"{prefix}wc-<i>"` and so on. Use distinct prefixes to build several systems in one environment.	`''`
`default_metrics`	`bool`	Attach the shop floor's default EMA collector (True by default).	`True`
`collect_workload`	`bool`	If True, attach a `CurrentWorkloadCollector` (listed in `env.collectors`) to the shopfloor for workload time-series.	`False`

Returns:

Type	Description
`BuiltSystem[SlarLimit]`	`BuiltSystem[SlarLimit]` whose `policy` is the wired `SlarLimit`
`BuiltSystem[SlarLimit]`	instance (inspect `wl_norm`). Unpacks as
`BuiltSystem[SlarLimit]`	`psp, servers, shop_floor, router, policy`.

Example

env = Environment() psp, servers, shop_floor, router, policy = build_slar_limit_system( ... env=env, allowance_factor=3.0, wl_norm_level=5.0 ... ) env.run(until=1000)

References

Thürer, M. & Stevenson, M. (2021). Improving superfluous load avoidance release (SLAR): A new load-based SLAR mechanism. International Journal of Production Economics, 231, 107881. https://doi.org/10.1016/j.ijpe.2020.107881

build_draco_system

build_draco_system(
    *,
    env: Environment,
    scenario: Scenario | None = None,
    prefix: str = "",
    default_metrics: bool = True,
    wip_target: int,
    loop_target: int,
    focus_weights: tuple[
        float, float, float, float, float
    ] = (0.25, 0.25, 0.25, 0.25, 0.0),
    total_impact_weights: tuple[float, float, float] = (
        0.25,
        0.25,
        0.5,
    ),
    collect_workload: bool = False,
) -> BuiltSystem[Draco]

Build a DRACO (non-hierarchical WIP control) pull system.

Creates a pull system using the DRACO policy (Kasper, Land, Teunter 2023, IJPE 257, 108768) that merges release, authorization, and dispatching into a single per-server decision. On every job completion at any server k, DRACO scores every candidate in Q_k ∪ P_k by a weighted total impact w^R·R + w^A·A + w^D·D and selects the maximum. The winner is dispatched to k next.

Wiring (performed by Draco.__init__): - priority_policy: Draco.priority_policy (queue-side DRACO score; returns -inf one-shot for forced PSP winners to preserve strict paper semantics). - shop_floor.on_processing_end: Draco.decide_next_job. - psp.on_arrival(starvation_avoidance): prevents idle-server starvation when a new arrival's first server is idle.

Parameters:

Name	Type	Description	Default
`env`	`Environment`	The simulation environment.	required
`wip_target`	`int`	Target shop WIP `τ` (count of jobs).	required
`loop_target`	`int`	Target overlapping loop `ε_{k,u}`. Scalar applied to every pair; for per-pair targets, instantiate `Draco` directly with a `dict[(Server, Server), int]`.	required
`focus_weights`	`tuple[float, float, float, float, float]`	FOCUS mechanism weights `(w1, w2, w3, w4, w5)`.	`(0.25, 0.25, 0.25, 0.25, 0.0)`
`total_impact_weights`	`tuple[float, float, float]`	`(w^R, w^A, w^D)` for the DRACO total impact; must sum to 1. Defaults to `(0.25, 0.25, 0.5)` — the paper's full DRACO configuration (Table 2).	`(0.25, 0.25, 0.5)`
`scenario`	`Scenario \| None`	Environment description (shop type, machine count, arrival process, service-time, due-date rule). Defaults to a fresh `Scenario()`: a 6-machine pure job shop at rho=0.90.	`None`
`prefix`	`str`	Prefix of every entity id the builder creates: servers `wc-<i>`, `shopfloor`, `router` and `psp` (if any) become `f"{prefix}wc-<i>"` and so on. Use distinct prefixes to build several systems in one environment.	`''`
`default_metrics`	`bool`	Attach the shop floor's default EMA collector (True by default).	`True`
`collect_workload`	`bool`	If True, attach a `CurrentWorkloadCollector` (listed in `env.collectors`).	`False`

Returns:

Type	Description
`BuiltSystem[Draco]`	`BuiltSystem[Draco]` whose `policy` is the wired `Draco` instance.
`BuiltSystem[Draco]`	Unpacks as `psp, servers, shop_floor, router, policy`.

Example

env = Environment() psp, servers, shop_floor, router, policy = build_draco_system( ... env=env, wip_target=8, loop_target=4 ... ) env.run(until=1000)

References

Kasper, A., Land, M., Teunter, R. (2023). Non-hierarchical work-in-progress control in manufacturing. International Journal of Production Economics, 257, 108768. https://doi.org/10.1016/j.ijpe.2022.108768

build_conwip_system

build_conwip_system(
    *,
    env: Environment,
    scenario: Scenario | None = None,
    prefix: str = "",
    default_metrics: bool = True,
    wip_cap: int,
    collect_workload: bool = False,
) -> BuiltSystem[ConWIP]

Build a ConWIP (Constant Work-In-Process) pull system.

Jobs wait in the Pre-Shop Pool and are released — earliest due date first — only while the shop holds fewer than wip_cap jobs. Release is re-checked on every job completion and on every PSP arrival.

Parameters:

Name	Type	Description	Default
`env`	`Environment`	The simulation environment.	required
`wip_cap`	`int`	Maximum number of jobs allowed on the shop floor at once.	required
`scenario`	`Scenario \| None`	Environment description (shop type, machine count, arrival process, service-time, due-date rule). Defaults to a fresh `Scenario()`: a 6-machine pure job shop at rho=0.90.	`None`
`prefix`	`str`	Prefix of every entity id the builder creates: servers `wc-<i>`, `shopfloor`, `router` and `psp` (if any) become `f"{prefix}wc-<i>"` and so on. Use distinct prefixes to build several systems in one environment.	`''`
`default_metrics`	`bool`	Attach the shop floor's default EMA collector (True by default).	`True`
`collect_workload`	`bool`	If True, attach a `CurrentWorkloadCollector` (listed in `env.collectors`).	`False`

Returns:

Type	Description
`BuiltSystem[ConWIP]`	`BuiltSystem[ConWIP]` whose `policy` is the wired `ConWIP` instance
`BuiltSystem[ConWIP]`	(retune `wip_cap` mid-run). Unpacks as
`BuiltSystem[ConWIP]`	`psp, servers, shop_floor, router, policy`.

Example

env = Environment() psp, servers, shop_floor, router, policy = build_conwip_system(env=env, wip_cap=8) env.run(until=1000)

References

Spearman, M. L., Woodruff, D. L. & Hopp, W. J. (1990). CONWIP: a pull alternative to kanban. International Journal of Production Research, 28(5), 879-894. https://doi.org/10.1080/00207549008942761

build_continuous_release_system

build_continuous_release_system(
    *,
    env: Environment,
    scenario: Scenario | None = None,
    prefix: str = "",
    default_metrics: bool = True,
    wl_norm_level: float,
    allowance_factor: int = 2,
    collect_workload: bool = False,
) -> BuiltSystem[ContinuousRelease]

Build a Continuous Release (workload-controlled) pull system.

Jobs are held in the Pre-Shop Pool and released continuously — on each job completion and on PSP arrival — when their corrected workload contribution keeps every server in their routing at or below wl_norm_level. Requires CorrectedWIPStrategy on the shopfloor.

Parameters:

Name	Type	Description	Default
`env`	`Environment`	The simulation environment.	required
`wl_norm_level`	`float`	Corrected workload norm applied uniformly to every server.	required
`allowance_factor`	`int`	Buffer time per server for due-date planning.	`2`
`scenario`	`Scenario \| None`	Environment description (shop type, machine count, arrival process, service-time, due-date rule). Defaults to a fresh `Scenario()`: a 6-machine pure job shop at rho=0.90.	`None`
`prefix`	`str`	Prefix of every entity id the builder creates: servers `wc-<i>`, `shopfloor`, `router` and `psp` (if any) become `f"{prefix}wc-<i>"` and so on. Use distinct prefixes to build several systems in one environment.	`''`
`default_metrics`	`bool`	Attach the shop floor's default EMA collector (True by default).	`True`
`collect_workload`	`bool`	If True, attach a `CurrentWorkloadCollector` (listed in `env.collectors`).	`False`

Returns:

Type	Description
`BuiltSystem[ContinuousRelease]`	`BuiltSystem[ContinuousRelease]` whose `policy` is the wired
`BuiltSystem[ContinuousRelease]`	`ContinuousRelease` instance (inspect `wl_norm`). Unpacks as
`BuiltSystem[ContinuousRelease]`	`psp, servers, shop_floor, router, policy`.

Example

env = Environment() psp, servers, shop_floor, router, policy = build_continuous_release_system( ... env=env, wl_norm_level=6.0 ... ) env.run(until=1000)

References

Fernandes, N. O. & Carmo-Silva, S. (2011). Workload control under continuous order release. International Journal of Production Economics, 131(1), 257-262. https://doi.org/10.1016/j.ijpe.2010.09.026

build_starvation_avoidance_system

build_starvation_avoidance_system(
    *,
    env: Environment,
    scenario: Scenario | None = None,
    prefix: str = "",
    default_metrics: bool = True,
    collect_workload: bool = False,
) -> BuiltSystem[None]

Build a starvation-avoidance-only pull system.

The simplest pull policy: a job is released from the Pre-Shop Pool only when its first routing server is idle — checked on PSP arrival and again on every job completion. There is no workload norm or WIP cap; release is driven purely by first-server starvation.

Parameters:

Name	Type	Description	Default
`env`	`Environment`	The simulation environment.	required
`scenario`	`Scenario \| None`	Environment description (shop type, machine count, arrival process, service-time, due-date rule). Defaults to a fresh `Scenario()`: a 6-machine pure job shop at rho=0.90.	`None`
`prefix`	`str`	Prefix of every entity id the builder creates: servers `wc-<i>`, `shopfloor`, `router` and `psp` (if any) become `f"{prefix}wc-<i>"` and so on. Use distinct prefixes to build several systems in one environment.	`''`
`default_metrics`	`bool`	Attach the shop floor's default EMA collector (True by default).	`True`
`collect_workload`	`bool`	If True, attach a `CurrentWorkloadCollector` (listed in `env.collectors`).	`False`

Returns:

Type	Description
`BuiltSystem[None]`	`BuiltSystem[None]` with `policy=None`: this builder wires plain
`BuiltSystem[None]`	starvation-avoidance callbacks, so there is no policy object. Unpacks as
`BuiltSystem[None]`	`psp, servers, shop_floor, router, policy`.

Example

env = Environment() psp, servers, shop_floor, router, _ = build_starvation_avoidance_system(env=env) env.run(until=1000)

Scenario

Scenario `dataclass`

Immutable description of a shop environment and its order stream.

Attributes:

Name	Type	Description
`arrival_process`	`Callable[[float], ScalarSource]`	Factory producing the inter-arrival distribution. It is called with the resolved arrival rate (orders per time unit, from `resolved_arrival_rate`) and must return a distribution description (or a zero-argument sampler) of inter-arrival times whose mean is `1 / rate`. The default `Exponential` satisfies this (Poisson arrivals: `Exponential(rate)` has mean `1 / rate`). Passing a factory whose mean is not `1 / rate` — e.g. `Erlang`, whose `Erlang(rate)` has mean `shape / rate` — silently breaks the `target_utilization` calibration even though it satisfies the declared type.
`arrival_rate`	`float \| None`	Explicit arrival rate (orders per time unit) that overrides the mix-weighted derivation when not `None`; `resolved_arrival_rate` returns it verbatim, skipping the ρ→λ derivation.
`target_utilization`	`float`	Target steady-state shop utilization ρ, a fraction in `(0, 1]`; drives the arrival-rate derivation unless `arrival_rate` is set.
`families`	`tuple[SkuFamily, ...]`	The product mix (one or more `SkuFamily` entries); each carries its own service time, due-date offset/rule, and mix weight.
`due_date_offset`	`Distribution`	Default due-date offset distribution applied to families that do not set their own `due_date_offset`.

build_floor

build_floor(
    env: Environment,
    *,
    prefix: str = "",
    default_metrics: bool = True,
    collect_workload: bool = False,
    collect_time_series: bool = False,
    retain_job_history: bool = False,
) -> tuple[ShopFloor, tuple[Server, ...]]

Create the ShopFloor and n_servers single-capacity servers.

The entities are named f"{prefix}shopfloor" and f"{prefix}wc-{i}" (i from 0), so systems built with different prefixes can share an environment. collect_workload attaches a :class:~simulatte.collectors.CurrentWorkloadCollector to the shop floor and collect_time_series a :class:~simulatte.collectors.ServerTimeSeries to each server; env.collectors lists them.

build_router

build_router(
    env: Environment,
    shop_floor: ShopFloor,
    servers: Sequence[Server],
    *,
    psp: PreShopPool | None,
    priority_policies: Callable[..., float] | None = None,
    prefix: str = "",
) -> Router

Assemble the Router from the family mix: arrival process, per-family routing, service-time distributions, and due-date offsets/rules.

Parameters:

Name	Type	Description	Default
`env`	`Environment`	The simulation environment.	required
`shop_floor`	`ShopFloor`	The shop floor the router feeds jobs into.	required
`servers`	`Sequence[Server]`	The server pool the routings and service-time maps are built over. Its length must equal `self.n_servers`: that count drives the arrival-rate derivation (via `resolved_arrival_rate`, used for both the expected routing length E[L] and the machine count M), so a mismatch would silently miscalibrate utilization.	required
`psp`	`PreShopPool \| None`	The pre-shop pool, or `None` for immediate release.	required
`priority_policies`	`Callable[..., float] \| None`	Optional priority policy callable for the router.	`None`
`prefix`	`str`	Prefix of the router's id, `f"{prefix}router"`.	`''`

Raises:

Type	Description
`ValueError`	If `len(servers) != self.n_servers`.

general_flow_shop `classmethod`

general_flow_shop(
    **overrides: Unpack[_PresetKwargs],
) -> Scenario

General Flow Shop preset (directed/sorted routing).

pure_flow_shop `classmethod`

pure_flow_shop(
    **overrides: Unpack[_PresetKwargs],
) -> Scenario

Pure Flow Shop preset (all machines, fixed order).

pure_job_shop `classmethod`

pure_job_shop(
    **overrides: Unpack[_PresetKwargs],
) -> Scenario

Pure Job Shop preset (undirected routing).

resolved_arrival_rate

resolved_arrival_rate() -> float

The exponential arrival rate (explicit override, else mix-weighted derivation).

The returned rate (orders per time unit) is fed to arrival_process in build_router, which expects a sampler of inter-arrival times with mean 1 / rate (the default Exponential satisfies this).

single `classmethod`

single(
    *,
    service_time: Distribution | None = None,
    due_date_offset: Distribution | None = None,
    twk_allowance_factor: float | None = None,
    name: str = "F1",
    **shop: Unpack[_SingleShopKwargs],
) -> Scenario

Convenience for the common one-product case: build a single-family Scenario.

name is always forwarded (it has a default), while service_time, due_date_offset, and twk_allowance_factor are forwarded only when non-None, so SkuFamily's own defaults fill the rest. **shop forwards shop-level kwargs (shop_type, n_servers, target_utilization, arrival_rate, ...).

ShopType

Bases: Enum

Standard workload-control benchmark shop types (by routing directedness).

Distributions

Distributions and routings are descriptions: sampler(rng) builds a sampler that draws from an environment stream, and components bind them with env.bind (see simulatte.rng).

PureJobShopRouting `dataclass`

Bases: _Routing

Pure Job Shop (PJS) routing description.

The Pure Job Shop — also called a randomly routed job shop (Conway et al., 1967) — is the least directed of the standard workload-control benchmark shops: each order has a random routing length and a random routing direction (Oosterman, Land & Gaalman, 2000; Kasper, Land & Teunter, 2023). A sampler draws a routing length k uniformly from [1, len(servers)] and returns k distinct servers in random order (sampling without replacement, so re-entrant flow is prohibited).

See GeneralFlowShopRouting (same length rule, but the subset is sorted into a directed flow) and FlowShopRouting (fixed length, fully directed) for the directed counterparts on the directedness spectrum.

Attributes:

Name	Type	Description
`servers`	`Sequence[Server]`	The pool of servers to sample from. Its order defines the canonical machine index used by the directed descriptions.

Example

routing = env.bind(PureJobShopRouting(servers), kind="routing", stream="r/routing/A", owner="r") routing() # Returns e.g., [server_2, server_5, server_1]

References

Oosterman, B., Land, M., & Gaalman, G. (2000). The influence of shop characteristics on workload control. International Journal of Production Economics, 68(1), 107-119. https://doi.org/10.1016/S0925-5273(99)00141-3

GeneralFlowShopRouting `dataclass`

Bases: _Routing

General Flow Shop (GFS) routing description.

The General Flow Shop is the directed counterpart of the Pure Job Shop: each order has the same random routing length k ~ U[1, len(servers)] and the same equal per-machine inclusion probability, but the selected machines are sorted into ascending index order so that orders flow in a single direction with typical upstream and downstream stations (Oosterman, Land & Gaalman, 2000; Kasper, Land & Teunter, 2023). Machines are drawn without replacement, so re-entrant flow is prohibited.

Attributes:

Name	Type	Description
`servers`	`Sequence[Server]`	The pool of servers to sample from. Their order defines the canonical machine index along which routings are directed.

Example

routing = env.bind(GeneralFlowShopRouting(servers), kind="routing", stream="r/routing/A", owner="r") routing() # Returns e.g., [server_1, server_4, server_5]

References

Oosterman, B., Land, M., & Gaalman, G. (2000). The influence of shop characteristics on workload control. International Journal of Production Economics, 68(1), 107-119. https://doi.org/10.1016/S0925-5273(99)00141-3

FlowShopRouting `dataclass`

Bases: _Routing

Pure Flow Shop (PFS) routing description.

The Pure Flow Shop is the most directed benchmark shop: every order has a fixed routing length equal to the number of machines and visits all servers in the same fixed (directed) sequence (Oosterman, Land & Gaalman, 2000; Kasper, Land & Teunter, 2023). The routing is deterministic — every job shares the identical routing — so the sampler ignores the RNG.

Note

Because every order visits every machine, the mean routing length is len(servers) rather than (len(servers) + 1) / 2 as for the pure job shop / general flow shop. Calibrate the arrival rate accordingly to hit a target utilization (see arrival_rate_for_utilization).

Attributes:

Name	Type	Description
`servers`	`Sequence[Server]`	The servers visited, in the fixed order they are visited.

References

Oosterman, B., Land, M., & Gaalman, G. (2000). The influence of shop characteristics on workload control. International Journal of Production Economics, 68(1), 107-119. https://doi.org/10.1016/S0925-5273(99)00141-3

pure_job_shop_routing

pure_job_shop_routing(
    servers: Sequence[Server],
) -> PureJobShopRouting

Pure Job Shop routing description over servers (see PureJobShopRouting).

general_flow_shop_routing

general_flow_shop_routing(
    servers: Sequence[Server],
) -> GeneralFlowShopRouting

General Flow Shop routing description over servers (see GeneralFlowShopRouting).

pure_flow_shop_routing

pure_flow_shop_routing(
    servers: Sequence[Server],
) -> FlowShopRouting

Pure Flow Shop routing description over servers (see FlowShopRouting).

Distribution

Bases: Protocol

A description of a random variate that also reports its analytical mean.

Any object with a sampler(rng) method returning a zero-argument float sampler that draws from rng, and a mean property, satisfies this protocol. The Router binds descriptions to its RNG streams (env.bind); Scenario reads .mean to derive the arrival rate for a target utilization.

Exponential `dataclass`

Exponential distribution with rate rate (mean 1/rate).

Erlang `dataclass`

Erlang (Gamma with integer shape) distribution, mean shape/rate.

TruncatedErlang `dataclass`

Erlang truncated to [0, max_value] by rejection resampling.

The default shape=2 reproduces the classic truncated 2-Erlang service process. mean is the TRUE conditional mean E[X | X <= max_value], strictly below the nominal shape/rate whenever max_value is finite.

LogNormal `dataclass`

Lognormal distribution with underlying-normal params mu/sigma.

Uniform `dataclass`

Continuous uniform distribution on [low, high], mean (low+high)/2.

Deterministic `dataclass`

Degenerate distribution that always returns value.

arrival_rate_for_utilization

arrival_rate_for_utilization(
    target_utilization: float,
    *,
    n_servers: int,
    mean_routing_length: float,
    mean_processing_time: float = 1.0,
) -> float

Derive the Poisson arrival rate that yields a target shop utilization.

Shop utilization couples to the mean routing length through rho = lambda * E[L] * E[p] / M, where lambda is the order arrival rate, E[L] the mean routing length, E[p] the mean operation processing time, and M the number of machines. Inverting gives::

lambda = rho * M / (E[L] * E[p])

Because E[L] differs by shop type — M for a pure flow shop versus (M + 1) / 2 for the pure job shop / general flow shop — the arrival rate is not portable across shop types: a pure flow shop reusing a job-shop arrival rate is driven unstable (rho > 1). Use this helper to recompute the rate whenever the shop type, machine count, or processing-time mean changes.

Parameters:

Name	Type	Description	Default
`target_utilization`	`float`	Desired steady-state utilization `rho` in (0, 1].	required
`n_servers`	`int`	Number of machines `M`.	required
`mean_routing_length`	`float`	Mean number of operations per order `E[L]`.	required
`mean_processing_time`	`float`	Mean operation processing time `E[p]`. Defaults to `1.0` (the nominal 2-Erlang mean; see `TruncatedErlang`).	`1.0`

Returns:

Type	Description
`float`	The arrival rate `lambda` (orders per time unit). Its reciprocal is
`float`	the mean inter-arrival time for an exponential arrival process.

Example

Pure flow shop, M = 6, rho = 0.90, E[p] = 1.0 -> mean IAT 1.111

round(1 / arrival_rate_for_utilization(0.9, n_servers=6, mean_routing_length=6), 3) 1.111

General flow shop / pure job shop, E[L] = 3.5 -> mean IAT 0.648

round(1 / arrival_rate_for_utilization(0.9, n_servers=6, mean_routing_length=3.5), 3) 0.648

twk_due_date

twk_due_date(
    allowance_factor: float,
) -> Callable[[Sequence[float]], float]

Create a Total Work Content (TWK) due-date offset rule.

The TWK procedure sets an order's due date proportional to its total work content: due_date = arrival_time + K * sum(p_ij), where K is the allowance factor and sum(p_ij) is the order's total processing time (Kasper, Land & Teunter, 2023). Larger, more work-intensive orders are granted proportionally more lead time than the flat-allowance rule.

The returned callable matches the Router due_date_rule contract: it receives the job's sampled operation processing times and returns the due-date offset (the allowance added to the arrival time).

Parameters:

Name	Type	Description	Default
`allowance_factor`	`float`	The constant `K` multiplying total work content.	required

Returns:

Type	Description
`Callable[[Sequence[float]], float]`	A callable `(processing_times) -> float` returning `K * sum(...)`.

Example

rule = twk_due_date(8.74) # FOCUS pure-job-shop K (6 work centres) rule([1.0, 2.0]) # offset for a 3.0 work-content order 26.22

References

Kasper, A., Land, M., & Teunter, R. (2023). Towards system state dispatching in high-variety manufacturing. Omega, 114, 102726. https://doi.org/10.1016/j.omega.2022.102726

RunningStats

Compute running mean, variance, and standard deviation using Welford's algorithm.

Welford's algorithm maintains numerical stability by avoiding catastrophic cancellation that occurs when computing variance via the naive formula (sum of squares minus squared sum). This is especially important for simulations with many observations or values of similar magnitude.

Attributes:

Name	Type	Description
`n`		Number of observations added.
`mean`		Current running mean of all observations.
`M2`		Sum of squared differences from the mean (internal state).

Example

stats = RunningStats() for value in [2.0, 4.0, 6.0]: ... stats.update(value) stats.mean 4.0 stats.std 2.0

std `property`

std: float

Sample standard deviation (square root of variance).

variance `property`

variance: float

Sample variance using Bessel's correction (n-1 denominator).

update

update(x: float) -> None

Add a new observation and update running statistics.

Parameters:

Name	Type	Description	Default
`x`	`float`	The new value to incorporate into the statistics.	required

z_norm

z_norm(x: float) -> float

Compute the z-score (standard score) for a given value.

Parameters:

Name	Type	Description	Default
`x`	`float`	The value to normalize.	required

Returns:

Type	Description
`float`	The z-score (x - mean) / std, or 0.0 if insufficient data
`float`	or zero standard deviation.

Logging

TextSink

Bases: _StreamSink

Write each record as time | LEVEL | component | message.

Parameters:

Name	Type	Description	Default
`target`	`str \| Path \| TextIO \| None`	File path (opened once in append mode, closed by :meth:`close`), a text stream (written to, not closed), or None for the current `sys.stderr`.	required
`level`	`str`	Lowest level kept.	`'INFO'`
`components`	`Iterable[str] \| None`	When given, only records of these components are kept.	`None`
`exclude`	`Iterable[str]`	Records of these components are dropped.	`()`
`render_domain`	`bool`	At `DEBUG`, also write every domain event (type name and payload).	`True`

JsonSink

Bases: _StreamSink

Write each record as one JSON object per line.

A log record has sim_time, sim_time_formatted, wall_time, seq, kind ("log"), type ("log"), level, message, component and extra. A domain event (DEBUG with render_domain) has kind "domain", its event type, level DEBUG, the namespace of its type as component, its one-line rendering as message and its payload as data. Arguments as for :class:TextSink.

SQLiteSink

Bases: LogSink

Insert each record as a row of the events table of an SQLite database.

The table has the columns env_id, seq, t, kind, type, level, component, message, data_json: kind is "log" for a log event (data_json holds its extra) or "domain" for a domain event written at DEBUG with render_domain (data_json holds its payload, component the namespace of its type). Several environments, each with its own sink, can share one database file; :attr:env_id tells their rows apart. Every row is committed when inserted.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	Database file, created if missing.	required
`level`	`str`	Lowest level kept.	`'INFO'`
`components`	`Iterable[str] \| None`	When given, only records of these components are kept.	`None`
`exclude`	`Iterable[str]`	Records of these components are dropped.	`()`
`render_domain`	`bool`	At `DEBUG`, also store every domain event.	`True`

path `property`

path: Path

The database file.

execute_sql

execute_sql(
    sql: str, params: tuple[Any, ...] = ()
) -> list[sqlite3.Row]

Execute sql with params on the database and return the rows (sqlite3.Row, indexable by name).

Raises:

Type	Description
`RuntimeError`	The sink is not attached or already closed.

query

query(
    *,
    level: str | None = None,
    component: str | None = None,
    since: float | None = None,
    until: float | None = None,
    limit: int | None = None,
    offset: int = 0,
) -> list[LogEvent]

The log records of the attached environment, ordered by time and emission.

Parameters:

Name	Type	Description	Default
`level`	`str \| None`	Keep this level only (case-insensitive).	`None`
`component`	`str \| None`	Keep this component only.	`None`
`since`	`float \| None`	Keep records with `t >= since`.	`None`
`until`	`float \| None`	Keep records with `t <= until`.	`None`
`limit`	`int \| None`	Maximum number of records returned.	`None`
`offset`	`int`	Number of records skipped.	`0`

Returns:

Type	Description
`list[LogEvent]`	class:`~simulatte.events.LogEvent` instances with their original `t` and `seq`.

Raises:

Type	Description
`RuntimeError`	The sink is not attached or already closed.

HistorySink

Bases: LogSink

Keep the latest maxlen log records in memory, oldest first (env.log_history).

Iterating yields :class:~simulatte.events.LogEvent instances. After :meth:close the records stay available, and no new record is kept.

Parameters:

Name	Type	Description	Default
`maxlen`	`int`	Number of records kept; older ones are dropped.	`1000`
`level`	`str`	Lowest level kept.	`'INFO'`
`components`	`Iterable[str] \| None`	When given, only records of these components are kept.	`None`
`exclude`	`Iterable[str]`	Records of these components are dropped.	`()`

maxlen `property`

maxlen: int | None

Number of records kept.

clear

clear() -> None

Drop every record.

query

query(
    *,
    level: str | None = None,
    component: str | None = None,
    since: float | None = None,
    until: float | None = None,
) -> list[LogEvent]

The kept records matching every given filter, oldest first.

Parameters:

Name	Type	Description	Default
`level`	`str \| None`	Keep this level only (case-insensitive), e.g. `"ERROR"`.	`None`
`component`	`str \| None`	Keep this component only, e.g. `"FleetCoordinator"`.	`None`
`since`	`float \| None`	Keep records with `t >= since`.	`None`
`until`	`float \| None`	Keep records with `t <= until`.	`None`

LogSink

Base of the log sinks: level and component filters, attachment and closing.

Parameters:

Name	Type	Description	Default
`level`	`str`	Lowest level kept (`DEBUG`, `INFO`, `WARNING`, `ERROR` or `CRITICAL`, case-insensitive).	`'INFO'`
`components`	`Iterable[str] \| None`	When given, only records of these components are kept.	`None`
`exclude`	`Iterable[str]`	Records of these components are dropped.	`()`

Records without a component always pass the component filters.

closed `property`

closed: bool

Whether :meth:close was called.

env_id `property`

env_id: str | None

Identifier of the attached environment (a fresh UUID hex string per attachment), None before.

level `property`

level: str

Lowest level kept, upper-case.

attach

attach(env: Environment) -> Self

Subscribe to the log events of env (and, see the module docstring, its domain events); return self.

The sink is added to env.sinks and closed by env.close(). Raises RuntimeError if the sink is already attached or closed.

close

close() -> None

Stop receiving events and release the sink's resources. Idempotent.

diagnostic

diagnostic(t: float, message: str, **extra: Any) -> None

Write a recorder diagnostic directly, without emitting a bus event.

Diagnostics have sequence -1: they are not simulation or observer events. Closed sinks ignore them. Component filters and the warning threshold still apply.

disable_component

disable_component(component: str) -> None

Drop the records of component.

enable_component

enable_component(component: str) -> None

Keep the records of component again (and add it to the components allow-list, if one was given).

LogEvent `dataclass`

Bases: ObserverEvent

A log record.

Utilities API

Builders

build_immediate_release_system

build_focus_system

build_lumscor_system

build_slar_system

build_slar_limit_system

build_draco_system

build_conwip_system

build_continuous_release_system

build_starvation_avoidance_system

Scenario

Scenario dataclass

build_floor

build_router

general_flow_shop classmethod

pure_flow_shop classmethod

pure_job_shop classmethod

resolved_arrival_rate

single classmethod

ShopType

Distributions

PureJobShopRouting dataclass

GeneralFlowShopRouting dataclass

FlowShopRouting dataclass

pure_job_shop_routing

general_flow_shop_routing

pure_flow_shop_routing

Distribution

Exponential dataclass

Erlang dataclass

TruncatedErlang dataclass

LogNormal dataclass

Uniform dataclass

Deterministic dataclass

arrival_rate_for_utilization

Pure flow shop, M = 6, rho = 0.90, E[p] = 1.0 -> mean IAT 1.111

General flow shop / pure job shop, E[L] = 3.5 -> mean IAT 0.648

twk_due_date

RunningStats

std property

variance property

update

z_norm

Logging

TextSink

JsonSink

SQLiteSink

path property

execute_sql

query

HistorySink

maxlen property

clear

query

LogSink

closed property

env_id property

level property

attach

close

diagnostic

disable_component

enable_component

LogEvent dataclass

Scenario `dataclass`

general_flow_shop `classmethod`

pure_flow_shop `classmethod`

pure_job_shop `classmethod`

single `classmethod`

PureJobShopRouting `dataclass`

GeneralFlowShopRouting `dataclass`

FlowShopRouting `dataclass`

Exponential `dataclass`

Erlang `dataclass`

TruncatedErlang `dataclass`

LogNormal `dataclass`

Uniform `dataclass`

Deterministic `dataclass`

std `property`

variance `property`

path `property`

maxlen `property`

closed `property`

env_id `property`

level `property`

LogEvent `dataclass`