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