API-Referenz — Übersicht¶

Automatisch generierte API-Dokumentation via mkdocstrings.

Module¶

pyadm1ode_estimation.estimation — Kern der Zustandsschätzung
- base — EstimationStep, StateEstimator-Protocol
- state_vector — StateChannel, StateVectorSpec
- specs — adm1da_full_spec() Factory, InputSpec,
  KineticSpec, SensorQualityProfile, Quality-Enum
- process_model — ADM1ProcessModel
- observation_model — ObservationChannel, ObservationModel,
  eingebaute Extractor-Funktionen
- measurement_calendar — MeasurementCalendar, SampleRate
- sensors — SensorAdapter, measure_truth_with_sensors
  (Adapter zu PyADM1ODE-Sensor-Klassen)
- twin — Twin-Experiment-Helfer (propagate_truth, run_filter,
  coverage_within_2sigma)
- filters.sr_ukf — UnscentedKalmanFilter (Square-Root UKF)
pyadm1ode_estimation.example_plants — Referenz-Anlagen
- simple — build_simple_plant() (1 Fermenter + Storage + BHKW)
- multi_stage — build_multi_stage_plant()
  (3 Fermenter-Kaskade mit 2 BHKW)
pyadm1ode_estimation.artifacts — Schnittstelle zur Kalibrierung
- calibration_artifact — CalibrationArtifact, load_artifact,
  save_artifact, apply_to_plant

Auto-Doc-Beispiel¶

`pyadm1ode_estimation.estimation.state_vector` ¶

State-vector specification: what the filter actually estimates.

Three kinds of channels are supported:

"adm1" - a slot of the 41-element ADM1da state vector (see pyadm1.core.adm1 module-level docstring for the index map). Pushed into digester.adm1_state before each simulation step, read back afterwards.
"input_flow" - an augmented substrate-input rate in m³/d FM. Propagated by a random-walk (or OU) drift inside the filter. No direct interaction with the ADM1 state. Read by the process model to set digester.Q_substrates.
"kinetic_param" - an augmented kinetic-rate override (Phase 3). Written into adm1._kinetic before each step.

Classes¶

`StateChannel(name, kind='adm1', adm1_index=None, input_substrate_index=None, initial=0.0, initial_std=0.1, process_noise_std=0.01, drift_model='random_walk', ou_mean=None, ou_theta=1.0, lower=0.0, upper=float('inf'))` `dataclass` ¶

One slot in the augmented state vector.

Attributes:

Name	Type	Description
`name`	`str`	Human-readable identifier (used in plots / logs).
`kind`	`ChannelKind`	Where the value is consumed by the process model. `"adm1"` requires `adm1_index`. `"input_flow"` is applied via `Q_substrates`. `"kinetic_param"` is applied via `adm1._kinetic[name]`.
`adm1_index`	`int \| None`	Index in the 41-element ADM1da state vector, required when `kind="adm1"`.
`input_substrate_index`	`int \| None`	Position in the plant's `Q_substrates` vector for `kind="input_flow"`. The n-slot vector is laid out as `[Q_total_substrate_1, …, Q_total_substrate_n]`, where `n` matches the schema substrate count.
`initial`	`float`	Initial state estimate.
`initial_std`	`float`	Initial standard deviation (used to build :math:`P_0`).
`process_noise_std`	`float`	Standard deviation of the per-step process noise. The filter scales this by `sqrt(dt)` for the `random_walk` drift model.
`drift_model`	`DriftModel`	Augmented dynamics. `"random_walk"` is the default; `"ou"` is an Ornstein-Uhlenbeck pull-back to `ou_mean` with rate `ou_theta` and is recommended for input flows that should not drift unbounded during long sensor-blind intervals; `"constant"` adds zero noise (used for diagnostics).
`ou_mean`	`float \| None`	Long-term mean for the OU process.
`ou_theta`	`float`	Mean-reversion rate for the OU process [1/d].
`lower,`	`upper`	Hard bounds enforced after every predict / update step.

`StateVectorSpec(digester_id, channels=list())` `dataclass` ¶

A list of channels, ordered as they appear in the state vector.

The order matters because the filter's covariance matrix and sigma-point construction use it. Reordering channels after the fact will silently break P.

Methods:¶

clip(x) ¶

Hard-clip a state vector to the configured channel bounds.

process_noise_cov(dt) ¶

Discrete process-noise covariance for one dt- step.

Random-walk / constant channels contribute (σ_w · sqrt(dt))^2. OU channels contribute the discrete equivalent σ²(1 - e^{-2θ·dt}) / (2θ) (Doob-Itō). The mean-reversion term itself is applied inside the process model, not here.

push_adm1_state(x, plant) ¶

Write the "adm1" channels of x into the digester.

read_adm1_state(plant) ¶

Read the "adm1" channels back from the digester state.

`pyadm1ode_estimation.estimation.base` ¶

Protocols and shared dataclasses for the estimation subpackage.

Classes¶

`BatchEstimator` ¶

Bases: Protocol

Common interface for offline/batch estimators (e.g. a PINN smoother).

Unlike :class:StateEstimator, which runs a recursive predict/update loop, a batch estimator is fitted to a whole window of (sparse) measurements at once and then queried for the full state trajectory:

.. code-block:: text

estimator.fit(...)                       # train over the window
traj = estimator.estimate(query_times)   # (T, n_state) + std

fit is estimator-specific (training schedule, collocation, priors); :meth:estimate is the shared output contract used for evaluation.

Methods:¶

estimate(times) ¶

Return the posterior trajectory + uncertainty at times.

`EstimationStep(t, x_hat, P, y_pred=dict(), y_std=dict(), innovation=dict(), nis=float('nan'), active_channels=list())` `dataclass` ¶

One filter step's output, suitable for time-series logging.

Attributes:

Name	Type	Description
`t`	`float`	Wall-clock time of the step (days since experiment start or any consistent monotonic reference).
`x_hat`	`ndarray`	Posterior state estimate, shape `(n_state,)`.
`P`	`ndarray`	Posterior covariance, shape `(n_state, n_state)`.
`y_pred`	`dict[str, float]`	Predicted measurement value for each channel that had an observation in this step. Empty dict on a pure predict-only step.
`y_std`	`dict[str, float]`	Standard deviation of the predicted measurement, `sqrt(diag(S))` where `S = Cov(h(sigma)) + R` is the full innovation covariance. Includes both state-driven uncertainty (state covariance propagated through `h`) and the sensor noise `R`. Useful for plotting predicted measurement bands.
`innovation`	`dict[str, float]`	`y_obs - y_pred` per active channel.
`nis`	`float`	Normalised innovation squared `ν^T S^{-1} ν` summed across active channels. `nan` when no channel was observed.
`active_channels`	`list`	Names of channels that contributed to the update. Useful for diagnostics with gated channels.

`StateEstimator` ¶

Bases: Protocol

Common interface for all filters in this subpackage.

The lifecycle is the usual two-step Bayesian filter loop:

.. code-block:: text

for each measurement time t:
    estimator.predict(dt)            # propagate prior
    step = estimator.update(y_dict)  # incorporate measurement
    log(step)

Augmented-input states (e.g. substrate flows) live inside the state vector and are picked up by predict from the latest x_hat. Sparse / gated channels are passed in y_dict with only the keys that are currently observable.

Methods:¶

predict(dt) ¶

Propagate the state distribution forward by dt days.

reset(x0, P0) ¶

Re-initialise the filter with a new prior.

update(y, t) ¶

Fuse observations y at time t into the estimate.

`TrajectoryEstimate(time, x_hat, std)` `dataclass` ¶

Output of a batch/offline estimator over a whole window.

The shared currency between recursive filters and batch smoothers: a posterior trajectory with a per-state uncertainty band, so the twin- experiment harness (coverage, plots, error metrics) can evaluate both estimator families the same way.

Attributes:

Name	Type	Description
`time`	`ndarray`	Query times, shape `(T,)` [days].
`x_hat`	`ndarray`	Posterior state estimate, shape `(T, n_state)`.
`std`	`ndarray`	Per-state posterior standard deviation, shape `(T, n_state)`. Zeros when the estimator provides no uncertainty.

`pyadm1ode_estimation.artifacts.calibration_artifact` ¶

Calibration artifact — handoff format between calibration and estimation.

Kalibrierungs-Artefakt — Übergabeformat zwischen Kalibrierung und Zustandsschätzung.

Workflow / Ablauf¶

pyadm1ode_calibration optimiert ADM1-Parameter gegen historische Messdaten und schreibt das Ergebnis als YAML via :func:save_artifact.
pyadm1ode_estimation lädt das YAML beim Hochfahren des Live-Filters via :func:load_artifact und legt die Parameter via :func:apply_to_plant auf einen frisch gebauten BiogasPlant.

Design / Designprinzipien¶

Versioniert — schema_version macht das Format zukunftssicher. Unbekannte Versionen verursachen einen Fehler statt einer stillen Fehlinterpretation.
Sektioniert und partiell — jede Sektion (kinetic, substrates, initial_state, residuals) ist optional. Eine Calibration die nur kinetische Parameter fittet schreibt nur diese Sektion.
Plant-agnostisch — Dictionaries statt fester Felder erlauben beliebige Anlagen und ADM1-Varianten ohne Schema-Bruch.
Tolerant beim Apply — unbekannte Keys oder fehlende Komponenten lösen eine Warnung aus, kein Crash. So bleibt ein älteres Artefakt gegen eine weiterentwickelte Topologie noch nutzbar (mit eingeschränkter Wirkung).

Example YAML¶

.. code-block:: yaml

schema_version: 1

metadata: plant_id: plant_id_xyz calibration_run_id: 2026-05-14-3w timestamp: 2026-05-14T12:30:00Z data_window_start: 2026-04-01 data_window_end: 2026-04-21 adm1_version: "1.2" calibration_version: "0.2.0" fitted_against: [Q_gas, P_el] notes: "3-week window"

kinetic: primary: k_dis: 0.4 k_hyd_ch: 10.0

initial_state: primary: _values: [0.012, 0.0053, ... ] # 37 ADM1 entries

Classes¶

`CalibrationArtifact(metadata, kinetic=dict(), substrates=dict(), initial_state=dict(), residuals=dict(), schema_version=SCHEMA_VERSION)` `dataclass` ¶

Calibrated simulation-model parameters, ready to apply to a plant.

Kalibrierte Simulationsmodell-Parameter, bereit zum Anwenden auf eine Anlage.

Sections / Sektionen:

metadata — provenance, see :class:CalibrationMetadata.
kinetic — per-digester overrides for adm1._kinetic. Outer key: digester component id. Inner dict: kinetic parameter name → value (units defined by pyadm1).
substrates — per-substrate composition / disintegration fractions. Outer key: substrate name as known to the plant builder. Inner dict: fraction name → value (e.g. f_ch_xc).
initial_state — per-digester ADM1 state at the end of the calibration window. Use as prior mean for the filter's initial x_hat. Inner dict accepts either named entries ({"S_su": 0.012, ...}) or the array shortcut {"_values": [...]} with 37 entries in canonical ADM1 order.
residuals — per-channel fit quality (RMSE, MAE, …). Informational only; not consumed by Estimation.
schema_version — must equal :data:SCHEMA_VERSION to be readable by this runtime.

`CalibrationMetadata(plant_id, calibration_run_id, timestamp='', data_window_start=None, data_window_end=None, adm1_version='', calibration_version='', fitted_against=list(), notes='')` `dataclass` ¶

Provenance information for a calibration run.

Provenance-Information eines Calibration-Runs.

Identifies which plant was calibrated, against which data window, by which software version. Estimation should refuse to apply an artifact whose plant_id does not match the runtime plant.

Attributes:

Name	Type	Description
`plant_id`	`str`	Stable identifier of the physical plant.
`calibration_run_id`	`str`	Identifier of the optimization run that produced this artifact. Should be unique per run.
`timestamp`	`str`	ISO8601 timestamp when the artifact was written.
`data_window_start`	`str \| None`	Start of the historical data window used in calibration (ISO date).
`data_window_end`	`str \| None`	End of the historical data window (ISO date).
`adm1_version`	`str`	Version of the ADM1 model these parameters target. Useful for detecting incompatible upgrades.
`calibration_version`	`str`	Version of the calibration software that produced the artifact.
`fitted_against`	`list[str]`	Names of the measurement channels the optimizer actually used. Informational only.
`notes`	`str`	Free-form operator notes (e.g. `"BHKW1 offline that week"`).

Functions:¶

`apply_to_plant(artifact, plant, *, strict=False)` ¶

Overlay the artifact's parameters onto a fresh :class:BiogasPlant.

Trägt die kalibrierten Parameter auf eine bereits gebaute Anlage auf.

The plant must have its topology built (components present) before this is called. apply_to_plant only writes calibrated values into existing slots; it does not create new components or substrate definitions.

Mismatches (unknown digester id, unknown kinetic key, …) trigger a warning in non-strict mode so an older artifact can still be applied to an evolved topology with graceful degradation. In strict mode the same mismatches raise KeyError — recommended for production startup, when you want to fail fast.

Parameters:

Name	Type	Description	Default
`artifact`	`CalibrationArtifact`	The calibration artifact to apply.	required
`plant`	`BiogasPlant`	A fully built `pyadm1.BiogasPlant`.	required
`strict`	`bool`	When `True`, raise on any mismatch; when `False` (default), emit warnings and skip.	`False`

Returns:

Type	Description
`dict[str, list[str]]`	A diagnostics dict with keys `"applied"` (parameters actually
`dict[str, list[str]]`	written) and `"skipped"` (mismatches encountered).

`load_artifact(path)` ¶

Load an artifact from a YAML file.

Lädt ein Artefakt aus einer YAML-Datei.

Validates that schema_version matches this runtime. Unknown versions raise ValueError so the operator is forced to upgrade rather than silently mis-reading the file.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	Source file path.	required

Returns:

Type	Description
`CalibrationArtifact`	The deserialized artifact.

Raises:

Type	Description
`ValueError`	If the YAML is not a mapping or carries an unknown `schema_version`.
`FileNotFoundError`	If the file does not exist.

`save_artifact(artifact, path)` ¶

Serialize an artifact to a YAML file.

Schreibt das Artefakt als YAML-Datei.

Parameters:

Name	Type	Description	Default
`artifact`	`CalibrationArtifact`	The artifact to write.	required
`path`	`str \| Path`	Destination file path. Parent directories are created if missing.	required

API-Referenz — Übersicht¶

Module¶

Auto-Doc-Beispiel¶

pyadm1ode_estimation.estimation.state_vector ¶

Classes¶

StateChannel(name, kind='adm1', adm1_index=None, input_substrate_index=None, initial=0.0, initial_std=0.1, process_noise_std=0.01, drift_model='random_walk', ou_mean=None, ou_theta=1.0, lower=0.0, upper=float('inf')) dataclass ¶

StateVectorSpec(digester_id, channels=list()) dataclass ¶

Methods:¶

pyadm1ode_estimation.estimation.base ¶

Classes¶

BatchEstimator ¶

Methods:¶

EstimationStep(t, x_hat, P, y_pred=dict(), y_std=dict(), innovation=dict(), nis=float('nan'), active_channels=list()) dataclass ¶

StateEstimator ¶

Methods:¶

TrajectoryEstimate(time, x_hat, std) dataclass ¶

pyadm1ode_estimation.artifacts.calibration_artifact ¶

Workflow / Ablauf¶

Design / Designprinzipien¶

Example YAML¶

Classes¶

CalibrationArtifact(metadata, kinetic=dict(), substrates=dict(), initial_state=dict(), residuals=dict(), schema_version=SCHEMA_VERSION) dataclass ¶

CalibrationMetadata(plant_id, calibration_run_id, timestamp='', data_window_start=None, data_window_end=None, adm1_version='', calibration_version='', fitted_against=list(), notes='') dataclass ¶

Functions:¶

apply_to_plant(artifact, plant, *, strict=False) ¶

load_artifact(path) ¶

save_artifact(artifact, path) ¶

`pyadm1ode_estimation.estimation.state_vector` ¶

`StateChannel(name, kind='adm1', adm1_index=None, input_substrate_index=None, initial=0.0, initial_std=0.1, process_noise_std=0.01, drift_model='random_walk', ou_mean=None, ou_theta=1.0, lower=0.0, upper=float('inf'))` `dataclass` ¶

`StateVectorSpec(digester_id, channels=list())` `dataclass` ¶

`pyadm1ode_estimation.estimation.base` ¶

`BatchEstimator` ¶

`EstimationStep(t, x_hat, P, y_pred=dict(), y_std=dict(), innovation=dict(), nis=float('nan'), active_channels=list())` `dataclass` ¶

`StateEstimator` ¶

`TrajectoryEstimate(time, x_hat, std)` `dataclass` ¶

`pyadm1ode_estimation.artifacts.calibration_artifact` ¶

`CalibrationArtifact(metadata, kinetic=dict(), substrates=dict(), initial_state=dict(), residuals=dict(), schema_version=SCHEMA_VERSION)` `dataclass` ¶

`CalibrationMetadata(plant_id, calibration_run_id, timestamp='', data_window_start=None, data_window_end=None, adm1_version='', calibration_version='', fitted_against=list(), notes='')` `dataclass` ¶

`apply_to_plant(artifact, plant, *, strict=False)` ¶

`load_artifact(path)` ¶

`save_artifact(artifact, path)` ¶