Calibration (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)

Functions

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)

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):
    super().__init__(plant, verbose)
    self.parameter_bounds = create_default_bounds()
    self.validator = CalibrationValidator(plant, verbose=False)
    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()

Functions

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, **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
**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: Optional[Dict[str, Tuple[float, float]]] = None,
    method: str = "differential_evolution",
    objectives: Optional[List[str]] = None,
    weights: Optional[Dict[str, float]] = 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,
    **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.
        **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)

    # 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:
        try:
            measurements_dict[obj] = train_data.get_measurement(obj).values
        except Exception:
            continue

    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:
        val_result = self.validator.validate(
            parameters=opt_result.parameter_dict, measurements=val_data, objectives=objectives
        )
        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)

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

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
) -> 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.

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

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: Optional[List[str]] = 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)

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: Optional[ParameterBounds] = None):
    super().__init__(plant, verbose)
    self.parameter_bounds: ParameterBounds = parameter_bounds or create_default_bounds()
    self.validator: CalibrationValidator = CalibrationValidator(plant, verbose=False)
    self.trigger: OnlineCalibrationTrigger = OnlineCalibrationTrigger()
    self.state: OnlineState = OnlineState()

Functions

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: Optional[List[str]] = None,
    current_parameters: Optional[Dict[str, float]] = 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: Optional[List[str]] = None,
    weights: Optional[Dict[str, float]] = 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()

    history_entry = ParameterChangeHistory(
        timestamp=datetime.now(),
        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: Optional[List[str]] = 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() - 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().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.

Functions

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)

Simulator

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

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):
    self.plant = plant
    self.verbose = verbose
    self._original_params: Dict[str, Dict[str, float]] = {}

Functions

simulate_with_parameters(parameters, measurements, restore_params=True)

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

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
) -> Dict[str, np.ndarray]:
    """
    Run a plant simulation using a specific set of parameters.

    Args:
        parameters (Dict[str, float]): Parameter values to apply {name: value}.
        measurements (MeasurementData): Input data for substrate feeds and timing.
        restore_params (bool): Whether to restore the original plant parameters
            after the simulation finishes. Defaults to True.

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

    try:
        self._apply_parameters(parameters)

        # Simulation settings
        n_steps = len(measurements)
        dt = 1.0 / 24.0
        duration = n_steps * dt

        # Apply substrate feeds from measurements
        Q_substrates = self._extract_substrate_feeds(measurements)
        self._apply_substrate_feeds(Q_substrates)

        # Run simulation
        results = self.plant.simulate(duration=duration, dt=dt, save_interval=dt)

        return self._extract_outputs_from_results(results)
    finally:
        if restore_params:
            self._restore_parameters()