API Reference

API Reference#

estim8.datatypes#

This model defines the datatypes used to store simulated and experimental data in the estim8 package. Further it implements methods for calculation of discrepancy measures or statistical likelhoods.

class estim8.datatypes.Constants#

Bases: object

Attributes:

SINGLE_ID

Methods

VALID_METRICS

SINGLE_ID = None#

VALID_METRICS#: alias of Literal[‘SS’, ‘WSS’, ‘negLL’]

class estim8.datatypes.Experiment(measurements: List[Measurement] | DataFrame, replicate_ID=None, errors: DataFrame = None, error_model: BaseErrorModel = <estim8.error_models.LinearErrorModel object>, observation_mapping: Dict[str, str] | None=None)#

Bases: object

A class to represent an experiment.

Attributes:

measurementslist of Measurement: The measurements of the experiment.
replicate_IDstr: The replicate ID of the experiment.
observation_mappingdict of str to str: The mapping of observation names to measurement names.

Methods

__getitem__(name)	Gets the measurement with the given name.
getbysimkey(name)	Gets the measurement by the simulation key.
calculate_loss(simulation, metric)	Calculates the loss between the experiment and the simulation.
generate_mc_samples(n_samples)	Generates Monte Carlo samples from the experiment.

calculate_loss(simulation: Simulation, metric: Literal['SS', 'WSS', 'negLL'] = 'SS') → float#

Calculates the loss between the experiment and the simulation.

Parameters:

simulationSimulation: The simulation.
metricstr, optional: The metric to use for calculating the loss (default is “SS”).

Returns:

float: The loss between the experiment and the simulation.

generate_mc_samples(n_samples: int) → List[Experiment]#

Generates Monte Carlo samples from the experiment.

Parameters:

n_samplesint: The number of samples to generate.

Returns:

list of Experiment: The generated samples.

getbysimkey(name)#

Gets the measurement by the simulation key.

Parameters:

namestr: The simulation key.

Returns:

Measurement: The measurement with the given simulation key.

class estim8.datatypes.Measurement(name, timepoints: array, values: array, replicate_ID: str | None = None, errors: array = None, error_model: BaseErrorModel = <estim8.error_models.LinearErrorModel object>)#

Bases: TimeSeries

A class to represent a measurement, inheriting from TimeSeries.

Attributes:

errorsnp.array: Gets the errors of the measurement.
error_modelerror_models.BaseErrorModel: The error model used for the measurement.

Methods

get_loss(model_prediction, metric)	Calculates the loss between the measurement and the model prediction.
get_sampling(n_samples)	Generates samples from the measurement.

property errors#

Gets the errors of the measurement.

Returns:

np.array: The errors of the measurement.

get_loss(model_prediction: ModelPrediction, metric: Literal['SS', 'WSS', 'negLL'] = 'SS') → float#

Calculates the loss between the measurement and the model prediction.

Parameters:

model_predictionModelPrediction: The model prediction.
metricstr, optional: The metric to use for calculating the loss (default is “SS”).

Returns:

float: The loss between the measurement and the model prediction.

Raises:

NotImplementedError: If the metric is not supported.

get_sampling(n_samples) → List[Measurement]#

Generates samples from the measurement.

Parameters:

n_samplesint: The number of samples to generate.

Returns:

list of Measurement: The generated samples.

class estim8.datatypes.ModelPrediction(name: str, timepoints: array, values: array, replicate_ID: str | None = None)#

Bases: TimeSeries

A class to represent model predictions, inheriting from TimeSeries.

Methods

interpolate(measurement_timepoints)

Interpolates the model predictions to the given measurement timepoints.

interpolate(measurement_timepoints: array) → TimeSeries#

Interpolates the model predictions to the given measurement timepoints.

Parameters:

measurement_timepointsnp.array: The time points at which to interpolate the model predictions.

Returns:

TimeSeries: The interpolated model predictions.

class estim8.datatypes.Simulation(simulation: Dict[str, array], replicate_ID: str | None = None)#

Bases: object

A class to represent a simulation.

Attributes:

model_predictionslist of ModelPrediction: The model predictions for the simulation.
replicate_IDstr: The replicate ID of the simulation.

Methods

__getitem__(name)

Gets the model prediction with the given name.

class estim8.datatypes.TimeSeries(name: str, timepoints: array, values: array, replicate_ID: str | None = None)#

Bases: object

A class to represent a time series.

Attributes:

namestr: The name of the time series.
timepointsnp.array: The time points of the time series.
valuesnp.array: The values of the time series.
replicate_IDstr: The replicate ID of the time series.

Methods

_equal_shapes(array1, array2)	Checks if two arrays have equal shapes.
_get_mask()	Creates a mask for non-NaN values.
drop_nans()	Drops NaN values from timepoints and values.

drop_nans()#: Drops NaN values from timepoints and values.

estim8.error_models#

This module defines the blueprint for modeling measurement noise.

class estim8.error_models.BaseErrorModel(error_distribution: rv_continuous = <scipy.stats._continuous_distns.norm_gen object>, error_distribution_kwargs: dict = {})#

Bases: ABC

Abstract base class for error modeling.

Methods

`generate_error_data`(values)	Abstract class method for calculating errors of experimental data given the datapoints.
`get_sampling`(values, errors, n_samples)	Resamples values of data given the class instance error_distribution.

abstractmethod generate_error_data(values: array) → array#

Abstract class method for calculating errors of experimental data given the datapoints.

Parameters:

valuesnp.array: Values of the data on which to apply error model

Returns:

errorsnp.array: Error calculated according to specified model

get_sampling(values: array, errors: array, n_samples: int) → List[array]#

Resamples values of data given the class instance error_distribution.

Parameters:

valuesnp.array: The values to resample.
n_samplesint: The number of samples to generate.

Returns:

resamplingList[np.array]: The generated Monte Carlo samples of values.

class estim8.error_models.LinearErrorModel(slope: float = 0, offset: float = 0, error_distribution: rv_continuous = <scipy.stats._continuous_distns.norm_gen object>, error_distribution_kwargs: dict = {})#

Bases: BaseErrorModel

An ErrorModel with linear relationship between measurement value and noise given by;

\[\sigma = slope \cdot y + offset\]

Attributes:

error_model_params: Error model parameters given by slope and offset.

Methods

`generate_error_data`(values)	Generates error data based on the linear error model.
`get_sampling`(values, errors, n_samples)	Resamples values of data given the class instance error_distribution.

property error_model_params: dict#: Error model parameters given by slope and offset.

generate_error_data(values: array) → array#

Generates error data based on the linear error model.

Parameters:

valuesnp.array: Values of the data on which to apply error model.

Returns:

errorsnp.array: Error calculated according to the linear model.

estim8.estimator#

This module implements the Estimator class for parameter estimation and uncertainty quantification.

class estim8.estimator.Estimator(model: ~estim8.models.Estim8Model, bounds: dict[str, ~typing.List[float]], data: ~estim8.datatypes.Experiment | ~typing.Dict[str, ~estim8.datatypes.Experiment] | ~pandas.core.frame.DataFrame | ~typing.Dict[str, ~pandas.core.frame.DataFrame], t: ~typing.List[float] | None = None, metric: ~typing.Literal['SS', 'WSS', 'negLL'] = 'SS', error_model: ~estim8.error_models.BaseErrorModel = <estim8.error_models.LinearErrorModel object>, parameter_mapping: list[~estim8.utils.ModelHelpers.ParameterMapper] | ~estim8.utils.ModelHelpers.ParameterMapping | None = None)#

Bases: object

Parameter estimation and uncertainty quantification manager.

This class provides a comprehensive interface for: - Parameter estimation using various optimization algorithms - Monte Carlo sampling for uncertainty quantification - Profile likelihood analysis - Handling multiple experimental replicates - Distributed computation through federated workers

Parameters:

modelmodels.Estim8Model: Model to fit to experimental data
boundsdict[str, List[float]]: Parameter bounds formatted as {parameter_name: [lower_bound, upper_bound]}
dataUnion[Experiment, Dict[str, Experiment], pd.DataFrame, Dict[str, pd.DataFrame]]: Experimental measurements to fit model against. Can be: - Single experiment as Experiment or DataFrame - Multiple replicates as Dict[str, Experiment] or Dict[str, DataFrame]
tList[float], optional: Simulation time settings as [t_start, t_end, stepsize]. If None, derived from data
metricLiteral[“SS”, “WSS”, “negLL”], optional: Loss function for optimization: - “SS”: Sum of squares - “WSS”: Weighted sum of squares - “negLL”: Negative log-likelihood Default is “SS”
error_modelerror_models.BaseErrorModel, optional: Error model for experimental data. Only used when data is DataFrame. Default is LinearErrorModel()
parameter_mappingUnion[List[utils.ModelHelpers.ParameterMapper], utils.ModelHelpers.ParameterMapping], optional: Maps parameters between replicates. Either: - List of ParameterMapper objects - Complete ParameterMapping object Default is None (no mapping)

Attributes:

replicate_IDsList[str]: IDs of experimental replicates
modelmodels.Estim8Model: Reference to model being fit
boundsdict: Parameter bounds
metricstr: Current loss function
error_modelerror_models.BaseErrorModel: Current error model
parameter_mappingutils.ModelHelpers.ParameterMapping: Current parameter mapping

Methods

estimate(method, max_iter, optimizer_kwargs={}, n_jobs=1, federated_workers=False)	Main parameter estimation method
mc_sampling(method, n_jobs, max_iter, n_samples=100, mcs_at_once=1)	Monte Carlo sampling for uncertainty quantification
profile_likelihood(p_opt, method, max_iter, n_points=3, dp_rel=0.1)	Profile likelihood analysis for parameter confidence intervals

Notes

Supports both single-core and parallel computation
Can use federated worker architecture for distributed computing, enabling parallel simulation of model replicates within a single objective function call
Handles replicate-specific parameters through parameter mapping
Provides multiple optimization algorithms through scipy, scikit-optimize and pygmo

Examples

Basic parameter estimation:

>>> estimator = Estimator(model=model, bounds=bounds, data=data)
>>> estimates, info = estimator.estimate(method='de', max_iter=1000)

Parallelized parameter estimation: >>> estimates, info = estimator.estimate(method=’de’, n_jobs=2, max_iter=1000)

With federated workers:

>>> estimates, info = estimator.estimate(method='de', max_iter=1000, federated_workers=4)

Monte Carlo sampling:

>>> results = estimator.mc_sampling(method='de', n_jobs=4, max_iter=100, n_samples=10)

check_problem_input() → None#

Validate estimation problem setup.

Checks: 1. Model observables match data observation mappings 2. Data completeness and consistency

Raises:

KeyError: If observation mappings don’t match model observables

property data#

estimate(method: str | List[str], max_iter: int, optimizer_kwargs: dict = {}, n_jobs: int = 1, federated_workers: int | bool = False, worker_kwargs: dict = {})#

Estimate the parameters of the model.

Parameters:

methodstr | List[str]: The optimization method(s).
max_iterint: The maximum number of iterations.
optimizer_kwargsdict, optional: Additional keyword arguments for the optimizer, by default {}.
n_jobs: int, optional: The number of parallel jobs, by default 1.
federated_workersint | bool, optional: The number of federated workers, by default False.
worker_kwargsdict, optional: Additional keyword arguments for federated workers. These are passed to the launch_workers() method (see launch_workers() for details), by default {}.

Returns:

tuple: The optimization result and additional information.

property hosts_and_ports#

launch_workers(n_workers: int | None = None, host: str = 'localhost', ports: List[int] | None = None, start_at_port: int = 9500, mc_sampling=False) → None#

Launch federated worker processes for distributed computation.

Parameters:

n_workersint, optional: Number of worker processes, by default None
hoststr, optional: Host address for workers, by default “localhost”
portsList[int], optional: Specific ports for workers, by default None
start_at_portint, optional: Starting port number when auto-assigning, by default 9500
mc_samplingbool, optional: Whether workers will be used for Monte Carlo sampling, by default False

Raises:

ValueError: If neither n_workers nor ports are specified

mc_sampling(method: str | List[str], n_jobs: int, max_iter: int, federated_workers: int = 0, optimizer_kwargs: dict = {}, n_samples: int = 100, mcs_at_once: int = 1, worker_kwargs: dict = {})#

Perform Monte Carlo sampling for parameter estimation problem.

Parameters:

methodstr | List[str]: The optimization method(s).
n_jobsint: The number of parallel jobs.
max_iterint: The maximum number of iterations.
federated_workersint, optional: The number of federated workers, by default 0.
optimizer_kwargsdict, optional: Additional keyword arguments for the optimizer, by default {}.
n_samplesint, optional: The number of Monte Carlo samples, by default 100.
mcs_at_onceint, optional: The number of Monte Carlo samples to process at once, by default 1.
worker_kwargsdict, optional: Additional keyword arguments for federated workers. These are passed to the launch_workers() method (see launch_workers() for details), by default {}.

Returns:

list: The results of the Monte Carlo sampling.

property metric#

objective(theta: array) → float#

Calculate total loss function value across all replicates.

Parameters:

thetanp.array: Parameter vector to evaluate

Returns:

float: Total loss function value

objective_for_replicate(parameters: dict, data: Experiment, metric: Literal['SS', 'WSS', 'negLL'] | None = None) → float#

Calculate loss function value for a single experimental replicate.

Parameters:

parametersdict: Model parameters as {parameter_name: value}
dataExperiment: Experimental data for this replicate
metricLiteral[“SS”, “WSS”, “negLL”], optional: Loss function type, by default None (uses instance metric)

Returns:

float: Loss function value. Returns np.inf if simulator crashes with RuntimeError, ValueError, ZeroDivisionError, OverflowError or fmpy.fmi1.FMICallException,.

property parameter_mapping#

profile_likelihood(p_opt: dict, method: str | List[str], max_iter: int, n_jobs: int = 1, federated_workers: int = 0, optimizer_kwargs: dict = {}, p_at_once=1, max_steps: int | None = None, stepsize: float = 0.02, p_inv: list | None = None, alpha: float = 0.05, worker_kwargs: dict = {})#

Calculate the profile likelihood for the estimated parameters.

Parameters:

p_optdict: The optimal parameter values.
methodstr | List[str]: The optimization method(s).
max_iterint: The maximum number of iterations.
n_jobsint, optional: The number of parallel jobs, by default 1.
federated_workersint, optional: The number of federated workers, by default 0.
optimizer_kwargsdict, optional: Additional keyword arguments for the optimizer, by default {}.
p_at_onceint, optional: The number of parameters to profile at once, by default 1.
n_pointsint, optional: The number of points to evaluate, by default 3.
dp_relfloat, optional: The relative parameter variation width, by default 0.1.
p_invlist, optional: The parameters to investigate, by default None.
worker_kwargsdict, optional: Additional keyword arguments for federated workers. These are passed to the launch_workers() method (see launch_workers() for details), by default {}.

Returns:

dict: The profile likelihood results.

Raises:

KeyError: If the parameter names in p_inv or p_opt are not specified in the bounds.
ValueError: If dp_rel is not in the range (0,1).

shutdown_workers()#: Shut down the federated workers.

property t#

test_workers()#

Test the responsiveness of the federated workers.

Raises:

utils.WorkerNotUpError: If any worker is not responsive.

static use_federated_workers(func: Callable)#

Decorator for managing federated worker lifecycle.

This decorator handles: 1. Worker initialization 2. Worker testing 3. Function execution 4. Worker cleanup 5. Error handling

Parameters:

funccallable: Function to wrap

Returns:

callable: Wrapped function with worker lifecycle managementy

estim8.generalized_islands#

This module implements helper functions for working with pygmo`s generalized islands approach.

class estim8.generalized_islands.Estim8_mp_island(use_pool=True, report_level=1)#

Bases: mp_island

A custom mp_island implementation with evolution logging capabilities.

Attributes:

pid: ID of the evolution process (read-only).
use_pool: Pool usage flag (read-only).

Methods

`get_extra_info`()	Island's extra info.
`get_name`()	Island's name.
`get_pool_size`()	Get the size of the process pool.
`init_pool`([processes])	Initialise the process pool.
`resize_pool`(processes)	Resize pool.
`run_evolve`(algo, pop)	Run evolution with logging of progress.
`shutdown_pool`()	Shutdown pool.

run_evolve(algo, pop)#: Run evolution with logging of progress.

class estim8.generalized_islands.PygmoArchipelago(*args, **kwargs)#

Bases: Protocol

Methods

evolve
get_champions_f
get_champions_x
push_back
set_migrant_handling
wait
wait_check

evolve(n_evolutions: int = 1) → None#

get_champions_f() → ndarray[List[float]]#

get_champions_x() → List[List[float]]#

push_back(**kwargs) → None#

set_migrant_handling(handler: Any) → None#

wait() → None#

wait_check() → None#

class estim8.generalized_islands.PygmoEstimationInfo(archi: PygmoArchipelago, udi_type=<class 'pygmo._py_islands.mp_island'>, fun: float = inf, n_evos: int = 0)#

Bases: object

An object to store additional information from an evolved archipelago as well as the archipelago itself.

Methods

get_f_evals()

Get the number of accumulated objective function evaluations of the archipelago.

get_f_evals() → int#

Get the number of accumulated objective function evaluations of the archipelago.

Returns:

int: The number of accumulated objective function evaluations.

class estim8.generalized_islands.PygmoHelpers#

Bases: object

Helper functions for working with pygmo.

Methods

`create_archipelago`(objective, bounds, algos, ...)	Creates a pygmo.archipelago object using the generalized islands model.
`create_pygmo_pop`(args)	Create a pygmo population.
`extract_archipelago_problem`(archi[, i])	Extracts the user defined problem from an archipelago, implemented as pygmo.problem(UDproblem).
`get_archipelago_results`(archi, estimation_info)	Extracts the results of an evolved archipelago and updates additional estimation info.
`get_estimates_from_archipelago`(archi)	Extracts the best estimates and corresponding value of the objective function from an archipelago object.
`get_pygmo_algorithm_instance`(name, **kwargs)	Creates an instance of a pygmo.algorithm given the name and algorithm kwargs.

algo_default_kwargs: Dict[str, Dict[str, Any]] = {'bee_colony': {'gen': 10, 'limit': 2}, 'cmaes': {'force_bounds': False, 'ftol': 1e-08, 'gen': 10, 'xtol': 1e-08}, 'compass_search': {'max_fevals': 100, 'start_range': 1, 'stop_range': 1e-06}, 'de': {'ftol': 1e-08, 'gen': 10, 'xtol': 1e-08}, 'de1220': {'ftol': 1e-08, 'gen': 10, 'variant_adptv': 2, 'xtol': 1e-08}, 'gaco': {'gen': 10}, 'ihs': {'gen': 40}, 'maco': {'gen': 10}, 'mbh': {'algo': 'compass_search', 'perturb': 0.1, 'stop': 2}, 'moead': {'gen': 10}, 'nlopt': {'solver': 'lbfgs'}, 'nsga2': {'gen': 10}, 'nspso': {'gen': 10}, 'pso': {'gen': 10}, 'pso_gen': {'gen': 10}, 'sade': {'ftol': 1e-08, 'gen': 10, 'variant_adptv': 2, 'xtol': 1e-08}, 'scipy_optimize': {}, 'sea': {'gen': 40}, 'sga': {'gen': 10}, 'simulated_annealing': {}, 'xnes': {'eta_mu': 0.05, 'ftol': 1e-08, 'gen': 10, 'xtol': 1e-08}}#

static create_archipelago(objective: ~typing.Callable, bounds: dict, algos: ~typing.List[str], algos_kwargs: ~typing.List[dict], pop_size: int, topology: ~typing.Any = <pygmo.core.fully_connected object>, report=0, n_processes=2, init_pool: bool = True) → Tuple[PygmoArchipelago, PygmoEstimationInfo]#

Creates a pygmo.archipelago object using the generalized islands model.

Parameters:

objectivecallable: An instance of an estim8.Objective function that is used to create a UDproblem.
boundsdict: The bounds for the parameters.
algoslist[str]: A list of optimization algorithms for the individual islands of the archipelago.
algos_kwargslist[dict]: A list of algorithm kwargs corresponding to passed optimizers.
pop_sizeint: Population size for each individual island.
topologypygmo.topology, optional: Represents the connection policy between the islands of the archipelago, by default pygmo.fully_connected().
report: int, optional: The report level during optimization, by default 0. In case of report > 0, the evoltions are runned with Estim8_mp_island which logs the evolution trace. A report level of 1 yields data on the archipelago’s island champions over evolutions. With a report level of 2 the islands current states are printed in the Log.
n_processesint, optional: The number of processes to use, by default joblib.cpu_count().

Returns:

pygmo.archipelago: The created archipelago.
PygmoEstimationInfo: An estimation info object containing the archipelago

static create_pygmo_pop(args)#

Create a pygmo population.

Parameters:

argstuple: The arguments for creating the population.

Returns:

pygmo.population: The created population.

static extract_archipelago_problem(archi: PygmoArchipelago, i=0) → PygmoProblem#

Extracts the user defined problem from an archipelago, implemented as pygmo.problem(UDproblem).

Parameters:

archiPygmoArchipelago: The evolved archipelago.
iint, optional: Index of the island from which the problem is extracted, by default 0.

Returns:

pygmo.problem: The extracted problem.

static get_archipelago_results(archi: PygmoArchipelago, estimation_info: PygmoEstimationInfo) → Tuple[dict, PygmoEstimationInfo]#

Extracts the results of an evolved archipelago and updates additional estimation info.

Parameters:

archiPygmoArchipelago: The evolved archipelago.
estimation_infoPygmoEstimationInfo: Additional information about the archipelago before evolution(s).

Returns:

Tuple[dict, PygmoEstimationInfo]: Dictionary of best estimates according to the smallest loss value. Updated additional information about the evolved archipelago containing the archipelago itself.

static get_estimates_from_archipelago(archi: PygmoArchipelago) → Tuple[dict, float]#

Extracts the best estimates and corresponding value of the objective function from an archipelago object.

Parameters:

archiPygmoArchipelago: The evolved archipelago.

Returns:

Tuple[dict, float]: Dictionary of best estimates according to the smallest loss aka objective function value. The smallest loss value among all islands.

static get_pygmo_algorithm_instance(name: str, **kwargs) → Any#

Creates an instance of a pygmo.algorithm given the name and algorithm kwargs.

Parameters:

namestr: The name of the optimization algorithm.
kwargsdict, optional: Keyword arguments for the algorithm, by default None, which means the default arguments of the respective pygmo.algorithm will be used.

Returns:

pygmo.algorithm: The pygmo algorithm instance.

Raises:

ValueError: If the algorithm name is not supported.

class estim8.generalized_islands.PygmoMpIsland(use_pool: bool = True)#

Bases: Protocol

Protocol for pygmo.mp_island type checking

Methods

init_pool
run_evolve
shutdown_pool

init_pool(n_processes: int) → None#

run_evolve(algo: Any, pop: Any) → Any#

shutdown_pool() → None#

class estim8.generalized_islands.PygmoProblem(*args, **kwargs)#

Bases: Protocol

Methods

extract
get_fevals

extract(cls: Any) → Any#

get_fevals() → int#

class estim8.generalized_islands.UDproblem(objective: Callable, bounds: dict)#

Bases: object

A wrapper class around an Objective function with functions required for creating a user defined pygmo.problem.

Methods

`fitness`(theta)	Evaluate the fitness of a solution.
`get_bounds`()	Get the bounds of the problem.
`get_extra_info`()	Get extra information about the problem.

fitness(theta) → array#

Evaluate the fitness of a solution.

Parameters:

thetanp.array: The solution to evaluate.

Returns:

np.array: The fitness value.

get_bounds() → tuple#

Get the bounds of the problem.

Returns:

tuple: The lower and upper bounds.

get_extra_info() → list#

Get extra information about the problem.

Returns:

list: The keys of the bounds dictionary.

estim8.generalized_islands.optional_import(module_name, mock_name=None)#

estim8.models#

This module implements the model classes for the Estim8 package.

class estim8.models.Estim8Model(default_parameters={}, r_tol=0.0001)#

Bases: ABC

Methods

`retrieve_variables`()	Retrieve model parameters and observables.
`simulate`(t0, t_end, stepsize[, parameters, ...])	Simulate the model.

abstractmethod retrieve_variables()#: Retrieve model parameters and observables.

abstractmethod simulate(t0: float, t_end: float, stepsize: float, parameters: Dict[str, float] = {}, observe: List[str] | None = None, replicate_ID: str | None = None) → Simulation#

Simulate the model.

Parameters:

t0float: Start time of the simulation.
t_endfloat: End time of the simulation.
stepsizefloat: Time step size for the simulation.
parametersdict: Parameters for the simulation.
observelist: List of observables to record during the simulation.
replicate_IDstr, optional: Replicate ID for the simulation, by default SINGLE_ID.

Returns:

Simulation: The simulation results.

class estim8.models.FmuModel(path: str, fmi_type: Literal['ModelExchange', 'CoSimulation'] = 'ModelExchange', default_parameters: dict = {}, r_tol=0.0001, memory_threshold_mb: float | None = None)#

Bases: Estim8Model

Attributes:

fmi_type: Get the FMI type.

Methods

`cleanup`()	Clean up temporary files and directories.
`freeInstance`()	Free the FMU instance.
`instantiate_fmu`()	Instantiate the FMU from the copied file.
`retrieve_variables`()	Retrieve model parameters and observables.
`simulate`(t0, t_end, stepsize[, parameters, ...])	Simulate the FMU model.

cleanup()#: Clean up temporary files and directories.

property fmi_type#

Get the FMI type.

Returns:

str: The FMI type.

freeInstance()#: Free the FMU instance.

instantiate_fmu()#: Instantiate the FMU from the copied file.

retrieve_variables()#: Retrieve model parameters and observables.

simulate(t0: float, t_end: float, stepsize: float, parameters: dict[str, float] = {}, observe: list[str] | None = None, replicate_ID: str | None = None, r_tol: float | None = None, solver: Literal['CVode', 'Euler'] = 'CVode') → Simulation#

Simulate the FMU model.

Parameters:

t0float: Start time of the simulation.
t_endfloat: End time of the simulation.
stepsizefloat: Time step size for the simulation.
parametersdict, optional: Parameters for the simulation, by default {}.
observelist, optional: List of observables to record during the simulation, by default None.
r_tolfloat, optional: Relative tolerance for the simulation, by default None.
solverLiteral[“CVode”, “Euler”], optional: Solver to use for the simulation, by default “CVode”.
replicate_IDstr, optional: Replicate ID for the simulation, by default SINGLE_ID.

Returns:

Simulation: The simulation results.

estim8.profile#

This module implements functionality for profiling likelihoods and likelihood based confidence intervals.

class estim8.profile.ProfileSampler(parameter, mle: float, mle_negll: float, negll_threshold: float, optimizer: Optimization, bounds: List[float], direction: Literal[-1, 1], stepsize=0.02, max_steps: int | None = None)#

Bases: object

Sample points along a profile likelihood curve.

Parameters:

parameterstr: Name of the parameter to profile
mlefloat: Maximum likelihood estimate of the parameter
mle_negllfloat: Negative log-likelihood at the MLE
negll_thresholdfloat: Threshold value for the negative log-likelihood
optimizerOptimization: Optimizer instance to use for likelihood calculations
boundsList[float]: Parameter bounds [lower, upper]
directionLiteral[-1, 1]: Direction to walk the profile (1 for positive, -1 for negative)
stepsizefloat, optional: Relative step size for parameter updates, by default 0.02
max_stepsint, optional: Maximum number of steps to take, by default None, in which case the sampler proceeds until the negll threshold is reached.

Methods

`next_pl_sample`()	Calculate the next profile likelihood sample point.
`next_step`()	Take a next fixed step along the profile.
`update_optimizer_objective`(value)	Update the optimizer's objective function with a new parameter value and set a new task_id.
`walk_profile`()	Walk the complete profile likelihood curve.

next_pl_sample()#

Calculate the next profile likelihood sample point.

Updates the internal samples array with the new point and checks if the profile likelihood threshold has been exceeded.

next_step()#

Take a next fixed step along the profile.

Returns:

float: The next value of the parameter to evaluate

update_optimizer_objective(value: float)#

Update the optimizer’s objective function with a new parameter value and set a new task_id.

Parameters:

valuefloat: New value for the profiled parameter

walk_profile()#

Walk the complete profile likelihood curve.

Returns:

tuple: Tuple containing: - numpy.ndarray: Array of sample points [(parameter_value, negll), …] - str: Name of the profiled parameter

estim8.profile.approximate_confidence_interval(xvalues, negll_values, threshold)#

Approximate the confidence interval from profile likelihood results.

Parameters:

xvaluesnumpy.ndarray: The parameter values of the profile likelihood curve
negll_valuesnumpy.ndarray: The negative log-likelihood values of the profile likelihood curve
thresholdfloat: The threshold value for the confidence interval

Returns:

tuple: Tuple containing: - float: Lower bound of the confidence interval - float: Upper bound of the confidence interval

Raises:

ValueError: If confidence interval bounds cannot be found

estim8.profile.calculate_negll_thresshold(alpha: float = 0.05, df: int = 1, mle_negll: float = 0)#

Calculate the negative log-likelihood threshold for profile likelihood.

Parameters:

alphafloat, optional: Significance level, by default 0.05
dfint, optional: Degrees of freedom, by default 1
mle_negllfloat, optional: Maximum likelihood estimate of negative log-likelihood, by default 0

Returns:

float: The threshold value for the negative log-likelihood

estim8.objective#

This module implements the objective functions that are passed to an optimization algorithm.

class estim8.objective.Objective(func: Callable, bounds: Dict[str, List[float]], parameter_mapping: ParameterMapping, mc_sample: int | None = 0)#

Bases: object

Objective function that is passed to an optimization algorithm. When called to evaluate a parameter set theta, conducts a replicate handling step and sends request for loss calculation to federated worker nodes via a compiled pytensor.function

Methods

__call__(theta)

Evaluate the objective function for a given parameter set.

estim8.objective.global_objective(*replicate_par_sets, local_objective: Callable)#

Calculates a global objective as the sum of local objective mapped to replicate parameter sets.

Parameters:

local_objectivecallable: The local objective function.

Returns:

list: The global objective value.

estim8.objective.objective_function_wrapper(theta: array, objective, task_id: str | None = None) → float#

A wrapper around the objective. Creates global copy of the objective function in the current process to avoid serialization on every function call, and stores it in a dictionary with a unique task ID.

Parameters:

thetanp.array: The parameter set to evaluate.
objective: tuple: A serialized callable.
task_idstr, optional: The task ID, by default None.

Returns:

float: The loss value.

estim8.optimizers#

This submodule implements the optimization of an objective function using various optimization algorithms.

class estim8.optimizers.Optimization(objective: Objective | Callable, method: str | List[str], bounds: dict, optimizer_kwargs: dict = {}, use_parallel=True, task_id: str = 'estimate')#

Bases: object

Class that manages the optimization of an objective function.

Methods

`get_optimization_method`(method)	Get the optimization method.
`optimize`()	Optimize the objective function.
`optimize_pygmo_archi`(objective, bounds, ...)	Optimize the objective function using a pygmo archipelago.
`optimize_pygmo_archipelago_continued`(_, ...)	Continue optimizing the objective function using a pygmo archipelago object.
`prepare_optimizer_kwargs`(method, ...)	Prepare the optimizer keyword arguments.

static get_optimization_method(method: str | List[str]) → Callable#

Get the optimization method.

Parameters:

methodstr | List[str]: The optimization method(s).

Returns:

callable: The optimization function.

Raises:

KeyError: If the method is not supported.
NotImplementedError: If the method is not implemented.

optimization_funcs = {'bh': <function basinhopping>, 'de': <function differential_evolution>, 'dual_annealing': <function dual_annealing>, 'gp': <function gp_minimize>, 'local': <function minimize>, 'shgo': <function shgo>}#

optimize()#

Optimize the objective function.

Returns:

tuple: The optimization result and additional information.

static optimize_pygmo_archi(objective: ~typing.Callable, bounds: dict, algos: ~typing.List[str], algos_kwargs: ~typing.List[dict], n_processes: int = 2, pop_size=50, topology=<pygmo.core.unconnected object>, max_iter: int = 10, report: int = 0, init_pool: bool = True) → Tuple[dict, PygmoEstimationInfo]#

Optimize the objective function using a pygmo archipelago.

Parameters:

objectivecallable: The objective function to be optimized.
boundsdict: The bounds for the parameters.
algosList[str]: The optimization algorithms to be used.
algos_kwargsList[dict]: The keyword arguments for the optimization algorithms.
n_processesint, optional: The number of processes to use, by default joblib.cpu_count().
pop_sizeint, optional: The population size, by default 50.
topologypygmo.topology, optional: The topology of the archipelago, by default pygmo.unconnected().
max_iterint, optional: The number of evolutions, by default 10.
report: int, optional: The report level during optimization, by default 0. A report level of 1 yields data on the archipelago’s island champions over evolutions. With a report level of 2 the islands current states are printed in the Log.

Returns:

Tuple[dict, generalized_islands.PygmoEstimationInfo]: The optimization result and additional information.

static optimize_pygmo_archipelago_continued(_, estimation_info: PygmoEstimationInfo, max_iter: int = 10) → Tuple[dict, PygmoEstimationInfo]#

Continue optimizing the objective function using a pygmo archipelago object.

Parameters:

estimation_infogeneralized_islands.PygmoEstimationInfo: The estimation information and the archipelago itself before evolution(s).
max_iterint, optional: The number of evolutions, by default 10.

Returns:

Tuple[dict, generalized_islands.PygmoEstimationInfo]: The optimization result and additional information.

prepare_optimizer_kwargs(method, optimizer_kwargs)#

Prepare the optimizer keyword arguments.

Parameters:

methodstr | List[str]: The optimization method(s).
optimizer_kwargsdict: The optimizer keyword arguments.

Returns:

dict: The prepared optimizer keyword arguments.

Raises:

ValueError: If the list of algorithms doesn’t match the list of algorithm kwargs.

pygmo_algos = {'bee_colony', 'cmaes', 'compass_search', 'de', 'de1220', 'gaco', 'gwo', 'ihs', 'mbh', 'nsga2', 'pso', 'scipy_optimize', 'sea', 'sga', 'simulated_annealing', 'xnes'}#

estim8.utils#

This module implements utility functions for the estim8 package.

class estim8.utils.EstimatorHelpers#

Bases: object

Methods

`generate_mc_samples`(estimator_data, n_samples)	Generate Monte Carlo samples from the data.
`get_t_from_data`(data)	Get the time vector from data.
`get_worker_loads`(hosts_and_ports)	Get the loads of the workers.
`make_replicate`(data, errors, error_model, ...)	Create a replicate from data.
`make_tensor_function`(functions, replicate_IDs)	Create a tensor function.
`test_worker`(host_and_port, theta_test, retries)	Test a worker.
`test_workers`(hosts_and_ports, theta_test[, ...])	Test multiple workers.
`update_optimizer_kwargs`(opt_kwargs, **kwargs)	Update optimizer keyword arguments.
`wait_for_worker_launch`(hosts_and_ports[, ...])	Wait for the federated workers to launch.
`workers_up`(hosts_and_ports)	Check if the workers are up.

static generate_mc_samples(estimator_data: dict[str, Experiment], n_samples: int)#

Generate Monte Carlo samples from the data.

Parameters:

estimator_datadict[str, Experiment]: The data.
n_samplesint: The number of samples.

Returns:

list: The Monte Carlo samples.

static get_t_from_data(data: Dict[str, Experiment]) → Tuple[float, float, float]#

Get the time vector from data.

Parameters:

dataDict[str, Experiment]: The data.

Returns:

Tuple[float, float, float]: The time vector for simulations given by (t_start, t_end, stepsize).

static get_worker_loads(hosts_and_ports: List[Tuple[str, int]]) → List[GetLoadResult]#

Get the loads of the workers.

Parameters:

hosts_and_portsList[Tuple[str, int]]: The hosts and ports of the workers.

Returns:

List[pytensor_federated.rpc.GetLoadResult]: The loads of the workers.

static make_replicate(data: Experiment | DataFrame, errors: DataFrame = None, error_model: BaseErrorModel = <estim8.error_models.LinearErrorModel object>, replicate_ID: str | None = None)#

Create a replicate from data.

Parameters:

dataExperiment or pd.DataFrame: The data for the replicate.
errorspd.DataFrame, optional: The errors for the data, by default None.
error_modelerror_models.BaseErrorModel, optional: The error model, by default None.
replicate_IDstr, optional: The replicate ID, by default SINGLE_ID.

Returns:

dict: The replicate data.

static make_tensor_function(functions: list[Callable], replicate_IDs: list[str | None]) → function#

Create a tensor function.

Parameters:

functionslist[callable]: The functions.
replicate_IDslist[str]: The replicate IDs.

Returns:

pytensor.function: The tensor function.

static test_worker(host_and_port: Tuple[str, int], theta_test: array, retries: int) → bool#

Test a worker.

Parameters:

host_and_portTuple[str, int]: The host and port of the worker.
theta_testnp.array: The test parameter vector.
retriesint: The number of retries.

Returns:

bool: Whether the worker is up.

static test_workers(hosts_and_ports, theta_test, retries=2)#

Test multiple workers.

Parameters:

hosts_and_portslist of Tuple[str, int]: The hosts and ports of the workers.
theta_testnp.array: The test parameter vector.
retriesint, optional: The number of retries, by default 2.

Returns:

bool or list: Whether all workers are up or the workers that are down.

static update_optimizer_kwargs(opt_kwargs: dict, **kwargs)#

Update optimizer keyword arguments.

Parameters:

opt_kwargsdict: The optimizer keyword arguments.
**kwargs: Additional keyword arguments.

Returns:

dict: The updated optimizer keyword arguments.

static wait_for_worker_launch(hosts_and_ports: List[Tuple[str, int]], timeout=30) → bool#

Wait for the federated workers to launch.

Parameters:

hosts_and_portsList[Tuple[str, int]]: The hosts and ports of the workers.
timeoutint, optional: The timeout in seconds, by default 30.

Returns:

bool: Whether the workers launched successfully.

static workers_up(hosts_and_ports: List[Tuple[str, int]]) → List[bool]#

Check if the workers are up.

Parameters:

hosts_and_portsList[Tuple[str, int]]: The hosts and ports of the workers.

Returns:

List[bool]: Whether the workers are up.

class estim8.utils.ModelHelpers#

Bases: object

Methods

ParameterMapper(global_name, replicate_ID[, ...])

Defines a replicate specific parameter.

ParameterMapping

class ParameterMapper(global_name: str, replicate_ID: str, value: float = 0, local_name: str | None = None)#

Bases: object

Defines a replicate specific parameter.

class ParameterMapping(mappings: List[ParameterMapper], default_parameters: dict, replicate_IDs: list[str | None] = [None])#

Bases: object

Attributes:

mapping: A pretty presentation of parameter mapping as a pandas.DataFrame.

Methods

`replicate_handling`(replicate_ID[, parameters])	Handle parameter mapping for a specific replicate.
`set_parameter`(name, value)	Set a parameter value by its name (either local or global).

property mapping: DataFrame#

A pretty presentation of parameter mapping as a pandas.DataFrame.

Returns:

pd.DataFrame: The parameter mapping.

replicate_handling(replicate_ID: str | None, parameters: Dict[str, float] = {})#

Handle parameter mapping for a specific replicate.

Parameters:

replicate_IDstr: The replicate ID.
parametersDict[str, float], optional: The parameters, by default dict().

Returns:

dict: The complete parameters for the replicate.

set_parameter(name: str, value: float)#

Set a parameter value by its name (either local or global).

Parameters:

namestr: The parameter name (can be local or global)
valuefloat: The new parameter value

Raises:

ValueError: If the parameter name is not found in either local or global parameters

exception estim8.utils.WorkerNotUpError(host_and_ports: List[Tuple[str, int]])#: Bases: Exception

estim8.visualization#

This module implements visualization functions for the estim8 package.

estim8.visualization.plot_distributions(mc_samples: DataFrame, ci_level: float = 0.95, kde=True)#

Plot the distributions of the Monte Carlo samples.

Parameters:

mc_samplespd.DataFrame: The Monte Carlo samples.
ci_levelint, optional: The confidence interval level, by default 0.95.
kdebool, optional: Whether to plot the kernel density estimate, by default True.

Returns:

matplotlib.figure.Figure: The figure containing the plots.

estim8.visualization.plot_estimates(estimates: dict, estimator: Estimator, only_measured: bool = False)#

Plot the estimates.

Parameters:

estimatesdict: The parameter estimates.
estimatorEstimator: The estimator object.
only_measuredbool, optional: Whether to plot only the measured observables, by default False.

Returns:

matplotlib.figure.Figure or list of matplotlib.figure.Figure: The figure(s) containing the plots.

estim8.visualization.plot_estimates_many(mc_samples: DataFrame, estimator: Estimator, only_measured: bool = False)#

Plot the estimates for many Monte Carlo samples.

Parameters:

mc_samplespd.DataFrame: The Monte Carlo samples.
estimatorEstimator: The estimator object.
only_measuredbool, optional: Whether to plot only the measured observables, by default False.

Returns:

matplotlib.figure.Figure or list of matplotlib.figure.Figure: The figure(s) containing the plots.

estim8.visualization.plot_heatmap(mc_samples: DataFrame, thresholds: int = 5, show_vals: bool = False)#

Plotting the correlations for parameter results of a Monte Carlo Sampling as a heatmap. The threshold defines the resolution of values, i.e. it defines a color map with discrete increments. The larger the thresholds argument is set, the finer is the resolution. Set show_vals True to write the exact correlation values into the plot.

Parameters:

mc_samplespd.DataFrame: Parameter information from Monte Carlo Sampling.
thresholdsint, optional: Number of increments for color map, by default 5.
show_valsbool, optional: Displays exact values into the plot, by default False.

Returns:

matplotlib.figure.Figure: The figure containing the heatmap.

estim8.visualization.plot_measurement(ax, measurement: Measurement, **kwargs)#

Plot the measurement.

Parameters:

axmatplotlib.axes.Axes: The axes to plot on.
measurementMeasurement: The measurement to plot.
**kwargs: Additional keyword arguments for the plot.

estim8.visualization.plot_model_prediction(ax, model_prediction: ModelPrediction, **kwargs)#

Plot the model prediction.

Parameters:

axmatplotlib.axes.Axes: The axes to plot on.
model_predictionModelPrediction: The model prediction to plot.
**kwargs: Additional keyword arguments for the plot.

estim8.visualization.plot_pairs(mc_samples: DataFrame, kind: Literal['scatter', 'kde', 'hist', 'reg'] = 'kde')#

Plot pairwise relationships in the Monte Carlo samples.

Parameters:

mc_samplespd.DataFrame: The Monte Carlo samples.
kindstr, optional: The kind of plot to draw, by default “kde”.

Returns:

seaborn.axisgrid.PairGrid: The pair grid containing the plots.

estim8.visualization.plot_predictives_many(ax, timepoints: ndarray, trajectories: ndarray, color)#

Plot the predictive distributions for many trajectories.

Parameters:

axmatplotlib.axes.Axes: The axes to plot on.
timepointsnp.ndarray: The time points.
trajectoriesnp.ndarray: The trajectories.
colorstr: The color for the plot.

estim8.visualization.plot_profile_likelihood(pl_results, p_opt, alpha=0.05, show_coi=True)#

Plot the profile likelihood results.

Parameters:

pl_resultsdict: The profile likelihood results.

Returns:

matplotlib.figure.Figure: The figure containing the plots.

estim8.visualization.plot_simulation(simulation: Simulation, observe: List[str] | None = None, experiment: Experiment | None = None)#

Plot the simulation results.

Parameters:

simulationSimulation: The simulation results to plot.
observeList[str], optional: List of observables to plot, by default None.
experimentExperiment, optional: The experiment data to plot alongside the simulation, by default None.

Returns:

matplotlib.figure.Figure: The figure containing the plots.

estim8.workers#

This module implements Workers that serve as a computaion backend for an objective function. Further it implements the federated Worker class and functions for launching a pool of such services.

class estim8.workers.FederatedWorker(estimator: Estimator, mc_sampling: bool = False)#

Bases: Worker

A subclass of Worker that adds error handling and logging for federated computation.

Methods

__call__(theta)

Calculates the objective function given the data by simulating the model with the parameter vector theta to evaluate.

logger = <Logger estim8.workers (WARNING)>#

class estim8.workers.Worker(estimator: Estimator, mc_sampling: bool = False)#

Bases: object

The Worker class wraps an instance of Estimator class and turns its objective for a single replicate into a callable.

Methods

__call__(theta)

Calculates the objective function given the data by simulating the model with the parameter vector theta to evaluate.

estim8.workers.init_logging()#: Configure logging.

estim8.workers.run_worker_pool(host: str, ports: List[int], estimator: Estimator, mc_sampling: bool = False) → List[multiprocessing.Process]#

Launches a pool of FederatedWorker services in parallel.

Parameters:

hoststr: The host address of worker services that are launched.
portsList[int]: The ports of worker services.
estimatorestim8.Estimator: An instance of estim8.Estimator.
mc_samplingbool, optional: If the Worker is set up for MC sampling or not, default is False.

Returns:

List[multiprocessing.Process]: The worker processes.

estim8.workers.run_worker_service(host: str, port: int, estimator: Estimator, mc_sampling: bool = False)#

Runs a FederatedWorker as a service bound to a host and port address.

Parameters:

hoststr: The host address of the service.
portint: The port number of the service.
estimatorestim8.Estimator: An instance of estim8.Estimator.
mc_samplingbool, optional: If the Worker is set up for MC sampling or not, default is False.

Notes

The loop is run forever if not stopped externally.

API Reference

Contents

API Reference#

estim8.datatypes#

estim8.error_models#

estim8.estimator#

estim8.generalized_islands#

estim8.models#

estim8.profile#

estim8.objective#

estim8.optimizers#

estim8.utils#

estim8.visualization#

estim8.workers#