epydemix.utils package

Submodules

epydemix.utils.abc_smc_utils module

class epydemix.utils.abc_smc_utils.DefaultPerturbationContinuous(param_name)[source]

Bases: Perturbation

Componenent-wise normal perturbation kernel with adaptive standard deviation (Beaumont et al. (2009)).

pdf(x, center)[source]

Evaluate the PDF of the kernel.

propose(x, rng=None)[source]

Propose a new value based on the current value.

update(particles, weights, param_names)[source]

Update the standard deviation based on previous generation variance.

class epydemix.utils.abc_smc_utils.DefaultPerturbationDiscrete(param_name, prior, jump_probability=0.3)[source]

Bases: Perturbation

pdf(x, center)[source]

Transition probability for the discrete parameter.

propose(x, rng=None)[source]

Propose a new value for the discrete parameter.

update(particles, weights, param_names)[source]

Update jump_probability or other characteristics if needed.

class epydemix.utils.abc_smc_utils.Perturbation(param_name)[source]

Bases: ABC

abstractmethod pdf(x, center)[source]

Evaluate the PDF of the kernel.

abstractmethod propose(x, rng=None)[source]

Propose a new value from the current value x.

Draw randomness from rng (a seed or np.random.Generator) so that proposals are reproducible

abstractmethod update(particles, weights, param_names)[source]

Update the kernel parameters based on particles and weights.

epydemix.utils.abc_smc_utils.compute_effective_sample_size(weights: ndarray) → float[source]

Computes the effective sample size (ESS) of a set of weights.

Parameters:

weights (np.ndarray) – A NumPy array of weights for the particles.

Returns:

The effective sample size, calculated as the reciprocal of the sum of squared weights.

Return type:

float

epydemix.utils.abc_smc_utils.fast_normal_pdf(x, mean, std)[source]
epydemix.utils.abc_smc_utils.sample_prior(priors, param_names, rng=None)[source]

Samples a parameter set from the given prior distributions. priors: dictionary mapping parameter names to scipy.stats distributions param_names: list of parameter names to maintain consistent order rng: optional np.random.Generator (or seed) used for sampling Returns: list of sampled parameter values in the order of param_names

epydemix.utils.abc_smc_utils.weighted_quantile(values: ndarray | list, weights: ndarray | list, quantile: float) → float[source]

Compute the weighted quantile of a dataset.

Parameters:

  • values (Union[np.ndarray, list]) – Data values for which the quantile is to be computed.

  • weights (Union[np.ndarray, list]) – Weights associated with the data values.

  • quantile (float) – The quantile to compute, must be between 0 and 1.

Returns:

The weighted quantile value.

Return type:

float

Raises:

ValueError – If quantile is not between 0 and 1.

epydemix.utils.utils module

epydemix.utils.utils.apply_initial_conditions(epimodel, initial_conditions_dict) → ndarray[source]

Applies initial conditions to the compartments of an epidemiological model.

Parameters:

  • epimodel (EpiModel) – An instance of an epidemiological model containing compartments and population information.

  • **kwargs – Keyword arguments where each key is a compartment name and each value is an array or list specifying the initial population for that compartment.

Returns:

A 2D array where rows correspond to compartments and columns correspond to demographic groups,

representing the initial conditions of the model. The shape of the array is (number_of_compartments, number_of_demographic_groups).

Return type:

np.ndarray

Raises:

ValueError – If initial_conditions_dict contains keys that are not compartments of the model.

epydemix.utils.utils.apply_overrides(definitions: Dict[str, ndarray], overrides: Dict[str, List[Dict[str, Any]]], dates: List[date]) → Dict[str, ndarray][source]

Applies parameter overrides to the definitions based on the specified date ranges.

Parameters:

  • definitions (dict) – A dictionary where keys are parameter names and values are 2D arrays representing the parameter values over time and demographics.

  • overrides (dict) – A dictionary where keys are parameter names and values are lists of override specifications. Each override specification is a dictionary containing: - ‘start_date’ (str): The start date of the override period in ‘YYYY-MM-DD’ format. - ‘end_date’ (str): The end date of the override period in ‘YYYY-MM-DD’ format. - ‘value’ (np.ndarray or scalar): The value to override within the specified date range.

  • dates (list) – A list of datetime.date objects corresponding to the time steps in the definitions arrays.

Returns:

A dictionary with the same keys as definitions, but with values updated according to the overrides.

Return type:

dict

Raises:

ValueError – If the override values do not match the expected shape for the specified date ranges.

epydemix.utils.utils.combine_simulation_outputs(simulation_outputs_list: List[Dict[str, ndarray]]) → Dict[str, List[ndarray]][source]

Combines multiple simulation outputs into a single dictionary by appending new outputs to existing keys.

Parameters:

simulation_outputs_list (List[Dict[str, np.ndarray]]) – A list of dictionaries containing the latest simulation output to be combined. Each dictionary contains keys corresponding to compartment-demographic names and values are arrays of simulation results.

Returns:

A dictionary where keys are compartment-demographic names and values are lists of arrays.

Each list contains simulation results accumulated from multiple runs.

Return type:

Dict[str, List[np.ndarray]]

epydemix.utils.utils.compute_days(start_date: str | Timestamp, end_date: str | Timestamp) → int[source]

Computes the number of days between two dates.

Parameters:

  • start_date (Union[str, pd.Timestamp]) – The start date in string format (e.g., “YYYY-MM-DD”) or as a pandas Timestamp.

  • end_date (Union[str, pd.Timestamp]) – The end date in string format (e.g., “YYYY-MM-DD”) or as a pandas Timestamp.

Returns:

The number of days between the start date and end date, inclusive.

Return type:

int

epydemix.utils.utils.compute_simulation_dates(start_date: str | date | datetime64, end_date: str | date | datetime64, steps: int | None = None, dt: float = 1.0) → ndarray[source]

Compute simulation dates including sub-day intervals.

Parameters:

  • start_date – Start date

  • end_date – End date

  • steps – Number of steps (if None, computed from dt)

  • dt – Time step in days (e.g., 1/3 for 8-hour intervals)

Returns:

numpy array of datetime64 objects for each simulation step

epydemix.utils.utils.convert_to_2Darray(lst: List[Any]) → ndarray[source]

Converts a list into a 2D NumPy array.

Parameters:

lst (List[Any]) – A list of elements to be converted into a 2D array. Elements can be of any type.

Returns:

A 2D NumPy array with shape (1, len(lst)), where len(lst) is the length of the input list.

Return type:

np.ndarray

epydemix.utils.utils.create_definitions(parameters: Dict[str, Any], T: int, n_age: int) → Dict[str, ndarray][source]

Generates a dictionary where each value is a 2D array based on the input dictionary and the parameters provided.

Parameters:

  • parameters (Dict[str, Union[np.ndarray, int, float, Iterable]]) – A dictionary where values can be either scalars or iterables.

  • T (int) – The length of the first dimension of the arrays to be created.

  • n_age (int) – The length of the second dimension of the arrays to be created.

Returns:

A dictionary where keys are the same as in parameters and values are 2D arrays of shape (T, n_age).

Return type:

Dict[str, np.ndarray]

Raises:

ValueError – If any parameter value does not meet the required shape criteria.

epydemix.utils.utils.evaluate(expr: str, env: dict) → any[source]

Evaluates the expression with the given environment, allowing only whitelisted operations.

This function uses a private copy of the base evaluation model that also whitelists the ‘Mult’ (multiplication) and ‘Pow’ (power) operations, ensuring that only these operations are permitted during the evaluation.

Parameters:

  • expr (str) – The expression to evaluate. It is expected to be a string containing the mathematical expression to be evaluated.

  • env (dict) – The environment containing variable values. Keys should be variable names and values should be their corresponding numeric values.

Returns:

The result of evaluating the expression. The result type depends on the expression

and its evaluation.

Return type:

any

Raises:

EvalException – If there is an error in evaluating the expression, such as an invalid operation or an undefined variable.

epydemix.utils.utils.format_simulation_output(compartments_evolution: ndarray, transitions_evolution: ndarray, compartments_idx: Dict[str, int], transitions_idx: Dict[str, int], demographics: List[str]) → Dict[str, ndarray][source]

Format simulation output into dictionary with proper naming.

Parameters:

  • compartments_evolution – Array of shape (timesteps, n_compartments, n_demographics)

  • transitions_evolution – Array of shape (timesteps, n_transitions, n_demographics)

  • compartments_idx – Dictionary mapping compartment names to their indices

  • transitions_idx – Dictionary mapping transition names to their indices

  • demographics – List of demographic group names

epydemix.utils.utils.generate_unique_string(length: int = 12) → str[source]

Generates a random unique string containing only letters.

Parameters:

length (int, optional) – The length of the generated string. Defaults to 12.

Returns:

A random unique string containing both uppercase and lowercase letters.

Return type:

str

epydemix.utils.utils.get_initial_conditions_dict(Nk, perc_dict)[source]

Helper function to get initial conditions dictionary from percentage dictionary.

Parameters:

perc_dict (dict) – Dictionary with percentage of population for each compartment

Returns:

Initial conditions dictionary

Return type:

dict

epydemix.utils.utils.is_iterable(value)[source]
epydemix.utils.utils.is_scalar(value)[source]
epydemix.utils.utils.multinomial(n, rates, stay_idx, mask, dt, apply_linear_approximation=False, rng=None, probs_out=None)[source]

Multinomial sample with a ‘stay’ compartment. Probability computation is JIT-compiled via Numba; the draw uses the numpy Generator passed in to preserve reproducibility.

Parameters:

  • n (int) – number of trials

  • rates (np.ndarray) – array of rates

  • stay_idx (int) – index of the stay compartment

  • mask (np.ndarray) – boolean array selecting the ‘leave’ destinations

  • dt (float) – time step size

  • apply_linear_approximation (bool) – whether to apply a linear approximation to the probabilities

  • rng (int, np.random.Generator, or None) – seed or random number generator

  • probs_out (np.ndarray) – unused, kept for API compatibility

Returns:

array of multinomial samples

Return type:

np.ndarray

epydemix.utils.utils.njit(*args, **kwargs)[source]
epydemix.utils.utils.resize_parameter(value: Any, T: int, n_age: int) → ndarray[source]

Resizes the input value to have the shape (T, n_age).

Parameters:

  • value (Union[np.ndarray, int, float]) – The value to be resized, which can be a NumPy array or a scalar.

  • T (int) – The length of the first dimension.

  • n_age (int) – The length of the second dimension.

Returns:

A 2D array with shape (T, n_age).

Return type:

np.ndarray

epydemix.utils.utils.str_to_date(date_str: str) → date[source]

Converts a date string in the format ‘YYYY-MM-DD’ to a datetime.date object.

Parameters:

date_str (str) – A string representing a date in the format ‘YYYY-MM-DD’.

Returns:

A datetime.date object corresponding to the input string.

Return type:

datetime.date

Raises:

ValueError – If the input string is not in the correct format.

epydemix.utils.utils.validate_parameter_shape(key: str, value: Any, T: int, n_age: int) → None[source]

Validates the shape of the input value based on its type and expected dimensions.

Parameters:

  • key (str) – The key associated with the value in the parameters dictionary.

  • value (Union[np.ndarray, Iterable]) – The value to be validated, which can be a NumPy array or an iterable.

  • T (int) – The expected length of the first dimension.

  • n_age (int) – The expected length of the second dimension.

Raises:

ValueError – If the value does not meet the required shape criteria or if the type is unsupported.

Module contents

class epydemix.utils.DefaultPerturbationContinuous(param_name)[source]

Bases: Perturbation

Componenent-wise normal perturbation kernel with adaptive standard deviation (Beaumont et al. (2009)).

pdf(x, center)[source]

Evaluate the PDF of the kernel.

propose(x, rng=None)[source]

Propose a new value based on the current value.

update(particles, weights, param_names)[source]

Update the standard deviation based on previous generation variance.

class epydemix.utils.DefaultPerturbationDiscrete(param_name, prior, jump_probability=0.3)[source]

Bases: Perturbation

pdf(x, center)[source]

Transition probability for the discrete parameter.

propose(x, rng=None)[source]

Propose a new value for the discrete parameter.

update(particles, weights, param_names)[source]

Update jump_probability or other characteristics if needed.

class epydemix.utils.Perturbation(param_name)[source]

Bases: ABC

abstractmethod pdf(x, center)[source]

Evaluate the PDF of the kernel.

abstractmethod propose(x, rng=None)[source]

Propose a new value from the current value x.

Draw randomness from rng (a seed or np.random.Generator) so that proposals are reproducible

abstractmethod update(particles, weights, param_names)[source]

Update the kernel parameters based on particles and weights.

epydemix.utils.combine_simulation_outputs(simulation_outputs_list: List[Dict[str, ndarray]]) → Dict[str, List[ndarray]][source]

Combines multiple simulation outputs into a single dictionary by appending new outputs to existing keys.

Parameters:

simulation_outputs_list (List[Dict[str, np.ndarray]]) – A list of dictionaries containing the latest simulation output to be combined. Each dictionary contains keys corresponding to compartment-demographic names and values are arrays of simulation results.

Returns:

A dictionary where keys are compartment-demographic names and values are lists of arrays.

Each list contains simulation results accumulated from multiple runs.

Return type:

Dict[str, List[np.ndarray]]

epydemix.utils.compute_days(start_date: str | Timestamp, end_date: str | Timestamp) → int[source]

Computes the number of days between two dates.

Parameters:

  • start_date (Union[str, pd.Timestamp]) – The start date in string format (e.g., “YYYY-MM-DD”) or as a pandas Timestamp.

  • end_date (Union[str, pd.Timestamp]) – The end date in string format (e.g., “YYYY-MM-DD”) or as a pandas Timestamp.

Returns:

The number of days between the start date and end date, inclusive.

Return type:

int

epydemix.utils.compute_effective_sample_size(weights: ndarray) → float[source]

Computes the effective sample size (ESS) of a set of weights.

Parameters:

weights (np.ndarray) – A NumPy array of weights for the particles.

Returns:

The effective sample size, calculated as the reciprocal of the sum of squared weights.

Return type:

float

epydemix.utils.compute_simulation_dates(start_date: str | date | datetime64, end_date: str | date | datetime64, steps: int | None = None, dt: float = 1.0) → ndarray[source]

Compute simulation dates including sub-day intervals.

Parameters:

  • start_date – Start date

  • end_date – End date

  • steps – Number of steps (if None, computed from dt)

  • dt – Time step in days (e.g., 1/3 for 8-hour intervals)

Returns:

numpy array of datetime64 objects for each simulation step

epydemix.utils.convert_to_2Darray(lst: List[Any]) → ndarray[source]

Converts a list into a 2D NumPy array.

Parameters:

lst (List[Any]) – A list of elements to be converted into a 2D array. Elements can be of any type.

Returns:

A 2D NumPy array with shape (1, len(lst)), where len(lst) is the length of the input list.

Return type:

np.ndarray

epydemix.utils.get_initial_conditions_dict(Nk, perc_dict)[source]

Helper function to get initial conditions dictionary from percentage dictionary.

Parameters:

perc_dict (dict) – Dictionary with percentage of population for each compartment

Returns:

Initial conditions dictionary

Return type:

dict

epydemix.utils.sample_prior(priors, param_names, rng=None)[source]

Samples a parameter set from the given prior distributions. priors: dictionary mapping parameter names to scipy.stats distributions param_names: list of parameter names to maintain consistent order rng: optional np.random.Generator (or seed) used for sampling Returns: list of sampled parameter values in the order of param_names

epydemix.utils.weighted_quantile(values: ndarray | list, weights: ndarray | list, quantile: float) → float[source]

Compute the weighted quantile of a dataset.

Parameters:

  • values (Union[np.ndarray, list]) – Data values for which the quantile is to be computed.

  • weights (Union[np.ndarray, list]) – Weights associated with the data values.

  • quantile (float) – The quantile to compute, must be between 0 and 1.

Returns:

The weighted quantile value.

Return type:

float

Raises:

ValueError – If quantile is not between 0 and 1.