Skip to content

Python API

The public API, from the docstrings. Scenarios are the main interface; the core classes are for building and driving plants directly (see Using the core in Python).

Running scenarios

simulate 

simulate(source: Any) -> Run

Load, compile and run a scenario (for a config with distributions, one drawn plant).

prepare 

prepare(source: Any, plant: int = 0) -> Prepared

Validate a scenario and compile it into a graph and events. If the config holds distributions, plant plant is drawn from them first.

load_scenario 

load_scenario(source: Any) -> Scenario

A validated Scenario from a Scenario, a dict, a YAML file path or YAML text.

Prepared dataclass

A scenario turned into a compiled graph and a list of events, ready to run.

config_hash property 

config_hash: str

compile 

compile(lanes: int) -> CompiledGraph

The same plant compiled for another number of lanes (for example twin runs).

engine 

engine(noise_keys: Any = None, compiled: CompiledGraph | None = None, events: list[Event] | None = None) -> Engine

An engine with the scenario's events (or events) scheduled, not yet initialized.

run 

run(engine: Engine | None = None) -> Run

Initialize and run an engine for the scenario's duration, then collect the result.

Distributions and families

draw 

draw(config: dict[str, Any], plant: int = 0, attempt: int = 0) -> tuple[dict[str, Any], list[dict[str, Any]]]

A concrete config for plant plant, and the trace of every value drawn.

attempt > 0 redraws (used when a draw is rejected); it keys separate streams.

family 

family(source: Any, plants: int | None = None, histories: int | None = None, max_attempts: int = 20) -> Family

Draw plants plants from a config's distributions and run histories histories of each.

Defaults come from the config's family section. A draw that fails validation (for example a non-stationary AR source) is rejected and redrawn; rejections are recorded.

Family dataclass

run 

run(plant: int, history: int = 0) -> Run

One plant's history as a single-lane run.

draws 

draws() -> pd.DataFrame

The drawn values, one row per plant and one column per address.

Results

Run

Values and ground truth of one run (one or more lanes).

truth property 

truth: DataFrame

measured property 

measured: DataFrame

observed property 

observed: DataFrame

seconds property 

seconds: ndarray

Simulation time of each row in seconds, whether the index is elapsed time or calendar time (with output.start).

lanes property 

lanes: int

frame 

frame(signals: list[str] | None = None, lane: int | None = None) -> pd.DataFrame

Values of any signals (all by default). One lane gives flat columns; several lanes give (lane, signal) columns.

lane 

lane(lane: int) -> Run

A single-lane view of this run; its event log and reproducibility info describe that lane only.

The core

Graph

A mutable collection of operators wired by signal names; compile it before running.

add 

add(op: Operator, inputs: dict[str, str] | None = None, outputs: dict[str, str] | None = None, meta: dict[str, Any] | None = None) -> Operator

Add an operator. Inputs map ports to signal names; outputs rename output signals (by default a single output is named after the operator, others name.port).

describe 

describe(signal: str, **meta: Any) -> None

Attach metadata to a signal (unit, doc, tag, ...).

modulate 

modulate(path: str, signal: str, how: str = 'scale') -> None

Let a signal modify a parameter during the run: "scale" multiplies it, "shift" adds to it. The parameter is promoted; events on it keep driving its base value.

set_meta 

set_meta(op_name: str, **meta: Any) -> None

Add metadata to an operator, for example its update period: set_meta("x", period=10.0).

compile 

compile(dt: float, batch: int = 1, promote: Iterable[str] = (), lane_values: dict[str, Any] | None = None) -> CompiledGraph

Validate and order the graph. promote lists parameter paths to turn into signals; lane_values maps parameter paths to one value per lane (plant variants in one batch).

signature 

signature() -> tuple

What must be identical for graphs to run as lanes of one batch: operators, wiring, metadata, modulations and every parameter that cannot differ between lanes.

lane_parameters 

lane_parameters() -> dict[str, float]

Every parameter that may differ between lanes, by path, with its value in this graph.

producer 

producer(signal: str) -> tuple[str, str] | None

(operator, port) writing signal, or None.

inputs_of 

inputs_of(op_name: str) -> dict[str, str]

The signals wired to an operator's inputs (unconnected optional inputs are absent).

meta_of 

meta_of(op_name: str) -> dict[str, Any]

modulations_of 

modulations_of(path: str) -> list[tuple[str, str]]

(signal, "scale" or "shift") pairs that modulate a parameter.

resolve_target 

resolve_target(target: str) -> tuple[str, Any] | None

('signal', name) or ('param', (operator, parameter)) or None.

CompiledGraph dataclass

schedulable property 

schedulable: set[str]

Targets that set, shift and scale events may change.

set_initial 

set_initial(target: str, value: float | ndarray) -> list[Issue]

Set the value a schedulable signal or a parameter has at the start of a run (before settling), for example the values of the first regime; an array gives one value per lane. Call before creating an engine.

level 

level(signal: str) -> str

'measured' for final outputs of observation chains, 'internal' for intermediate observation stages, else 'true'. Operators marked truth-only (meta "truth_only", for example labelling tracers) are always 'true'.

loop_of 

loop_of(signal: str, role: str) -> Loop | None

Engine

Runs a compiled graph over time.

seed: the base seed; every random stream is keyed by (seed, operator path, noise key). noise_keys: one integer per lane; lanes with the same key share noise (twin runs). Defaults to 0, 1, 2, ... (independent lanes). record_every: record one step in every record_every steps.

schedule 

schedule(*events: Event) -> None

Add events; each is validated against the compiled graph.

initialize 

initialize(init: str | dict[str, float] = 'steady', tol: float = 1e-09, max_steps: int | None = None) -> None

Bring the plant to its starting state.

"steady": noise off, stochastic sources at their means and quantization passed through, stepped at t = 0 until every state settles; stochastic sources then start from their stationary distributions. Operators marked as ageing (graph meta "aging": True, for example a degradation integrator) are held at their initial state meanwhile. {"burn_in": seconds}: run with noise for that long before t = 0 and discard it. "none": start from each operator's initial state.

run 

run(until: float) -> None

Advance the clock to until seconds (exclusive), recording as configured.

fork 

fork() -> Engine

A copy of this engine; schedule different events on each to make twin runs.

result 

result(start: Any = None) -> Run

Collect everything recorded so far into a Run.

Event dataclass

A timed change.

at: time in seconds from the start of the run. target: a schedule-driven signal (for example "TIC-101.sp"), a parameter path ("TIC-101.plant.K"), which is then promoted to a signal at compile time, or, for "reset", an operator name. action: "set" to a value, "shift" by a value, "scale" by a factor, or "reset" an operator's state to its initial value (the value is ignored). profile: "step", or "ramp" over over seconds. label: "planned", or "fault:" for unplanned changes. origin: the alias that produced the event, for example "fault:disturbance@unit-1". lanes: the lanes the event applies to (None = all lanes).

Labels

visibility 

visibility(source: Any, window: float = 1800.0, z: float = 3.0, alpha: float = 0.001, persistence: float | None = None, signals: list[str] | None = None) -> Visibility

Run a scenario with one twin lane per fault and label when each fault becomes visible.

window, persistence: seconds (persistence defaults to half the window). signals: which signals to test (default: every observed signal).

Visibility dataclass

summary 

summary() -> pd.DataFrame

Per fault: when it first becomes visible on any signal, and on which.

settle_times 

settle_times(run: Run, band: float | None = None, hold: float = 600.0) -> pd.DataFrame

For each transition in run.meta["timeline"] and each loop, the first time after the transition's ramp ends at which the loop's true regulated variable is within band of its new setpoint and stays there for hold seconds. settle_time counts from the start of the transition, so it is never shorter than the ramp.

band: in the regulated variable's units; by default 1 % of the new setpoint. Settling is searched for before the next transition starts.

feasibility

Feasibility at the start: can each loop hold its setpoint without saturating?

A drawn plant can be valid yet unable to reach its setpoint (for example a low process gain with the valve fully open). This labeler checks, for every lane and loop, the first recorded step: the regulated variable must be within band of its setpoint (a fraction of the setpoint) and the controller output strictly inside its limits.

lineage 

lineage(source: Any, sample: str, units: list[str] | None = None) -> Lineage

Run a scenario with lineage tracers from units to the measured signal sample (for example a lab tag). By default, every unit with an inlet whose material reaches the sample.

Lineage dataclass

Errors

Issue dataclass

One problem found while compiling a graph, loading a scenario or running the engine.

HomeostatError

Bases: Exception

Base error; holds one or more issues.

Catalog

catalog 

catalog(library: str | None = 'process') -> dict[str, Any]

Describe the core operators and, optionally, a domain library, for people and AI assistants.