Kalibrierung (API)¶

Calibrator Facade¶

`pyadm1ode_calibration.calibration.Calibrator(plant, verbose=True)` ¶

Orchestration layer for calibration workflows.

Provides a simplified interface for running both initial and online calibrations on a plant model.

Parameters:

Name	Type	Description	Default
`plant`	`Any`	The PyADM1ODE plant model instance.	required
`verbose`	`bool`	Whether to enable verbose logging. Defaults to True.	`True`

Source code in pyadm1ode_calibration/calibration/__init__.py

def __init__(self, plant: Any, verbose: bool = True):
    self.plant = plant
    self.verbose = verbose
    self.initial_calibrator = InitialCalibrator(plant, verbose)
    self.online_calibrator = OnlineCalibrator(plant, verbose)

Methods:¶

`apply_calibration(result)` ¶

Apply calibration results to the plant model.

Parameters:

Name	Type	Description	Default
`result`	`CalibrationResult`	Result containing new parameters.	required

Source code in pyadm1ode_calibration/calibration/__init__.py

def apply_calibration(self, result: CalibrationResult) -> None:
    """
    Apply calibration results to the plant model.

    Args:
        result (CalibrationResult): Result containing new parameters.
    """
    self.online_calibrator.apply_calibration(result)

`run_initial_calibration(measurements, parameters, **kwargs)` ¶

Run initial batch calibration from historical data.

Parameters:

Name	Type	Description	Default
`measurements`	`MeasurementData`	Historical measurement data.	required
`parameters`	`list[str]`	List of parameter names to calibrate.	required
`**kwargs`	`Any`	Additional settings for InitialCalibrator.	`{}`

Returns:

Name	Type	Description
`CalibrationResult`	`CalibrationResult`	Results of the calibration.

Source code in pyadm1ode_calibration/calibration/__init__.py

def run_initial_calibration(
    self, measurements: "MeasurementData", parameters: list[str], **kwargs: Any
) -> CalibrationResult:
    """
    Run initial batch calibration from historical data.

    Args:
        measurements: Historical measurement data.
        parameters: List of parameter names to calibrate.
        **kwargs: Additional settings for InitialCalibrator.

    Returns:
        CalibrationResult: Results of the calibration.
    """
    return self.initial_calibrator.calibrate(measurements, parameters, **kwargs)

`run_online_calibration(measurements, parameters, **kwargs)` ¶

Run online re-calibration for real-time adjustments.

Parameters:

Name	Type	Description	Default
`measurements`	`MeasurementData`	Recent measurement data.	required
`parameters`	`list[str]`	List of parameter names to calibrate.	required
`**kwargs`	`Any`	Additional settings for OnlineCalibrator.	`{}`

Returns:

Name	Type	Description
`CalibrationResult`	`CalibrationResult`	Results of the calibration.

Source code in pyadm1ode_calibration/calibration/__init__.py

def run_online_calibration(
    self, measurements: "MeasurementData", parameters: list[str], **kwargs: Any
) -> CalibrationResult:
    """
    Run online re-calibration for real-time adjustments.

    Args:
        measurements: Recent measurement data.
        parameters: List of parameter names to calibrate.
        **kwargs: Additional settings for OnlineCalibrator.

    Returns:
        CalibrationResult: Results of the calibration.
    """
    return self.online_calibrator.calibrate(measurements, parameters, **kwargs)

Initial Calibrator¶

`pyadm1ode_calibration.calibration.InitialCalibrator(plant, verbose=True, time_varying_feed=False)` ¶

Bases: BaseCalibrator

Initial calibrator for ADM1 parameters from historical data.

This calibrator is designed for batch optimization using a window of historical measurement data. It supports multi-objective optimization, sensitivity analysis, and cross-validation.

Parameters:

Name	Type	Description	Default
`plant`	`Any`	The PyADM1ODE plant model to calibrate.	required
`verbose`	`bool`	Whether to enable verbose output. Defaults to True.	`True`

Source code in pyadm1ode_calibration/calibration/methods/initial.py

def __init__(self, plant: Any, verbose: bool = True, time_varying_feed: bool = False):
    super().__init__(plant, verbose, time_varying_feed=time_varying_feed)
    self.parameter_bounds = create_default_bounds()
    self.validator = CalibrationValidator(plant, verbose=False, time_varying_feed=time_varying_feed)
    self.sensitivity_analyzer = SensitivityAnalyzer(plant, self.simulator, verbose)
    self.identifiability_analyzer = IdentifiabilityAnalyzer(plant, self.sensitivity_analyzer, verbose)
    self._optimization_history: list[dict[str, Any]] = []
    self._best_objective_value: float = float("inf")
    self._original_parameters: dict[str, float] = self._get_current_parameters()

Methods:¶

`calibrate(measurements, parameters, bounds=None, method='differential_evolution', objectives=None, weights=None, validation_split=0.2, max_iterations=100, population_size=15, tolerance=0.0001, sensitivity_analysis=True, use_constraints=False, check_identifiability=False, **kwargs)` ¶

Run the initial calibration workflow.

Parameters:

Name	Type	Description	Default
`measurements`	`MeasurementData`	Historical measurement data for calibration.	required
`parameters`	`List[str]`	Names of parameters to optimize.	required
`bounds`	`Optional[Dict[str, Tuple[float, float]]]`	Custom search bounds for parameters.	`None`
`method`	`str`	Optimization algorithm name. Defaults to 'differential_evolution'.	`'differential_evolution'`
`objectives`	`Optional[List[str]]`	List of objective variables (e.g., ['Q_ch4', 'pH']).	`None`
`weights`	`Optional[Dict[str, float]]`	Weights for different objectives in the cost function.	`None`
`validation_split`	`float`	Fraction of data to use for out-of-sample validation. Defaults to 0.2.	`0.2`
`max_iterations`	`int`	Maximum number of optimizer iterations. Defaults to 100.	`100`
`population_size`	`int`	Population size for evolutionary algorithms. Defaults to 15.	`15`
`tolerance`	`float`	Convergence tolerance. Defaults to 1e-4.	`0.0001`
`sensitivity_analysis`	`bool`	Whether to perform sensitivity analysis after calibration.	`True`
`use_constraints`	`bool`	Whether to apply parameter constraints. Defaults to False.	`False`
`check_identifiability`	`bool`	Screen the parameter set for collinearity before optimizing and warn when it is unidentifiable as a set. Costs `2 * len(parameters) + 1` simulations against the several hundred a global search needs, so it is worth it for anything but a one-parameter fit. Defaults to False to leave existing behaviour untouched.	`False`
`**kwargs`	`Any`	Additional keyword arguments passed to the optimizer.	`{}`

Returns:

Name	Type	Description
`CalibrationResult`	`CalibrationResult`	The calibration results including optimized parameters and metrics.

Source code in pyadm1ode_calibration/calibration/methods/initial.py

def calibrate(
    self,
    measurements: MeasurementData,
    parameters: list[str],
    bounds: dict[str, tuple[float, float]] | None = None,
    method: str = "differential_evolution",
    objectives: list[str] | None = None,
    weights: dict[str, float] | None = None,
    validation_split: float = 0.2,
    max_iterations: int = 100,
    population_size: int = 15,
    tolerance: float = 1e-4,
    sensitivity_analysis: bool = True,
    use_constraints: bool = False,
    check_identifiability: bool = False,
    **kwargs: Any,
) -> CalibrationResult:
    """
    Run the initial calibration workflow.

    Args:
        measurements (MeasurementData): Historical measurement data for calibration.
        parameters (List[str]): Names of parameters to optimize.
        bounds (Optional[Dict[str, Tuple[float, float]]]): Custom search bounds for parameters.
        method (str): Optimization algorithm name. Defaults to 'differential_evolution'.
        objectives (Optional[List[str]]): List of objective variables (e.g., ['Q_ch4', 'pH']).
        weights (Optional[Dict[str, float]]): Weights for different objectives in the cost function.
        validation_split (float): Fraction of data to use for out-of-sample validation. Defaults to 0.2.
        max_iterations (int): Maximum number of optimizer iterations. Defaults to 100.
        population_size (int): Population size for evolutionary algorithms. Defaults to 15.
        tolerance (float): Convergence tolerance. Defaults to 1e-4.
        sensitivity_analysis (bool): Whether to perform sensitivity analysis after calibration.
        use_constraints (bool): Whether to apply parameter constraints. Defaults to False.
        check_identifiability (bool): Screen the parameter set for collinearity before
            optimizing and warn when it is unidentifiable as a set. Costs
            ``2 * len(parameters) + 1`` simulations against the several hundred a
            global search needs, so it is worth it for anything but a one-parameter
            fit. Defaults to False to leave existing behaviour untouched.
        **kwargs (Any): Additional keyword arguments passed to the optimizer.

    Returns:
        CalibrationResult: The calibration results including optimized parameters and metrics.
    """
    start_time = time.time()
    if objectives is None:
        objectives = ["Q_ch4"]

    # Split data
    train_data, val_data = self._split_data(measurements, validation_split)

    initial_params = self.parameter_bounds.get_default_values(parameters)
    param_bounds = self._setup_bounds(parameters, bounds)

    # Finding out that a set could never have been identified is worth far more
    # before the optimizer runs than after it.
    self.subset_identifiability: SubsetIdentifiability | None = None
    if check_identifiability and len(parameters) > 1:
        self.subset_identifiability = self.identifiability_analyzer.analyze_subset(initial_params, train_data, objectives)
        if not self.subset_identifiability.is_identifiable:
            warnings.warn(
                f"Parameter set {parameters} is not identifiable together: " f"{self.subset_identifiability.reason}",
                stacklevel=2,
            )
        elif self.verbose:
            print(f"  {self.subset_identifiability.reason}")

    # Create objective function
    def simulator_wrapper(params: dict[str, float]) -> dict[str, np.ndarray]:
        return self.simulator.simulate_with_parameters(params, train_data)

    measurements_dict: dict[str, np.ndarray] = {}
    for obj in objectives:
        if obj in train_data.data.columns:
            measurements_dict[obj] = train_data.get_measurement(obj).values

    objective_func: Callable[[np.ndarray], float]
    if weights is None:
        objective_func = WeightedSumObjective(
            simulator=simulator_wrapper,
            measurements_dict=measurements_dict,
            objectives=objectives,
            parameter_names=parameters,
            error_metric="rmse",
            normalize=True,
        )
    else:
        objective_func = MultiObjectiveFunction(
            simulator=simulator_wrapper,
            measurements_dict=measurements_dict,
            objectives=objectives,
            weights=weights,
            parameter_names=parameters,
            error_metric="rmse",
            normalize=True,
        )

    # Constraints
    obj_func_final: Callable[[np.ndarray], float]
    if use_constraints:
        constraints = ParameterConstraints()
        for param, (lb, ub) in param_bounds.items():
            constraints.add_box_constraint(param, lb, ub, hard=True)

        def penalized_objective(x: np.ndarray) -> float:
            params = {name: val for name, val in zip(parameters, x)}
            return objective_func(x) + constraints.calculate_penalty(params)

        obj_func_final = penalized_objective
    else:
        obj_func_final = objective_func

    optimizer_kwargs = {**kwargs}
    optimizer_kwargs["tolerance"] = tolerance
    if method in ["differential_evolution", "de"]:
        optimizer_kwargs["population_size"] = population_size

    optimizer = create_optimizer(
        method=method,
        bounds=param_bounds,
        max_iterations=max_iterations,
        verbose=self.verbose,
        **optimizer_kwargs,
    )

    initial_guess = (
        np.array([initial_params[p] for p in parameters]) if method in ["nelder_mead", "nm", "lbfgsb", "powell"] else None
    )
    opt_result = optimizer.optimize(obj_func_final, initial_guess=initial_guess)

    # Validation
    validation_metrics: dict[str, float] = {}
    if len(val_data) > 0:
        # Without the warmup the validation window would be simulated from the
        # plant's initial state, so a late stretch of a record could never line
        # up with its measurements no matter how good the parameters are.
        val_result = self.validator.validate(
            parameters=opt_result.parameter_dict,
            measurements=val_data,
            objectives=objectives,
            warmup=train_data,
        )
        for obj, metrics in val_result.items():
            validation_metrics.update(
                {f"{obj}_rmse": float(metrics.rmse), f"{obj}_r2": float(metrics.r2), f"{obj}_nse": float(metrics.nse)}
            )

    # Sensitivity
    sensitivity_results: dict[str, float] = {}
    if sensitivity_analysis and opt_result.success:
        sens = self.sensitivity_analyzer.analyze(opt_result.parameter_dict, train_data, objectives)
        sensitivity_results = {p: float(max(abs(s) for s in r.sensitivity_indices.values())) for p, r in sens.items()}

    return CalibrationResult(
        success=opt_result.success,
        parameters=opt_result.parameter_dict,
        initial_parameters=initial_params,
        objective_value=float(opt_result.fun),
        n_iterations=int(opt_result.nit),
        execution_time=time.time() - start_time,
        method=method,
        message=str(opt_result.message) if hasattr(opt_result, "message") else "Optimization completed",
        validation_metrics=validation_metrics,
        sensitivity=sensitivity_results,
        history=opt_result.history,
    )

`identifiability_analysis(parameters, measurements, optimization_history=None, confidence_level=0.95, correlation_threshold=0.8)` ¶

Perform parameter identifiability analysis.

Checks for parameter correlations and information content in the data.

Parameters:

Name	Type	Description	Default
`parameters`	`Dict[str, float]`	Parameter set to analyze.	required
`measurements`	`MeasurementData`	Data window for simulation.	required
`optimization_history`	`Optional[List[Dict[str, Any]]]`	History from a previous optimizer run, used to widen the confidence intervals.	`None`
`confidence_level`	`float`	Confidence level for the intervals.	`0.95`
`correlation_threshold`	`float`	Above this, two parameters count as correlated and neither is reported as identifiable on its own.	`0.8`

Returns:

Name	Type	Description
`IdentifiabilityResult`	`dict[str, IdentifiabilityResult]`	Analysis results including correlation matrix.

Source code in pyadm1ode_calibration/calibration/methods/initial.py

def identifiability_analysis(
    self,
    parameters: dict[str, float],
    measurements: MeasurementData,
    optimization_history: list[dict[str, Any]] | None = None,
    confidence_level: float = 0.95,
    correlation_threshold: float = 0.8,
) -> dict[str, IdentifiabilityResult]:
    """
    Perform parameter identifiability analysis.

    Checks for parameter correlations and information content in the data.

    Args:
        parameters (Dict[str, float]): Parameter set to analyze.
        measurements (MeasurementData): Data window for simulation.
        optimization_history (Optional[List[Dict[str, Any]]]): History from a previous
            optimizer run, used to widen the confidence intervals.
        confidence_level (float): Confidence level for the intervals.
        correlation_threshold (float): Above this, two parameters count as correlated
            and neither is reported as identifiable on its own.

    Returns:
        IdentifiabilityResult: Analysis results including correlation matrix.
    """
    return self.identifiability_analyzer.analyze(
        parameters,
        measurements,
        optimization_history=optimization_history,
        confidence_level=confidence_level,
        correlation_threshold=correlation_threshold,
    )

`sensitivity_analysis(parameters, measurements, objectives=None)` ¶

Perform local sensitivity analysis for given parameters.

Parameters:

Name	Type	Description	Default
`parameters`	`Dict[str, float]`	Parameter set to analyze.	required
`measurements`	`MeasurementData`	Data window for simulation.	required
`objectives`	`Optional[List[str]]`	List of objective variables.	`None`

Returns:

Type	Description
`dict[str, SensitivityResult]`	Dict[str, SensitivityResult]: Mapping of parameter names to sensitivity indices.

Source code in pyadm1ode_calibration/calibration/methods/initial.py

def sensitivity_analysis(
    self, parameters: dict[str, float], measurements: MeasurementData, objectives: list[str] | None = None
) -> dict[str, SensitivityResult]:
    """
    Perform local sensitivity analysis for given parameters.

    Args:
        parameters (Dict[str, float]): Parameter set to analyze.
        measurements (MeasurementData): Data window for simulation.
        objectives (Optional[List[str]]): List of objective variables.

    Returns:
        Dict[str, SensitivityResult]: Mapping of parameter names to sensitivity indices.
    """
    return self.sensitivity_analyzer.analyze(parameters, measurements, objectives)

Online Calibrator¶

`pyadm1ode_calibration.calibration.OnlineCalibrator(plant, verbose=True, parameter_bounds=None, time_varying_feed=False)` ¶

Bases: BaseCalibrator

Online calibrator for real-time parameter adjustment.

Performs fast, bounded re-calibration when model predictions deviate from measurements. It is optimized for speed and stability, ensuring that parameters do not drift too far from their physical meanings.

Parameters:

Name	Type	Description	Default
`plant`	`Any`	The PyADM1ODE plant model to calibrate.	required
`verbose`	`bool`	Whether to enable verbose logging. Defaults to True.	`True`
`parameter_bounds`	`Optional[ParameterBounds]`	Custom parameter bounds manager.	`None`

Source code in pyadm1ode_calibration/calibration/methods/online.py

def __init__(
    self,
    plant: Any,
    verbose: bool = True,
    parameter_bounds: ParameterBounds | None = None,
    time_varying_feed: bool = False,
):
    super().__init__(plant, verbose, time_varying_feed=time_varying_feed)
    self.parameter_bounds: ParameterBounds = parameter_bounds or create_default_bounds()
    self.validator: CalibrationValidator = CalibrationValidator(plant, verbose=False, time_varying_feed=time_varying_feed)
    self.trigger: OnlineCalibrationTrigger = OnlineCalibrationTrigger()
    self.state: OnlineState = OnlineState()

Methods:¶

`apply_calibration(result)` ¶

Apply calibrated parameters to the underlying plant model.

Parameters:

Name	Type	Description	Default
`result`	`CalibrationResult`	The result containing new parameter values.	required

Source code in pyadm1ode_calibration/calibration/methods/online.py

def apply_calibration(self, result: CalibrationResult) -> None:
    """
    Apply calibrated parameters to the underlying plant model.

    Args:
        result (CalibrationResult): The result containing new parameter values.
    """
    for component in self.plant.components.values():
        if component.component_type.value == "digester":
            component.apply_calibration_parameters(result.parameters)

`calibrate(measurements, parameters=None, current_parameters=None, variance_threshold=0.15, max_parameter_change=0.2, time_window=7, method='nelder_mead', max_iterations=50, objectives=None, weights=None, use_constraints=True, **kwargs)` ¶

Perform online re-calibration with bounded parameter adjustments.

Parameters:

Name	Type	Description	Default
`measurements`	`MeasurementData`	Recent measurement data window.	required
`parameters`	`Optional[List[str]]`	Parameters to adjust. If None, uses history.	`None`
`current_parameters`	`Optional[Dict[str, float]]`	Current values. If None, gets from plant.	`None`
`variance_threshold`	`float`	Variance threshold for triggering (0-1).	`0.15`
`max_parameter_change`	`float`	Max relative parameter change (0.0 to 1.0).	`0.2`
`time_window`	`int`	Number of days of recent data to use for optimization.	`7`
`method`	`str`	Optimization method name. Defaults to 'nelder_mead'.	`'nelder_mead'`
`max_iterations`	`int`	Maximum optimization iterations. Defaults to 50.	`50`
`objectives`	`Optional[List[str]]`	List of outputs to match (e.g., ['Q_ch4']).	`None`
`weights`	`Optional[Dict[str, float]]`	Objective weights for multi-objective cost.	`None`
`use_constraints`	`bool`	Whether to apply parameter box constraints.	`True`
`**kwargs`	`Any`	Extra settings passed to the optimizer.	`{}`

Returns:

Name	Type	Description
`CalibrationResult`	`CalibrationResult`	Results of the online calibration step.

Raises:

Type	Description
`ValueError`	If no parameters are specified and no history is available.

Source code in pyadm1ode_calibration/calibration/methods/online.py

def calibrate(
    self,
    measurements: MeasurementData,
    parameters: list[str] | None = None,
    current_parameters: dict[str, float] | None = None,
    variance_threshold: float = 0.15,
    max_parameter_change: float = 0.20,
    time_window: int = 7,
    method: str = "nelder_mead",
    max_iterations: int = 50,
    objectives: list[str] | None = None,
    weights: dict[str, float] | None = None,
    use_constraints: bool = True,
    **kwargs: Any,
) -> CalibrationResult:
    """
    Perform online re-calibration with bounded parameter adjustments.

    Args:
        measurements (MeasurementData): Recent measurement data window.
        parameters (Optional[List[str]]): Parameters to adjust. If None, uses history.
        current_parameters (Optional[Dict[str, float]]): Current values. If None, gets from plant.
        variance_threshold (float): Variance threshold for triggering (0-1).
        max_parameter_change (float): Max relative parameter change (0.0 to 1.0).
        time_window (int): Number of days of recent data to use for optimization.
        method (str): Optimization method name. Defaults to 'nelder_mead'.
        max_iterations (int): Maximum optimization iterations. Defaults to 50.
        objectives (Optional[List[str]]): List of outputs to match (e.g., ['Q_ch4']).
        weights (Optional[Dict[str, float]]): Objective weights for multi-objective cost.
        use_constraints (bool): Whether to apply parameter box constraints.
        **kwargs (Any): Extra settings passed to the optimizer.

    Returns:
        CalibrationResult: Results of the online calibration step.

    Raises:
        ValueError: If no parameters are specified and no history is available.
    """
    start_time = time.time()
    if objectives is None:
        objectives = ["Q_ch4", "pH"]

    if parameters is None:
        if self.state.parameter_history:
            parameters = list(self.state.parameter_history[-1].parameters.keys())
        else:
            raise ValueError("No parameters specified and no calibration history available")

    if current_parameters is None:
        current_parameters = self._get_current_parameters()

    windowed_data = self._extract_time_window(measurements, time_window)
    current_variance = self._calculate_prediction_variance(windowed_data, current_parameters, objectives)
    self.state.current_variance = current_variance

    param_bounds = self._setup_online_bounds(parameters, current_parameters, max_parameter_change)

    def simulator_wrapper(params: dict[str, float]) -> dict[str, np.ndarray]:
        return self.simulator.simulate_with_parameters(params, windowed_data)

    measurements_dict: dict[str, np.ndarray] = {
        obj: windowed_data.get_measurement(obj).values for obj in objectives if obj in windowed_data.data.columns
    }

    objective_func: Callable[[np.ndarray], float] = MultiObjectiveFunction(
        simulator=simulator_wrapper,
        measurements_dict=measurements_dict,
        objectives=objectives,
        weights=weights or {obj: 1.0 / len(objectives) for obj in objectives},
        parameter_names=parameters,
        error_metric="rmse",
        normalize=True,
    )

    obj_func_final: Callable[[np.ndarray], float]
    if use_constraints:
        constraints = ParameterConstraints()
        for p, (lb, ub) in param_bounds.items():
            constraints.add_box_constraint(p, lb, ub, hard=True)

        def penalized_objective(x: np.ndarray) -> float:
            params = {name: val for name, val in zip(parameters, x)}
            return objective_func(x) + constraints.calculate_penalty(params)

        obj_func_final = penalized_objective
    else:
        obj_func_final = objective_func

    optimizer = create_optimizer(
        method=method, bounds=param_bounds, max_iterations=max_iterations, verbose=self.verbose, **kwargs
    )

    initial_guess = np.array(
        [current_parameters.get(p, self.parameter_bounds.get_default_values([p])[p]) for p in parameters]
    )
    opt_result = optimizer.optimize(obj_func_final, initial_guess=initial_guess)

    validation_metrics: dict[str, float] = {}
    if opt_result.success:
        val_res = self.validator.validate(
            parameters=opt_result.parameter_dict, measurements=windowed_data, objectives=objectives
        )
        validation_metrics = {f"{obj}_{k}": float(getattr(m, k)) for obj, m in val_res.items() for k in ["rmse", "r2"]}

    self.state.total_calibrations += 1
    self.state.last_calibration_time = datetime.now(timezone.utc)

    history_entry = ParameterChangeHistory(
        timestamp=datetime.now(timezone.utc),
        parameters=opt_result.parameter_dict.copy(),
        trigger_reason="variance_threshold" if current_variance > variance_threshold else "manual",
        objective_value=float(opt_result.fun),
        variance=float(current_variance),
        success=bool(opt_result.success),
    )
    self.state.parameter_history.append(history_entry)

    return CalibrationResult(
        success=opt_result.success,
        parameters=opt_result.parameter_dict,
        initial_parameters=current_parameters,
        objective_value=float(opt_result.fun),
        n_iterations=int(opt_result.nit),
        execution_time=time.time() - start_time,
        method=method,
        message=str(getattr(opt_result, "message", "Online calibration completed")),
        validation_metrics=validation_metrics,
    )

`should_recalibrate(recent_measurements, objectives=None)` ¶

Check if re-calibration should be triggered based on prediction error.

Parameters:

Name	Type	Description	Default
`recent_measurements`	`MeasurementData`	Recent plant data.	required
`objectives`	`Optional[List[str]]`	Variables to monitor for error.	`None`

Returns:

Type	Description
`tuple[bool, str]`	Tuple[bool, str]: (should_recalibrate, reason_string).

Source code in pyadm1ode_calibration/calibration/methods/online.py

def should_recalibrate(
    self, recent_measurements: MeasurementData, objectives: list[str] | None = None
) -> tuple[bool, str]:
    """
    Check if re-calibration should be triggered based on prediction error.

    Args:
        recent_measurements (MeasurementData): Recent plant data.
        objectives (Optional[List[str]]): Variables to monitor for error.

    Returns:
        Tuple[bool, str]: (should_recalibrate, reason_string).
    """
    if not self.trigger.enabled:
        return False, "Disabled"

    if objectives is None:
        objectives = ["Q_ch4", "pH"]

    if self.state.last_calibration_time:
        hours = (datetime.now(timezone.utc) - self.state.last_calibration_time).total_seconds() / 3600
        if hours < self.trigger.time_threshold:
            return False, f"Too soon since last calibration ({hours:.1f}h)"

    variance = self._calculate_prediction_variance(recent_measurements, self._get_current_parameters(), objectives)
    self.state.current_variance = variance

    if variance > self.trigger.variance_threshold:
        self.state.consecutive_violations += 1
        if self.state.consecutive_violations >= self.trigger.consecutive_violations:
            return True, f"Variance {variance:.4f} > {self.trigger.variance_threshold}"
    else:
        self.state.consecutive_violations = 0

    return False, "Prediction within accuracy threshold"

Calibration Result¶

`pyadm1ode_calibration.calibration.CalibrationResult(success, parameters, initial_parameters, objective_value, n_iterations, execution_time, method, message, validation_metrics=dict(), sensitivity=dict(), history=list(), timestamp=(lambda: datetime.now(timezone.utc).isoformat())())` `dataclass` ¶

Result from a calibration run.

Stores the output of an optimization process, including calibrated values, performance metrics, and diagnostic information.

Attributes:

Name	Type	Description
`success`	`bool`	Whether calibration converged successfully.
`parameters`	`Dict[str, float]`	Calibrated parameter values.
`initial_parameters`	`Dict[str, float]`	Initial parameter values before calibration.
`objective_value`	`float`	Final objective function value.
`n_iterations`	`int`	Number of optimization iterations.
`execution_time`	`float`	Wall clock time in seconds.
`method`	`str`	Optimization method used (e.g., 'differential_evolution').
`message`	`str`	Status message from the optimizer.
`validation_metrics`	`Dict[str, float]`	Metrics on validation data (RMSE, R2, etc.).
`sensitivity`	`Dict[str, Any]`	Parameter sensitivity analysis results.
`history`	`List[Dict[str, Any]]`	Optimization history if tracking was enabled.
`timestamp`	`str`	Calibration timestamp in ISO format.

Methods:¶

`from_dict(data)` `classmethod` ¶

Create a CalibrationResult instance from a dictionary.

Parameters:

Name	Type	Description	Default
`data`	`Dict[str, Any]`	Dictionary with result attributes.	required

Returns:

Name	Type	Description
`CalibrationResult`	`CalibrationResult`	A new instance populated with the data.

Source code in pyadm1ode_calibration/calibration/core/result.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "CalibrationResult":
    """
    Create a CalibrationResult instance from a dictionary.

    Args:
        data (Dict[str, Any]): Dictionary with result attributes.

    Returns:
        CalibrationResult: A new instance populated with the data.
    """
    return cls(**data)

`to_dict()` ¶

Convert the result to a dictionary.

Returns:

Type	Description
`dict[str, Any]`	Dict[str, Any]: Dictionary containing all result data.

Source code in pyadm1ode_calibration/calibration/core/result.py

def to_dict(self) -> dict[str, Any]:
    """
    Convert the result to a dictionary.

    Returns:
        Dict[str, Any]: Dictionary containing all result data.
    """
    return {
        "success": self.success,
        "parameters": self.parameters,
        "initial_parameters": self.initial_parameters,
        "objective_value": self.objective_value,
        "n_iterations": self.n_iterations,
        "execution_time": self.execution_time,
        "method": self.method,
        "message": self.message,
        "validation_metrics": self.validation_metrics,
        "sensitivity": self.sensitivity,
        "history": self.history,
        "timestamp": self.timestamp,
    }

`to_json(filepath)` ¶

Save the result to a JSON file.

Parameters:

Name	Type	Description	Default
`filepath`	`str`	Destination file path.	required

Source code in pyadm1ode_calibration/calibration/core/result.py

def to_json(self, filepath: str) -> None:
    """
    Save the result to a JSON file.

    Args:
        filepath (str): Destination file path.
    """
    with open(filepath, "w") as f:
        json.dump(self.to_dict(), f, indent=2)

Parameter Bounds¶

`pyadm1ode_calibration.calibration.parameter_bounds.ParameterBounds()` ¶

Manager for parameter bounds in ADM1 calibration.

Provides methods for accessing bounds, validating parameters, and calculating penalties for constraint violations.

Attributes:

Name	Type	Description
`bounds`	`dict[str, ParameterBound]`	Dictionary mapping parameter names to ParameterBound objects

Example

bounds = ParameterBounds() bounds.add_bound("k_dis", lower=0.3, upper=0.8, default=0.5) is_valid = bounds.is_within_bounds("k_dis", 0.6)

Initialize empty parameter bounds manager.

Source code in pyadm1ode_calibration/calibration/parameter_bounds.py

def __init__(self):
    """Initialize empty parameter bounds manager."""
    self.bounds: dict[str, ParameterBound] = {}

Methods:¶

`add_bound(name, lower, upper, default, bound_type=BoundType.HARD, penalty_weight=1.0, description='', unit='', substrate_dependent=False)` ¶

Add parameter bound.

Parameters:

Name	Type	Description	Default
`name`	`str`	Parameter name	required
`lower`	`float`	Lower bound	required
`upper`	`float`	Upper bound	required
`default`	`float`	Default value	required
`bound_type`	`BoundType`	Type of bound	`HARD`
`penalty_weight`	`float`	Penalty weight for soft constraints	`1.0`
`description`	`str`	Parameter description	`''`
`unit`	`str`	Parameter unit	`''`
`substrate_dependent`	`bool`	Whether bound varies by substrate	`False`

Source code in pyadm1ode_calibration/calibration/parameter_bounds.py

def add_bound(
    self,
    name: str,
    lower: float,
    upper: float,
    default: float,
    bound_type: BoundType = BoundType.HARD,
    penalty_weight: float = 1.0,
    description: str = "",
    unit: str = "",
    substrate_dependent: bool = False,
) -> None:
    """
    Add parameter bound.

    Args:
        name: Parameter name
        lower: Lower bound
        upper: Upper bound
        default: Default value
        bound_type: Type of bound
        penalty_weight: Penalty weight for soft constraints
        description: Parameter description
        unit: Parameter unit
        substrate_dependent: Whether bound varies by substrate
    """
    self.bounds[name] = ParameterBound(
        name=name,
        lower=lower,
        upper=upper,
        default=default,
        bound_type=bound_type,
        penalty_weight=penalty_weight,
        description=description,
        unit=unit,
        substrate_dependent=substrate_dependent,
    )

`calculate_penalty(name, value, penalty_type='quadratic')` ¶

Calculate penalty for parameter value.

Parameters:

Name	Type	Description	Default
`name`	`str`	Parameter name	required
`value`	`float`	Parameter value	required
`penalty_type`	`str`	Type of penalty function	`'quadratic'`

Returns:

Type	Description
`float`	Penalty value

Source code in pyadm1ode_calibration/calibration/parameter_bounds.py

def calculate_penalty(self, name: str, value: float, penalty_type: str = "quadratic") -> float:
    """
    Calculate penalty for parameter value.

    Args:
        name: Parameter name
        value: Parameter value
        penalty_type: Type of penalty function

    Returns:
        Penalty value
    """
    bound = self.get_bounds(name)
    if bound is None:
        return 0.0
    return bound.calculate_penalty(value, penalty_type)

`calculate_total_penalty(parameters, penalty_type='quadratic')` ¶

Calculate total penalty for all parameters.

Parameters:

Name	Type	Description	Default
`parameters`	`dict[str, float]`	Dictionary of parameter values	required
`penalty_type`	`str`	Type of penalty function	`'quadratic'`

Returns:

Type	Description
`float`	Total penalty

Source code in pyadm1ode_calibration/calibration/parameter_bounds.py

def calculate_total_penalty(self, parameters: dict[str, float], penalty_type: str = "quadratic") -> float:
    """
    Calculate total penalty for all parameters.

    Args:
        parameters: Dictionary of parameter values
        penalty_type: Type of penalty function

    Returns:
        Total penalty
    """
    total_penalty = 0.0
    for name, value in parameters.items():
        penalty = self.calculate_penalty(name, value, penalty_type)
        if np.isinf(penalty):
            return np.inf
        total_penalty += penalty
    return total_penalty

`clip_to_bounds(name, value)` ¶

Clip parameter value to bounds.

Parameters:

Name	Type	Description	Default
`name`	`str`	Parameter name	required
`value`	`float`	Parameter value	required

Returns:

Type	Description
`float`	Clipped value

Source code in pyadm1ode_calibration/calibration/parameter_bounds.py

def clip_to_bounds(self, name: str, value: float) -> float:
    """
    Clip parameter value to bounds.

    Args:
        name: Parameter name
        value: Parameter value

    Returns:
        Clipped value
    """
    bound = self.get_bounds(name)
    if bound is None:
        return value
    return bound.clip_to_bounds(value)

`get_bounds(name)` ¶

Get bounds for parameter.

Parameters:

Name	Type	Description	Default
`name`	`str`	Parameter name	required

Returns:

Type	Description
`ParameterBound \| None`	ParameterBound object or None if not found

Source code in pyadm1ode_calibration/calibration/parameter_bounds.py

def get_bounds(self, name: str) -> ParameterBound | None:
    """
    Get bounds for parameter.

    Args:
        name: Parameter name

    Returns:
        ParameterBound object or None if not found
    """
    return self.bounds.get(name)

`get_bounds_tuple(name)` ¶

Get bounds as tuple (lower, upper).

Parameters:

Name	Type	Description	Default
`name`	`str`	Parameter name	required

Returns:

Type	Description
`tuple[float, float] \| None`	Tuple of (lower, upper) or None

Source code in pyadm1ode_calibration/calibration/parameter_bounds.py

def get_bounds_tuple(self, name: str) -> tuple[float, float] | None:
    """
    Get bounds as tuple (lower, upper).

    Args:
        name: Parameter name

    Returns:
        Tuple of (lower, upper) or None
    """
    bound = self.get_bounds(name)
    if bound is None:
        return None
    return (bound.lower, bound.upper)

`get_default_values(parameter_names)` ¶

Get default values for parameters.

Parameters:

Name	Type	Description	Default
`parameter_names`	`list[str]`	List of parameter names	required

Returns:

Type	Description
`dict[str, float]`	Dictionary of default values

Source code in pyadm1ode_calibration/calibration/parameter_bounds.py

def get_default_values(self, parameter_names: list[str]) -> dict[str, float]:
    """
    Get default values for parameters.

    Args:
        parameter_names: List of parameter names

    Returns:
        Dictionary of default values
    """
    defaults = {}
    for name in parameter_names:
        bound = self.get_bounds(name)
        if bound is not None:
            defaults[name] = bound.default
    return defaults

`is_within_bounds(name, value, tolerance=0.0)` ¶

Check if parameter value is within bounds.

Parameters:

Name	Type	Description	Default
`name`	`str`	Parameter name	required
`value`	`float`	Parameter value	required
`tolerance`	`float`	Tolerance for boundary	`0.0`

Returns:

Type	Description
`bool`	True if within bounds, False otherwise

Source code in pyadm1ode_calibration/calibration/parameter_bounds.py

def is_within_bounds(self, name: str, value: float, tolerance: float = 0.0) -> bool:
    """
    Check if parameter value is within bounds.

    Args:
        name: Parameter name
        value: Parameter value
        tolerance: Tolerance for boundary

    Returns:
        True if within bounds, False otherwise
    """
    bound = self.get_bounds(name)
    if bound is None:
        return True  # No bounds defined, assume valid
    return bound.is_within_bounds(value, tolerance)

`scale_to_unit_interval(name, value)` ¶

Scale parameter value to unit interval [0, 1].

Useful for optimization algorithms that work better with normalized parameters.

Parameters:

Name	Type	Description	Default
`name`	`str`	Parameter name	required
`value`	`float`	Parameter value	required

Returns:

Type	Description
`float`	Scaled value in [0, 1]

Source code in pyadm1ode_calibration/calibration/parameter_bounds.py

def scale_to_unit_interval(self, name: str, value: float) -> float:
    """
    Scale parameter value to unit interval [0, 1].

    Useful for optimization algorithms that work better with
    normalized parameters.

    Args:
        name: Parameter name
        value: Parameter value

    Returns:
        Scaled value in [0, 1]
    """
    bound = self.get_bounds(name)
    if bound is None:
        return value

    if bound.upper == bound.lower:
        return 0.5

    return (value - bound.lower) / (bound.upper - bound.lower)

`unscale_from_unit_interval(name, scaled_value)` ¶

Unscale parameter value from unit interval [0, 1].

Parameters:

Name	Type	Description	Default
`name`	`str`	Parameter name	required
`scaled_value`	`float`	Scaled value in [0, 1]	required

Returns:

Type	Description
`float`	Unscaled parameter value

Source code in pyadm1ode_calibration/calibration/parameter_bounds.py

def unscale_from_unit_interval(self, name: str, scaled_value: float) -> float:
    """
    Unscale parameter value from unit interval [0, 1].

    Args:
        name: Parameter name
        scaled_value: Scaled value in [0, 1]

    Returns:
        Unscaled parameter value
    """
    bound = self.get_bounds(name)
    if bound is None:
        return scaled_value

    return bound.lower + scaled_value * (bound.upper - bound.lower)

`validate_parameters(parameters, raise_on_invalid=False)` ¶

Validate all parameters against bounds.

Parameters:

Name	Type	Description	Default
`parameters`	`dict[str, float]`	Dictionary of parameter values	required
`raise_on_invalid`	`bool`	Raise exception if invalid	`False`

Returns:

Type	Description
`tuple[bool, list[str]]`	Tuple of (all_valid, list of error messages)

Source code in pyadm1ode_calibration/calibration/parameter_bounds.py

def validate_parameters(self, parameters: dict[str, float], raise_on_invalid: bool = False) -> tuple[bool, list[str]]:
    """
    Validate all parameters against bounds.

    Args:
        parameters: Dictionary of parameter values
        raise_on_invalid: Raise exception if invalid

    Returns:
        Tuple of (all_valid, list of error messages)
    """
    errors = []

    for name, value in parameters.items():
        bound = self.get_bounds(name)
        if bound is None:
            continue

        if not bound.is_within_bounds(value):
            error = f"Parameter '{name}' = {value:.4f} is outside bounds [{bound.lower:.4f}, {bound.upper:.4f}]"
            errors.append(error)

    if errors and raise_on_invalid:
        raise ValueError("\n".join(errors))

    return (len(errors) == 0, errors)

`pyadm1ode_calibration.calibration.parameter_bounds.create_default_bounds()` ¶

Create parameter bounds with default ADM1 values.

Bounds are based on: - Batstone et al. (2002): ADM1 base values - Rosen et al. (2006): BSM2 values - Gaida (2014): Agricultural substrate calibrations - Literature ranges for agricultural biogas plants

Returns:

Type	Description
`ParameterBounds`	ParameterBounds with default ADM1 parameter bounds

Example

bounds = create_default_bounds() k_dis_bounds = bounds.get_bounds("k_dis")

Source code in pyadm1ode_calibration/calibration/parameter_bounds.py

def create_default_bounds() -> ParameterBounds:
    """
    Create parameter bounds with default ADM1 values.

    Bounds are based on:
    - Batstone et al. (2002): ADM1 base values
    - Rosen et al. (2006): BSM2 values
    - Gaida (2014): Agricultural substrate calibrations
    - Literature ranges for agricultural biogas plants

    Returns:
        ParameterBounds with default ADM1 parameter bounds

    Example:
        >>> bounds = create_default_bounds()
        >>> k_dis_bounds = bounds.get_bounds("k_dis")
    """
    bounds = ParameterBounds()

    # Disintegration rate [1/d] - substrate dependent
    bounds.add_bound(
        "k_dis",
        lower=0.1,
        upper=1.0,
        default=0.5,
        bound_type=BoundType.SOFT,
        penalty_weight=2.0,
        description="Disintegration rate constant",
        unit="1/d",
        substrate_dependent=True,
    )

    # Hydrolysis rates [1/d] - substrate dependent
    bounds.add_bound(
        "k_hyd_ch",
        lower=1.0,
        upper=15.0,
        default=4.0,
        bound_type=BoundType.SOFT,
        penalty_weight=1.0,
        description="Hydrolysis rate for carbohydrates",
        unit="1/d",
        substrate_dependent=True,
    )

    bounds.add_bound(
        "k_hyd_pr",
        lower=1.0,
        upper=15.0,
        default=4.0,
        bound_type=BoundType.SOFT,
        penalty_weight=1.0,
        description="Hydrolysis rate for proteins",
        unit="1/d",
        substrate_dependent=True,
    )

    bounds.add_bound(
        "k_hyd_li",
        lower=1.0,
        upper=15.0,
        default=4.0,
        bound_type=BoundType.SOFT,
        penalty_weight=1.0,
        description="Hydrolysis rate for lipids",
        unit="1/d",
        substrate_dependent=True,
    )

    # Yield coefficients [kg COD/kg COD]
    bounds.add_bound(
        "Y_su",
        lower=0.05,
        upper=0.15,
        default=0.10,
        bound_type=BoundType.HARD,
        description="Yield of sugar degraders",
        unit="kg COD/kg COD",
    )

    bounds.add_bound(
        "Y_aa",
        lower=0.04,
        upper=0.12,
        default=0.08,
        bound_type=BoundType.HARD,
        description="Yield of amino acid degraders",
        unit="kg COD/kg COD",
    )

    bounds.add_bound(
        "Y_fa",
        lower=0.03,
        upper=0.10,
        default=0.06,
        bound_type=BoundType.HARD,
        description="Yield of LCFA degraders",
        unit="kg COD/kg COD",
    )

    bounds.add_bound(
        "Y_c4",
        lower=0.03,
        upper=0.10,
        default=0.06,
        bound_type=BoundType.HARD,
        description="Yield of valerate and butyrate degraders",
        unit="kg COD/kg COD",
    )

    bounds.add_bound(
        "Y_pro",
        lower=0.02,
        upper=0.08,
        default=0.04,
        bound_type=BoundType.HARD,
        description="Yield of propionate degraders",
        unit="kg COD/kg COD",
    )

    bounds.add_bound(
        "Y_ac",
        lower=0.03,
        upper=0.08,
        default=0.05,
        bound_type=BoundType.HARD,
        description="Yield of acetate degraders",
        unit="kg COD/kg COD",
    )

    bounds.add_bound(
        "Y_h2",
        lower=0.03,
        upper=0.10,
        default=0.06,
        bound_type=BoundType.HARD,
        description="Yield of hydrogen degraders",
        unit="kg COD/kg COD",
    )

    # Maximum uptake rates [1/d] - substrate dependent
    bounds.add_bound(
        "k_m_su",
        lower=20.0,
        upper=40.0,
        default=30.0,
        bound_type=BoundType.SOFT,
        penalty_weight=1.0,
        description="Maximum uptake rate for sugars",
        unit="1/d",
    )

    bounds.add_bound(
        "k_m_aa",
        lower=35.0,
        upper=65.0,
        default=50.0,
        bound_type=BoundType.SOFT,
        penalty_weight=1.0,
        description="Maximum uptake rate for amino acids",
        unit="1/d",
    )

    bounds.add_bound(
        "k_m_fa",
        lower=3.0,
        upper=10.0,
        default=6.0,
        bound_type=BoundType.SOFT,
        penalty_weight=1.0,
        description="Maximum uptake rate for LCFA",
        unit="1/d",
    )

    bounds.add_bound(
        "k_m_c4",
        lower=15.0,
        upper=30.0,
        default=20.0,
        bound_type=BoundType.SOFT,
        penalty_weight=1.5,
        description="Maximum uptake rate for valerate and butyrate",
        unit="1/d",
        substrate_dependent=True,
    )

    bounds.add_bound(
        "k_m_pro",
        lower=8.0,
        upper=18.0,
        default=13.0,
        bound_type=BoundType.SOFT,
        penalty_weight=1.5,
        description="Maximum uptake rate for propionate",
        unit="1/d",
        substrate_dependent=True,
    )

    bounds.add_bound(
        "k_m_ac",
        lower=4.0,
        upper=12.0,
        default=8.0,
        bound_type=BoundType.SOFT,
        penalty_weight=2.0,
        description="Maximum uptake rate for acetate",
        unit="1/d",
        substrate_dependent=True,
    )

    bounds.add_bound(
        "k_m_h2",
        lower=25.0,
        upper=45.0,
        default=35.0,
        bound_type=BoundType.SOFT,
        penalty_weight=1.5,
        description="Maximum uptake rate for hydrogen",
        unit="1/d",
        substrate_dependent=True,
    )

    # Half-saturation constants [kg COD/m³]
    bounds.add_bound(
        "K_S_su",
        lower=0.3,
        upper=0.7,
        default=0.5,
        bound_type=BoundType.SOFT,
        penalty_weight=1.0,
        description="Half-saturation constant for sugars",
        unit="kg COD/m³",
    )

    bounds.add_bound(
        "K_S_aa",
        lower=0.2,
        upper=0.5,
        default=0.3,
        bound_type=BoundType.SOFT,
        penalty_weight=1.0,
        description="Half-saturation constant for amino acids",
        unit="kg COD/m³",
    )

    bounds.add_bound(
        "K_S_fa",
        lower=0.3,
        upper=0.6,
        default=0.4,
        bound_type=BoundType.SOFT,
        penalty_weight=1.0,
        description="Half-saturation constant for LCFA",
        unit="kg COD/m³",
    )

    bounds.add_bound(
        "K_S_c4",
        lower=0.15,
        upper=0.4,
        default=0.2,
        bound_type=BoundType.SOFT,
        penalty_weight=1.0,
        description="Half-saturation constant for C4",
        unit="kg COD/m³",
    )

    bounds.add_bound(
        "K_S_pro",
        lower=0.05,
        upper=0.15,
        default=0.1,
        bound_type=BoundType.SOFT,
        penalty_weight=1.0,
        description="Half-saturation constant for propionate",
        unit="kg COD/m³",
    )

    bounds.add_bound(
        "K_S_ac",
        lower=0.1,
        upper=0.25,
        default=0.15,
        bound_type=BoundType.SOFT,
        penalty_weight=1.0,
        description="Half-saturation constant for acetate",
        unit="kg COD/m³",
    )

    bounds.add_bound(
        "K_S_h2",
        lower=5e-6,
        upper=1e-5,
        default=7e-6,
        bound_type=BoundType.SOFT,
        penalty_weight=1.0,
        description="Half-saturation constant for hydrogen",
        unit="kg COD/m³",
    )

    # Decay rates [1/d]
    bounds.add_bound(
        "k_dec_X_su",
        lower=0.01,
        upper=0.04,
        default=0.02,
        bound_type=BoundType.SOFT,
        penalty_weight=0.5,
        description="Decay rate for sugar degraders",
        unit="1/d",
    )

    bounds.add_bound(
        "k_dec_X_aa",
        lower=0.01,
        upper=0.04,
        default=0.02,
        bound_type=BoundType.SOFT,
        penalty_weight=0.5,
        description="Decay rate for amino acid degraders",
        unit="1/d",
    )

    bounds.add_bound(
        "k_dec_X_fa",
        lower=0.01,
        upper=0.04,
        default=0.02,
        bound_type=BoundType.SOFT,
        penalty_weight=0.5,
        description="Decay rate for LCFA degraders",
        unit="1/d",
    )

    bounds.add_bound(
        "k_dec_X_c4",
        lower=0.01,
        upper=0.04,
        default=0.02,
        bound_type=BoundType.SOFT,
        penalty_weight=0.5,
        description="Decay rate for C4 degraders",
        unit="1/d",
    )

    bounds.add_bound(
        "k_dec_X_pro",
        lower=0.01,
        upper=0.04,
        default=0.02,
        bound_type=BoundType.SOFT,
        penalty_weight=0.5,
        description="Decay rate for propionate degraders",
        unit="1/d",
    )

    bounds.add_bound(
        "k_dec_X_ac",
        lower=0.01,
        upper=0.04,
        default=0.02,
        bound_type=BoundType.SOFT,
        penalty_weight=0.5,
        description="Decay rate for acetate degraders",
        unit="1/d",
    )

    bounds.add_bound(
        "k_dec_X_h2",
        lower=0.01,
        upper=0.04,
        default=0.02,
        bound_type=BoundType.SOFT,
        penalty_weight=0.5,
        description="Decay rate for hydrogen degraders",
        unit="1/d",
    )

    # ------------------------------------------------------------------
    # ADM1da (Schlattmann 2011) parameter names — these are the names
    # pyadm1.core.adm1 actually reads from ``ADM1._kinetic`` at every
    # ODE step. The bounds above use the classical ADM1 names
    # (``k_dis``, ``k_dec_X_*``); pyadm1 ignores those, so calibrating
    # them was a no-op. The block below adds the actual ADM1da names
    # with defaults matching ``ADMParams.get_kinetic_params()``.
    # ------------------------------------------------------------------

    # Two-pool disintegration: slow (PS, particulate substrate) and
    # fast (PF, particulate feed) pools.
    bounds.add_bound(
        "k_dis_PS",
        lower=0.01,
        upper=0.20,
        default=0.04,
        bound_type=BoundType.SOFT,
        penalty_weight=2.0,
        description="Disintegration rate, slow particulate pool (ADM1da)",
        unit="1/d",
        substrate_dependent=True,
    )

    bounds.add_bound(
        "k_dis_PF",
        lower=0.1,
        upper=2.0,
        default=0.4,
        bound_type=BoundType.SOFT,
        penalty_weight=2.0,
        description="Disintegration rate, fast particulate pool (ADM1da)",
        unit="1/d",
        substrate_dependent=True,
    )

    # Per-organism decay rates [1/d] — pyadm1 reads ``k_dec_<group>``
    # without the ``X_`` prefix used in the classical-ADM1 bounds above.
    for group in ("su", "aa", "fa", "c4", "pro", "h2"):
        bounds.add_bound(
            f"k_dec_{group}",
            lower=0.01,
            upper=0.04,
            default=0.02,
            bound_type=BoundType.SOFT,
            penalty_weight=0.5,
            description=f"Decay rate for {group} degraders (ADM1da)",
            unit="1/d",
        )
    # Acetate decay is doubled in ADM1da (Schlattmann 2011): 0.04 /d.
    bounds.add_bound(
        "k_dec_ac",
        lower=0.02,
        upper=0.08,
        default=0.04,
        bound_type=BoundType.SOFT,
        penalty_weight=0.5,
        description="Decay rate for acetate degraders (ADM1da: 0.04 /d)",
        unit="1/d",
    )

    return bounds

Simulator¶

`pyadm1ode_calibration.calibration.core.simulator.PlantSimulator(plant, verbose=True, time_varying_feed=False)` ¶

Handles plant simulation with parameter variations.

This class separates the simulation logic from calibration algorithms, providing a consistent interface for running the ADM1 model with modified parameter sets.

Parameters:

Name	Type	Description	Default
`plant`	`Any`	The PyADM1ODE plant model instance.	required
`verbose`	`bool`	Whether to enable progress output and logging. Defaults to True.	`True`

Source code in pyadm1ode_calibration/calibration/core/simulator.py

def __init__(self, plant: Any, verbose: bool = True, time_varying_feed: bool = False):
    self.plant = plant
    self.verbose = verbose
    self.time_varying_feed = time_varying_feed
    self._original_params: dict[str, dict[str, float]] = {}
    self._original_state: dict[str, dict[str, Any]] = {}

Methods:¶

`simulate_with_parameters(parameters, measurements, restore_params=True, chp_load_setpoints=None, restore_state=True, warmup=None)` ¶

Run a plant simulation using a specific set of parameters.

Parameters:

Name	Type	Description	Default
`parameters`	`dict[str, float]`	Parameter values to apply {name: value}.	required
`measurements`	`MeasurementData`	Input data for substrate feeds and timing.	required
`restore_params`	`bool`	Whether to restore the original plant parameters after the simulation finishes. Defaults to True.	`True`
`restore_state`	`bool`	Whether to rewind the plant to the state it had before the run. Defaults to True, and calibration depends on it: a plant keeps integrating where the last simulation stopped, so without the rewind the second evaluation of a parameter set starts from a different digester state than the first and returns a different error. The optimizer then sees a landscape that changes under its feet.	`True`
`warmup`	`Optional[MeasurementData]`	Optional window to run before the scored one, keeping its end state. Use it when `measurements` is a later stretch of a record: a digester that has been running for weeks is not in the state the model starts from, so scoring such a window cold measures the start-up transient instead of the parameters. The warm-up itself is not returned, and the plant is still rewound afterwards when `restore_state` is set.	`None`
`chp_load_setpoints`	`dict[str, ndarray] \| None`	Optional mapping from CHP component ID to an array of load setpoints (one value per simulation step, in [0, 1]). When provided, the simulator drives each CHP from this time series instead of letting the plant engine default to full load. Useful for replaying measured CHP operation so the comparison isolates the gas-production model.	`None`

Returns:

Type	Description
`dict[str, ndarray]`	Dict[str, np.ndarray]: Dictionary mapping output names (e.g., 'Q_ch4') to simulated numpy arrays.

Source code in pyadm1ode_calibration/calibration/core/simulator.py

def simulate_with_parameters(
    self,
    parameters: dict[str, float],
    measurements: "MeasurementData",
    restore_params: bool = True,
    chp_load_setpoints: dict[str, np.ndarray] | None = None,
    restore_state: bool = True,
    warmup: Optional["MeasurementData"] = None,
) -> dict[str, np.ndarray]:
    """
    Run a plant simulation using a specific set of parameters.

    Args:
        parameters: Parameter values to apply {name: value}.
        measurements: Input data for substrate feeds and timing.
        restore_params: Whether to restore the original plant parameters
            after the simulation finishes. Defaults to True.
        restore_state: Whether to rewind the plant to the state it had
            before the run. Defaults to True, and calibration depends on
            it: a plant keeps integrating where the last simulation
            stopped, so without the rewind the second evaluation of a
            parameter set starts from a different digester state than the
            first and returns a different error. The optimizer then sees a
            landscape that changes under its feet.
        warmup: Optional window to run *before* the scored one, keeping its
            end state. Use it when ``measurements`` is a later stretch of a
            record: a digester that has been running for weeks is not in the
            state the model starts from, so scoring such a window cold
            measures the start-up transient instead of the parameters. The
            warm-up itself is not returned, and the plant is still rewound
            afterwards when ``restore_state`` is set.
        chp_load_setpoints: Optional mapping from CHP component ID to an
            array of load setpoints (one value per simulation step, in
            [0, 1]). When provided, the simulator drives each CHP from
            this time series instead of letting the plant engine default
            to full load. Useful for replaying measured CHP operation
            so the comparison isolates the gas-production model.

    Returns:
        Dict[str, np.ndarray]: Dictionary mapping output names (e.g., 'Q_ch4')
            to simulated numpy arrays.
    """
    if restore_params:
        self._backup_parameters()
    if restore_state:
        self._backup_state()

    try:
        self._apply_parameters(parameters)

        if warmup is not None and len(warmup) > 0:
            self._run_window(warmup, chp_load_setpoints=None)

        results = self._run_window(measurements, chp_load_setpoints)
        return self._extract_outputs_from_results(results)
    finally:
        if restore_params:
            self._restore_parameters()
        if restore_state:
            self._restore_state()

Analysis¶

`pyadm1ode_calibration.calibration.analysis.sensitivity.SensitivityAnalyzer(plant, simulator=None, verbose=True)` ¶

Performs local sensitivity analysis on ADM1 plant parameters.

Calculates how changes in input parameters affect specific model outputs, helping to identify which parameters are most influential.

Parameters:

Name	Type	Description	Default
`plant`		The PyADM1ODE plant model instance.	required
`simulator`	`Optional[PlantSimulator]`	Simulator instance to use.	`None`
`verbose`	`bool`	Whether to enable progress output. Defaults to True.	`True`

Source code in pyadm1ode_calibration/calibration/analysis/sensitivity.py

def __init__(self, plant, simulator: PlantSimulator | None = None, verbose: bool = True):
    self.plant = plant
    self.simulator = simulator or PlantSimulator(plant, verbose)
    self.verbose = verbose

Methods:¶

`analyze(parameters, measurements, objectives=None, perturbation=0.01)` ¶

Perform local sensitivity analysis using finite differences.

Parameters:

Name	Type	Description	Default
`parameters`	`Dict[str, float]`	Baseline parameter set.	required
`measurements`	`Any`	Input data for simulation.	required
`objectives`	`Optional[List[str]]`	List of objective variables to analyze.	`None`
`perturbation`	`float`	Relative step size for finite differences. Defaults to 0.01.	`0.01`

Returns:

Type	Description
`dict[str, SensitivityResult]`	Dict[str, SensitivityResult]: Mapping of parameter names to their respective sensitivity analysis results.

Source code in pyadm1ode_calibration/calibration/analysis/sensitivity.py

def analyze(
    self,
    parameters: dict[str, float],
    measurements: Any,
    objectives: list[str] | None = None,
    perturbation: float = 0.01,
) -> dict[str, SensitivityResult]:
    """
    Perform local sensitivity analysis using finite differences.

    Args:
        parameters (Dict[str, float]): Baseline parameter set.
        measurements (Any): Input data for simulation.
        objectives (Optional[List[str]]): List of objective variables to analyze.
        perturbation (float): Relative step size for finite differences. Defaults to 0.01.

    Returns:
        Dict[str, SensitivityResult]: Mapping of parameter names to their
            respective sensitivity analysis results.
    """
    if objectives is None:
        objectives = ["Q_ch4"]

    if self.verbose:
        print(f"\nSensitivity analysis for {len(parameters)} parameters")
        print(f"Objectives: {objectives}")
        print(f"Perturbation: {perturbation * 100:.1f}%")

    results = {}

    # The base case does not depend on which parameter is being perturbed, so run
    # it once instead of once per parameter. 2N+1 simulations rather than 3N.
    outputs_base = self.simulator.simulate_with_parameters(parameters, measurements)

    for param_name, base_value in parameters.items():
        delta = base_value * perturbation if base_value != 0 else perturbation

        param_plus = parameters.copy()
        param_plus[param_name] = base_value + delta

        param_minus = parameters.copy()
        param_minus[param_name] = base_value - delta

        outputs_plus = self.simulator.simulate_with_parameters(param_plus, measurements)
        outputs_minus = self.simulator.simulate_with_parameters(param_minus, measurements)

        local_gradient, sensitivity_indices, normalized_sensitivity = {}, {}, {}
        column_parts: list[np.ndarray] = []

        for obj in objectives:
            if all(obj in out for out in [outputs_base, outputs_plus, outputs_minus]):
                series_base = np.asarray(outputs_base[obj], dtype=float)
                series_plus = np.asarray(outputs_plus[obj], dtype=float)
                series_minus = np.asarray(outputs_minus[obj], dtype=float)

                # A channel the simulation did not produce carries no information.
                # Averaging it would yield NaN and poison every index built on it.
                if min(series_base.size, series_plus.size, series_minus.size) == 0:
                    continue

                base_val = np.mean(series_base)
                plus_val = np.mean(series_plus)
                minus_val = np.mean(series_minus)

                gradient = (plus_val - minus_val) / (2 * delta)
                local_gradient[obj] = gradient
                sensitivity_indices[obj] = gradient * (base_value / base_val) if base_val != 0 else 0.0

                base_std = np.std(series_base)
                normalized_sensitivity[obj] = abs(gradient * delta / base_std) if base_std > 0 else 0.0

                # Per-observation sensitivity, scaled the way Brun et al. 2001 do:
                # by the parameter value so the columns are comparable across
                # parameters, and by the spread of the channel so they are
                # comparable across channels of different units. A flat channel
                # carries no information and contributes zeros rather than noise
                # blown up by a near-zero divisor.
                scale = base_std if base_std > 0 else 0.0
                if scale > 0:
                    column_parts.append((series_plus - series_minus) / (2 * delta) * base_value / scale)
                else:
                    column_parts.append(np.zeros_like(series_base))

        variance_contrib = sum(s**2 for s in sensitivity_indices.values())
        column = np.concatenate(column_parts) if column_parts else np.empty(0)

        results[param_name] = SensitivityResult(
            parameter=param_name,
            base_value=base_value,
            sensitivity_indices=sensitivity_indices,
            local_gradient=local_gradient,
            normalized_sensitivity=normalized_sensitivity,
            variance_contribution=variance_contrib,
            sensitivity_column=column,
        )

    return results

`pyadm1ode_calibration.calibration.analysis.identifiability.IdentifiabilityAnalyzer(plant, sensitivity_analyzer=None, verbose=True)` ¶

Assesses parameter identifiability for ADM1 calibration.

Analyzes whether parameters have enough influence on model outputs and whether they are correlated with each other, which can hinder accurate estimation.

Parameters:

Name	Type	Description	Default
`plant`		The PyADM1ODE plant model instance.	required
`sensitivity_analyzer`	`Optional[SensitivityAnalyzer]`	Analyzer for calculating gradients.	`None`
`verbose`	`bool`	Whether to enable verbose output. Defaults to True.	`True`

Source code in pyadm1ode_calibration/calibration/analysis/identifiability.py

def __init__(self, plant, sensitivity_analyzer: SensitivityAnalyzer | None = None, verbose: bool = True):
    self.plant = plant
    self.sensitivity_analyzer = sensitivity_analyzer or SensitivityAnalyzer(plant, verbose=verbose)
    self.verbose = verbose

Methods:¶

`analyze(parameters, measurements, optimization_history=None, confidence_level=0.95, correlation_threshold=0.8)` ¶

Assess parameter identifiability based on sensitivity and correlation.

Parameters:

Name	Type	Description	Default
`parameters`	`Dict[str, float]`	Parameter set to analyze.	required
`measurements`	`Any`	Measurement data for simulation.	required
`optimization_history`	`Optional[List[Dict[str, Any]]]`	History from optimizer.	`None`
`confidence_level`	`float`	Level for confidence intervals. Defaults to 0.95.	`0.95`
`correlation_threshold`	`float`	Threshold for identifying high correlations.	`0.8`

Returns:

Type	Description
`dict[str, IdentifiabilityResult]`	Dict[str, IdentifiabilityResult]: Results for each parameter. For whether
`dict[str, IdentifiabilityResult]`	the parameters can be estimated together, see :meth:`analyze_subset`.

Source code in pyadm1ode_calibration/calibration/analysis/identifiability.py

def analyze(
    self,
    parameters: dict[str, float],
    measurements: Any,
    optimization_history: list[dict[str, Any]] | None = None,
    confidence_level: float = 0.95,
    correlation_threshold: float = 0.8,
) -> dict[str, IdentifiabilityResult]:
    """
    Assess parameter identifiability based on sensitivity and correlation.

    Args:
        parameters (Dict[str, float]): Parameter set to analyze.
        measurements (Any): Measurement data for simulation.
        optimization_history (Optional[List[Dict[str, Any]]]): History from optimizer.
        confidence_level (float): Level for confidence intervals. Defaults to 0.95.
        correlation_threshold (float): Threshold for identifying high correlations.

    Returns:
        Dict[str, IdentifiabilityResult]: Results for each parameter. For whether
        the parameters can be estimated *together*, see :meth:`analyze_subset`.
    """
    sensitivity = self.sensitivity_analyzer.analyze(parameters, measurements, objectives=["Q_ch4", "pH", "VFA"])
    return self._verdicts(parameters, sensitivity, correlation_threshold)

`analyze_subset(parameters, measurements, objectives=None)` ¶

Can this set of parameters be estimated together?

Individual sensitivity is necessary but not sufficient. Two parameters that each move the outputs strongly are still unidentifiable as a pair when one can undo what the other did, which is what the collinearity index measures.

Costs 2 * len(parameters) + 1 simulations, so it is cheap enough to run before a calibration rather than after it.

Parameters:

Name	Type	Description	Default
`parameters`	`dict[str, float]`	The candidate set, name to nominal value.	required
`measurements`	`Any`	Data window the sensitivities are evaluated over.	required
`objectives`	`list[str] \| None`	Channels to score against. Defaults to gas, pH and VFA.	`None`

Returns:

Name	Type	Description
`SubsetIdentifiability`	`SubsetIdentifiability`	The verdict for the set as a whole.

Source code in pyadm1ode_calibration/calibration/analysis/identifiability.py

def analyze_subset(
    self,
    parameters: dict[str, float],
    measurements: Any,
    objectives: list[str] | None = None,
) -> SubsetIdentifiability:
    """Can this set of parameters be estimated together?

    Individual sensitivity is necessary but not sufficient. Two parameters that
    each move the outputs strongly are still unidentifiable as a pair when one can
    undo what the other did, which is what the collinearity index measures.

    Costs ``2 * len(parameters) + 1`` simulations, so it is cheap enough to run
    before a calibration rather than after it.

    Args:
        parameters: The candidate set, name to nominal value.
        measurements: Data window the sensitivities are evaluated over.
        objectives: Channels to score against. Defaults to gas, pH and VFA.

    Returns:
        SubsetIdentifiability: The verdict for the set as a whole.
    """
    sensitivity = self.sensitivity_analyzer.analyze(
        parameters, measurements, objectives=objectives or ["Q_ch4", "pH", "VFA"]
    )
    return self.subset_verdict(sensitivity, self._verdicts(parameters, sensitivity))

`subset_verdict(sensitivity, per_parameter)` `staticmethod` ¶

Judge a parameter set from an existing sensitivity analysis.

Two stages, in the order Brun et al. 2001 prescribe. First each parameter is screened on its own, because one that fails alone cannot be rescued by the company it keeps. The collinearity index is then computed over the survivors only. Doing it the other way round lets a parameter made of solver noise pass as an independent direction, since normalising its column to unit length hides how small it was.

Source code in pyadm1ode_calibration/calibration/analysis/identifiability.py

@staticmethod
def subset_verdict(
    sensitivity: dict[str, Any],
    per_parameter: dict[str, IdentifiabilityResult],
) -> SubsetIdentifiability:
    """Judge a parameter set from an existing sensitivity analysis.

    Two stages, in the order Brun et al. 2001 prescribe. First each parameter is
    screened on its own, because one that fails alone cannot be rescued by the
    company it keeps. The collinearity index is then computed over the survivors
    only. Doing it the other way round lets a parameter made of solver noise pass
    as an independent direction, since normalising its column to unit length hides
    how small it was.
    """
    names = list(sensitivity)
    weak = [n for n in names if n in per_parameter and not per_parameter[n].is_identifiable]
    survivors = [n for n in names if n not in weak]

    columns = {n: np.asarray(sensitivity[n].sensitivity_column, dtype=float) for n in survivors}
    sizes = {c.size for c in columns.values()}
    if len(survivors) < 2 or not sizes or 0 in sizes or len(sizes) != 1:
        gamma = 1.0 if len(survivors) == 1 else float("inf")
    else:
        gamma = collinearity_index(np.column_stack([columns[n] for n in survivors]))

    collinear = gamma > MAX_COLLINEARITY_INDEX
    parts = []
    if weak:
        parts.append(f"{weak} not identifiable on their own")
    if collinear and len(survivors) >= 2:
        parts.append(
            f"collinearity index {gamma:.1f} above {MAX_COLLINEARITY_INDEX:.0f} for {survivors}, "
            f"they compensate each other, drop one"
        )
    elif len(survivors) >= 2:
        parts.append(f"collinearity index {gamma:.1f} for {survivors}")

    identifiable = not weak and not collinear
    reason = "; ".join(parts) if parts else "Nothing to judge"
    return SubsetIdentifiability(names, gamma, identifiable, reason)

`pyadm1ode_calibration.calibration.analysis.identifiability.IdentifiabilityResult(parameter, is_identifiable, confidence_interval, correlation_with, objective_sensitivity, reason)` `dataclass` ¶

Result from parameter identifiability analysis.

Determines if a parameter can be reliably estimated from the available data.

Attributes:

Name	Type	Description
`parameter`	`str`	Name of the parameter.
`is_identifiable`	`bool`	Whether the parameter is considered identifiable.
`confidence_interval`	`Tuple[float, float]`	Estimated 95% confidence interval.
`correlation_with`	`Dict[str, float]`	Correlation coefficients with other parameters.
`objective_sensitivity`	`float`	Maximum sensitivity across all objectives.
`reason`	`str`	Explanation for the identifiability status.

`pyadm1ode_calibration.calibration.analysis.identifiability.SubsetIdentifiability(parameters, collinearity_index, is_identifiable, reason)` `dataclass` ¶

Whether a set of parameters can be estimated together.

Attributes:

Name	Type	Description
`parameters`	`List[str]`	The subset this verdict is about.
`collinearity_index`	`float`	Brun's index, see :func:`collinearity_index`.
`is_identifiable`	`bool`	Whether the index stays below :data:`MAX_COLLINEARITY_INDEX`.
`reason`	`str`	Explanation of the verdict.

`pyadm1ode_calibration.calibration.analysis.identifiability.collinearity_index(sensitivity_matrix)` ¶

Brun's collinearity index for the given sensitivity matrix.

Parameters:

Name	Type	Description	Default
`sensitivity_matrix`	`ndarray`	One column per parameter, one row per observation.	required

Returns:

Name	Type	Description
`float`	`float`	`1 / smallest singular value` of the column-normalised matrix.
	`float`	1.0 means the parameters act on the outputs in completely independent
	`float`	directions, and the value grows without bound as they become linearly
	`float`	dependent. `inf` when a column is all zeros, that is when a parameter does
	`float`	nothing at all.

Source code in pyadm1ode_calibration/calibration/analysis/identifiability.py

def collinearity_index(sensitivity_matrix: np.ndarray) -> float:
    """Brun's collinearity index for the given sensitivity matrix.

    Args:
        sensitivity_matrix: One column per parameter, one row per observation.

    Returns:
        float: ``1 / smallest singular value`` of the column-normalised matrix.
        1.0 means the parameters act on the outputs in completely independent
        directions, and the value grows without bound as they become linearly
        dependent. ``inf`` when a column is all zeros, that is when a parameter does
        nothing at all.
    """
    matrix = np.atleast_2d(np.asarray(sensitivity_matrix, dtype=float))
    if matrix.size == 0 or matrix.shape[1] == 0:
        return float("inf")

    norms = np.linalg.norm(matrix, axis=0)
    # Normalising to unit length is what makes the columns comparable, but it also
    # stretches a column of pure rounding noise into a full-blown direction, which
    # would make a parameter that does nothing look perfectly independent. Anything
    # negligible next to the strongest column is treated as the zero it is.
    if np.all(norms == 0) or np.any(norms <= norms.max() * DEAD_COLUMN_RATIO):
        return float("inf")

    smallest = np.linalg.svd(matrix / norms, compute_uv=False)[-1]
    return float("inf") if smallest <= 0 else float(1.0 / smallest)

Validation¶

`pyadm1ode_calibration.calibration.validation.CalibrationValidator(plant, verbose=True, time_varying_feed=False)` ¶

Validator for calibrated model parameters.

Provides methods to evaluate the goodness-of-fit of calibrated parameters on both training and out-of-sample validation data.

Parameters:

Name	Type	Description	Default
`plant`	`Any`	The PyADM1ODE plant model instance.	required
`verbose`	`bool`	Whether to enable verbose output. Defaults to True.	`True`

Source code in pyadm1ode_calibration/calibration/validation.py

def __init__(self, plant: Any, verbose: bool = True, time_varying_feed: bool = False):
    self.plant = plant
    self.verbose = verbose
    self.time_varying_feed = time_varying_feed

Methods:¶

`analyze_residuals(measurements, simulated, objectives=None)` ¶

Perform detailed residual analysis.

Parameters:

Name	Type	Description	Default
`measurements`	`MeasurementData`	Reference measurements.	required
`simulated`	`Dict[str, ndarray]`	Simulated outputs.	required
`objectives`	`Optional[List[str]]`	Variables to analyze.	`None`

Returns:

Type	Description
`dict[str, ResidualAnalysis]`	Dict[str, ResidualAnalysis]: Analysis results for each objective.

Source code in pyadm1ode_calibration/calibration/validation.py

def analyze_residuals(
    self,
    measurements: MeasurementData,
    simulated: dict[str, np.ndarray],
    objectives: list[str] | None = None,
) -> dict[str, ResidualAnalysis]:
    """
    Perform detailed residual analysis.

    Args:
        measurements (MeasurementData): Reference measurements.
        simulated (Dict[str, np.ndarray]): Simulated outputs.
        objectives (Optional[List[str]]): Variables to analyze.

    Returns:
        Dict[str, ResidualAnalysis]: Analysis results for each objective.
    """
    if objectives is None:
        objectives = list(simulated.keys())

    results = {}
    for objective in objectives:
        if objective not in simulated:
            continue

        observed = self._extract_measurements(measurements, objective)
        predicted = simulated[objective]

        observed, predicted = self._align_arrays(observed, predicted)

        if len(observed) < 3:
            continue

        residuals = observed - predicted
        std_residuals = self._standardize_residuals(residuals)
        normality = self._test_normality(residuals)
        autocorr = self._calculate_autocorrelation(residuals)
        hetero = self._test_heteroscedasticity(residuals, predicted)
        outliers = np.where(np.abs(std_residuals) > 3)[0].tolist()

        results[objective] = ResidualAnalysis(
            objective=objective,
            residuals=residuals,
            standardized_residuals=std_residuals,
            normality_test=normality,
            autocorrelation=autocorr,
            heteroscedasticity_test=hetero,
            outlier_indices=outliers,
        )

    return results

`cross_validate(parameters, measurements, n_folds=5, objectives=None)` ¶

Perform k-fold cross-validation.

Parameters:

Name	Type	Description	Default
`parameters`	`Dict[str, float]`	Calibrated parameters.	required
`measurements`	`MeasurementData`	Full dataset to split.	required
`n_folds`	`int`	Number of folds. Defaults to 5.	`5`
`objectives`	`Optional[List[str]]`	Variables to validate.	`None`

Returns:

Type	Description
`dict[str, list[ValidationMetrics]]`	Dict[str, List[ValidationMetrics]]: List of metrics for each fold.

Source code in pyadm1ode_calibration/calibration/validation.py

def cross_validate(
    self,
    parameters: dict[str, float],
    measurements: MeasurementData,
    n_folds: int = 5,
    objectives: list[str] | None = None,
) -> dict[str, list[ValidationMetrics]]:
    """
    Perform k-fold cross-validation.

    Args:
        parameters (Dict[str, float]): Calibrated parameters.
        measurements (MeasurementData): Full dataset to split.
        n_folds (int): Number of folds. Defaults to 5.
        objectives (Optional[List[str]]): Variables to validate.

    Returns:
        Dict[str, List[ValidationMetrics]]: List of metrics for each fold.
    """
    if objectives is None:
        objectives = ["Q_ch4", "pH", "VFA"]

    n_samples = len(measurements)
    fold_size = n_samples // n_folds
    cv_results: dict[str, list[ValidationMetrics]] = {obj: [] for obj in objectives}

    for fold in range(n_folds):
        start_idx = fold * fold_size
        end_idx = start_idx + fold_size if fold < n_folds - 1 else n_samples

        val_data = measurements.data.iloc[start_idx:end_idx].copy()
        val_measurements = type(measurements)(val_data)

        fold_metrics = self.validate(parameters, val_measurements, objectives)
        for obj, metrics in fold_metrics.items():
            cv_results[obj].append(metrics)

    return cv_results

`validate(parameters, measurements, objectives=None, simulation_duration=None, warmup=None)` ¶

Validate parameters against measurement data.

Parameters:

Name	Type	Description	Default
`parameters`	`Dict[str, float]`	Calibrated parameters to test.	required
`measurements`	`MeasurementData`	Reference measurement data.	required
`objectives`	`Optional[List[str]]`	Variables to validate.	`None`
`simulation_duration`	`Optional[float]`	Duration in days.	`None`
`warmup`	`Optional[MeasurementData]`	Window to run before the scored one, so a later stretch of a record is not judged from the plant's initial state. Pass the training window when validating on the test window that follows it; without it the score contains the start-up transient rather than the model error.	`None`

Returns:

Type	Description
`dict[str, ValidationMetrics]`	Dict[str, ValidationMetrics]: Metrics for each objective.

Source code in pyadm1ode_calibration/calibration/validation.py

def validate(
    self,
    parameters: dict[str, float],
    measurements: MeasurementData,
    objectives: list[str] | None = None,
    simulation_duration: float | None = None,
    warmup: MeasurementData | None = None,
) -> dict[str, ValidationMetrics]:
    """
    Validate parameters against measurement data.

    Args:
        parameters (Dict[str, float]): Calibrated parameters to test.
        measurements (MeasurementData): Reference measurement data.
        objectives (Optional[List[str]]): Variables to validate.
        simulation_duration (Optional[float]): Duration in days.
        warmup (Optional[MeasurementData]): Window to run before the scored
            one, so a later stretch of a record is not judged from the
            plant's initial state. Pass the training window when validating
            on the test window that follows it; without it the score
            contains the start-up transient rather than the model error.

    Returns:
        Dict[str, ValidationMetrics]: Metrics for each objective.
    """
    if objectives is None:
        objectives = ["Q_ch4", "pH", "VFA"]

    simulated_outputs = self._simulate_plant(measurements, parameters, warmup=warmup)

    metrics = {}
    for objective in objectives:
        if objective not in simulated_outputs:
            warnings.warn(f"Objective '{objective}' not in simulation outputs")
            continue

        observed = self._extract_measurements(measurements, objective)
        predicted = simulated_outputs[objective]

        observed, predicted = self._align_arrays(observed, predicted)

        if len(observed) == 0:
            warnings.warn(f"No valid data for objective '{objective}'")
            continue

        obj_metrics = self._calculate_metrics(objective, observed, predicted)
        metrics[objective] = obj_metrics

    return metrics

Kalibrierung (API)¶

Calibrator Facade¶

pyadm1ode_calibration.calibration.Calibrator(plant, verbose=True) ¶

Methods:¶

apply_calibration(result) ¶

run_initial_calibration(measurements, parameters, **kwargs) ¶

run_online_calibration(measurements, parameters, **kwargs) ¶

Initial Calibrator¶

pyadm1ode_calibration.calibration.InitialCalibrator(plant, verbose=True, time_varying_feed=False) ¶

Methods:¶

calibrate(measurements, parameters, bounds=None, method='differential_evolution', objectives=None, weights=None, validation_split=0.2, max_iterations=100, population_size=15, tolerance=0.0001, sensitivity_analysis=True, use_constraints=False, check_identifiability=False, **kwargs) ¶

identifiability_analysis(parameters, measurements, optimization_history=None, confidence_level=0.95, correlation_threshold=0.8) ¶

sensitivity_analysis(parameters, measurements, objectives=None) ¶

Online Calibrator¶

pyadm1ode_calibration.calibration.OnlineCalibrator(plant, verbose=True, parameter_bounds=None, time_varying_feed=False) ¶

Methods:¶

apply_calibration(result) ¶

calibrate(measurements, parameters=None, current_parameters=None, variance_threshold=0.15, max_parameter_change=0.2, time_window=7, method='nelder_mead', max_iterations=50, objectives=None, weights=None, use_constraints=True, **kwargs) ¶

should_recalibrate(recent_measurements, objectives=None) ¶

Calibration Result¶

pyadm1ode_calibration.calibration.CalibrationResult(success, parameters, initial_parameters, objective_value, n_iterations, execution_time, method, message, validation_metrics=dict(), sensitivity=dict(), history=list(), timestamp=(lambda: datetime.now(timezone.utc).isoformat())()) dataclass ¶

Methods:¶

from_dict(data) classmethod ¶

to_dict() ¶

to_json(filepath) ¶

Parameter Bounds¶

pyadm1ode_calibration.calibration.parameter_bounds.ParameterBounds() ¶

Methods:¶

add_bound(name, lower, upper, default, bound_type=BoundType.HARD, penalty_weight=1.0, description='', unit='', substrate_dependent=False) ¶

calculate_penalty(name, value, penalty_type='quadratic') ¶

calculate_total_penalty(parameters, penalty_type='quadratic') ¶

clip_to_bounds(name, value) ¶

get_bounds(name) ¶

get_bounds_tuple(name) ¶

get_default_values(parameter_names) ¶

is_within_bounds(name, value, tolerance=0.0) ¶

scale_to_unit_interval(name, value) ¶

unscale_from_unit_interval(name, scaled_value) ¶

validate_parameters(parameters, raise_on_invalid=False) ¶

pyadm1ode_calibration.calibration.parameter_bounds.create_default_bounds() ¶

Simulator¶

pyadm1ode_calibration.calibration.core.simulator.PlantSimulator(plant, verbose=True, time_varying_feed=False) ¶

Methods:¶

simulate_with_parameters(parameters, measurements, restore_params=True, chp_load_setpoints=None, restore_state=True, warmup=None) ¶

Analysis¶

pyadm1ode_calibration.calibration.analysis.sensitivity.SensitivityAnalyzer(plant, simulator=None, verbose=True) ¶

Methods:¶

analyze(parameters, measurements, objectives=None, perturbation=0.01) ¶

pyadm1ode_calibration.calibration.analysis.identifiability.IdentifiabilityAnalyzer(plant, sensitivity_analyzer=None, verbose=True) ¶

Methods:¶

analyze(parameters, measurements, optimization_history=None, confidence_level=0.95, correlation_threshold=0.8) ¶

analyze_subset(parameters, measurements, objectives=None) ¶

subset_verdict(sensitivity, per_parameter) staticmethod ¶

pyadm1ode_calibration.calibration.analysis.identifiability.IdentifiabilityResult(parameter, is_identifiable, confidence_interval, correlation_with, objective_sensitivity, reason) dataclass ¶

pyadm1ode_calibration.calibration.analysis.identifiability.SubsetIdentifiability(parameters, collinearity_index, is_identifiable, reason) dataclass ¶

pyadm1ode_calibration.calibration.analysis.identifiability.collinearity_index(sensitivity_matrix) ¶

Validation¶

pyadm1ode_calibration.calibration.validation.CalibrationValidator(plant, verbose=True, time_varying_feed=False) ¶

Methods:¶

analyze_residuals(measurements, simulated, objectives=None) ¶

cross_validate(parameters, measurements, n_folds=5, objectives=None) ¶

validate(parameters, measurements, objectives=None, simulation_duration=None, warmup=None) ¶

`pyadm1ode_calibration.calibration.Calibrator(plant, verbose=True)` ¶

`apply_calibration(result)` ¶

`run_initial_calibration(measurements, parameters, **kwargs)` ¶

`run_online_calibration(measurements, parameters, **kwargs)` ¶

`pyadm1ode_calibration.calibration.InitialCalibrator(plant, verbose=True, time_varying_feed=False)` ¶

`calibrate(measurements, parameters, bounds=None, method='differential_evolution', objectives=None, weights=None, validation_split=0.2, max_iterations=100, population_size=15, tolerance=0.0001, sensitivity_analysis=True, use_constraints=False, check_identifiability=False, **kwargs)` ¶

`identifiability_analysis(parameters, measurements, optimization_history=None, confidence_level=0.95, correlation_threshold=0.8)` ¶

`sensitivity_analysis(parameters, measurements, objectives=None)` ¶

`pyadm1ode_calibration.calibration.OnlineCalibrator(plant, verbose=True, parameter_bounds=None, time_varying_feed=False)` ¶

`apply_calibration(result)` ¶

`calibrate(measurements, parameters=None, current_parameters=None, variance_threshold=0.15, max_parameter_change=0.2, time_window=7, method='nelder_mead', max_iterations=50, objectives=None, weights=None, use_constraints=True, **kwargs)` ¶

`should_recalibrate(recent_measurements, objectives=None)` ¶

`pyadm1ode_calibration.calibration.CalibrationResult(success, parameters, initial_parameters, objective_value, n_iterations, execution_time, method, message, validation_metrics=dict(), sensitivity=dict(), history=list(), timestamp=(lambda: datetime.now(timezone.utc).isoformat())())` `dataclass` ¶

`from_dict(data)` `classmethod` ¶

`to_dict()` ¶

`to_json(filepath)` ¶

`pyadm1ode_calibration.calibration.parameter_bounds.ParameterBounds()` ¶

`add_bound(name, lower, upper, default, bound_type=BoundType.HARD, penalty_weight=1.0, description='', unit='', substrate_dependent=False)` ¶

`calculate_penalty(name, value, penalty_type='quadratic')` ¶

`calculate_total_penalty(parameters, penalty_type='quadratic')` ¶

`clip_to_bounds(name, value)` ¶

`get_bounds(name)` ¶

`get_bounds_tuple(name)` ¶

`get_default_values(parameter_names)` ¶

`is_within_bounds(name, value, tolerance=0.0)` ¶

`scale_to_unit_interval(name, value)` ¶

`unscale_from_unit_interval(name, scaled_value)` ¶

`validate_parameters(parameters, raise_on_invalid=False)` ¶

`pyadm1ode_calibration.calibration.parameter_bounds.create_default_bounds()` ¶

`pyadm1ode_calibration.calibration.core.simulator.PlantSimulator(plant, verbose=True, time_varying_feed=False)` ¶

`simulate_with_parameters(parameters, measurements, restore_params=True, chp_load_setpoints=None, restore_state=True, warmup=None)` ¶

`pyadm1ode_calibration.calibration.analysis.sensitivity.SensitivityAnalyzer(plant, simulator=None, verbose=True)` ¶

`analyze(parameters, measurements, objectives=None, perturbation=0.01)` ¶

`pyadm1ode_calibration.calibration.analysis.identifiability.IdentifiabilityAnalyzer(plant, sensitivity_analyzer=None, verbose=True)` ¶

`analyze(parameters, measurements, optimization_history=None, confidence_level=0.95, correlation_threshold=0.8)` ¶

`analyze_subset(parameters, measurements, objectives=None)` ¶

`subset_verdict(sensitivity, per_parameter)` `staticmethod` ¶

`pyadm1ode_calibration.calibration.analysis.identifiability.IdentifiabilityResult(parameter, is_identifiable, confidence_interval, correlation_with, objective_sensitivity, reason)` `dataclass` ¶

`pyadm1ode_calibration.calibration.analysis.identifiability.SubsetIdentifiability(parameters, collinearity_index, is_identifiable, reason)` `dataclass` ¶

`pyadm1ode_calibration.calibration.analysis.identifiability.collinearity_index(sensitivity_matrix)` ¶

`pyadm1ode_calibration.calibration.validation.CalibrationValidator(plant, verbose=True, time_varying_feed=False)` ¶

`analyze_residuals(measurements, simulated, objectives=None)` ¶

`cross_validate(parameters, measurements, n_folds=5, objectives=None)` ¶

`validate(parameters, measurements, objectives=None, simulation_duration=None, warmup=None)` ¶