CLT Base Package Code API Reference

Docstrings and references for clt_base package.

Compartment

Bases: StateVariable

Class for epidemiological compartments (e.g. Susceptible, Exposed, Infected, etc...).

Inherits attributes from StateVariable.

Attributes:

Name	Type	Description
current_inflow	ndarray	same size as self.current_val, used to sum up all transition variable realizations incoming to this compartment for age-risk groups.
current_outflow	ndarray	same size of self.current_val, used to sum up all transition variable realizations outgoing from this compartment for age-risk groups.

Source code in CLT_BaseModel/clt_base/base_components.py

class Compartment(StateVariable):
    """
    Class for epidemiological compartments (e.g. Susceptible,
        Exposed, Infected, etc...).

    Inherits attributes from `StateVariable`.

    Attributes:
        current_inflow (np.ndarray):
            same size as `self.current_val`, used to sum up all
            transition variable realizations incoming to this compartment
            for age-risk groups.
        current_outflow (np.ndarray):
            same size of `self.current_val`, used to sum up all
            transition variable realizations outgoing from this compartment
            for age-risk groups.
    """

    def __init__(self,
                 init_val):
        super().__init__(np.asarray(init_val, dtype=float))

        self.current_inflow = np.zeros(np.shape(init_val))
        self.current_outflow = np.zeros(np.shape(init_val))

    def update_current_val(self) -> None:
        """
        Updates `self.current_val` attribute in-place by adding
            `self.current_inflow` (sum of all incoming transition variables'
            realizations) and subtracting current outflow (sum of all
            outgoing transition variables' realizations).
        """
        self.current_val = self.current_val + self.current_inflow - self.current_outflow

    def reset_inflow(self) -> None:
        """
        Resets `self.current_inflow` attribute to np.ndarray of zeros.
        """
        self.current_inflow = np.zeros(np.shape(self.current_inflow))

    def reset_outflow(self) -> None:
        """
        Resets `self.current_outflow` attribute to np.ndarray of zeros.
        """
        self.current_outflow = np.zeros(np.shape(self.current_outflow))

reset_inflow() -> None

Resets self.current_inflow attribute to np.ndarray of zeros.

Source code in CLT_BaseModel/clt_base/base_components.py

def reset_inflow(self) -> None:
    """
    Resets `self.current_inflow` attribute to np.ndarray of zeros.
    """
    self.current_inflow = np.zeros(np.shape(self.current_inflow))

reset_outflow() -> None

Resets self.current_outflow attribute to np.ndarray of zeros.

Source code in CLT_BaseModel/clt_base/base_components.py

def reset_outflow(self) -> None:
    """
    Resets `self.current_outflow` attribute to np.ndarray of zeros.
    """
    self.current_outflow = np.zeros(np.shape(self.current_outflow))

update_current_val() -> None

Updates self.current_val attribute in-place by adding self.current_inflow (sum of all incoming transition variables' realizations) and subtracting current outflow (sum of all outgoing transition variables' realizations).

Source code in CLT_BaseModel/clt_base/base_components.py

def update_current_val(self) -> None:
    """
    Updates `self.current_val` attribute in-place by adding
        `self.current_inflow` (sum of all incoming transition variables'
        realizations) and subtracting current outflow (sum of all
        outgoing transition variables' realizations).
    """
    self.current_val = self.current_val + self.current_inflow - self.current_outflow

Config

Stores simulation configuration values.

Attributes:

Name	Type	Description
timesteps_per_day	int	number of discretized timesteps within a simulation day -- more timesteps_per_day mean smaller discretization time intervals, which may cause the model to run slower.
transition_type	str	valid value must be from TransitionTypes, specifying the probability distribution of transitions between compartments.
start_real_date	date	actual date that aligns with the beginning of the simulation.
save_daily_history	bool	set to True to save StateVariable state to history after each simulation day -- set to False if want speedier performance.

Source code in CLT_BaseModel/clt_base/base_components.py

@dataclass
class Config:
    """
    Stores simulation configuration values.

    Attributes:
        timesteps_per_day (int):
            number of discretized timesteps within a simulation
            day -- more `timesteps_per_day` mean smaller discretization
            time intervals, which may cause the model to run slower.
        transition_type (str):
            valid value must be from `TransitionTypes`, specifying
            the probability distribution of transitions between
            compartments.
        start_real_date (datetime.date):
            actual date that aligns with the beginning of the simulation.
        save_daily_history (bool):
            set to `True` to save `StateVariable` state to history after each
            simulation day -- set to `False` if want speedier performance.
    """

    timesteps_per_day: int = 7
    transition_type: str = TransitionTypes.BINOMIAL
    start_real_date: datetime.time = datetime.datetime.strptime("2024-10-31",
                                                                "%Y-%m-%d").date()
    save_daily_history: bool = True

DataClassProtocol

Bases: Protocol

Source code in CLT_BaseModel/clt_base/input_parsers.py

class DataClassProtocol(Protocol):
    __dataclass_fields__: dict

DynamicVal

Bases: StateVariable, ABC

Abstract base class for variables that dynamically adjust their values based the current values of other StateVariable instances.

This class should model social distancing (and more broadly, staged-alert policies). For example, if we consider a case where transmission rates decrease when number infected increase above a certain level, we can create a subclass of DynamicVal that models a coefficient that modifies transmission rates, depending on the epi compartments corresponding to infected people.

Inherits attributes from StateVariable.

See __init__ docstring for other attributes.

Source code in CLT_BaseModel/clt_base/base_components.py

class DynamicVal(StateVariable, ABC):
    """
    Abstract base class for variables that dynamically adjust
    their values based the current values of other `StateVariable`
    instances.

    This class should model social distancing (and more broadly,
    staged-alert policies). For example, if we consider a
    case where transmission rates decrease when number infected
    increase above a certain level, we can create a subclass of
    DynamicVal that models a coefficient that modifies transmission
    rates, depending on the epi compartments corresponding to
    infected people.

    Inherits attributes from `StateVariable`.

    See `__init__` docstring for other attributes.
    """

    def __init__(self,
                 init_val: Optional[np.ndarray | float] = None,
                 is_enabled: Optional[bool] = False):
        """

        Args:
            init_val (Optional[np.ndarray | float]):
                starting value(s) at the beginning of the simulation.
            is_enabled (Optional[bool]):
                if `False`, this dynamic value does not get updated
                during the simulation and defaults to its `self.init_val`.
                This is designed to allow easy toggling of
                simulations with or without staged alert policies
                and other interventions.
        """

        super().__init__(init_val)
        self.is_enabled = is_enabled

    @abstractmethod
    def update_current_val(self,
                           state: SubpopState,
                           params: SubpopParams) -> None:
        """
        Args:
            state (SubpopState):
                holds subpopulation simulation state (current values of
                `StateVariable` instances).
            params (SubpopParams):
                holds values of epidemiological parameters.
        """

__init__(init_val: Optional[np.ndarray | float] = None, is_enabled: Optional[bool] = False)

Parameters:

Name	Type	Description	Default
init_val	Optional[ndarray | float]	starting value(s) at the beginning of the simulation.	None
is_enabled	Optional[bool]	if False, this dynamic value does not get updated during the simulation and defaults to its self.init_val. This is designed to allow easy toggling of simulations with or without staged alert policies and other interventions.	False

Source code in CLT_BaseModel/clt_base/base_components.py

def __init__(self,
             init_val: Optional[np.ndarray | float] = None,
             is_enabled: Optional[bool] = False):
    """

    Args:
        init_val (Optional[np.ndarray | float]):
            starting value(s) at the beginning of the simulation.
        is_enabled (Optional[bool]):
            if `False`, this dynamic value does not get updated
            during the simulation and defaults to its `self.init_val`.
            This is designed to allow easy toggling of
            simulations with or without staged alert policies
            and other interventions.
    """

    super().__init__(init_val)
    self.is_enabled = is_enabled

update_current_val(state: SubpopState, params: SubpopParams) -> None

Parameters:

Name	Type	Description	Default
state	SubpopState	holds subpopulation simulation state (current values of StateVariable instances).	required
params	SubpopParams	holds values of epidemiological parameters.	required

Source code in CLT_BaseModel/clt_base/base_components.py

@abstractmethod
def update_current_val(self,
                       state: SubpopState,
                       params: SubpopParams) -> None:
    """
    Args:
        state (SubpopState):
            holds subpopulation simulation state (current values of
            `StateVariable` instances).
        params (SubpopParams):
            holds values of epidemiological parameters.
    """

EpiMetric

Bases: StateVariable, ABC

Abstract base class for epi metrics in epidemiological model.

This is intended for variables that are aggregate deterministic functions of the SubpopState (including Compartment current_val's, other parameters, and time.)

For example, population-level immunity variables should be modeled as a EpiMetric subclass, with a concrete implementation of the abstract method self.get_change_in_current_val.

Inherits attributes from StateVariable.

Attributes:

Name	Type	Description
current_val	ndarray	same size as init_val, holds current value of StateVariable for age-risk groups.
change_in_current_val		(np.ndarray): initialized to None, but during simulation holds change in current value of EpiMetric for age-risk groups (size |A| x |R|, where |A| is the number of risk groups and |R| is number of age groups).

See __init__ docstring for other attributes.

Source code in CLT_BaseModel/clt_base/base_components.py

class EpiMetric(StateVariable, ABC):
    """
    Abstract base class for epi metrics in epidemiological model.

    This is intended for variables that are aggregate deterministic functions of
    the `SubpopState` (including `Compartment` `current_val`'s, other parameters,
    and time.)

    For example, population-level immunity variables should be
    modeled as a `EpiMetric` subclass, with a concrete
    implementation of the abstract method `self.get_change_in_current_val`.

    Inherits attributes from `StateVariable`.

    Attributes:
        current_val (np.ndarray):
            same size as init_val, holds current value of `StateVariable`
            for age-risk groups.
        change_in_current_val : (np.ndarray):
            initialized to None, but during simulation holds change in
            current value of `EpiMetric` for age-risk groups
            (size |A| x |R|, where |A| is the number of risk groups and |R| is number
            of age groups).

    See `__init__` docstring for other attributes.
    """

    def __init__(self,
                 init_val):
        """
        Args:
            init_val (np.ndarray):
                2D array that contains nonnegative floats,
                corresponding to initial value of dynamic val,
                where i,jth entry corresponds to age group i and
                risk group j.
        """

        super().__init__(init_val)

        self.change_in_current_val = None

    @abstractmethod
    def get_change_in_current_val(self,
                                  state: SubpopState,
                                  params: SubpopParams,
                                  num_timesteps: int) -> np.ndarray:
        """
        Computes and returns change in current value of dynamic val,
        based on current state of the simulation and epidemiological parameters.

        NOTE:
            OUTPUT SHOULD ALREADY BE SCALED BY NUM_TIMESTEPS.

        Output should be a numpy array of size |A| x |R|, where A
        is number of age groups and |R| is number of risk groups.

        Args:
            state (SubpopState):
                holds subpopulation simulation state (current values of
                `StateVariable` instances).
            params (SubpopParams):
                holds values of epidemiological parameters.
            num_timesteps (int):
                number of timesteps per day -- used to determine time interval
                length for discretization.

        Returns:
            np.ndarray:
                size |A| x |R|, where |A| is the number of age groups and
                |R| is number of risk groups.
        """
        pass

    def update_current_val(self) -> None:
        """
        Adds `self.change_in_current_val` attribute to
            `self.current_val` attribute in-place.
        """

        self.current_val += self.change_in_current_val

__init__(init_val)

Parameters:

Name	Type	Description	Default
init_val	ndarray	2D array that contains nonnegative floats, corresponding to initial value of dynamic val, where i,jth entry corresponds to age group i and risk group j.	required

Source code in CLT_BaseModel/clt_base/base_components.py

def __init__(self,
             init_val):
    """
    Args:
        init_val (np.ndarray):
            2D array that contains nonnegative floats,
            corresponding to initial value of dynamic val,
            where i,jth entry corresponds to age group i and
            risk group j.
    """

    super().__init__(init_val)

    self.change_in_current_val = None

get_change_in_current_val(state: SubpopState, params: SubpopParams, num_timesteps: int) -> np.ndarray

Computes and returns change in current value of dynamic val, based on current state of the simulation and epidemiological parameters.

NOTE

OUTPUT SHOULD ALREADY BE SCALED BY NUM_TIMESTEPS.

Output should be a numpy array of size |A| x |R|, where A is number of age groups and |R| is number of risk groups.

Parameters:

Name	Type	Description	Default
state	SubpopState	holds subpopulation simulation state (current values of StateVariable instances).	required
params	SubpopParams	holds values of epidemiological parameters.	required
num_timesteps	int	number of timesteps per day -- used to determine time interval length for discretization.	required

Returns:

Type	Description
ndarray	np.ndarray: size |A| x |R|, where |A| is the number of age groups and |R| is number of risk groups.

Source code in CLT_BaseModel/clt_base/base_components.py

@abstractmethod
def get_change_in_current_val(self,
                              state: SubpopState,
                              params: SubpopParams,
                              num_timesteps: int) -> np.ndarray:
    """
    Computes and returns change in current value of dynamic val,
    based on current state of the simulation and epidemiological parameters.

    NOTE:
        OUTPUT SHOULD ALREADY BE SCALED BY NUM_TIMESTEPS.

    Output should be a numpy array of size |A| x |R|, where A
    is number of age groups and |R| is number of risk groups.

    Args:
        state (SubpopState):
            holds subpopulation simulation state (current values of
            `StateVariable` instances).
        params (SubpopParams):
            holds values of epidemiological parameters.
        num_timesteps (int):
            number of timesteps per day -- used to determine time interval
            length for discretization.

    Returns:
        np.ndarray:
            size |A| x |R|, where |A| is the number of age groups and
            |R| is number of risk groups.
    """
    pass

update_current_val() -> None

Adds self.change_in_current_val attribute to self.current_val attribute in-place.

Source code in CLT_BaseModel/clt_base/base_components.py

def update_current_val(self) -> None:
    """
    Adds `self.change_in_current_val` attribute to
        `self.current_val` attribute in-place.
    """

    self.current_val += self.change_in_current_val

Experiment

Class to manage running multiple simulation replications on a SubpopModel or MetapopModel instance and query its results.

Also allows running a batch of simulation replications on a deterministic sequence of values for a given input (for example, to see how output changes as a function of a given input).

Also handles random sampling of inputs from a uniform distribution.

NOTE

If an input is an |A| x |R| array (for age-risk), the current functionality does not support sampling individual age-risk elements separately. Instead, a single scalar value is sampled at a time for the entire input. See self.sample_random_inputs method for more details.

Parameters:

Name	Type	Description	Default
experiment_subpop_models	tuple	tuple of SubpopModel instances associated with the Experiment. If the Experiment is for a MetapopModel, then this tuple contains all the associated SubpopModel instances that comprise that MetapopModel. If the Experiment is for a SubpopModel only, then this tuple contains only that particular SubpopModel.	required
inputs_realizations	dict	dictionary of dictionaries that stores user-specified deterministic sequences for inputs or realizations of random input sampling -- keys are SubpopModel names, values are dictionaries. These second-layer dictionaries' keys are strings corresponding to input names (they must match names in each SubpopModel's SubpopParams to be valid) and values are list-like, where the ith element corresponds to the ith random sample for that input.	required
results_df	DataFrame	DataFrame holding simulation results from each simulation replication	required
has_been_run	bool	indicates if self.run_static_inputs, self.run_random_inputs, or self.run_sequences_of_inputs has been executed.	required

See __init__ docstring for other attributes.

Source code in CLT_BaseModel/clt_base/experiments.py

class Experiment:
    """
    Class to manage running multiple simulation replications
    on a `SubpopModel` or `MetapopModel` instance and query its results.

    Also allows running a batch of simulation replications on a
    deterministic sequence of values for a given input
    (for example, to see how output changes as a function of
    a given input).

    Also handles random sampling of inputs from a uniform
    distribution.

    NOTE:
        If an input is an |A| x |R| array (for age-risk),
        the current functionality does not support sampling individual
        age-risk elements separately. Instead, a single scalar value
        is sampled at a time for the entire input.
        See `self.sample_random_inputs` method for more details.

    Params:
        experiment_subpop_models (tuple):
            tuple of `SubpopModel` instances associated with the `Experiment`.
            If the `Experiment` is for a `MetapopModel`, then this tuple
            contains all the associated `SubpopModel` instances
            that comprise that `MetapopModel.` If the `Experiment` is for
            a `SubpopModel` only, then this tuple contains only that
            particular `SubpopModel`.
        inputs_realizations (dict):
            dictionary of dictionaries that stores user-specified deterministic
            sequences for inputs or realizations of random input sampling --
            keys are `SubpopModel` names, values are dictionaries.
            These second-layer dictionaries' keys are strings corresponding
            to input names (they must match names in each `SubpopModel`'s
            `SubpopParams` to be valid) and values are list-like, where the ith
            element corresponds to the ith random sample for that input.
        results_df (pd.DataFrame):
            DataFrame holding simulation results from each
            `simulation` replication
        has_been_run (bool):
            indicates if `self.run_static_inputs`, `self.run_random_inputs`,
            or `self.run_sequences_of_inputs` has been executed.

    See `__init__` docstring for other attributes.
    """

    def __init__(self,
                 model: SubpopModel | MetapopModel,
                 state_variables_to_record: list,
                 database_filename: str):

        """
        Params:
            model (SubpopModel | MetapopModel):
                SubpopModel or MetapopModel instance on which to
                run multiple replications.
            state_variables_to_record (list[str]):
                list or list-like of strings corresponding to
                state variables to record -- each string must match
                a state variable name on each SubpopModel in
                the MetapopModel.
            database_filename (str):
                must be valid filename with suffix ".db" --
                experiment results are saved to this SQL database
        """

        self.model = model
        self.state_variables_to_record = state_variables_to_record
        self.database_filename = database_filename

        self.has_been_run = False

        # Create experiment_subpop_models tuple
        # If model is MetapopModel instance, then this tuple is a list
        #   of all associated SubpopModel instances
        # If model is a SubpopModel instance, then this tuple
        #   only contains that SubpopModel.
        if isinstance(model, MetapopModel):
            experiment_subpop_models = tuple(model.subpop_models.values())
        elif isinstance(model, SubpopModel):
            experiment_subpop_models = (model,)
        else:
            raise ExperimentError("\"model\" argument must be an instance of SubpopModel "
                                  "or MetapopModel class.")
        self.experiment_subpop_models = experiment_subpop_models

        # Initialize results_df attribute -- this will store
        #   results of experiment run
        self.results_df = None

        # User-specified deterministic sequences for inputs or randomly
        #   sampled inputs' realizations are stored in this dictionary
        self.inputs_realizations = {}

        for subpop_model in self.experiment_subpop_models:
            self.inputs_realizations[subpop_model.name] = {}

            # Make sure the state variables to record are valid -- the names
            #   of the state variables to record must match actual state variables
            #   on each SubpopModel
            if not check_is_subset_list(state_variables_to_record,
                                        subpop_model.all_state_variables.keys()):
                raise ExperimentError(
                    f"\"state_variables_to_record\" list is not a subset "
                    "of the state variables on SubpopModel \"{subpop_name}\" -- "
                    "modify \"state_variables_to_record\" and re-initialize experiment.")

    def run_static_inputs(self,
                          num_reps: int,
                          simulation_end_day: int,
                          days_between_save_history: int = 1,
                          results_filename: str = None):
        """
        Runs the associated `SubpopModel` or `MetapopModel` for a
        given number of independent replications until `simulation_end_day`.
        Parameter values and initial values are the same across
        simulation replications. User can specify how often to save the
        history and a CSV file in which to store this history.

        Params:
            num_reps (positive int):
                number of independent simulation replications
                to run in an experiment.
            simulation_end_day (positive int):
                stop simulation at simulation_end_day (i.e. exclusive,
                simulate up to but not including simulation_end_day).
            days_between_save_history (positive int):
                indicates how often to save simulation results.
            results_filename (str):
                if specified, must be valid filename with suffix ".csv" --
                experiment results are saved to this CSV file.
        """

        if self.has_been_run:
            raise ExperimentError("Experiment has already been run. "
                                  "Create a new Experiment instance to simulate "
                                  "more replications.")

        else:
            self.has_been_run = True

            self.create_results_sql_table()

            self.simulate_reps_and_save_results(reps=num_reps,
                                                end_day=simulation_end_day,
                                                days_per_save=days_between_save_history,
                                                inputs_are_static=True,
                                                filename=results_filename)

    def run_random_inputs(self,
                          num_reps: int,
                          simulation_end_day: int,
                          random_inputs_RNG: np.random.Generator,
                          random_inputs_spec: dict,
                          days_between_save_history: int = 1,
                          results_filename: str = None,
                          inputs_filename_suffix: str = None):
        """
        Runs the associated `SubpopModel` or `MetapopModel` for a
        given number of independent replications until `simulation_end_day`,
        where certain parameter values or initial values are
        independently randomly sampled for each replication.
        Random sampling details (which inputs to sample and what
        Uniform distribution lower and upper bounds to use for each input)
        are specified in `random_inputs_spec`, and random sampling of inputs
        uses `random_inputs_RNG`. User can specify how often to save the
        history and a CSV file in which to store this history.
        User can also specify a filename suffix to name CSV files
        in which to store realizations of inputs.

        Params:
            num_reps (positive int):
                number of independent simulation replications
                to run in an experiment.
            simulation_end_day (positive int):
                stop simulation at simulation_end_day (i.e. exclusive,
                simulate up to but not including simulation_end_day).
            random_inputs_RNG (np.random.Generator):
                random number generator used to sample random
                inputs -- for reproducibility, it is recommended to
                use a distinct RNG for sampling random inputs different
                from the RNG for simulating the experiment/model.
            random_inputs_spec (dict):
                random inputs' specification -- stores details
                for random input sampling -- keys are strings
                corresponding to input names (they must match
                names in each `SubpopModel`'s `SubpopParams` or associated
                `EpiMetric` or `Compartment` instances to be valid)
                and values are 2-tuples of nonnegative floats corresponding
                to lower and upper bounds for sampling that input from a
                uniform distribution.
            days_between_save_history (positive int):
                indicates how often to save simulation results.
            results_filename (str):
                if specified, must be valid filename with suffix ".csv" --
                experiment results are saved to this CSV file
            inputs_filename_suffix (str):
                if specified, must be a valid filename with suffix ".csv" --
                one inputs CSV is generated for each `SubpopModel` with the
                filename "{name}_{inputs_filename}" where name is the name
                of the `SubpopModel` -- saves the values of each parameter that
                varies between replications
        """

        if self.has_been_run:
            raise ExperimentError("Experiment has already been run. "
                                  "Create a new Experiment instance to simulate "
                                  "more replications.")

        else:
            self.has_been_run = True

            self.sample_random_inputs(num_reps,
                                      random_inputs_RNG,
                                      random_inputs_spec)

            self.create_results_sql_table()
            self.create_inputs_realizations_sql_tables()

            self.simulate_reps_and_save_results(reps=num_reps,
                                                end_day=simulation_end_day,
                                                days_per_save=days_between_save_history,
                                                inputs_are_static=False,
                                                filename=results_filename)

            if inputs_filename_suffix:
                self.write_inputs_csvs(inputs_filename_suffix)

    def run_sequences_of_inputs(self,
                                num_reps: int,
                                simulation_end_day: int,
                                sequences_of_inputs: dict,
                                days_between_save_history: int = 1,
                                results_filename: str = None,
                                inputs_filename_suffix: str = None):
        """
        Runs the associated `SubpopModel` or `MetapopModel` for a
        given number of independent replications until `simulation_end_day`,
        where certain parameter values or initial values deterministically
        change between replications, according to `sequences_of_inputs`.
        User can specify how often to save the history and a CSV file
        in which to store this history. User can also specify a filename
        suffix to name CSV files in which to store sequences of inputs.

        Params:
            num_reps (positive int):
                number of independent simulation replications
                to run in an experiment.
            simulation_end_day (positive int):
                stop simulation at `simulation_end_day` (i.e. exclusive,
                simulate up to but not including `simulation_end_day`).
            sequences_of_inputs (dict):
                dictionary of dictionaries that stores user-specified deterministic
                sequences for inputs -- must follow specific structure.
                Keys are `SubpopModel` names, values are dictionaries.
                These second-layer dictionaries' keys are strings corresponding
                to input names (they must match names in each `SubpopModel`'s
                `SubpopParams` to be valid) and values are list-like, where the ith
                element corresponds to the ith random sample for that input.
            days_between_save_history (positive int):
                indicates how often to save simulation results.
            results_filename (str):
                if specified, must be valid filename with suffix ".csv" --
                experiment results are saved to this CSV file
            inputs_filename_suffix (str):
                if specified, must be a valid filename with suffix ".csv" --
                one inputs CSV is generated for each `SubpopModel` with the
                filename "{name}_{inputs_filename}" where name is the name
                of the `SubpopModel` -- saves the values of each parameter that
                varies between replications
        """

        if self.has_been_run:
            raise ExperimentError("Experiment has already been run. "
                                  "Create a new Experiment instance to simulate "
                                  "more replications.")

        else:
            self.has_been_run = True

            self.inputs_realizations = sequences_of_inputs

            self.create_results_sql_table()
            self.create_inputs_realizations_sql_tables()

            self.simulate_reps_and_save_results(reps=num_reps,
                                                end_day=simulation_end_day,
                                                days_per_save=days_between_save_history,
                                                inputs_are_static=False,
                                                filename=results_filename)

            if inputs_filename_suffix:
                self.write_inputs_csvs(inputs_filename_suffix)

    def sample_random_inputs(self,
                             total_reps: int,
                             RNG: np.random.Generator,
                             spec: dict):
        """
        Randomly and independently samples inputs specified by keys of
        `spec` according to uniform distribution with lower
        and upper bounds specified by values of `spec`.
        Stores random realizations in `self.inputs_realizations` attribute.
        Uses `RNG` to sample inputs.

        Params:
            total_reps (positive int):
                number of independent simulation replications
                to run in an experiment -- corresponds to number of
                Uniform random variables to draw for each
                state variable.
            RNG (np.random.Generator):
                random number generator used to sample random
                inputs -- for reproducibility, it is recommended to
                use a distinct RNG for sampling random inputs different
                from the RNG for simulating the experiment/model.
            spec (dict):
                random inputs' specification -- stores details
                for random input sampling -- keys are strings
                corresponding to input names (they must match
                names in each `SubpopModel`'s `SubpopParams` or associated
                `EpiMetric` or `Compartment` instances to be valid)
                and values are 2-tuples of nonnegative floats corresponding
                to lower and upper bounds for sampling that input from a
                uniform distribution.

        NOTE:
            If an input is an |A| x |R| array (for age-risk),
            the current functionality does not support sampling individual
            age-risk elements separately. Instead, a single scalar value
            is sampled at a time for the entire input. Consequently,
            if an |A| x |R| input is chosen to be randomly sampled,
            all its elements will have the same sampled value.

            If a user wants to sample some age-risk elements separately,
            they should create new inputs for these elements. "Inputs"
            refers to both parameters (in `SubpopParams`) and initial values
            of `Compartment` and `EpiMetric` instances. For example, if the model
            has a parameter `H_to_R_rate` that is 2x1 (2 age groups, 1 risk group)
            and the user wants to sample each element separately, they should create
            two parameters: `H_to_R_rate_age_group_1` and `H_to_R_rate_age_group_2.`
            These should be added to the relevant `SubpopParams` instance and
            input dictionary/file used to create the `SubpopParams` instance.
            The user can then specify both parameters to be randomly sampled
            and specify the lower and upper bounds accordingly.

            TODO: allow sampling individual age-risk elements separately
            without creating new parameters for each element.

            (Developer note: the difficulty is not with randomly sampling
            arrays, but rather storing arrays in SQL -- SQL tables only
            support atomic values.)
        """

        inputs_realizations = self.inputs_realizations

        for subpop_model in self.experiment_subpop_models:

            subpop_name = subpop_model.name

            compartments_names_list = list(subpop_model.compartments.keys())
            epi_metrics_names_list = list(subpop_model.epi_metrics.keys())
            params_names_list = [field.name for field in fields(subpop_model.params)]

            if not check_is_subset_list(spec[subpop_name],
                                        compartments_names_list + epi_metrics_names_list + params_names_list):
                raise (f"\"random_inputs_spec[\"{subpop_name}\"]\" keys are not a subset "
                       "of the state variables and parameters on SubpopModel \"{subpop_name}\" -- "
                       "modify \"random_inputs_spec\" and re-initialize experiment.")

            for state_var_name, bounds in spec[subpop_name].items():
                lower_bd = bounds[0]
                upper_bd = bounds[1]
                inputs_realizations[subpop_name][state_var_name] = \
                    RNG.uniform(low=lower_bd,
                                high=upper_bd,
                                size=total_reps)

    def get_state_var_df(self,
                         state_var_name: str,
                         subpop_name: str = None,
                         age_group: int = None,
                         risk_group: int = None,
                         results_filename: str = None) -> pd.DataFrame:
        """
        Get pandas DataFrame of recorded values of `StateVariable` given by
        `state_var_name`, in the `SubpopModel` given by `subpop_name`,
        for the age-risk group given by `age_group` and `risk_group`.
        If `subpop_name` is not specified, then values are summed across all
        associated subpopulations. Similarly, if `age_group` (or `risk_group`)
        is not specified, then values are summed across all age groups
        (or risk groups).

        Args:
            state_var_name (str):
                Name of the `StateVariable` to retrieve.
            subpop_name (Optional[str]):
                The name of the `SubpopModel` for filtering. If None, values are
                summed across all `SubpopModel` instances.
            age_group (Optional[int]):
                The age group to select. If None, values are summed across
                all age groups.
            risk_group (Optional[int]):
                The risk group to select. If None, values are summed across
                all risk groups.
            results_filename (Optional[str]):
                If provided, saves the resulting DataFrame as a CSV.

        Returns:
            A pandas DataFrame where rows represent the replication and columns indicate the
            simulation day (timepoint) of recording. DataFrame values are the `StateVariable`'s
            current_val or the sum of the `StateVariable`'s current_val across subpopulations,
            age groups, or risk groups (the combination of what is summed over is
            specified by the user -- details are in the part of this docstring describing
            this function's parameters).
        """

        if state_var_name not in self.state_variables_to_record:
            raise ExperimentError("\"state_var_name\" is not in \"self.state_variables_to_record\" --"
                                  "function call is invalid.")

        conn = sqlite3.connect(self.database_filename)

        # Query all results table entries where state_var_name matches
        # This will return results across all subpopulations, age groups,
        #   and risk groups
        df = get_sql_table_as_df(conn,
                                 "SELECT * FROM results WHERE state_var_name = ?",
                                 chunk_size=int(1e4),
                                 sql_query_params=(state_var_name,))

        conn.close()

        # Define filter conditions
        filters = {
            "subpop_name": subpop_name,
            "age_group": age_group,
            "risk_group": risk_group
        }

        # Filter DataFrame based on user-specified conditions
        #   (for example, if user specifies subpop_name, return subset of
        #   DataFrame where subpop_name matches)
        conditions = [(df[col] == value) for col, value in filters.items() if value is not None]
        df_filtered = df if not conditions else df[np.logical_and.reduce(conditions)]

        # Group DataFrame based on unique combinations of "rep" and "timepoint" columns
        # Then sum (numeric values only), return the "value" column, and reset the index
        #   so that "rep" and "timepoint" become regular columns and are not the index
        df_aggregated = \
            df_filtered.groupby(["rep",
                                 "timepoint"]).sum(numeric_only=True)["value"].reset_index()

        # breakpoint()

        # Use pivot() function to reshape the DataFrame for its final form
        # The "timepoint" values are spread across new columns
        #   (creating a column for each unique timepoint).
        # The "value" column populates the corresponding cells.
        df_final = df_aggregated.pivot(index="rep",
                                       columns="timepoint",
                                       values="value")

        if results_filename:
            df_final.to_csv(results_filename)

        return df_final

    def log_current_vals_to_sql(self,
                                rep_counter: int,
                                experiment_cursor: sqlite3.Cursor) -> None:
        """
        For each subpopulation and state variable to record
        associated with this `Experiment`, save current values to
        "results" table in SQL database specified by `experiment_cursor`.

        Params:
            rep_counter (int):
                Current replication ID.
            experiment_cursor (sqlite3.Cursor):
                Cursor object connected to the database
                where results should be inserted.
        """

        for subpop_model in self.experiment_subpop_models:
            for state_var_name in self.state_variables_to_record:
                data = format_current_val_for_sql(subpop_model,
                                                  state_var_name,
                                                  rep_counter)
                experiment_cursor.executemany(
                    "INSERT INTO results VALUES (?, ?, ?, ?, ?, ?, ?)", data)

    def log_inputs_to_sql(self,
                          experiment_cursor: sqlite3.Cursor):
        """
        For each subpopulation, add a new table to SQL
        database specified by `experiment_cursor`. Each table
        contains information on inputs that vary across
        replications (either due to random sampling or
        user-specified deterministic sequence). Each table
        contains inputs information from `Experiment` attribute
        `self.inputs_realizations` for a given subpopulation.

        Params:
            experiment_cursor (sqlite3.Cursor):
                Cursor object connected to the database
                where results should be inserted.
        """

        for subpop_model in self.experiment_subpop_models:
            table_name = f'"{subpop_model.name}_inputs"'

            # Get the column names (dynamically, based on table)
            experiment_cursor.execute(f"PRAGMA table_info({table_name})")

            # Extract column names from the table info
            # But exclude the column name "rep"
            columns_info = experiment_cursor.fetchall()
            column_names = [col[1] for col in columns_info if col[1] != "rep"]

            # Create a placeholder string for the dynamic query
            placeholders = ", ".join(["?" for _ in column_names])  # Number of placeholders matches number of columns

            # Create the dynamic INSERT statement
            sql_statement = f"INSERT INTO {table_name} ({', '.join(column_names)}) VALUES ({placeholders})"

            # Create list of lists -- each nested list contains a sequence of values
            #   for that particular input
            subpop_inputs_realizations = self.inputs_realizations[subpop_model.name]
            inputs_vals_over_reps_list = \
                [np.array(subpop_inputs_realizations[input_name]).reshape(-1,1) for input_name in column_names]
            inputs_vals_over_reps_list = np.hstack(inputs_vals_over_reps_list)

            experiment_cursor.executemany(sql_statement, inputs_vals_over_reps_list)

    def apply_inputs_to_model(self,
                              rep_counter: int):
        """
        Changes inputs (parameters or initial values) for a given
        replication according to `self.inputs_realizations` attribute.

        Specifically, for each subpopulation, this function retrieves
        the corresponding values from `self.inputs_realizations` and applies them to either:
        (a) the subpopulation's `StateVariable`, if the input corresponds to a state variable,
        or (b) the subpopulation's `params` attribute, if the input corresponds to a model
        parameter. If the parameter or state variable is multidimensional
        (e.g., varies across age or risk groups), it is assigned a numpy array of
        the appropriate shape with the replicated input value.

        Params:
            rep_counter (int):
                Replication ID, used to retrieve the correct
                realizations of inputs for the current run.
        """

        for subpop_model in self.experiment_subpop_models:

            params = subpop_model.params

            for input_name, input_val in self.inputs_realizations[subpop_model.name].items():

                dimensions = (params.num_age_groups, params.num_risk_groups)

                if input_name in subpop_model.all_state_variables.keys():
                    subpop_model.all_state_variables[input_name].current_val = np.full(dimensions,
                                                                                       input_val[rep_counter])
                else:
                    if np.isscalar(getattr(subpop_model.params, input_name)):
                        setattr(subpop_model.params, input_name, input_val[rep_counter])
                    else:
                        setattr(subpop_model.params, input_name, np.full(dimensions, input_val[rep_counter]))

    def simulate_reps_and_save_results(self,
                                       reps: int,
                                       end_day: int,
                                       days_per_save: int,
                                       inputs_are_static: bool,
                                       filename: str = None):
        """
        Helper function that executes main loop over
        replications in `Experiment` and saves results.

        Params:
            reps (int):
                number of independent simulation replications
                to run in an experiment.
            end_day (int):
                stop simulation at end_day (i.e. exclusive,
                simulate up to but not including end_day).
            days_per_save (int):
                indicates how often to save simulation results.
            inputs_are_static (bool):
                indicates if inputs are same across replications.
            filename (str):
                if specified, must be valid filename with suffix ".csv" --
                experiment results are saved to this CSV file.
        """

        # Override each subpop config's save_daily_history attribute --
        #   set it to False -- because we will manually save history
        #   to results database according to user-defined
        #   days_between_save_history for all subpops
        for subpop_model in self.experiment_subpop_models:
            subpop_model.config.save_daily_history = False

        model = self.model

        # Connect to SQL database
        conn = sqlite3.connect(self.database_filename)
        cursor = conn.cursor()

        # Loop through replications
        for rep in range(reps):

            # Reset model and clear its history
            model.reset_simulation()

            # Apply new values of inputs, if some
            #   inputs change between replications
            if not inputs_are_static:
                self.apply_inputs_to_model(rep)
                self.log_inputs_to_sql(cursor)

            # Simulate model and save results every `days_per_save` days
            while model.current_simulation_day < end_day:
                model.simulate_until_day(min(model.current_simulation_day + days_per_save,
                                             end_day))

                self.log_current_vals_to_sql(rep, cursor)

        self.results_df = get_sql_table_as_df(conn, "SELECT * FROM results", chunk_size=int(1e4))

        if filename:
            self.results_df.to_csv(filename)

        # Commit changes to database and close
        conn.commit()
        conn.close()

    def create_results_sql_table(self):
        """
        Create SQL database and save to `self.database_filename`.
        Create table named `results` with columns `subpop_name`,
        `state_var_name`, `age_group`, `risk_group`, `rep`, `timepoint`,
        and `value` to store results from each replication of experiment.
        """

        # Make sure user is not overwriting database
        if os.path.exists(self.database_filename):
            raise ExperimentError("Database already exists! Overwriting is not allowed. "
                                  "Delete existing .db file or change database_filename "
                                  "attribute.")

        # Connect to the SQLite database and create database
        # Create a cursor object to execute SQL commands
        # Initialize a table with columns given by column_names
        # Commit changes and close the connection
        conn = sqlite3.connect(self.database_filename)
        cursor = conn.cursor()
        cursor.execute("""
        CREATE TABLE IF NOT EXISTS results (
            subpop_name TEXT,
            state_var_name TEXT,
            age_group INT,
            risk_group INT,
            rep INT,
            timepoint INT,
            value FLOAT,
            PRIMARY KEY (subpop_name, state_var_name, age_group, risk_group, rep, timepoint)
        )
        """)
        conn.commit()
        conn.close()

    def create_inputs_realizations_sql_tables(self):
        """
        Create tables in SQL database given by `self.database_filename`
        to store realizations of inputs that change across
        replications. There is one table per associated subpopulation.
        """

        conn = sqlite3.connect(self.database_filename)
        cursor = conn.cursor()

        for subpop_name in self.inputs_realizations.keys():
            table_name = f'"{subpop_name}_inputs"'

            column_names = self.inputs_realizations[subpop_name].keys()

            # Construct the column definitions dynamically
            column_definitions = ", ".join([f'"{col}" FLOAT' for col in column_names])

            # SQL statement to create table
            # Each table has a column called "rep" which is the primary
            #   key and automatically increments every time a row is added
            sql_statement = f"""
            CREATE TABLE IF NOT EXISTS {table_name} (
                rep INTEGER PRIMARY KEY AUTOINCREMENT,
                {column_definitions}
            )
            """

            cursor.execute(sql_statement)

        conn.commit()
        conn.close()

    def write_inputs_csvs(self,
                          suffix: str):
        """
        For each subpopulation, writes a CSV (with a filename
        based on provided suffix) containing values of inputs
        across replications.

        Params:
            suffix (str):
                Common suffix used to generate CSV filenames.
        """

        for subpop_model in self.experiment_subpop_models:
            conn = sqlite3.connect(self.database_filename)

            table_name = f"{subpop_model.name}_inputs"

            subpop_inputs_df = get_sql_table_as_df(conn, f"SELECT * FROM {table_name}", chunk_size=int(1e4))

            subpop_inputs_df.to_csv(f"{subpop_model.name}_{suffix}", index=False)

__init__(model: SubpopModel | MetapopModel, state_variables_to_record: list, database_filename: str)

Parameters:

Name	Type	Description	Default
model	SubpopModel | MetapopModel	SubpopModel or MetapopModel instance on which to run multiple replications.	required
state_variables_to_record	list[str]	list or list-like of strings corresponding to state variables to record -- each string must match a state variable name on each SubpopModel in the MetapopModel.	required
database_filename	str	must be valid filename with suffix ".db" -- experiment results are saved to this SQL database	required

Source code in CLT_BaseModel/clt_base/experiments.py

def __init__(self,
             model: SubpopModel | MetapopModel,
             state_variables_to_record: list,
             database_filename: str):

    """
    Params:
        model (SubpopModel | MetapopModel):
            SubpopModel or MetapopModel instance on which to
            run multiple replications.
        state_variables_to_record (list[str]):
            list or list-like of strings corresponding to
            state variables to record -- each string must match
            a state variable name on each SubpopModel in
            the MetapopModel.
        database_filename (str):
            must be valid filename with suffix ".db" --
            experiment results are saved to this SQL database
    """

    self.model = model
    self.state_variables_to_record = state_variables_to_record
    self.database_filename = database_filename

    self.has_been_run = False

    # Create experiment_subpop_models tuple
    # If model is MetapopModel instance, then this tuple is a list
    #   of all associated SubpopModel instances
    # If model is a SubpopModel instance, then this tuple
    #   only contains that SubpopModel.
    if isinstance(model, MetapopModel):
        experiment_subpop_models = tuple(model.subpop_models.values())
    elif isinstance(model, SubpopModel):
        experiment_subpop_models = (model,)
    else:
        raise ExperimentError("\"model\" argument must be an instance of SubpopModel "
                              "or MetapopModel class.")
    self.experiment_subpop_models = experiment_subpop_models

    # Initialize results_df attribute -- this will store
    #   results of experiment run
    self.results_df = None

    # User-specified deterministic sequences for inputs or randomly
    #   sampled inputs' realizations are stored in this dictionary
    self.inputs_realizations = {}

    for subpop_model in self.experiment_subpop_models:
        self.inputs_realizations[subpop_model.name] = {}

        # Make sure the state variables to record are valid -- the names
        #   of the state variables to record must match actual state variables
        #   on each SubpopModel
        if not check_is_subset_list(state_variables_to_record,
                                    subpop_model.all_state_variables.keys()):
            raise ExperimentError(
                f"\"state_variables_to_record\" list is not a subset "
                "of the state variables on SubpopModel \"{subpop_name}\" -- "
                "modify \"state_variables_to_record\" and re-initialize experiment.")

apply_inputs_to_model(rep_counter: int)

Changes inputs (parameters or initial values) for a given replication according to self.inputs_realizations attribute.

Specifically, for each subpopulation, this function retrieves the corresponding values from self.inputs_realizations and applies them to either: (a) the subpopulation's StateVariable, if the input corresponds to a state variable, or (b) the subpopulation's params attribute, if the input corresponds to a model parameter. If the parameter or state variable is multidimensional (e.g., varies across age or risk groups), it is assigned a numpy array of the appropriate shape with the replicated input value.

Parameters:

Name	Type	Description	Default
rep_counter	int	Replication ID, used to retrieve the correct realizations of inputs for the current run.	required

Source code in CLT_BaseModel/clt_base/experiments.py

def apply_inputs_to_model(self,
                          rep_counter: int):
    """
    Changes inputs (parameters or initial values) for a given
    replication according to `self.inputs_realizations` attribute.

    Specifically, for each subpopulation, this function retrieves
    the corresponding values from `self.inputs_realizations` and applies them to either:
    (a) the subpopulation's `StateVariable`, if the input corresponds to a state variable,
    or (b) the subpopulation's `params` attribute, if the input corresponds to a model
    parameter. If the parameter or state variable is multidimensional
    (e.g., varies across age or risk groups), it is assigned a numpy array of
    the appropriate shape with the replicated input value.

    Params:
        rep_counter (int):
            Replication ID, used to retrieve the correct
            realizations of inputs for the current run.
    """

    for subpop_model in self.experiment_subpop_models:

        params = subpop_model.params

        for input_name, input_val in self.inputs_realizations[subpop_model.name].items():

            dimensions = (params.num_age_groups, params.num_risk_groups)

            if input_name in subpop_model.all_state_variables.keys():
                subpop_model.all_state_variables[input_name].current_val = np.full(dimensions,
                                                                                   input_val[rep_counter])
            else:
                if np.isscalar(getattr(subpop_model.params, input_name)):
                    setattr(subpop_model.params, input_name, input_val[rep_counter])
                else:
                    setattr(subpop_model.params, input_name, np.full(dimensions, input_val[rep_counter]))

create_inputs_realizations_sql_tables()

Create tables in SQL database given by self.database_filename to store realizations of inputs that change across replications. There is one table per associated subpopulation.

Source code in CLT_BaseModel/clt_base/experiments.py

def create_inputs_realizations_sql_tables(self):
    """
    Create tables in SQL database given by `self.database_filename`
    to store realizations of inputs that change across
    replications. There is one table per associated subpopulation.
    """

    conn = sqlite3.connect(self.database_filename)
    cursor = conn.cursor()

    for subpop_name in self.inputs_realizations.keys():
        table_name = f'"{subpop_name}_inputs"'

        column_names = self.inputs_realizations[subpop_name].keys()

        # Construct the column definitions dynamically
        column_definitions = ", ".join([f'"{col}" FLOAT' for col in column_names])

        # SQL statement to create table
        # Each table has a column called "rep" which is the primary
        #   key and automatically increments every time a row is added
        sql_statement = f"""
        CREATE TABLE IF NOT EXISTS {table_name} (
            rep INTEGER PRIMARY KEY AUTOINCREMENT,
            {column_definitions}
        )
        """

        cursor.execute(sql_statement)

    conn.commit()
    conn.close()

create_results_sql_table()

Create SQL database and save to self.database_filename. Create table named results with columns subpop_name, state_var_name, age_group, risk_group, rep, timepoint, and value to store results from each replication of experiment.

Source code in CLT_BaseModel/clt_base/experiments.py

def create_results_sql_table(self):
    """
    Create SQL database and save to `self.database_filename`.
    Create table named `results` with columns `subpop_name`,
    `state_var_name`, `age_group`, `risk_group`, `rep`, `timepoint`,
    and `value` to store results from each replication of experiment.
    """

    # Make sure user is not overwriting database
    if os.path.exists(self.database_filename):
        raise ExperimentError("Database already exists! Overwriting is not allowed. "
                              "Delete existing .db file or change database_filename "
                              "attribute.")

    # Connect to the SQLite database and create database
    # Create a cursor object to execute SQL commands
    # Initialize a table with columns given by column_names
    # Commit changes and close the connection
    conn = sqlite3.connect(self.database_filename)
    cursor = conn.cursor()
    cursor.execute("""
    CREATE TABLE IF NOT EXISTS results (
        subpop_name TEXT,
        state_var_name TEXT,
        age_group INT,
        risk_group INT,
        rep INT,
        timepoint INT,
        value FLOAT,
        PRIMARY KEY (subpop_name, state_var_name, age_group, risk_group, rep, timepoint)
    )
    """)
    conn.commit()
    conn.close()

get_state_var_df(state_var_name: str, subpop_name: str = None, age_group: int = None, risk_group: int = None, results_filename: str = None) -> pd.DataFrame

Get pandas DataFrame of recorded values of StateVariable given by state_var_name, in the SubpopModel given by subpop_name, for the age-risk group given by age_group and risk_group. If subpop_name is not specified, then values are summed across all associated subpopulations. Similarly, if age_group (or risk_group) is not specified, then values are summed across all age groups (or risk groups).

Parameters:

Name	Type	Description	Default
state_var_name	str	Name of the StateVariable to retrieve.	required
subpop_name	Optional[str]	The name of the SubpopModel for filtering. If None, values are summed across all SubpopModel instances.	None
age_group	Optional[int]	The age group to select. If None, values are summed across all age groups.	None
risk_group	Optional[int]	The risk group to select. If None, values are summed across all risk groups.	None
results_filename	Optional[str]	If provided, saves the resulting DataFrame as a CSV.	None

Returns:

Type	Description
DataFrame	A pandas DataFrame where rows represent the replication and columns indicate the
DataFrame	simulation day (timepoint) of recording. DataFrame values are the StateVariable's
DataFrame	current_val or the sum of the StateVariable's current_val across subpopulations,
DataFrame	age groups, or risk groups (the combination of what is summed over is
DataFrame	specified by the user -- details are in the part of this docstring describing
DataFrame	this function's parameters).

Source code in CLT_BaseModel/clt_base/experiments.py

def get_state_var_df(self,
                     state_var_name: str,
                     subpop_name: str = None,
                     age_group: int = None,
                     risk_group: int = None,
                     results_filename: str = None) -> pd.DataFrame:
    """
    Get pandas DataFrame of recorded values of `StateVariable` given by
    `state_var_name`, in the `SubpopModel` given by `subpop_name`,
    for the age-risk group given by `age_group` and `risk_group`.
    If `subpop_name` is not specified, then values are summed across all
    associated subpopulations. Similarly, if `age_group` (or `risk_group`)
    is not specified, then values are summed across all age groups
    (or risk groups).

    Args:
        state_var_name (str):
            Name of the `StateVariable` to retrieve.
        subpop_name (Optional[str]):
            The name of the `SubpopModel` for filtering. If None, values are
            summed across all `SubpopModel` instances.
        age_group (Optional[int]):
            The age group to select. If None, values are summed across
            all age groups.
        risk_group (Optional[int]):
            The risk group to select. If None, values are summed across
            all risk groups.
        results_filename (Optional[str]):
            If provided, saves the resulting DataFrame as a CSV.

    Returns:
        A pandas DataFrame where rows represent the replication and columns indicate the
        simulation day (timepoint) of recording. DataFrame values are the `StateVariable`'s
        current_val or the sum of the `StateVariable`'s current_val across subpopulations,
        age groups, or risk groups (the combination of what is summed over is
        specified by the user -- details are in the part of this docstring describing
        this function's parameters).
    """

    if state_var_name not in self.state_variables_to_record:
        raise ExperimentError("\"state_var_name\" is not in \"self.state_variables_to_record\" --"
                              "function call is invalid.")

    conn = sqlite3.connect(self.database_filename)

    # Query all results table entries where state_var_name matches
    # This will return results across all subpopulations, age groups,
    #   and risk groups
    df = get_sql_table_as_df(conn,
                             "SELECT * FROM results WHERE state_var_name = ?",
                             chunk_size=int(1e4),
                             sql_query_params=(state_var_name,))

    conn.close()

    # Define filter conditions
    filters = {
        "subpop_name": subpop_name,
        "age_group": age_group,
        "risk_group": risk_group
    }

    # Filter DataFrame based on user-specified conditions
    #   (for example, if user specifies subpop_name, return subset of
    #   DataFrame where subpop_name matches)
    conditions = [(df[col] == value) for col, value in filters.items() if value is not None]
    df_filtered = df if not conditions else df[np.logical_and.reduce(conditions)]

    # Group DataFrame based on unique combinations of "rep" and "timepoint" columns
    # Then sum (numeric values only), return the "value" column, and reset the index
    #   so that "rep" and "timepoint" become regular columns and are not the index
    df_aggregated = \
        df_filtered.groupby(["rep",
                             "timepoint"]).sum(numeric_only=True)["value"].reset_index()

    # breakpoint()

    # Use pivot() function to reshape the DataFrame for its final form
    # The "timepoint" values are spread across new columns
    #   (creating a column for each unique timepoint).
    # The "value" column populates the corresponding cells.
    df_final = df_aggregated.pivot(index="rep",
                                   columns="timepoint",
                                   values="value")

    if results_filename:
        df_final.to_csv(results_filename)

    return df_final

log_current_vals_to_sql(rep_counter: int, experiment_cursor: sqlite3.Cursor) -> None

For each subpopulation and state variable to record associated with this Experiment, save current values to "results" table in SQL database specified by experiment_cursor.

Parameters:

Name	Type	Description	Default
rep_counter	int	Current replication ID.	required
experiment_cursor	Cursor	Cursor object connected to the database where results should be inserted.	required

Source code in CLT_BaseModel/clt_base/experiments.py

def log_current_vals_to_sql(self,
                            rep_counter: int,
                            experiment_cursor: sqlite3.Cursor) -> None:
    """
    For each subpopulation and state variable to record
    associated with this `Experiment`, save current values to
    "results" table in SQL database specified by `experiment_cursor`.

    Params:
        rep_counter (int):
            Current replication ID.
        experiment_cursor (sqlite3.Cursor):
            Cursor object connected to the database
            where results should be inserted.
    """

    for subpop_model in self.experiment_subpop_models:
        for state_var_name in self.state_variables_to_record:
            data = format_current_val_for_sql(subpop_model,
                                              state_var_name,
                                              rep_counter)
            experiment_cursor.executemany(
                "INSERT INTO results VALUES (?, ?, ?, ?, ?, ?, ?)", data)

log_inputs_to_sql(experiment_cursor: sqlite3.Cursor)

For each subpopulation, add a new table to SQL database specified by experiment_cursor. Each table contains information on inputs that vary across replications (either due to random sampling or user-specified deterministic sequence). Each table contains inputs information from Experiment attribute self.inputs_realizations for a given subpopulation.

Parameters:

Name	Type	Description	Default
experiment_cursor	Cursor	Cursor object connected to the database where results should be inserted.	required

Source code in CLT_BaseModel/clt_base/experiments.py

def log_inputs_to_sql(self,
                      experiment_cursor: sqlite3.Cursor):
    """
    For each subpopulation, add a new table to SQL
    database specified by `experiment_cursor`. Each table
    contains information on inputs that vary across
    replications (either due to random sampling or
    user-specified deterministic sequence). Each table
    contains inputs information from `Experiment` attribute
    `self.inputs_realizations` for a given subpopulation.

    Params:
        experiment_cursor (sqlite3.Cursor):
            Cursor object connected to the database
            where results should be inserted.
    """

    for subpop_model in self.experiment_subpop_models:
        table_name = f'"{subpop_model.name}_inputs"'

        # Get the column names (dynamically, based on table)
        experiment_cursor.execute(f"PRAGMA table_info({table_name})")

        # Extract column names from the table info
        # But exclude the column name "rep"
        columns_info = experiment_cursor.fetchall()
        column_names = [col[1] for col in columns_info if col[1] != "rep"]

        # Create a placeholder string for the dynamic query
        placeholders = ", ".join(["?" for _ in column_names])  # Number of placeholders matches number of columns

        # Create the dynamic INSERT statement
        sql_statement = f"INSERT INTO {table_name} ({', '.join(column_names)}) VALUES ({placeholders})"

        # Create list of lists -- each nested list contains a sequence of values
        #   for that particular input
        subpop_inputs_realizations = self.inputs_realizations[subpop_model.name]
        inputs_vals_over_reps_list = \
            [np.array(subpop_inputs_realizations[input_name]).reshape(-1,1) for input_name in column_names]
        inputs_vals_over_reps_list = np.hstack(inputs_vals_over_reps_list)

        experiment_cursor.executemany(sql_statement, inputs_vals_over_reps_list)

run_random_inputs(num_reps: int, simulation_end_day: int, random_inputs_RNG: np.random.Generator, random_inputs_spec: dict, days_between_save_history: int = 1, results_filename: str = None, inputs_filename_suffix: str = None)

Runs the associated SubpopModel or MetapopModel for a given number of independent replications until simulation_end_day, where certain parameter values or initial values are independently randomly sampled for each replication. Random sampling details (which inputs to sample and what Uniform distribution lower and upper bounds to use for each input) are specified in random_inputs_spec, and random sampling of inputs uses random_inputs_RNG. User can specify how often to save the history and a CSV file in which to store this history. User can also specify a filename suffix to name CSV files in which to store realizations of inputs.

Parameters:

Name	Type	Description	Default
num_reps	positive int	number of independent simulation replications to run in an experiment.	required
simulation_end_day	positive int	stop simulation at simulation_end_day (i.e. exclusive, simulate up to but not including simulation_end_day).	required
random_inputs_RNG	Generator	random number generator used to sample random inputs -- for reproducibility, it is recommended to use a distinct RNG for sampling random inputs different from the RNG for simulating the experiment/model.	required
random_inputs_spec	dict	random inputs' specification -- stores details for random input sampling -- keys are strings corresponding to input names (they must match names in each SubpopModel's SubpopParams or associated EpiMetric or Compartment instances to be valid) and values are 2-tuples of nonnegative floats corresponding to lower and upper bounds for sampling that input from a uniform distribution.	required
days_between_save_history	positive int	indicates how often to save simulation results.	1
results_filename	str	if specified, must be valid filename with suffix ".csv" -- experiment results are saved to this CSV file	None
inputs_filename_suffix	str	if specified, must be a valid filename with suffix ".csv" -- one inputs CSV is generated for each SubpopModel with the filename "{name}_{inputs_filename}" where name is the name of the SubpopModel -- saves the values of each parameter that varies between replications	None

Source code in CLT_BaseModel/clt_base/experiments.py

def run_random_inputs(self,
                      num_reps: int,
                      simulation_end_day: int,
                      random_inputs_RNG: np.random.Generator,
                      random_inputs_spec: dict,
                      days_between_save_history: int = 1,
                      results_filename: str = None,
                      inputs_filename_suffix: str = None):
    """
    Runs the associated `SubpopModel` or `MetapopModel` for a
    given number of independent replications until `simulation_end_day`,
    where certain parameter values or initial values are
    independently randomly sampled for each replication.
    Random sampling details (which inputs to sample and what
    Uniform distribution lower and upper bounds to use for each input)
    are specified in `random_inputs_spec`, and random sampling of inputs
    uses `random_inputs_RNG`. User can specify how often to save the
    history and a CSV file in which to store this history.
    User can also specify a filename suffix to name CSV files
    in which to store realizations of inputs.

    Params:
        num_reps (positive int):
            number of independent simulation replications
            to run in an experiment.
        simulation_end_day (positive int):
            stop simulation at simulation_end_day (i.e. exclusive,
            simulate up to but not including simulation_end_day).
        random_inputs_RNG (np.random.Generator):
            random number generator used to sample random
            inputs -- for reproducibility, it is recommended to
            use a distinct RNG for sampling random inputs different
            from the RNG for simulating the experiment/model.
        random_inputs_spec (dict):
            random inputs' specification -- stores details
            for random input sampling -- keys are strings
            corresponding to input names (they must match
            names in each `SubpopModel`'s `SubpopParams` or associated
            `EpiMetric` or `Compartment` instances to be valid)
            and values are 2-tuples of nonnegative floats corresponding
            to lower and upper bounds for sampling that input from a
            uniform distribution.
        days_between_save_history (positive int):
            indicates how often to save simulation results.
        results_filename (str):
            if specified, must be valid filename with suffix ".csv" --
            experiment results are saved to this CSV file
        inputs_filename_suffix (str):
            if specified, must be a valid filename with suffix ".csv" --
            one inputs CSV is generated for each `SubpopModel` with the
            filename "{name}_{inputs_filename}" where name is the name
            of the `SubpopModel` -- saves the values of each parameter that
            varies between replications
    """

    if self.has_been_run:
        raise ExperimentError("Experiment has already been run. "
                              "Create a new Experiment instance to simulate "
                              "more replications.")

    else:
        self.has_been_run = True

        self.sample_random_inputs(num_reps,
                                  random_inputs_RNG,
                                  random_inputs_spec)

        self.create_results_sql_table()
        self.create_inputs_realizations_sql_tables()

        self.simulate_reps_and_save_results(reps=num_reps,
                                            end_day=simulation_end_day,
                                            days_per_save=days_between_save_history,
                                            inputs_are_static=False,
                                            filename=results_filename)

        if inputs_filename_suffix:
            self.write_inputs_csvs(inputs_filename_suffix)

run_sequences_of_inputs(num_reps: int, simulation_end_day: int, sequences_of_inputs: dict, days_between_save_history: int = 1, results_filename: str = None, inputs_filename_suffix: str = None)

Runs the associated SubpopModel or MetapopModel for a given number of independent replications until simulation_end_day, where certain parameter values or initial values deterministically change between replications, according to sequences_of_inputs. User can specify how often to save the history and a CSV file in which to store this history. User can also specify a filename suffix to name CSV files in which to store sequences of inputs.

Parameters:

Name	Type	Description	Default
num_reps	positive int	number of independent simulation replications to run in an experiment.	required
simulation_end_day	positive int	stop simulation at simulation_end_day (i.e. exclusive, simulate up to but not including simulation_end_day).	required
sequences_of_inputs	dict	dictionary of dictionaries that stores user-specified deterministic sequences for inputs -- must follow specific structure. Keys are SubpopModel names, values are dictionaries. These second-layer dictionaries' keys are strings corresponding to input names (they must match names in each SubpopModel's SubpopParams to be valid) and values are list-like, where the ith element corresponds to the ith random sample for that input.	required
days_between_save_history	positive int	indicates how often to save simulation results.	1
results_filename	str	if specified, must be valid filename with suffix ".csv" -- experiment results are saved to this CSV file	None
inputs_filename_suffix	str	if specified, must be a valid filename with suffix ".csv" -- one inputs CSV is generated for each SubpopModel with the filename "{name}_{inputs_filename}" where name is the name of the SubpopModel -- saves the values of each parameter that varies between replications	None

Source code in CLT_BaseModel/clt_base/experiments.py

def run_sequences_of_inputs(self,
                            num_reps: int,
                            simulation_end_day: int,
                            sequences_of_inputs: dict,
                            days_between_save_history: int = 1,
                            results_filename: str = None,
                            inputs_filename_suffix: str = None):
    """
    Runs the associated `SubpopModel` or `MetapopModel` for a
    given number of independent replications until `simulation_end_day`,
    where certain parameter values or initial values deterministically
    change between replications, according to `sequences_of_inputs`.
    User can specify how often to save the history and a CSV file
    in which to store this history. User can also specify a filename
    suffix to name CSV files in which to store sequences of inputs.

    Params:
        num_reps (positive int):
            number of independent simulation replications
            to run in an experiment.
        simulation_end_day (positive int):
            stop simulation at `simulation_end_day` (i.e. exclusive,
            simulate up to but not including `simulation_end_day`).
        sequences_of_inputs (dict):
            dictionary of dictionaries that stores user-specified deterministic
            sequences for inputs -- must follow specific structure.
            Keys are `SubpopModel` names, values are dictionaries.
            These second-layer dictionaries' keys are strings corresponding
            to input names (they must match names in each `SubpopModel`'s
            `SubpopParams` to be valid) and values are list-like, where the ith
            element corresponds to the ith random sample for that input.
        days_between_save_history (positive int):
            indicates how often to save simulation results.
        results_filename (str):
            if specified, must be valid filename with suffix ".csv" --
            experiment results are saved to this CSV file
        inputs_filename_suffix (str):
            if specified, must be a valid filename with suffix ".csv" --
            one inputs CSV is generated for each `SubpopModel` with the
            filename "{name}_{inputs_filename}" where name is the name
            of the `SubpopModel` -- saves the values of each parameter that
            varies between replications
    """

    if self.has_been_run:
        raise ExperimentError("Experiment has already been run. "
                              "Create a new Experiment instance to simulate "
                              "more replications.")

    else:
        self.has_been_run = True

        self.inputs_realizations = sequences_of_inputs

        self.create_results_sql_table()
        self.create_inputs_realizations_sql_tables()

        self.simulate_reps_and_save_results(reps=num_reps,
                                            end_day=simulation_end_day,
                                            days_per_save=days_between_save_history,
                                            inputs_are_static=False,
                                            filename=results_filename)

        if inputs_filename_suffix:
            self.write_inputs_csvs(inputs_filename_suffix)

run_static_inputs(num_reps: int, simulation_end_day: int, days_between_save_history: int = 1, results_filename: str = None)

Runs the associated SubpopModel or MetapopModel for a given number of independent replications until simulation_end_day. Parameter values and initial values are the same across simulation replications. User can specify how often to save the history and a CSV file in which to store this history.

Parameters:

Name	Type	Description	Default
num_reps	positive int	number of independent simulation replications to run in an experiment.	required
simulation_end_day	positive int	stop simulation at simulation_end_day (i.e. exclusive, simulate up to but not including simulation_end_day).	required
days_between_save_history	positive int	indicates how often to save simulation results.	1
results_filename	str	if specified, must be valid filename with suffix ".csv" -- experiment results are saved to this CSV file.	None

Source code in CLT_BaseModel/clt_base/experiments.py

def run_static_inputs(self,
                      num_reps: int,
                      simulation_end_day: int,
                      days_between_save_history: int = 1,
                      results_filename: str = None):
    """
    Runs the associated `SubpopModel` or `MetapopModel` for a
    given number of independent replications until `simulation_end_day`.
    Parameter values and initial values are the same across
    simulation replications. User can specify how often to save the
    history and a CSV file in which to store this history.

    Params:
        num_reps (positive int):
            number of independent simulation replications
            to run in an experiment.
        simulation_end_day (positive int):
            stop simulation at simulation_end_day (i.e. exclusive,
            simulate up to but not including simulation_end_day).
        days_between_save_history (positive int):
            indicates how often to save simulation results.
        results_filename (str):
            if specified, must be valid filename with suffix ".csv" --
            experiment results are saved to this CSV file.
    """

    if self.has_been_run:
        raise ExperimentError("Experiment has already been run. "
                              "Create a new Experiment instance to simulate "
                              "more replications.")

    else:
        self.has_been_run = True

        self.create_results_sql_table()

        self.simulate_reps_and_save_results(reps=num_reps,
                                            end_day=simulation_end_day,
                                            days_per_save=days_between_save_history,
                                            inputs_are_static=True,
                                            filename=results_filename)

sample_random_inputs(total_reps: int, RNG: np.random.Generator, spec: dict)

Randomly and independently samples inputs specified by keys of spec according to uniform distribution with lower and upper bounds specified by values of spec. Stores random realizations in self.inputs_realizations attribute. Uses RNG to sample inputs.

Parameters:

Name	Type	Description	Default
total_reps	positive int	number of independent simulation replications to run in an experiment -- corresponds to number of Uniform random variables to draw for each state variable.	required
RNG	Generator	random number generator used to sample random inputs -- for reproducibility, it is recommended to use a distinct RNG for sampling random inputs different from the RNG for simulating the experiment/model.	required
spec	dict	random inputs' specification -- stores details for random input sampling -- keys are strings corresponding to input names (they must match names in each SubpopModel's SubpopParams or associated EpiMetric or Compartment instances to be valid) and values are 2-tuples of nonnegative floats corresponding to lower and upper bounds for sampling that input from a uniform distribution.	required

NOTE

If an input is an |A| x |R| array (for age-risk), the current functionality does not support sampling individual age-risk elements separately. Instead, a single scalar value is sampled at a time for the entire input. Consequently, if an |A| x |R| input is chosen to be randomly sampled, all its elements will have the same sampled value.

If a user wants to sample some age-risk elements separately, they should create new inputs for these elements. "Inputs" refers to both parameters (in SubpopParams) and initial values of Compartment and EpiMetric instances. For example, if the model has a parameter H_to_R_rate that is 2x1 (2 age groups, 1 risk group) and the user wants to sample each element separately, they should create two parameters: H_to_R_rate_age_group_1 and H_to_R_rate_age_group_2. These should be added to the relevant SubpopParams instance and input dictionary/file used to create the SubpopParams instance. The user can then specify both parameters to be randomly sampled and specify the lower and upper bounds accordingly.

TODO: allow sampling individual age-risk elements separately without creating new parameters for each element.

(Developer note: the difficulty is not with randomly sampling arrays, but rather storing arrays in SQL -- SQL tables only support atomic values.)

Source code in CLT_BaseModel/clt_base/experiments.py

def sample_random_inputs(self,
                         total_reps: int,
                         RNG: np.random.Generator,
                         spec: dict):
    """
    Randomly and independently samples inputs specified by keys of
    `spec` according to uniform distribution with lower
    and upper bounds specified by values of `spec`.
    Stores random realizations in `self.inputs_realizations` attribute.
    Uses `RNG` to sample inputs.

    Params:
        total_reps (positive int):
            number of independent simulation replications
            to run in an experiment -- corresponds to number of
            Uniform random variables to draw for each
            state variable.
        RNG (np.random.Generator):
            random number generator used to sample random
            inputs -- for reproducibility, it is recommended to
            use a distinct RNG for sampling random inputs different
            from the RNG for simulating the experiment/model.
        spec (dict):
            random inputs' specification -- stores details
            for random input sampling -- keys are strings
            corresponding to input names (they must match
            names in each `SubpopModel`'s `SubpopParams` or associated
            `EpiMetric` or `Compartment` instances to be valid)
            and values are 2-tuples of nonnegative floats corresponding
            to lower and upper bounds for sampling that input from a
            uniform distribution.

    NOTE:
        If an input is an |A| x |R| array (for age-risk),
        the current functionality does not support sampling individual
        age-risk elements separately. Instead, a single scalar value
        is sampled at a time for the entire input. Consequently,
        if an |A| x |R| input is chosen to be randomly sampled,
        all its elements will have the same sampled value.

        If a user wants to sample some age-risk elements separately,
        they should create new inputs for these elements. "Inputs"
        refers to both parameters (in `SubpopParams`) and initial values
        of `Compartment` and `EpiMetric` instances. For example, if the model
        has a parameter `H_to_R_rate` that is 2x1 (2 age groups, 1 risk group)
        and the user wants to sample each element separately, they should create
        two parameters: `H_to_R_rate_age_group_1` and `H_to_R_rate_age_group_2.`
        These should be added to the relevant `SubpopParams` instance and
        input dictionary/file used to create the `SubpopParams` instance.
        The user can then specify both parameters to be randomly sampled
        and specify the lower and upper bounds accordingly.

        TODO: allow sampling individual age-risk elements separately
        without creating new parameters for each element.

        (Developer note: the difficulty is not with randomly sampling
        arrays, but rather storing arrays in SQL -- SQL tables only
        support atomic values.)
    """

    inputs_realizations = self.inputs_realizations

    for subpop_model in self.experiment_subpop_models:

        subpop_name = subpop_model.name

        compartments_names_list = list(subpop_model.compartments.keys())
        epi_metrics_names_list = list(subpop_model.epi_metrics.keys())
        params_names_list = [field.name for field in fields(subpop_model.params)]

        if not check_is_subset_list(spec[subpop_name],
                                    compartments_names_list + epi_metrics_names_list + params_names_list):
            raise (f"\"random_inputs_spec[\"{subpop_name}\"]\" keys are not a subset "
                   "of the state variables and parameters on SubpopModel \"{subpop_name}\" -- "
                   "modify \"random_inputs_spec\" and re-initialize experiment.")

        for state_var_name, bounds in spec[subpop_name].items():
            lower_bd = bounds[0]
            upper_bd = bounds[1]
            inputs_realizations[subpop_name][state_var_name] = \
                RNG.uniform(low=lower_bd,
                            high=upper_bd,
                            size=total_reps)

simulate_reps_and_save_results(reps: int, end_day: int, days_per_save: int, inputs_are_static: bool, filename: str = None)

Helper function that executes main loop over replications in Experiment and saves results.

Parameters:

Name	Type	Description	Default
reps	int	number of independent simulation replications to run in an experiment.	required
end_day	int	stop simulation at end_day (i.e. exclusive, simulate up to but not including end_day).	required
days_per_save	int	indicates how often to save simulation results.	required
inputs_are_static	bool	indicates if inputs are same across replications.	required
filename	str	if specified, must be valid filename with suffix ".csv" -- experiment results are saved to this CSV file.	None

Source code in CLT_BaseModel/clt_base/experiments.py

def simulate_reps_and_save_results(self,
                                   reps: int,
                                   end_day: int,
                                   days_per_save: int,
                                   inputs_are_static: bool,
                                   filename: str = None):
    """
    Helper function that executes main loop over
    replications in `Experiment` and saves results.

    Params:
        reps (int):
            number of independent simulation replications
            to run in an experiment.
        end_day (int):
            stop simulation at end_day (i.e. exclusive,
            simulate up to but not including end_day).
        days_per_save (int):
            indicates how often to save simulation results.
        inputs_are_static (bool):
            indicates if inputs are same across replications.
        filename (str):
            if specified, must be valid filename with suffix ".csv" --
            experiment results are saved to this CSV file.
    """

    # Override each subpop config's save_daily_history attribute --
    #   set it to False -- because we will manually save history
    #   to results database according to user-defined
    #   days_between_save_history for all subpops
    for subpop_model in self.experiment_subpop_models:
        subpop_model.config.save_daily_history = False

    model = self.model

    # Connect to SQL database
    conn = sqlite3.connect(self.database_filename)
    cursor = conn.cursor()

    # Loop through replications
    for rep in range(reps):

        # Reset model and clear its history
        model.reset_simulation()

        # Apply new values of inputs, if some
        #   inputs change between replications
        if not inputs_are_static:
            self.apply_inputs_to_model(rep)
            self.log_inputs_to_sql(cursor)

        # Simulate model and save results every `days_per_save` days
        while model.current_simulation_day < end_day:
            model.simulate_until_day(min(model.current_simulation_day + days_per_save,
                                         end_day))

            self.log_current_vals_to_sql(rep, cursor)

    self.results_df = get_sql_table_as_df(conn, "SELECT * FROM results", chunk_size=int(1e4))

    if filename:
        self.results_df.to_csv(filename)

    # Commit changes to database and close
    conn.commit()
    conn.close()

write_inputs_csvs(suffix: str)

For each subpopulation, writes a CSV (with a filename based on provided suffix) containing values of inputs across replications.

Parameters:

Name	Type	Description	Default
suffix	str	Common suffix used to generate CSV filenames.	required

Source code in CLT_BaseModel/clt_base/experiments.py

def write_inputs_csvs(self,
                      suffix: str):
    """
    For each subpopulation, writes a CSV (with a filename
    based on provided suffix) containing values of inputs
    across replications.

    Params:
        suffix (str):
            Common suffix used to generate CSV filenames.
    """

    for subpop_model in self.experiment_subpop_models:
        conn = sqlite3.connect(self.database_filename)

        table_name = f"{subpop_model.name}_inputs"

        subpop_inputs_df = get_sql_table_as_df(conn, f"SELECT * FROM {table_name}", chunk_size=int(1e4))

        subpop_inputs_df.to_csv(f"{subpop_model.name}_{suffix}", index=False)

ExperimentError

Bases: Exception

Custom exceptions for experiment errors.

Source code in CLT_BaseModel/clt_base/experiments.py

class ExperimentError(Exception):
    """Custom exceptions for experiment errors."""
    pass

InterSubpopRepo

Bases: ABC

Holds collection of SubpopState instances, with actions to query and interact with them.

Attributes:

Name	Type	Description
subpop_models	objdict	keys are unique names of subpopulation models, values are their respective SubpopModel instances -- this dictionary contains all SubpopModel instances that comprise a MetapopModel instance.

Source code in CLT_BaseModel/clt_base/base_components.py

class InterSubpopRepo(ABC):
    """
    Holds collection of `SubpopState` instances, with
        actions to query and interact with them.

    Attributes:
        subpop_models (sc.objdict):
            keys are unique names of subpopulation models,
            values are their respective `SubpopModel` instances --
            this dictionary contains all `SubpopModel` instances
            that comprise a `MetapopModel` instance.
    """

    def __init__(self,
                 subpop_models: Optional[dict] = None):
        self.subpop_models = sc.objdict(subpop_models)

        # The "name" argument for instantiating a SubpopModel
        #   is optional -- but SubpopModel instances must have
        #   names when used in a MetapopModel
        # So, we assign names to each SubpopModel based on
        #   the keys of the dictionary that creates the InterSubpopRepo
        for name, model in subpop_models.items():
            model.name = name

    @abstractmethod
    def compute_shared_quantities(self):
        """
        Subclasses must provide concrete implementation. This method
        is called by the `MetapopModel` instance at the beginning of
        each simulation day, before each `SubpopModel` simulates that day.

        Note: often, `InteractionTerm`s across `SubpopModel`s share similar
        terms in their computation. This `self.compute_shared_quantities`
        method computes such similar terms up front to reduce redundant
        computation.
        """

        pass

    def update_all_interaction_terms(self):
        """
        Updates `SubpopState` of each `SubpopModel` in
        `self.subpop_models` to reflect current values of each
        `InteractionTerm` on that `SubpopModel`.
        """

        for subpop_model in self.subpop_models.values():
            for iterm in subpop_model.interaction_terms.values():
                iterm.update_current_val(self,
                                         subpop_model.params)
            subpop_model.state.sync_to_current_vals(subpop_model.interaction_terms)

compute_shared_quantities()

Subclasses must provide concrete implementation. This method is called by the MetapopModel instance at the beginning of each simulation day, before each SubpopModel simulates that day.

Note: often, InteractionTerms across SubpopModels share similar terms in their computation. This self.compute_shared_quantities method computes such similar terms up front to reduce redundant computation.

Source code in CLT_BaseModel/clt_base/base_components.py

@abstractmethod
def compute_shared_quantities(self):
    """
    Subclasses must provide concrete implementation. This method
    is called by the `MetapopModel` instance at the beginning of
    each simulation day, before each `SubpopModel` simulates that day.

    Note: often, `InteractionTerm`s across `SubpopModel`s share similar
    terms in their computation. This `self.compute_shared_quantities`
    method computes such similar terms up front to reduce redundant
    computation.
    """

    pass

update_all_interaction_terms()

Updates SubpopState of each SubpopModel in self.subpop_models to reflect current values of each InteractionTerm on that SubpopModel.

Source code in CLT_BaseModel/clt_base/base_components.py

def update_all_interaction_terms(self):
    """
    Updates `SubpopState` of each `SubpopModel` in
    `self.subpop_models` to reflect current values of each
    `InteractionTerm` on that `SubpopModel`.
    """

    for subpop_model in self.subpop_models.values():
        for iterm in subpop_model.interaction_terms.values():
            iterm.update_current_val(self,
                                     subpop_model.params)
        subpop_model.state.sync_to_current_vals(subpop_model.interaction_terms)

InteractionTerm

Bases: StateVariable, ABC

Abstract base class for variables that depend on the state of more than one SubpopModel (i.e., that depend on more than one SubpopState). These variables are functions of how subpopulations interact.

In contrast to other state variables, each InteractionTerm takes in an InterSubpopRepo instance to update its self.current_val. Other state variables that are "local" and depend on exactly one subpopulation only need to take in one SubpopState and one SubpopParams instance to update its current value.

Inherits attributes from StateVariable.

See __init__ docstring for other attributes.

Source code in CLT_BaseModel/clt_base/base_components.py

class InteractionTerm(StateVariable, ABC):

    """
    Abstract base class for variables that depend on the state of
    more than one `SubpopModel` (i.e., that depend on more than one
    `SubpopState`). These variables are functions of how subpopulations
    interact.

    In contrast to other state variables, each `InteractionTerm`
    takes in an `InterSubpopRepo` instance to update its `self.current_val`.
    Other state variables that are "local" and depend on
    exactly one subpopulation only need to take in one `SubpopState`
    and one `SubpopParams` instance to update its current value.

    Inherits attributes from `StateVariable`.

    See `__init__` docstring for other attributes.
    """

    @abstractmethod
    def update_current_val(self,
                           inter_subpop_repo: InterSubpopRepo,
                           subpop_params: SubpopParams) -> None:
        """
        Subclasses must provide a concrete implementation of
        updating `self.current_val` in-place.

        Args:
            inter_subpop_repo (InterSubpopRepo):
                manages collection of subpop models with
                methods for querying information.
            subpop_params (SubpopParams):
                holds values of subpopulation's epidemiological parameters.
        """

        pass

update_current_val(inter_subpop_repo: InterSubpopRepo, subpop_params: SubpopParams) -> None

Subclasses must provide a concrete implementation of updating self.current_val in-place.

Parameters:

Name	Type	Description	Default
inter_subpop_repo	InterSubpopRepo	manages collection of subpop models with methods for querying information.	required
subpop_params	SubpopParams	holds values of subpopulation's epidemiological parameters.	required

Source code in CLT_BaseModel/clt_base/base_components.py

@abstractmethod
def update_current_val(self,
                       inter_subpop_repo: InterSubpopRepo,
                       subpop_params: SubpopParams) -> None:
    """
    Subclasses must provide a concrete implementation of
    updating `self.current_val` in-place.

    Args:
        inter_subpop_repo (InterSubpopRepo):
            manages collection of subpop models with
            methods for querying information.
        subpop_params (SubpopParams):
            holds values of subpopulation's epidemiological parameters.
    """

    pass

JointTransitionTypes

Bases: str, Enum

Source code in CLT_BaseModel/clt_base/base_components.py

class JointTransitionTypes(str, Enum):
    MULTINOMIAL = "multinomial"
    MULTINOMIAL_DETERMINISTIC = "multinomial_deterministic"
    MULTINOMIAL_TAYLOR_APPROX = "multinomial_taylor_approx"
    MULTINOMIAL_TAYLOR_APPROX_DETERMINISTIC = "multinomial_taylor_approx_deterministic"
    POISSON = "poisson"
    POISSON_DETERMINISTIC = "poisson_deterministic"

MetapopModel

Bases: ABC

Abstract base class that bundles SubpopModels linked using a travel model.

Parameters:

Name	Type	Description	Default
inter_subpop_repo	InterSubpopRepo	Accesses and manages SubpopState instances of corresponding SubpopModels, and provides methods to query current values.	required

See __init__ docstring for other attributes.

Source code in CLT_BaseModel/clt_base/base_components.py

class MetapopModel(ABC):
    """
    Abstract base class that bundles `SubpopModel`s linked using
        a travel model.

    Params:
        inter_subpop_repo (InterSubpopRepo):
            Accesses and manages `SubpopState` instances
            of corresponding `SubpopModel`s, and provides
            methods to query current values.

    See `__init__` docstring for other attributes.
    """

    def __init__(self,
                 inter_subpop_repo,
                 name: str = ""):
        """
        Params:
            inter_subpop_repo (InterSubpopRepo):
                manages collection of subpopulation models with
                methods for querying information.
            name (str):
                unique identifier for `MetapopModel`.
        """

        self.subpop_models = inter_subpop_repo.subpop_models

        self.inter_subpop_repo = inter_subpop_repo

        self.name = name

        for model in self.subpop_models.values():
            model.metapop_model = self
            model.interaction_terms = model.create_interaction_terms()
            model.state.interaction_terms = model.interaction_terms

    def extract_states_dict_from_models_dict(self,
                                             models_dict: sc.objdict) -> sc.objdict:
        """
        (Currently unused utility function.)

        Takes objdict of subpop models, where keys are subpop model names and
            values are the subpop model instances, and returns objdict of
            subpop model states, where keys are subpop model names and
            values are the subpop model `SubpopState` instances.
        """

        states_dict = \
            sc.objdict({name: model.state for name, model in models_dict.items()})

        return states_dict

    def simulate_until_day(self,
                           simulation_end_day: int) -> None:
        """
        Advance simulation model time until `simulation_end_day` in
        `MetapopModel`.

        NOT just the same as looping through each `SubpopModel`'s
        `simulate_until_day` method. On the `MetapopModel`,
        because `SubpopModel` instances are linked with `InteractionTerm`s
        and are not independent of each other, this `MetapopModel`'s
        `simulate_until_day` method has additional functionality.

        Note: the update order at the beginning of each day is very important!

        - First, each `SubpopModel` updates its daily state (computing
            `Schedule` and `DynamicVal` instances).
        - Second, the `MetapopModel`'s `InterSubpopRepo` computes any shared
            terms used across subpopulations (to reduce computational overhead),
            and then updates each `SubpopModel`'s associated `InteractionTerm`
            instances.
        - Third, each `SubpopModel` simulates discretized timesteps (sampling
            `TransitionVariable`s, updating `EpiMetric`s, and updating `Compartment`s).

        Args:
            simulation_end_day (positive int):
                stop simulation at `simulation_end_day` (i.e. exclusive,
                simulate up to but not including `simulation_end_day`).
        """

        if self.current_simulation_day > simulation_end_day:
            raise MetapopModelError(f"Current day counter ({self.current_simulation_day}) "
                                   f"exceeds last simulation day ({simulation_end_day}).")

        while self.current_simulation_day < simulation_end_day:

            for subpop_model in self.subpop_models.values():
                subpop_model.prepare_daily_state()

            self.inter_subpop_repo.compute_shared_quantities()
            self.inter_subpop_repo.update_all_interaction_terms()

            for subpop_model in self.subpop_models.values():

                save_daily_history = subpop_model.config.save_daily_history
                timesteps_per_day = subpop_model.config.timesteps_per_day

                subpop_model.simulate_timesteps(timesteps_per_day)

                if save_daily_history:
                    subpop_model.save_daily_history()

                subpop_model.increment_simulation_day()

    def display(self):
        """
        Prints structure (compartments and linkages), transition variables,
        epi metrics, schedules, and dynamic values for each `SubpopModel`
        instance in `self.subpop_models`.
        """
        for subpop_model in self.subpop_models.values():
            subpop_model.display()

    def reset_simulation(self):
        """
        Resets `MetapopModel` by resetting and clearing
            history on all `SubpopModel` instances in
            `self.subpop_models`.
        """

        for subpop_model in self.subpop_models.values():
            subpop_model.reset_simulation()

    @property
    def current_simulation_day(self) -> int:
        """
        Returns:
            Current simulation day. The current simulation day of the
            `MetapopModel` should be the same as each individual `SubpopModel`
            in the `MetapopModel`. Otherwise, an error is raised.
        """

        current_simulation_days_list = []

        for subpop_model in self.subpop_models.values():
            current_simulation_days_list.append(subpop_model.current_simulation_day)

        if len(set(current_simulation_days_list)) > 1:
            raise MetapopModelError("Subpopulation models are on different simulation days "
                                    "and are out-of-sync. This may be caused by simulating "
                                    "a subpopulation model independently from the "
                                    "metapopulation model. Fix error and try again.")
        else:
            return current_simulation_days_list[0]

    @property
    def current_real_date(self) -> datetime.date:
        """
        Returns:
            Current real date corresponding to current simulation day.
            The current real date of the `MetapopModel` should be the same as
            each individual `SubpopModel` in the `MetapopModel`.
            Otherwise, an error is raised.
        """

        current_real_dates_list = []

        for subpop_model in self.subpop_models.values():
            current_real_dates_list.append(subpop_model.current_real_date)

        if len(set(current_real_dates_list)) > 1:
            raise MetapopModelError("Subpopulation models are on different real dates "
                                    "and are out-of-sync. This may be caused by simulating "
                                    "a subpopulation model independently from the "
                                    "metapopulation model. Fix error and try again.")
        else:
            return current_real_dates_list[0]

current_real_date: datetime.date property

Returns:

Type	Description
date	Current real date corresponding to current simulation day.
date	The current real date of the MetapopModel should be the same as
date	each individual SubpopModel in the MetapopModel.
date	Otherwise, an error is raised.

current_simulation_day: int property

Returns:

Type	Description
int	Current simulation day. The current simulation day of the
int	MetapopModel should be the same as each individual SubpopModel
int	in the MetapopModel. Otherwise, an error is raised.

__init__(inter_subpop_repo, name: str = '')

Parameters:

Name	Type	Description	Default
inter_subpop_repo	InterSubpopRepo	manages collection of subpopulation models with methods for querying information.	required
name	str	unique identifier for MetapopModel.	''

Source code in CLT_BaseModel/clt_base/base_components.py

def __init__(self,
             inter_subpop_repo,
             name: str = ""):
    """
    Params:
        inter_subpop_repo (InterSubpopRepo):
            manages collection of subpopulation models with
            methods for querying information.
        name (str):
            unique identifier for `MetapopModel`.
    """

    self.subpop_models = inter_subpop_repo.subpop_models

    self.inter_subpop_repo = inter_subpop_repo

    self.name = name

    for model in self.subpop_models.values():
        model.metapop_model = self
        model.interaction_terms = model.create_interaction_terms()
        model.state.interaction_terms = model.interaction_terms

display()

Prints structure (compartments and linkages), transition variables, epi metrics, schedules, and dynamic values for each SubpopModel instance in self.subpop_models.

Source code in CLT_BaseModel/clt_base/base_components.py

def display(self):
    """
    Prints structure (compartments and linkages), transition variables,
    epi metrics, schedules, and dynamic values for each `SubpopModel`
    instance in `self.subpop_models`.
    """
    for subpop_model in self.subpop_models.values():
        subpop_model.display()

extract_states_dict_from_models_dict(models_dict: sc.objdict) -> sc.objdict

(Currently unused utility function.)

Takes objdict of subpop models, where keys are subpop model names and values are the subpop model instances, and returns objdict of subpop model states, where keys are subpop model names and values are the subpop model SubpopState instances.

Source code in CLT_BaseModel/clt_base/base_components.py

def extract_states_dict_from_models_dict(self,
                                         models_dict: sc.objdict) -> sc.objdict:
    """
    (Currently unused utility function.)

    Takes objdict of subpop models, where keys are subpop model names and
        values are the subpop model instances, and returns objdict of
        subpop model states, where keys are subpop model names and
        values are the subpop model `SubpopState` instances.
    """

    states_dict = \
        sc.objdict({name: model.state for name, model in models_dict.items()})

    return states_dict

reset_simulation()

Resets MetapopModel by resetting and clearing history on all SubpopModel instances in self.subpop_models.

Source code in CLT_BaseModel/clt_base/base_components.py

def reset_simulation(self):
    """
    Resets `MetapopModel` by resetting and clearing
        history on all `SubpopModel` instances in
        `self.subpop_models`.
    """

    for subpop_model in self.subpop_models.values():
        subpop_model.reset_simulation()

simulate_until_day(simulation_end_day: int) -> None

Advance simulation model time until simulation_end_day in MetapopModel.

NOT just the same as looping through each SubpopModel's simulate_until_day method. On the MetapopModel, because SubpopModel instances are linked with InteractionTerms and are not independent of each other, this MetapopModel's simulate_until_day method has additional functionality.

Note: the update order at the beginning of each day is very important!

• First, each SubpopModel updates its daily state (computing Schedule and DynamicVal instances).
• Second, the MetapopModel's InterSubpopRepo computes any shared terms used across subpopulations (to reduce computational overhead), and then updates each SubpopModel's associated InteractionTerm instances.
• Third, each SubpopModel simulates discretized timesteps (sampling TransitionVariables, updating EpiMetrics, and updating Compartments).

Parameters:

Name	Type	Description	Default
simulation_end_day	positive int	stop simulation at simulation_end_day (i.e. exclusive, simulate up to but not including simulation_end_day).	required

Source code in CLT_BaseModel/clt_base/base_components.py

def simulate_until_day(self,
                       simulation_end_day: int) -> None:
    """
    Advance simulation model time until `simulation_end_day` in
    `MetapopModel`.

    NOT just the same as looping through each `SubpopModel`'s
    `simulate_until_day` method. On the `MetapopModel`,
    because `SubpopModel` instances are linked with `InteractionTerm`s
    and are not independent of each other, this `MetapopModel`'s
    `simulate_until_day` method has additional functionality.

    Note: the update order at the beginning of each day is very important!

    - First, each `SubpopModel` updates its daily state (computing
        `Schedule` and `DynamicVal` instances).
    - Second, the `MetapopModel`'s `InterSubpopRepo` computes any shared
        terms used across subpopulations (to reduce computational overhead),
        and then updates each `SubpopModel`'s associated `InteractionTerm`
        instances.
    - Third, each `SubpopModel` simulates discretized timesteps (sampling
        `TransitionVariable`s, updating `EpiMetric`s, and updating `Compartment`s).

    Args:
        simulation_end_day (positive int):
            stop simulation at `simulation_end_day` (i.e. exclusive,
            simulate up to but not including `simulation_end_day`).
    """

    if self.current_simulation_day > simulation_end_day:
        raise MetapopModelError(f"Current day counter ({self.current_simulation_day}) "
                               f"exceeds last simulation day ({simulation_end_day}).")

    while self.current_simulation_day < simulation_end_day:

        for subpop_model in self.subpop_models.values():
            subpop_model.prepare_daily_state()

        self.inter_subpop_repo.compute_shared_quantities()
        self.inter_subpop_repo.update_all_interaction_terms()

        for subpop_model in self.subpop_models.values():

            save_daily_history = subpop_model.config.save_daily_history
            timesteps_per_day = subpop_model.config.timesteps_per_day

            subpop_model.simulate_timesteps(timesteps_per_day)

            if save_daily_history:
                subpop_model.save_daily_history()

            subpop_model.increment_simulation_day()

MetapopModelError

Bases: Exception

Custom exceptions for metapopulation simulation model errors.

Source code in CLT_BaseModel/clt_base/base_components.py

class MetapopModelError(Exception):
    """Custom exceptions for metapopulation simulation model errors."""
    pass

Schedule

Bases: StateVariable, ABC

Abstract base class for variables that are functions of real-world dates -- for example, contact matrices (which depend on the day of the week and whether the current day is a holiday), historical vaccination data, and seasonality.

Inherits attributes from StateVariable.

See __init__ docstring for other attributes.

Source code in CLT_BaseModel/clt_base/base_components.py

@dataclass
class Schedule(StateVariable, ABC):
    """
    Abstract base class for variables that are functions of real-world
    dates -- for example, contact matrices (which depend on the day of
    the week and whether the current day is a holiday), historical
    vaccination data, and seasonality.

    Inherits attributes from `StateVariable`.

    See `__init__` docstring for other attributes.
    """

    def __init__(self,
                 init_val: Optional[np.ndarray | float] = None,
                 timeseries_df: Optional[dict] = None):
        """
        Args:
            init_val (Optional[np.ndarray | float]):
                starting value(s) at the beginning of the simulation
            timeseries_df (Optional[pd.DataFrame] = None):
                has a "date" column with strings in format `"YYYY-MM-DD"`
                of consecutive calendar days, and other columns
                corresponding to values on those days
        """

        super().__init__(init_val)
        self.timeseries_df = timeseries_df

    @abstractmethod
    def update_current_val(self,
                           params: SubpopParams,
                           current_date: datetime.date) -> None:
        """
        Subpop classes must provide a concrete implementation of
        updating `self.current_val` in-place.

        Args:
            params (SubpopParams):
                fixed parameters of subpopulation model.
            current_date (date):
                real-world date corresponding to
                model's current simulation day.
        """
        pass

__init__(init_val: Optional[np.ndarray | float] = None, timeseries_df: Optional[dict] = None)

Parameters:

Name	Type	Description	Default
init_val	Optional[ndarray | float]	starting value(s) at the beginning of the simulation	None
timeseries_df	Optional[pd.DataFrame] = None	has a "date" column with strings in format "YYYY-MM-DD" of consecutive calendar days, and other columns corresponding to values on those days	None

Source code in CLT_BaseModel/clt_base/base_components.py

def __init__(self,
             init_val: Optional[np.ndarray | float] = None,
             timeseries_df: Optional[dict] = None):
    """
    Args:
        init_val (Optional[np.ndarray | float]):
            starting value(s) at the beginning of the simulation
        timeseries_df (Optional[pd.DataFrame] = None):
            has a "date" column with strings in format `"YYYY-MM-DD"`
            of consecutive calendar days, and other columns
            corresponding to values on those days
    """

    super().__init__(init_val)
    self.timeseries_df = timeseries_df

update_current_val(params: SubpopParams, current_date: datetime.date) -> None

Subpop classes must provide a concrete implementation of updating self.current_val in-place.

Parameters:

Name	Type	Description	Default
params	SubpopParams	fixed parameters of subpopulation model.	required
current_date	date	real-world date corresponding to model's current simulation day.	required

Source code in CLT_BaseModel/clt_base/base_components.py

@abstractmethod
def update_current_val(self,
                       params: SubpopParams,
                       current_date: datetime.date) -> None:
    """
    Subpop classes must provide a concrete implementation of
    updating `self.current_val` in-place.

    Args:
        params (SubpopParams):
            fixed parameters of subpopulation model.
        current_date (date):
            real-world date corresponding to
            model's current simulation day.
    """
    pass

StateVariable

Parent class of InteractionTerm, Compartment, EpiMetric, DynamicVal, and Schedule classes. All subclasses have the common attributes self.init_val and self.current_val.

Attributes:

Name	Type	Description
init_val	ndarray	holds initial value of StateVariable for age-risk groups.
current_val	ndarray	same size as self.init_val, holds current value of StateVariable for age-risk groups.
history_vals_list	list[ndarray]	each element is the same size of self.current_val, holds history of compartment states for age-risk groups -- element t corresponds to previous self.current_val value at end of simulation day t.

Source code in CLT_BaseModel/clt_base/base_components.py

class StateVariable:
    """
    Parent class of `InteractionTerm`, `Compartment`, `EpiMetric`,
    `DynamicVal`, and `Schedule` classes. All subclasses have the
    common attributes `self.init_val` and `self.current_val`.

    Attributes:
        init_val (np.ndarray):
            holds initial value of `StateVariable` for age-risk groups.
        current_val (np.ndarray):
            same size as `self.init_val`, holds current value of `StateVariable`
            for age-risk groups.
        history_vals_list (list[np.ndarray]):
            each element is the same size of `self.current_val`, holds
            history of compartment states for age-risk groups --
            element t corresponds to previous `self.current_val` value at
            end of simulation day t.
    """

    def __init__(self, init_val=None):
        self.init_val = init_val
        self.current_val = copy.deepcopy(init_val)
        self.history_vals_list = []

    def save_history(self) -> None:
        """
        Saves current value to history by appending `self.current_val` attribute
            to `self.history_vals_list` in place.

        Deep copying is CRUCIAL because `self.current_val` is a mutable
            `np.ndarray` -- without deep copying, `self.history_vals_list` would
            have the same value for all elements.
        """
        self.history_vals_list.append(copy.deepcopy(self.current_val))

    def reset(self) -> None:
        """
        Resets `self.current_val` to `self.init_val`
        and resets `self.history_vals_list` attribute to empty list.
        """

        self.current_val = copy.deepcopy(self.init_val)
        self.history_vals_list = []

reset() -> None

Resets self.current_val to self.init_val and resets self.history_vals_list attribute to empty list.

Source code in CLT_BaseModel/clt_base/base_components.py

def reset(self) -> None:
    """
    Resets `self.current_val` to `self.init_val`
    and resets `self.history_vals_list` attribute to empty list.
    """

    self.current_val = copy.deepcopy(self.init_val)
    self.history_vals_list = []

save_history() -> None

Saves current value to history by appending self.current_val attribute to self.history_vals_list in place.

Deep copying is CRUCIAL because self.current_val is a mutable np.ndarray -- without deep copying, self.history_vals_list would have the same value for all elements.

Source code in CLT_BaseModel/clt_base/base_components.py

def save_history(self) -> None:
    """
    Saves current value to history by appending `self.current_val` attribute
        to `self.history_vals_list` in place.

    Deep copying is CRUCIAL because `self.current_val` is a mutable
        `np.ndarray` -- without deep copying, `self.history_vals_list` would
        have the same value for all elements.
    """
    self.history_vals_list.append(copy.deepcopy(self.current_val))

SubpopModel

Bases: ABC

Contains and manages all necessary components for simulating a compartmental model for a given subpopulation.

Each SubpopModel instance includes compartments, epi metrics, dynamic vals, a data container for the current simulation state, transition variables and transition variable groups, epidemiological parameters, simulation experiment configuration parameters, and a random number generator.

All city-level subpopulation models, regardless of disease type and compartment/transition structure, are instances of this class.

When creating an instance, the order of elements does not matter within self.compartments, self.epi_metrics, self.dynamic_vals, self.transition_variables, and self.transition_variable_groups. The "flow" and "physics" information are stored on the objects.

Attributes:

Name	Type	Description
interaction_terms	objdict	objdict of all the subpop model's InteractionTerm instances.
compartments	objdict	objdict of all the subpop model's Compartment instances.
transition_variables	objdict	objdict of all the subpop model's TransitionVariable instances.
transition_variable_groups	objdict	objdict of all the subpop model's TransitionVariableGroup instances.
epi_metrics	objdict	objdict of all the subpop model's EpiMetric instances.
dynamic_vals	objdict	objdict of all the subpop model's DynamicVal instances.
schedules	objdict	objdict of all the subpop model's Schedule instances.
current_simulation_day	int	tracks current simulation day -- incremented by +1 when config.timesteps_per_day discretized timesteps have completed.
current_real_date	date	tracks real-world date -- advanced by +1 day when config.timesteps_per_day discretized timesteps have completed.

See __init__ docstring for other attributes.

Source code in CLT_BaseModel/clt_base/base_components.py

class SubpopModel(ABC):
    """
    Contains and manages all necessary components for
    simulating a compartmental model for a given subpopulation.

    Each `SubpopModel` instance includes compartments,
    epi metrics, dynamic vals, a data container for the current simulation
    state, transition variables and transition variable groups,
    epidemiological parameters, simulation experiment configuration
    parameters, and a random number generator.

    All city-level subpopulation models, regardless of disease type and
    compartment/transition structure, are instances of this class.

    When creating an instance, the order of elements does not matter
    within `self.compartments`, `self.epi_metrics`, `self.dynamic_vals`,
    `self.transition_variables`, and `self.transition_variable_groups`.
    The "flow" and "physics" information are stored on the objects.

    Attributes:
        interaction_terms (sc.objdict):
            objdict of all the subpop model's `InteractionTerm` instances.
        compartments (sc.objdict):
            objdict of all the subpop model's `Compartment` instances.
        transition_variables (sc.objdict):
            objdict of all the subpop model's `TransitionVariable` instances.
        transition_variable_groups (sc.objdict):
            objdict of all the subpop model's `TransitionVariableGroup` instances.
        epi_metrics (sc.objdict):
            objdict of all the subpop model's `EpiMetric` instances.
        dynamic_vals (sc.objdict):
            objdict of all the subpop model's `DynamicVal` instances.
        schedules (sc.objdict):
            objdict of all the subpop model's `Schedule` instances.
        current_simulation_day (int):
            tracks current simulation day -- incremented by +1
            when `config.timesteps_per_day` discretized timesteps
            have completed.
        current_real_date (datetime.date):
            tracks real-world date -- advanced by +1 day when
            `config.timesteps_per_day` discretized timesteps
            have completed.

    See `__init__` docstring for other attributes.
    """

    def __init__(self,
                 state: SubpopState,
                 params: SubpopParams,
                 config: Config,
                 RNG: np.random.Generator,
                 name: str = "",
                 metapop_model: MetapopModel = None):

        """
        Params:
            state (SubpopState):
                holds current values of `SubpopModel`'s state variables.
            params (SubpopParams):
                data container for the model's epidemiological parameters,
                such as the "Greek letters" characterizing sojourn times
                in compartments.
            config (Config):
                data container for the model's simulation configuration values.
            RNG (np.random.Generator):
                 used to generate stochastic transitions in the model and control
                 reproducibility.
            name (str):
                unique identifier of `SubpopModel`.
            metapop_model (Optional[MetapopModel]):
                if not `None`, is the `MetapopModel` instance
                associated with this `SubpopModel`.
        """

        self.state = copy.deepcopy(state)
        self.params = copy.deepcopy(params)
        self.config = copy.deepcopy(config)

        self.RNG = RNG

        self.current_simulation_day = 0
        self.start_real_date = self.get_start_real_date()
        self.current_real_date = self.start_real_date

        self.metapop_model = None
        self.name = name

        self.interaction_terms = self.create_interaction_terms()
        self.compartments = self.create_compartments()
        self.transition_variables = self.create_transition_variables()
        self.transition_variable_groups = self.create_transition_variable_groups()

        # Some epi metrics depend on transition variables, so
        #   set up epi metrics after transition variables
        self.epi_metrics = self.create_epi_metrics()
        self.dynamic_vals = self.create_dynamic_vals()
        self.schedules = self.create_schedules()

        self.all_state_variables = {**self.interaction_terms,
                                    **self.compartments,
                                    **self.epi_metrics,
                                    **self.dynamic_vals,
                                    **self.schedules}

        # The model's state also has access to the model's
        #   compartments, epi_metrics, dynamic_vals, and schedules --
        #   so that state can easily retrieve each object's
        #   current_val and store it
        self.state.interaction_terms = self.interaction_terms
        self.state.compartments = self.compartments
        self.state.epi_metrics = self.epi_metrics
        self.state.dynamic_vals = self.dynamic_vals
        self.state.schedules = self.schedules

        self.params.total_pop_age_risk = self.compute_total_pop_age_risk()

    def compute_total_pop_age_risk(self) -> np.ndarray:
        """
        Returns:
            np.ndarray:
                |A| x |R| array, where |A| is the number of age groups
                and |R| is the number of risk groups, corresponding to
                total population for that age-risk group (summed
                over all compartments in the subpop model).
        """

        total_pop_age_risk = np.zeros((self.params.num_age_groups,
                                       self.params.num_risk_groups))

        # At initialization (before simulation is run), each
        #   compartment's current val is equivalent to the initial val
        #   specified in the state variables' init val JSON.
        for compartment in self.compartments.values():
            total_pop_age_risk += compartment.current_val

        return total_pop_age_risk

    def get_start_real_date(self):
        """
        Fetches `start_real_date` from `self.config` -- converts to
            proper datetime.date format if originally given as
            string.

        Returns:
            start_real_date (datetime.date):
                real-world date that corresponds to start of
                simulation.
        """

        start_real_date = self.config.start_real_date

        if not isinstance(start_real_date, datetime.date):
            try:
                start_real_date = \
                    datetime.datetime.strptime(start_real_date, "%Y-%m-%d").date()
            except ValueError:
                print("Error: The date format should be YYYY-MM-DD.")

        return start_real_date

    @abstractmethod
    def create_interaction_terms(self) -> sc.objdict:
        pass

    @abstractmethod
    def create_compartments(self) -> sc.objdict:
        pass

    @abstractmethod
    def create_transition_variables(self) -> sc.objdict:
        pass

    @abstractmethod
    def create_transition_variable_groups(self) -> sc.objdict:
        pass

    @abstractmethod
    def create_epi_metrics(self) -> sc.objdict:
        pass

    @abstractmethod
    def create_dynamic_vals(self) -> sc.objdict:
        pass

    @abstractmethod
    def create_schedules(self) -> sc.objdict:
        pass

    def modify_random_seed(self, new_seed_number) -> None:
        """
        Modifies model's `self.RNG` attribute in-place to new generator
        seeded at `new_seed_number`.

        Args:
            new_seed_number (int):
                used to re-seed model's random number generator.
        """

        self._bit_generator = np.random.MT19937(seed=new_seed_number)
        self.RNG = np.random.Generator(self._bit_generator)

    def simulate_until_day(self,
                           simulation_end_day: int) -> None:
        """
        Advance simulation model time until `simulation_end_day`.

        Advance time by iterating through simulation days,
        which are simulated by iterating through discretized
        timesteps.

        Save daily simulation data as history on each `Compartment`
        instance.

        Args:
            simulation_end_day (positive int):
                stop simulation at `simulation_end_day` (i.e. exclusive,
                simulate up to but not including `simulation_end_day`).
        """

        if self.current_simulation_day > simulation_end_day:
            raise SubpopModelError(f"Current day counter ({self.current_simulation_day}) "
                                   f"exceeds last simulation day ({simulation_end_day}).")

        save_daily_history = self.config.save_daily_history
        timesteps_per_day = self.config.timesteps_per_day

        # simulation_end_day is exclusive endpoint
        while self.current_simulation_day < simulation_end_day:

            self.prepare_daily_state()

            self.simulate_timesteps(timesteps_per_day)

            if save_daily_history:
                self.save_daily_history()

            self.increment_simulation_day()

    def simulate_timesteps(self,
                           num_timesteps: int) -> None:
        """
        Subroutine for `self.simulate_until_day`.

        Iterates through discretized timesteps to simulate next
        simulation day. Granularity of discretization is given by
        attribute `self.config.timesteps_per_day`.

        Properly scales transition variable realizations and changes
        in dynamic vals by specified timesteps per day.

        Args:
            num_timesteps (int):
                number of timesteps per day -- used to determine time interval
                length for discretization.
        """

        for timestep in range(num_timesteps):

            self.update_transition_rates()

            self.sample_transitions()

            self.update_epi_metrics()

            self.update_compartments()

            self.state.sync_to_current_vals(self.epi_metrics)
            self.state.sync_to_current_vals(self.compartments)

    def prepare_daily_state(self) -> None:
        """
        At beginning of each day, update current value of
        interaction terms, schedules, dynamic values --
        note that these are only updated once a day, not
        for every discretized timestep.
        """

        subpop_state = self.state
        subpop_params = self.params
        current_real_date = self.current_real_date

        # Important note: this order of updating is important,
        #   because schedules do not depend on other state variables,
        #   but dynamic vals may depend on schedules
        # Interaction terms may depend on both schedules
        #   and dynamic vals (but interaction terms are updated by
        #   the InterSubpopRepo, not on individual SubpopModel
        #   instances).

        schedules = self.schedules
        dynamic_vals = self.dynamic_vals

        # Update schedules for current day
        for schedule in schedules.values():
            schedule.update_current_val(subpop_params,
                                        current_real_date)

        self.state.sync_to_current_vals(schedules)

        # Update dynamic values for current day
        for dval in dynamic_vals.values():
            if dval.is_enabled:
                dval.update_current_val(subpop_state, subpop_params)

        self.state.sync_to_current_vals(dynamic_vals)

    def update_epi_metrics(self) -> None:
        """
        Update current value attribute on each associated
            `EpiMetric` instance.
        """

        state = self.state
        params = self.params
        timesteps_per_day = self.config.timesteps_per_day

        for metric in self.epi_metrics.values():
            metric.change_in_current_val = \
                metric.get_change_in_current_val(state,
                                                 params,
                                                 timesteps_per_day)
            metric.update_current_val()

    def update_transition_rates(self) -> None:
        """
        Compute current transition rates for each transition variable,
            and store this updated value on each variable's
            current_rate attribute.
        """

        state = self.state
        params = self.params

        for tvar in self.transition_variables.values():
            tvar.current_rate = tvar.get_current_rate(state, params)

    def sample_transitions(self) -> None:
        """
        For each transition variable, sample a random realization
            using its current rate. Handle jointly distributed transition
            variables first (using `TransitionVariableGroup` logic), then
            handle marginally distributed transition variables.
            Use `SubpopModel`'s `RNG` to generate random variables.
        """

        RNG = self.RNG
        timesteps_per_day = self.config.timesteps_per_day

        # Obtain transition variable realizations for jointly distributed transition variables
        #   (i.e. when there are multiple transition variable outflows from an epi compartment)
        for tvargroup in self.transition_variable_groups.values():
            tvargroup.current_vals_list = tvargroup.get_joint_realization(RNG,
                                                                          timesteps_per_day)
            tvargroup.update_transition_variable_realizations()

        # Obtain transition variable realizations for marginally distributed transition variables
        #   (i.e. when there is only one transition variable outflow from an epi compartment)
        # If transition variable is jointly distributed, then its realization has already
        #   been computed by its transition variable group container previously,
        #   so skip the marginal computation
        for tvar in self.transition_variables.values():
            if not tvar.is_jointly_distributed:
                tvar.current_val = tvar.get_realization(RNG, timesteps_per_day)

    def update_compartments(self) -> None:
        """
        Update current value of each `Compartment`, by
            looping through all `TransitionVariable` instances
            and subtracting/adding their current values
            from origin/destination compartments respectively.
        """

        for tvar in self.transition_variables.values():
            tvar.update_origin_outflow()
            tvar.update_destination_inflow()

        for compartment in self.compartments.values():
            compartment.update_current_val()

            # After updating the compartment's current value,
            #   reset its inflow and outflow attributes, to
            #   prepare for the next iteration.
            compartment.reset_inflow()
            compartment.reset_outflow()

    def increment_simulation_day(self) -> None:
        """
        Move day counters to next simulation day, both
            for integer simulation day and real date.
        """

        self.current_simulation_day += 1
        self.current_real_date += datetime.timedelta(days=1)

    def save_daily_history(self) -> None:
        """
        Update history at end of each day, not at end of every
           discretization timestep, to be efficient.
        Update history of state variables other than `Schedule`
           instances -- schedules do not have history
           `TransitionVariableGroup` instances also do not
           have history, so do not include.
        """
        for svar in self.interaction_terms.values() + \
                    self.compartments.values() + \
                    self.epi_metrics.values() + \
                    self.dynamic_vals.values():
            svar.save_history()

    def reset_simulation(self) -> None:
        """
        Reset simulation in-place. Subsequent method calls of
        `self.simulate_until_day` start from day 0, with original
        day 0 state.

        Returns `self.current_simulation_day` to 0.
        Restores state values to initial values.
        Clears history on model's state variables.
        Resets transition variables' `current_val` attribute to 0.

        WARNING:
            DOES NOT RESET THE MODEL'S RANDOM NUMBER GENERATOR TO
            ITS INITIAL STARTING SEED. RANDOM NUMBER GENERATOR WILL CONTINUE
            WHERE IT LEFT OFF.

        Use method `self.modify_random_seed` to reset model's `RNG` to its
        initial starting seed.
        """

        self.current_simulation_day = 0
        self.current_real_date = self.start_real_date

        # AGAIN, MUST BE CAREFUL ABOUT MUTABLE NUMPY ARRAYS -- MUST USE DEEP COPY
        for svar in self.all_state_variables.values():
            setattr(svar, "current_val", copy.deepcopy(svar.init_val))

        self.state.sync_to_current_vals(self.all_state_variables)
        self.reset()

    def reset(self) -> None:
        """
        Resets `self.history_vals_list` attribute of each `InteractionTerm`,
            `Compartment`, `EpiMetric`, and `DynamicVal` to an empty list.
            Clears current rates and current values of
            `TransitionVariable` and `TransitionVariableGroup` instances.
        """

        # Schedules do not have history since they are deterministic
        for svar in self.all_state_variables.values():
            svar.reset()

        for tvar in self.transition_variables.values():
            tvar.current_rate = None
            tvar.current_val = 0.0

        for tvargroup in self.transition_variable_groups.values():
            tvargroup.current_vals_list = []

    def find_name_by_compartment(self,
                                 target_compartment: Compartment) -> str:
        """
        Given `Compartment`, returns name of that `Compartment`.

        Args:
            target_compartment (Compartment):
                Compartment object with a name to look up

        Returns:
            str:
                Compartment name, given by the key to look
                it up in the `SubpopModel`'s compartments objdict
        """

        for name, compartment in self.compartments.items():
            if compartment == target_compartment:
                return name

    def display(self) -> None:
        """
        Prints structure of model (compartments and linkages),
            transition variables, epi metrics, schedules,
            and dynamic values.
        """

        # We build origin_dict so that we can print
        #   compartment transitions in an easy-to-read way --
        #   for connections between origin --> destination,
        #   we print all connections with the same origin
        #   consecutively
        origin_dict = defaultdict(list)

        # Each key in origin_dict is a string corresponding to
        #   an origin (Compartment) name
        # Each val in origin_dict is a list of 3-tuples
        # Each 3-tuple has the name of a destination (Compartment)
        #   connected to the given origin, the name of the transition
        #   variable connecting the origin and destination,
        #   and Boolean indicating if the transition variable is jointly
        #   distributed
        for tvar_name, tvar in self.transition_variables.items():
            origin_dict[self.find_name_by_compartment(tvar.origin)].append(
                (self.find_name_by_compartment(tvar.destination),
                 tvar_name, tvar.is_jointly_distributed))

        print(f"\n>>> Displaying SubpopModel {self.name}")

        print("\nCompartments and transition variables")
        print("=====================================")
        for origin_name, origin in self.compartments.items():
            for output in origin_dict[origin_name]:
                if output[2]:
                    print(f"{origin_name} --> {output[0]}, via {output[1]}: jointly distributed")
                else:
                    print(f"{origin_name} --> {output[0]}, via {output[1]}")

        print("\nEpi metrics")
        print("===========")
        for name in self.epi_metrics.keys():
            print(f"{name}")

        print("\nSchedules")
        print("=========")
        for name in self.schedules.keys():
            print(f"{name}")

        print("\nDynamic values")
        print("==============")
        for name, dynamic_val in self.dynamic_vals.items():
            if dynamic_val.is_enabled:
                print(f"{name}: enabled")
            else:
                print(f"{name}: disabled")
        print("\n")

__init__(state: SubpopState, params: SubpopParams, config: Config, RNG: np.random.Generator, name: str = '', metapop_model: MetapopModel = None)

Parameters:

Name	Type	Description	Default
state	SubpopState	holds current values of SubpopModel's state variables.	required
params	SubpopParams	data container for the model's epidemiological parameters, such as the "Greek letters" characterizing sojourn times in compartments.	required
config	Config	data container for the model's simulation configuration values.	required
RNG	Generator	used to generate stochastic transitions in the model and control reproducibility.	required
name	str	unique identifier of SubpopModel.	''
metapop_model	Optional[MetapopModel]	if not None, is the MetapopModel instance associated with this SubpopModel.	None

Source code in CLT_BaseModel/clt_base/base_components.py

def __init__(self,
             state: SubpopState,
             params: SubpopParams,
             config: Config,
             RNG: np.random.Generator,
             name: str = "",
             metapop_model: MetapopModel = None):

    """
    Params:
        state (SubpopState):
            holds current values of `SubpopModel`'s state variables.
        params (SubpopParams):
            data container for the model's epidemiological parameters,
            such as the "Greek letters" characterizing sojourn times
            in compartments.
        config (Config):
            data container for the model's simulation configuration values.
        RNG (np.random.Generator):
             used to generate stochastic transitions in the model and control
             reproducibility.
        name (str):
            unique identifier of `SubpopModel`.
        metapop_model (Optional[MetapopModel]):
            if not `None`, is the `MetapopModel` instance
            associated with this `SubpopModel`.
    """

    self.state = copy.deepcopy(state)
    self.params = copy.deepcopy(params)
    self.config = copy.deepcopy(config)

    self.RNG = RNG

    self.current_simulation_day = 0
    self.start_real_date = self.get_start_real_date()
    self.current_real_date = self.start_real_date

    self.metapop_model = None
    self.name = name

    self.interaction_terms = self.create_interaction_terms()
    self.compartments = self.create_compartments()
    self.transition_variables = self.create_transition_variables()
    self.transition_variable_groups = self.create_transition_variable_groups()

    # Some epi metrics depend on transition variables, so
    #   set up epi metrics after transition variables
    self.epi_metrics = self.create_epi_metrics()
    self.dynamic_vals = self.create_dynamic_vals()
    self.schedules = self.create_schedules()

    self.all_state_variables = {**self.interaction_terms,
                                **self.compartments,
                                **self.epi_metrics,
                                **self.dynamic_vals,
                                **self.schedules}

    # The model's state also has access to the model's
    #   compartments, epi_metrics, dynamic_vals, and schedules --
    #   so that state can easily retrieve each object's
    #   current_val and store it
    self.state.interaction_terms = self.interaction_terms
    self.state.compartments = self.compartments
    self.state.epi_metrics = self.epi_metrics
    self.state.dynamic_vals = self.dynamic_vals
    self.state.schedules = self.schedules

    self.params.total_pop_age_risk = self.compute_total_pop_age_risk()

compute_total_pop_age_risk() -> np.ndarray

Returns:

Type	Description
ndarray	np.ndarray: |A| x |R| array, where |A| is the number of age groups and |R| is the number of risk groups, corresponding to total population for that age-risk group (summed over all compartments in the subpop model).

Source code in CLT_BaseModel/clt_base/base_components.py

def compute_total_pop_age_risk(self) -> np.ndarray:
    """
    Returns:
        np.ndarray:
            |A| x |R| array, where |A| is the number of age groups
            and |R| is the number of risk groups, corresponding to
            total population for that age-risk group (summed
            over all compartments in the subpop model).
    """

    total_pop_age_risk = np.zeros((self.params.num_age_groups,
                                   self.params.num_risk_groups))

    # At initialization (before simulation is run), each
    #   compartment's current val is equivalent to the initial val
    #   specified in the state variables' init val JSON.
    for compartment in self.compartments.values():
        total_pop_age_risk += compartment.current_val

    return total_pop_age_risk

display() -> None

Prints structure of model (compartments and linkages), transition variables, epi metrics, schedules, and dynamic values.

Source code in CLT_BaseModel/clt_base/base_components.py

def display(self) -> None:
    """
    Prints structure of model (compartments and linkages),
        transition variables, epi metrics, schedules,
        and dynamic values.
    """

    # We build origin_dict so that we can print
    #   compartment transitions in an easy-to-read way --
    #   for connections between origin --> destination,
    #   we print all connections with the same origin
    #   consecutively
    origin_dict = defaultdict(list)

    # Each key in origin_dict is a string corresponding to
    #   an origin (Compartment) name
    # Each val in origin_dict is a list of 3-tuples
    # Each 3-tuple has the name of a destination (Compartment)
    #   connected to the given origin, the name of the transition
    #   variable connecting the origin and destination,
    #   and Boolean indicating if the transition variable is jointly
    #   distributed
    for tvar_name, tvar in self.transition_variables.items():
        origin_dict[self.find_name_by_compartment(tvar.origin)].append(
            (self.find_name_by_compartment(tvar.destination),
             tvar_name, tvar.is_jointly_distributed))

    print(f"\n>>> Displaying SubpopModel {self.name}")

    print("\nCompartments and transition variables")
    print("=====================================")
    for origin_name, origin in self.compartments.items():
        for output in origin_dict[origin_name]:
            if output[2]:
                print(f"{origin_name} --> {output[0]}, via {output[1]}: jointly distributed")
            else:
                print(f"{origin_name} --> {output[0]}, via {output[1]}")

    print("\nEpi metrics")
    print("===========")
    for name in self.epi_metrics.keys():
        print(f"{name}")

    print("\nSchedules")
    print("=========")
    for name in self.schedules.keys():
        print(f"{name}")

    print("\nDynamic values")
    print("==============")
    for name, dynamic_val in self.dynamic_vals.items():
        if dynamic_val.is_enabled:
            print(f"{name}: enabled")
        else:
            print(f"{name}: disabled")
    print("\n")

find_name_by_compartment(target_compartment: Compartment) -> str

Given Compartment, returns name of that Compartment.

Parameters:

Name	Type	Description	Default
target_compartment	Compartment	Compartment object with a name to look up	required

Returns:

Name	Type	Description
str	str	Compartment name, given by the key to look it up in the SubpopModel's compartments objdict

Source code in CLT_BaseModel/clt_base/base_components.py

def find_name_by_compartment(self,
                             target_compartment: Compartment) -> str:
    """
    Given `Compartment`, returns name of that `Compartment`.

    Args:
        target_compartment (Compartment):
            Compartment object with a name to look up

    Returns:
        str:
            Compartment name, given by the key to look
            it up in the `SubpopModel`'s compartments objdict
    """

    for name, compartment in self.compartments.items():
        if compartment == target_compartment:
            return name

get_start_real_date()

Fetches start_real_date from self.config -- converts to proper datetime.date format if originally given as string.

Returns:

Name	Type	Description
start_real_date	date	real-world date that corresponds to start of simulation.

Source code in CLT_BaseModel/clt_base/base_components.py

def get_start_real_date(self):
    """
    Fetches `start_real_date` from `self.config` -- converts to
        proper datetime.date format if originally given as
        string.

    Returns:
        start_real_date (datetime.date):
            real-world date that corresponds to start of
            simulation.
    """

    start_real_date = self.config.start_real_date

    if not isinstance(start_real_date, datetime.date):
        try:
            start_real_date = \
                datetime.datetime.strptime(start_real_date, "%Y-%m-%d").date()
        except ValueError:
            print("Error: The date format should be YYYY-MM-DD.")

    return start_real_date

increment_simulation_day() -> None

Move day counters to next simulation day, both for integer simulation day and real date.

Source code in CLT_BaseModel/clt_base/base_components.py

def increment_simulation_day(self) -> None:
    """
    Move day counters to next simulation day, both
        for integer simulation day and real date.
    """

    self.current_simulation_day += 1
    self.current_real_date += datetime.timedelta(days=1)

modify_random_seed(new_seed_number) -> None

Modifies model's self.RNG attribute in-place to new generator seeded at new_seed_number.

Parameters:

Name	Type	Description	Default
new_seed_number	int	used to re-seed model's random number generator.	required

Source code in CLT_BaseModel/clt_base/base_components.py

def modify_random_seed(self, new_seed_number) -> None:
    """
    Modifies model's `self.RNG` attribute in-place to new generator
    seeded at `new_seed_number`.

    Args:
        new_seed_number (int):
            used to re-seed model's random number generator.
    """

    self._bit_generator = np.random.MT19937(seed=new_seed_number)
    self.RNG = np.random.Generator(self._bit_generator)

prepare_daily_state() -> None

At beginning of each day, update current value of interaction terms, schedules, dynamic values -- note that these are only updated once a day, not for every discretized timestep.

Source code in CLT_BaseModel/clt_base/base_components.py

def prepare_daily_state(self) -> None:
    """
    At beginning of each day, update current value of
    interaction terms, schedules, dynamic values --
    note that these are only updated once a day, not
    for every discretized timestep.
    """

    subpop_state = self.state
    subpop_params = self.params
    current_real_date = self.current_real_date

    # Important note: this order of updating is important,
    #   because schedules do not depend on other state variables,
    #   but dynamic vals may depend on schedules
    # Interaction terms may depend on both schedules
    #   and dynamic vals (but interaction terms are updated by
    #   the InterSubpopRepo, not on individual SubpopModel
    #   instances).

    schedules = self.schedules
    dynamic_vals = self.dynamic_vals

    # Update schedules for current day
    for schedule in schedules.values():
        schedule.update_current_val(subpop_params,
                                    current_real_date)

    self.state.sync_to_current_vals(schedules)

    # Update dynamic values for current day
    for dval in dynamic_vals.values():
        if dval.is_enabled:
            dval.update_current_val(subpop_state, subpop_params)

    self.state.sync_to_current_vals(dynamic_vals)

reset() -> None

Resets self.history_vals_list attribute of each InteractionTerm, Compartment, EpiMetric, and DynamicVal to an empty list. Clears current rates and current values of TransitionVariable and TransitionVariableGroup instances.

Source code in CLT_BaseModel/clt_base/base_components.py

def reset(self) -> None:
    """
    Resets `self.history_vals_list` attribute of each `InteractionTerm`,
        `Compartment`, `EpiMetric`, and `DynamicVal` to an empty list.
        Clears current rates and current values of
        `TransitionVariable` and `TransitionVariableGroup` instances.
    """

    # Schedules do not have history since they are deterministic
    for svar in self.all_state_variables.values():
        svar.reset()

    for tvar in self.transition_variables.values():
        tvar.current_rate = None
        tvar.current_val = 0.0

    for tvargroup in self.transition_variable_groups.values():
        tvargroup.current_vals_list = []

reset_simulation() -> None

Reset simulation in-place. Subsequent method calls of self.simulate_until_day start from day 0, with original day 0 state.

Returns self.current_simulation_day to 0. Restores state values to initial values. Clears history on model's state variables. Resets transition variables' current_val attribute to 0.

WARNING

DOES NOT RESET THE MODEL'S RANDOM NUMBER GENERATOR TO ITS INITIAL STARTING SEED. RANDOM NUMBER GENERATOR WILL CONTINUE WHERE IT LEFT OFF.

Use method self.modify_random_seed to reset model's RNG to its initial starting seed.

Source code in CLT_BaseModel/clt_base/base_components.py

def reset_simulation(self) -> None:
    """
    Reset simulation in-place. Subsequent method calls of
    `self.simulate_until_day` start from day 0, with original
    day 0 state.

    Returns `self.current_simulation_day` to 0.
    Restores state values to initial values.
    Clears history on model's state variables.
    Resets transition variables' `current_val` attribute to 0.

    WARNING:
        DOES NOT RESET THE MODEL'S RANDOM NUMBER GENERATOR TO
        ITS INITIAL STARTING SEED. RANDOM NUMBER GENERATOR WILL CONTINUE
        WHERE IT LEFT OFF.

    Use method `self.modify_random_seed` to reset model's `RNG` to its
    initial starting seed.
    """

    self.current_simulation_day = 0
    self.current_real_date = self.start_real_date

    # AGAIN, MUST BE CAREFUL ABOUT MUTABLE NUMPY ARRAYS -- MUST USE DEEP COPY
    for svar in self.all_state_variables.values():
        setattr(svar, "current_val", copy.deepcopy(svar.init_val))

    self.state.sync_to_current_vals(self.all_state_variables)
    self.reset()

sample_transitions() -> None

For each transition variable, sample a random realization using its current rate. Handle jointly distributed transition variables first (using TransitionVariableGroup logic), then handle marginally distributed transition variables. Use SubpopModel's RNG to generate random variables.

Source code in CLT_BaseModel/clt_base/base_components.py

def sample_transitions(self) -> None:
    """
    For each transition variable, sample a random realization
        using its current rate. Handle jointly distributed transition
        variables first (using `TransitionVariableGroup` logic), then
        handle marginally distributed transition variables.
        Use `SubpopModel`'s `RNG` to generate random variables.
    """

    RNG = self.RNG
    timesteps_per_day = self.config.timesteps_per_day

    # Obtain transition variable realizations for jointly distributed transition variables
    #   (i.e. when there are multiple transition variable outflows from an epi compartment)
    for tvargroup in self.transition_variable_groups.values():
        tvargroup.current_vals_list = tvargroup.get_joint_realization(RNG,
                                                                      timesteps_per_day)
        tvargroup.update_transition_variable_realizations()

    # Obtain transition variable realizations for marginally distributed transition variables
    #   (i.e. when there is only one transition variable outflow from an epi compartment)
    # If transition variable is jointly distributed, then its realization has already
    #   been computed by its transition variable group container previously,
    #   so skip the marginal computation
    for tvar in self.transition_variables.values():
        if not tvar.is_jointly_distributed:
            tvar.current_val = tvar.get_realization(RNG, timesteps_per_day)

save_daily_history() -> None

Update history at end of each day, not at end of every discretization timestep, to be efficient. Update history of state variables other than Schedule instances -- schedules do not have history TransitionVariableGroup instances also do not have history, so do not include.

Source code in CLT_BaseModel/clt_base/base_components.py

def save_daily_history(self) -> None:
    """
    Update history at end of each day, not at end of every
       discretization timestep, to be efficient.
    Update history of state variables other than `Schedule`
       instances -- schedules do not have history
       `TransitionVariableGroup` instances also do not
       have history, so do not include.
    """
    for svar in self.interaction_terms.values() + \
                self.compartments.values() + \
                self.epi_metrics.values() + \
                self.dynamic_vals.values():
        svar.save_history()

simulate_timesteps(num_timesteps: int) -> None

Subroutine for self.simulate_until_day.

Iterates through discretized timesteps to simulate next simulation day. Granularity of discretization is given by attribute self.config.timesteps_per_day.

Properly scales transition variable realizations and changes in dynamic vals by specified timesteps per day.

Parameters:

Name	Type	Description	Default
num_timesteps	int	number of timesteps per day -- used to determine time interval length for discretization.	required

Source code in CLT_BaseModel/clt_base/base_components.py

def simulate_timesteps(self,
                       num_timesteps: int) -> None:
    """
    Subroutine for `self.simulate_until_day`.

    Iterates through discretized timesteps to simulate next
    simulation day. Granularity of discretization is given by
    attribute `self.config.timesteps_per_day`.

    Properly scales transition variable realizations and changes
    in dynamic vals by specified timesteps per day.

    Args:
        num_timesteps (int):
            number of timesteps per day -- used to determine time interval
            length for discretization.
    """

    for timestep in range(num_timesteps):

        self.update_transition_rates()

        self.sample_transitions()

        self.update_epi_metrics()

        self.update_compartments()

        self.state.sync_to_current_vals(self.epi_metrics)
        self.state.sync_to_current_vals(self.compartments)

simulate_until_day(simulation_end_day: int) -> None

Advance simulation model time until simulation_end_day.

Advance time by iterating through simulation days, which are simulated by iterating through discretized timesteps.

Save daily simulation data as history on each Compartment instance.

Parameters:

Name	Type	Description	Default
simulation_end_day	positive int	stop simulation at simulation_end_day (i.e. exclusive, simulate up to but not including simulation_end_day).	required

Source code in CLT_BaseModel/clt_base/base_components.py

def simulate_until_day(self,
                       simulation_end_day: int) -> None:
    """
    Advance simulation model time until `simulation_end_day`.

    Advance time by iterating through simulation days,
    which are simulated by iterating through discretized
    timesteps.

    Save daily simulation data as history on each `Compartment`
    instance.

    Args:
        simulation_end_day (positive int):
            stop simulation at `simulation_end_day` (i.e. exclusive,
            simulate up to but not including `simulation_end_day`).
    """

    if self.current_simulation_day > simulation_end_day:
        raise SubpopModelError(f"Current day counter ({self.current_simulation_day}) "
                               f"exceeds last simulation day ({simulation_end_day}).")

    save_daily_history = self.config.save_daily_history
    timesteps_per_day = self.config.timesteps_per_day

    # simulation_end_day is exclusive endpoint
    while self.current_simulation_day < simulation_end_day:

        self.prepare_daily_state()

        self.simulate_timesteps(timesteps_per_day)

        if save_daily_history:
            self.save_daily_history()

        self.increment_simulation_day()

update_compartments() -> None

Update current value of each Compartment, by looping through all TransitionVariable instances and subtracting/adding their current values from origin/destination compartments respectively.

Source code in CLT_BaseModel/clt_base/base_components.py

def update_compartments(self) -> None:
    """
    Update current value of each `Compartment`, by
        looping through all `TransitionVariable` instances
        and subtracting/adding their current values
        from origin/destination compartments respectively.
    """

    for tvar in self.transition_variables.values():
        tvar.update_origin_outflow()
        tvar.update_destination_inflow()

    for compartment in self.compartments.values():
        compartment.update_current_val()

        # After updating the compartment's current value,
        #   reset its inflow and outflow attributes, to
        #   prepare for the next iteration.
        compartment.reset_inflow()
        compartment.reset_outflow()

update_epi_metrics() -> None

Update current value attribute on each associated EpiMetric instance.

Source code in CLT_BaseModel/clt_base/base_components.py

def update_epi_metrics(self) -> None:
    """
    Update current value attribute on each associated
        `EpiMetric` instance.
    """

    state = self.state
    params = self.params
    timesteps_per_day = self.config.timesteps_per_day

    for metric in self.epi_metrics.values():
        metric.change_in_current_val = \
            metric.get_change_in_current_val(state,
                                             params,
                                             timesteps_per_day)
        metric.update_current_val()

update_transition_rates() -> None

Compute current transition rates for each transition variable, and store this updated value on each variable's current_rate attribute.

Source code in CLT_BaseModel/clt_base/base_components.py

def update_transition_rates(self) -> None:
    """
    Compute current transition rates for each transition variable,
        and store this updated value on each variable's
        current_rate attribute.
    """

    state = self.state
    params = self.params

    for tvar in self.transition_variables.values():
        tvar.current_rate = tvar.get_current_rate(state, params)

SubpopModelError

Bases: Exception

Custom exceptions for subpopulation simulation model errors.

Source code in CLT_BaseModel/clt_base/base_components.py

class SubpopModelError(Exception):
    """Custom exceptions for subpopulation simulation model errors."""
    pass

SubpopParams

Bases: ABC

Data container for pre-specified and fixed epidemiological parameters in model.

Assume that SubpopParams fields are constant or piecewise constant throughout the simulation. For variables that are more complicated and time-dependent, use an EpiMetric instead.

Source code in CLT_BaseModel/clt_base/base_components.py

@dataclass
class SubpopParams(ABC):
    """
    Data container for pre-specified and fixed epidemiological
    parameters in model.

    Assume that `SubpopParams` fields are constant or piecewise
    constant throughout the simulation. For variables that
    are more complicated and time-dependent, use an `EpiMetric`
    instead.
    """

    pass

SubpopState

Bases: ABC

Holds current values of SubpopModel's simulation state.

Source code in CLT_BaseModel/clt_base/base_components.py

@dataclass
class SubpopState(ABC):
    """
    Holds current values of `SubpopModel`'s simulation state.
    """

    def sync_to_current_vals(self, lookup_dict: dict):
        """
        Updates `SubpopState`'s attributes according to
        data in `lookup_dict.` Keys of `lookup_dict` must match
        names of attributes of `SubpopState` instance.
        """

        for name, item in lookup_dict.items():
            setattr(self, name, item.current_val)

sync_to_current_vals(lookup_dict: dict)

Updates SubpopState's attributes according to data in lookup_dict. Keys of lookup_dict must match names of attributes of SubpopState instance.

Source code in CLT_BaseModel/clt_base/base_components.py

def sync_to_current_vals(self, lookup_dict: dict):
    """
    Updates `SubpopState`'s attributes according to
    data in `lookup_dict.` Keys of `lookup_dict` must match
    names of attributes of `SubpopState` instance.
    """

    for name, item in lookup_dict.items():
        setattr(self, name, item.current_val)

TransitionTypes

Bases: str, Enum

Source code in CLT_BaseModel/clt_base/base_components.py

class TransitionTypes(str, Enum):
    BINOMIAL = "binomial"
    BINOMIAL_DETERMINISTIC = "binomial_deterministic"
    BINOMIAL_TAYLOR_APPROX = "binomial_taylor_approx"
    BINOMIAL_TAYLOR_APPROX_DETERMINISTIC = "binomial_taylor_approx_deterministic"
    POISSON = "poisson"
    POISSON_DETERMINISTIC = "poisson_deterministic"

TransitionVariable

Bases: ABC

Abstract base class for transition variables in epidemiological model.

For example, in an S-I-R model, the new number infected every iteration (the number going from S to I) in an iteration is modeled as a TransitionVariable subclass, with a concrete implementation of the abstract method self.get_current_rate.

When an instance is initialized, its self.get_realization attribute is dynamically assigned, just like in the case of TransitionVariableGroup instantiation.

Attributes:

Name	Type	Description
_transition_type	str	only values defined in TransitionTypes are valid, specifying probability distribution of transitions between compartments.
get_current_rate	function	provides specific implementation for computing current rate as a function of current subpopulation simulation state and epidemiological parameters.
current_rate	ndarray	holds output from self.get_current_rate method -- used to generate random variable realizations for transitions between compartments.
current_val	ndarray	holds realization of random variable parameterized by self.current_rate.
history_vals_list	list[ndarray]	each element is the same size of self.current_val, holds history of transition variable realizations for age-risk groups -- element t corresponds to previous self.current_val value at end of simulation day t.

See __init__ docstring for other attributes.

Source code in CLT_BaseModel/clt_base/base_components.py

class TransitionVariable(ABC):
    """
    Abstract base class for transition variables in
    epidemiological model.

    For example, in an S-I-R model, the new number infected
    every iteration (the number going from S to I) in an iteration
    is modeled as a `TransitionVariable` subclass, with a concrete
    implementation of the abstract method `self.get_current_rate`.

    When an instance is initialized, its `self.get_realization` attribute
    is dynamically assigned, just like in the case of
    `TransitionVariableGroup` instantiation.

    Attributes:
        _transition_type (str):
            only values defined in `TransitionTypes` are valid, specifying
            probability distribution of transitions between compartments.
        get_current_rate (function):
            provides specific implementation for computing current rate
            as a function of current subpopulation simulation state and
            epidemiological parameters.
        current_rate (np.ndarray):
            holds output from `self.get_current_rate` method -- used to generate
            random variable realizations for transitions between compartments.
        current_val (np.ndarray):
            holds realization of random variable parameterized by
            `self.current_rate`.
        history_vals_list (list[np.ndarray]):
            each element is the same size of `self.current_val`, holds
            history of transition variable realizations for age-risk
            groups -- element t corresponds to previous `self.current_val`
            value at end of simulation day t.

    See `__init__` docstring for other attributes.
    """

    def __init__(self,
                 origin: Compartment,
                 destination: Compartment,
                 transition_type: TransitionTypes,
                 is_jointly_distributed: str=False):
        """
        Parameters:
            origin (Compartment):
                `Compartment` from which `TransitionVariable` exits.
            destination (Compartment):
                `Compartment` that the `TransitionVariable` enters.
            transition_type (str):
                only values defined in `TransitionTypes` are valid, specifying
                probability distribution of transitions between compartments.
            is_jointly_distributed (bool):
                indicates if transition quantity must be jointly computed
                (i.e. if there are multiple outflows from the origin compartment).
        """

        self.origin = origin
        self.destination = destination

        # Also see __init__ method in TransitionVariableGroup class.
        #   The structure is similar.
        self._transition_type = transition_type
        self._is_jointly_distributed = is_jointly_distributed

        if is_jointly_distributed:
            self.get_realization = None
        else:
            self.get_realization = getattr(self, "get_" + transition_type + "_realization")

        self.current_rate = None
        self.current_val = 0

        self.history_vals_list = []

    @property
    def transition_type(self) -> TransitionTypes:
        return self._transition_type

    @property
    def is_jointly_distributed(self) -> bool:
        return self._is_jointly_distributed

    @abstractmethod
    def get_current_rate(self,
                         state: SubpopState,
                         params: SubpopParams) -> np.ndarray:
        """
        Computes and returns current rate of transition variable,
        based on current state of the simulation and epidemiological parameters.
        Output should be a numpy array of size |A| x |R|, where |A| is the
        number of age groups and |R| is number of risk groups.

        Args:
            state (SubpopState):
                holds subpopulation simulation state
                (current values of `StateVariable` instances).
            params (SubpopParams):
                holds values of epidemiological parameters.

        Returns:
            np.ndarray:
                holds age-risk transition rate,
                must be same shape as origin.init_val,
                i.e. be size |A| x |R|, where |A| is the number of age groups
                and |R| is number of risk groups.
        """
        pass

    def update_origin_outflow(self) -> None:
        """
        Adds current realization of `TransitionVariable` to
            its origin `Compartment`'s current_outflow.
            Used to compute total number leaving that
            origin `Compartment`.
        """

        self.origin.current_outflow = self.origin.current_outflow + self.current_val

    def update_destination_inflow(self) -> None:
        """
        Adds current realization of `TransitionVariable` to
            its destination `Compartment`'s `current_inflow`.
            Used to compute total number leaving that
            destination `Compartment`.
        """

        self.destination.current_inflow = self.destination.current_inflow + self.current_val

    def save_history(self) -> None:
        """
        Saves current value to history by appending `self.current_val`
            attribute to `self.history_vals_list` in place.

        Deep copying is CRUCIAL because `self.current_val` is a mutable
            np.ndarray -- without deep copying, `self.history_vals_list` would
            have the same value for all elements.
        """
        self.history_vals_list.append(copy.deepcopy(self.current_val))

    def reset(self) -> None:
        """
        Resets `self.history_vals_list` attribute to empty list.
        """

        self.history_vals_list = []

    def get_realization(self,
                        RNG: np.random.Generator,
                        num_timesteps: int) -> np.ndarray:
        """
        This method gets assigned to one of the following methods
            based on the `TransitionVariable` transition type:
            `self.get_binomial_realization`, `self.get_binomial_taylor_approx_realization`,
            `self.get_poisson_realization`, `self.get_binomial_deterministic_realization`,
            `self.get_binomial_taylor_approx_deterministic_realization`,
            `self.get_poisson_deterministic_realization`. This is done so that
            the same method `self.get_realization` can be called regardless of
            transition type.

        Parameters:
            RNG (np.random.Generator object):
                 used to generate stochastic transitions in the model and control
                 reproducibility.
            num_timesteps (int):
                number of timesteps per day -- used to determine time interval
                length for discretization.
        """

        pass

    def get_binomial_realization(self,
                                 RNG: np.random.Generator,
                                 num_timesteps: int) -> np.ndarray:
        """
        Uses `RNG` to generate binomial random variable with
            number of trials equal to population count in the
            origin `Compartment` and probability computed from
            a function of the `TransitionVariable`'s current rate
            -- see the `approx_binomial_probability_from_rate`
            function for details.

        Parameters:
            RNG (np.random.Generator object):
                 used to generate stochastic transitions in the model and control
                 reproducibility.
            num_timesteps (int):
                number of timesteps per day -- used to determine time interval
                length for discretization.

        Returns:
            np.ndarray:
                size |A| x |R|, where |A| is the number of age groups and
                |R| is number of risk groups.
        """

        return RNG.binomial(n=np.asarray(self.base_count, dtype=int),
                            p=approx_binomial_probability_from_rate(self.current_rate, 1.0 / num_timesteps))

    def get_binomial_taylor_approx_realization(self,
                                               RNG: np.random.Generator,
                                               num_timesteps: int) -> np.ndarray:
        """
        Uses `RNG` to generate binomial random variable with
            number of trials equal to population count in the
            origin `Compartment` and probability equal to
            the `TransitionVariable`'s `current_rate` / `num_timesteps`.

        Parameters:
            RNG (np.random.Generator object):
                 used to generate stochastic transitions in the model and control
                 reproducibility.
            num_timesteps (int):
                number of timesteps per day -- used to determine time interval
                length for discretization.

        Returns:
            np.ndarray:
                size |A| x |R|, where |A| is the number of age groups and L
                is number of risk groups.
        """
        return RNG.binomial(n=np.asarray(self.base_count, dtype=int),
                            p=self.current_rate * (1.0 / num_timesteps))

    def get_poisson_realization(self,
                                RNG: np.random.Generator,
                                num_timesteps: int) -> np.ndarray:
        """
        Uses `RNG` to generate Poisson random variable with
            rate equal to (population count in the
            origin `Compartment` x the `TransitionVariable`'s
            `current_rate` / `num_timesteps`)

        Parameters:
            RNG (np.random.Generator object):
                 used to generate stochastic transitions in the model and control
                 reproducibility.
            num_timesteps (int):
                number of timesteps per day -- used to determine time interval
                length for discretization.

        Returns:
            np.ndarray:
                size |A| x |R|, where |A| is the number of age groups and
                |R| is number of risk groups.
        """
        return RNG.poisson(self.base_count * self.current_rate / float(num_timesteps))

    def get_binomial_deterministic_realization(self,
                                               RNG: np.random.Generator,
                                               num_timesteps: int) -> np.ndarray:
        """
        Deterministically returns mean of binomial distribution
            (number of trials x probability), where number of trials
            equals population count in the origin `Compartment` and
            probability is computed from a function of the `TransitionVariable`'s
            current rate -- see the `approx_binomial_probability_from_rate`
            function for details.

        Parameters:
            RNG (np.random.Generator object):
                NOT USED -- only included so that get_realization has
                the same function arguments regardless of transition type.
            num_timesteps (int):
                number of timesteps per day -- used to determine time interval
                length for discretization.

        Returns:
            np.ndarray:
                size |A| x |R|, where |A| is the number of age groups and
                |R| is number of risk groups.
        """

        return np.asarray(self.base_count *
                          approx_binomial_probability_from_rate(self.current_rate, 1.0 / num_timesteps),
                          dtype=int)

    def get_binomial_taylor_approx_deterministic_realization(self,
                                                             RNG: np.random.Generator,
                                                             num_timesteps: int) -> np.ndarray:
        """
        Deterministically returns mean of binomial distribution
            (number of trials x probability), where number of trials
            equals population count in the origin `Compartment` and
            probability equals the `TransitionVariable`'s `current_rate` /
            `num_timesteps`.

        Parameters:
            RNG (np.random.Generator object):
                NOT USED -- only included so that get_realization has
                the same function arguments regardless of transition type.
            num_timesteps (int):
                number of timesteps per day -- used to determine time interval
                length for discretization.

        Returns:
            np.ndarray:
                size |A| x |R|, where |A| is the number of age groups and
                |R| is number of risk groups.
        """

        return np.asarray(self.base_count * self.current_rate / num_timesteps, dtype=int)

    def get_poisson_deterministic_realization(self,
                                              RNG: np.random.Generator,
                                              num_timesteps: int) -> np.ndarray:
        """
        Deterministically returns mean of Poisson distribution,
            given by (population count in the origin `Compartment` x
            `TransitionVariable`'s `current_rate` / `num_timesteps`).

        Parameters:
            RNG (np.random.Generator object):
                NOT USED -- only included so that get_realization has
                the same function arguments regardless of transition type.
            num_timesteps (int):
                number of timesteps per day -- used to determine time interval
                length for discretization.

        Returns:
            np.ndarray:
                size |A| x |R|, where |A| is the number of age groups and
                |R| is number of risk groups.
        """

        return np.asarray(self.base_count * self.current_rate / num_timesteps, dtype=int)

    @property
    def base_count(self) -> np.ndarray:
        return self.origin.current_val

__init__(origin: Compartment, destination: Compartment, transition_type: TransitionTypes, is_jointly_distributed: str = False)

Parameters:

Name	Type	Description	Default
origin	Compartment	Compartment from which TransitionVariable exits.	required
destination	Compartment	Compartment that the TransitionVariable enters.	required
transition_type	str	only values defined in TransitionTypes are valid, specifying probability distribution of transitions between compartments.	required
is_jointly_distributed	bool	indicates if transition quantity must be jointly computed (i.e. if there are multiple outflows from the origin compartment).	False

Source code in CLT_BaseModel/clt_base/base_components.py

def __init__(self,
             origin: Compartment,
             destination: Compartment,
             transition_type: TransitionTypes,
             is_jointly_distributed: str=False):
    """
    Parameters:
        origin (Compartment):
            `Compartment` from which `TransitionVariable` exits.
        destination (Compartment):
            `Compartment` that the `TransitionVariable` enters.
        transition_type (str):
            only values defined in `TransitionTypes` are valid, specifying
            probability distribution of transitions between compartments.
        is_jointly_distributed (bool):
            indicates if transition quantity must be jointly computed
            (i.e. if there are multiple outflows from the origin compartment).
    """

    self.origin = origin
    self.destination = destination

    # Also see __init__ method in TransitionVariableGroup class.
    #   The structure is similar.
    self._transition_type = transition_type
    self._is_jointly_distributed = is_jointly_distributed

    if is_jointly_distributed:
        self.get_realization = None
    else:
        self.get_realization = getattr(self, "get_" + transition_type + "_realization")

    self.current_rate = None
    self.current_val = 0

    self.history_vals_list = []

get_binomial_deterministic_realization(RNG: np.random.Generator, num_timesteps: int) -> np.ndarray

Deterministically returns mean of binomial distribution (number of trials x probability), where number of trials equals population count in the origin Compartment and probability is computed from a function of the TransitionVariable's current rate -- see the approx_binomial_probability_from_rate function for details.

Parameters:

Name	Type	Description	Default
RNG	np.random.Generator object	NOT USED -- only included so that get_realization has the same function arguments regardless of transition type.	required
num_timesteps	int	number of timesteps per day -- used to determine time interval length for discretization.	required

Returns:

Type	Description
ndarray	np.ndarray: size |A| x |R|, where |A| is the number of age groups and |R| is number of risk groups.

Source code in CLT_BaseModel/clt_base/base_components.py

def get_binomial_deterministic_realization(self,
                                           RNG: np.random.Generator,
                                           num_timesteps: int) -> np.ndarray:
    """
    Deterministically returns mean of binomial distribution
        (number of trials x probability), where number of trials
        equals population count in the origin `Compartment` and
        probability is computed from a function of the `TransitionVariable`'s
        current rate -- see the `approx_binomial_probability_from_rate`
        function for details.

    Parameters:
        RNG (np.random.Generator object):
            NOT USED -- only included so that get_realization has
            the same function arguments regardless of transition type.
        num_timesteps (int):
            number of timesteps per day -- used to determine time interval
            length for discretization.

    Returns:
        np.ndarray:
            size |A| x |R|, where |A| is the number of age groups and
            |R| is number of risk groups.
    """

    return np.asarray(self.base_count *
                      approx_binomial_probability_from_rate(self.current_rate, 1.0 / num_timesteps),
                      dtype=int)

get_binomial_realization(RNG: np.random.Generator, num_timesteps: int) -> np.ndarray

Uses RNG to generate binomial random variable with number of trials equal to population count in the origin Compartment and probability computed from a function of the TransitionVariable's current rate -- see the approx_binomial_probability_from_rate function for details.

Parameters:

Name	Type	Description	Default
RNG	np.random.Generator object	used to generate stochastic transitions in the model and control reproducibility.	required
num_timesteps	int	number of timesteps per day -- used to determine time interval length for discretization.	required

Returns:

Type	Description
ndarray	np.ndarray: size |A| x |R|, where |A| is the number of age groups and |R| is number of risk groups.

Source code in CLT_BaseModel/clt_base/base_components.py

def get_binomial_realization(self,
                             RNG: np.random.Generator,
                             num_timesteps: int) -> np.ndarray:
    """
    Uses `RNG` to generate binomial random variable with
        number of trials equal to population count in the
        origin `Compartment` and probability computed from
        a function of the `TransitionVariable`'s current rate
        -- see the `approx_binomial_probability_from_rate`
        function for details.

    Parameters:
        RNG (np.random.Generator object):
             used to generate stochastic transitions in the model and control
             reproducibility.
        num_timesteps (int):
            number of timesteps per day -- used to determine time interval
            length for discretization.

    Returns:
        np.ndarray:
            size |A| x |R|, where |A| is the number of age groups and
            |R| is number of risk groups.
    """

    return RNG.binomial(n=np.asarray(self.base_count, dtype=int),
                        p=approx_binomial_probability_from_rate(self.current_rate, 1.0 / num_timesteps))

get_binomial_taylor_approx_deterministic_realization(RNG: np.random.Generator, num_timesteps: int) -> np.ndarray

Deterministically returns mean of binomial distribution (number of trials x probability), where number of trials equals population count in the origin Compartment and probability equals the TransitionVariable's current_rate / num_timesteps.

Parameters:

Name	Type	Description	Default
RNG	np.random.Generator object	NOT USED -- only included so that get_realization has the same function arguments regardless of transition type.	required
num_timesteps	int	number of timesteps per day -- used to determine time interval length for discretization.	required

Returns:

Type	Description
ndarray	np.ndarray: size |A| x |R|, where |A| is the number of age groups and |R| is number of risk groups.

Source code in CLT_BaseModel/clt_base/base_components.py

def get_binomial_taylor_approx_deterministic_realization(self,
                                                         RNG: np.random.Generator,
                                                         num_timesteps: int) -> np.ndarray:
    """
    Deterministically returns mean of binomial distribution
        (number of trials x probability), where number of trials
        equals population count in the origin `Compartment` and
        probability equals the `TransitionVariable`'s `current_rate` /
        `num_timesteps`.

    Parameters:
        RNG (np.random.Generator object):
            NOT USED -- only included so that get_realization has
            the same function arguments regardless of transition type.
        num_timesteps (int):
            number of timesteps per day -- used to determine time interval
            length for discretization.

    Returns:
        np.ndarray:
            size |A| x |R|, where |A| is the number of age groups and
            |R| is number of risk groups.
    """

    return np.asarray(self.base_count * self.current_rate / num_timesteps, dtype=int)

get_binomial_taylor_approx_realization(RNG: np.random.Generator, num_timesteps: int) -> np.ndarray

Uses RNG to generate binomial random variable with number of trials equal to population count in the origin Compartment and probability equal to the TransitionVariable's current_rate / num_timesteps.

Parameters:

Name	Type	Description	Default
RNG	np.random.Generator object	used to generate stochastic transitions in the model and control reproducibility.	required
num_timesteps	int	number of timesteps per day -- used to determine time interval length for discretization.	required

Returns:

Type	Description
ndarray	np.ndarray: size |A| x |R|, where |A| is the number of age groups and L is number of risk groups.

Source code in CLT_BaseModel/clt_base/base_components.py

def get_binomial_taylor_approx_realization(self,
                                           RNG: np.random.Generator,
                                           num_timesteps: int) -> np.ndarray:
    """
    Uses `RNG` to generate binomial random variable with
        number of trials equal to population count in the
        origin `Compartment` and probability equal to
        the `TransitionVariable`'s `current_rate` / `num_timesteps`.

    Parameters:
        RNG (np.random.Generator object):
             used to generate stochastic transitions in the model and control
             reproducibility.
        num_timesteps (int):
            number of timesteps per day -- used to determine time interval
            length for discretization.

    Returns:
        np.ndarray:
            size |A| x |R|, where |A| is the number of age groups and L
            is number of risk groups.
    """
    return RNG.binomial(n=np.asarray(self.base_count, dtype=int),
                        p=self.current_rate * (1.0 / num_timesteps))

get_current_rate(state: SubpopState, params: SubpopParams) -> np.ndarray

Computes and returns current rate of transition variable, based on current state of the simulation and epidemiological parameters. Output should be a numpy array of size |A| x |R|, where |A| is the number of age groups and |R| is number of risk groups.

Parameters:

Name	Type	Description	Default
state	SubpopState	holds subpopulation simulation state (current values of StateVariable instances).	required
params	SubpopParams	holds values of epidemiological parameters.	required

Returns:

Type	Description
ndarray	np.ndarray: holds age-risk transition rate, must be same shape as origin.init_val, i.e. be size |A| x |R|, where |A| is the number of age groups and |R| is number of risk groups.

Source code in CLT_BaseModel/clt_base/base_components.py

@abstractmethod
def get_current_rate(self,
                     state: SubpopState,
                     params: SubpopParams) -> np.ndarray:
    """
    Computes and returns current rate of transition variable,
    based on current state of the simulation and epidemiological parameters.
    Output should be a numpy array of size |A| x |R|, where |A| is the
    number of age groups and |R| is number of risk groups.

    Args:
        state (SubpopState):
            holds subpopulation simulation state
            (current values of `StateVariable` instances).
        params (SubpopParams):
            holds values of epidemiological parameters.

    Returns:
        np.ndarray:
            holds age-risk transition rate,
            must be same shape as origin.init_val,
            i.e. be size |A| x |R|, where |A| is the number of age groups
            and |R| is number of risk groups.
    """
    pass

get_poisson_deterministic_realization(RNG: np.random.Generator, num_timesteps: int) -> np.ndarray

Deterministically returns mean of Poisson distribution, given by (population count in the origin Compartment x TransitionVariable's current_rate / num_timesteps).

Parameters:

Name	Type	Description	Default
RNG	np.random.Generator object	NOT USED -- only included so that get_realization has the same function arguments regardless of transition type.	required
num_timesteps	int	number of timesteps per day -- used to determine time interval length for discretization.	required

Returns:

Type	Description
ndarray	np.ndarray: size |A| x |R|, where |A| is the number of age groups and |R| is number of risk groups.

Source code in CLT_BaseModel/clt_base/base_components.py

def get_poisson_deterministic_realization(self,
                                          RNG: np.random.Generator,
                                          num_timesteps: int) -> np.ndarray:
    """
    Deterministically returns mean of Poisson distribution,
        given by (population count in the origin `Compartment` x
        `TransitionVariable`'s `current_rate` / `num_timesteps`).

    Parameters:
        RNG (np.random.Generator object):
            NOT USED -- only included so that get_realization has
            the same function arguments regardless of transition type.
        num_timesteps (int):
            number of timesteps per day -- used to determine time interval
            length for discretization.

    Returns:
        np.ndarray:
            size |A| x |R|, where |A| is the number of age groups and
            |R| is number of risk groups.
    """

    return np.asarray(self.base_count * self.current_rate / num_timesteps, dtype=int)

get_poisson_realization(RNG: np.random.Generator, num_timesteps: int) -> np.ndarray

Uses RNG to generate Poisson random variable with rate equal to (population count in the origin Compartment x the TransitionVariable's current_rate / num_timesteps)

Parameters:

Name	Type	Description	Default
RNG	np.random.Generator object	used to generate stochastic transitions in the model and control reproducibility.	required
num_timesteps	int	number of timesteps per day -- used to determine time interval length for discretization.	required

Returns:

Type	Description
ndarray	np.ndarray: size |A| x |R|, where |A| is the number of age groups and |R| is number of risk groups.

Source code in CLT_BaseModel/clt_base/base_components.py

def get_poisson_realization(self,
                            RNG: np.random.Generator,
                            num_timesteps: int) -> np.ndarray:
    """
    Uses `RNG` to generate Poisson random variable with
        rate equal to (population count in the
        origin `Compartment` x the `TransitionVariable`'s
        `current_rate` / `num_timesteps`)

    Parameters:
        RNG (np.random.Generator object):
             used to generate stochastic transitions in the model and control
             reproducibility.
        num_timesteps (int):
            number of timesteps per day -- used to determine time interval
            length for discretization.

    Returns:
        np.ndarray:
            size |A| x |R|, where |A| is the number of age groups and
            |R| is number of risk groups.
    """
    return RNG.poisson(self.base_count * self.current_rate / float(num_timesteps))

get_realization(RNG: np.random.Generator, num_timesteps: int) -> np.ndarray

This method gets assigned to one of the following methods based on the TransitionVariable transition type: self.get_binomial_realization, self.get_binomial_taylor_approx_realization, self.get_poisson_realization, self.get_binomial_deterministic_realization, self.get_binomial_taylor_approx_deterministic_realization, self.get_poisson_deterministic_realization. This is done so that the same method self.get_realization can be called regardless of transition type.

Parameters:

Name	Type	Description	Default
RNG	np.random.Generator object	used to generate stochastic transitions in the model and control reproducibility.	required
num_timesteps	int	number of timesteps per day -- used to determine time interval length for discretization.	required

Source code in CLT_BaseModel/clt_base/base_components.py

def get_realization(self,
                    RNG: np.random.Generator,
                    num_timesteps: int) -> np.ndarray:
    """
    This method gets assigned to one of the following methods
        based on the `TransitionVariable` transition type:
        `self.get_binomial_realization`, `self.get_binomial_taylor_approx_realization`,
        `self.get_poisson_realization`, `self.get_binomial_deterministic_realization`,
        `self.get_binomial_taylor_approx_deterministic_realization`,
        `self.get_poisson_deterministic_realization`. This is done so that
        the same method `self.get_realization` can be called regardless of
        transition type.

    Parameters:
        RNG (np.random.Generator object):
             used to generate stochastic transitions in the model and control
             reproducibility.
        num_timesteps (int):
            number of timesteps per day -- used to determine time interval
            length for discretization.
    """

    pass

reset() -> None

Resets self.history_vals_list attribute to empty list.

Source code in CLT_BaseModel/clt_base/base_components.py

def reset(self) -> None:
    """
    Resets `self.history_vals_list` attribute to empty list.
    """

    self.history_vals_list = []

save_history() -> None

Saves current value to history by appending self.current_val attribute to self.history_vals_list in place.

Deep copying is CRUCIAL because self.current_val is a mutable np.ndarray -- without deep copying, self.history_vals_list would have the same value for all elements.

Source code in CLT_BaseModel/clt_base/base_components.py

def save_history(self) -> None:
    """
    Saves current value to history by appending `self.current_val`
        attribute to `self.history_vals_list` in place.

    Deep copying is CRUCIAL because `self.current_val` is a mutable
        np.ndarray -- without deep copying, `self.history_vals_list` would
        have the same value for all elements.
    """
    self.history_vals_list.append(copy.deepcopy(self.current_val))

update_destination_inflow() -> None

Adds current realization of TransitionVariable to its destination Compartment's current_inflow. Used to compute total number leaving that destination Compartment.

Source code in CLT_BaseModel/clt_base/base_components.py

def update_destination_inflow(self) -> None:
    """
    Adds current realization of `TransitionVariable` to
        its destination `Compartment`'s `current_inflow`.
        Used to compute total number leaving that
        destination `Compartment`.
    """

    self.destination.current_inflow = self.destination.current_inflow + self.current_val

update_origin_outflow() -> None

Adds current realization of TransitionVariable to its origin Compartment's current_outflow. Used to compute total number leaving that origin Compartment.

Source code in CLT_BaseModel/clt_base/base_components.py

def update_origin_outflow(self) -> None:
    """
    Adds current realization of `TransitionVariable` to
        its origin `Compartment`'s current_outflow.
        Used to compute total number leaving that
        origin `Compartment`.
    """

    self.origin.current_outflow = self.origin.current_outflow + self.current_val

TransitionVariableGroup

Container for TransitionVariable objects to handle joint sampling, when there are multiple outflows from a single compartment.

For example, if all outflows of compartment H are: R and D, i.e. from the hospital, people either recover or die, a TransitionVariableGroup that holds both R and D handles the correct correlation structure between R and D.

When an instance is initialized, its self.get_joint_realization attribute is dynamically assigned to a method according to its self.transition_type attribute. This enables all instances to use the same method during simulation.

Attributes:

Name	Type	Description
origin	Compartment	specifies origin of TransitionVariableGroup -- corresponding populations leave this compartment.
_transition_type	str	only values defined in JointTransitionTypes are valid, specifies joint probability distribution of all outflows from origin.
transition_variables	list[`TransitionVariable`]	specifying TransitionVariable instances that outflow from origin -- order does not matter.
get_joint_realization	function	assigned at initialization, generates realizations according to probability distribution given by self._transition_type attribute, returns either (M x |A| x |R|) or ((M+1) x |A| x |R|) np.ndarray, where M is the length of self.transition_variables (i.e., number of outflows from origin), |A| is the number of age groups, |R| is number of risk groups.
current_vals_list	list	used to store results from self.get_joint_realization -- has either M or M+1 arrays of size |A| x |R|.

See __init__ docstring for other attributes.

Source code in CLT_BaseModel/clt_base/base_components.py

class TransitionVariableGroup:
    """
    Container for `TransitionVariable` objects to handle joint sampling,
    when there are multiple outflows from a single compartment.

    For example, if all outflows of compartment `H` are: `R` and `D`,
    i.e. from the hospital, people either recover or die,
    a `TransitionVariableGroup` that holds both `R` and `D` handles
    the correct correlation structure between `R` and `D.`

    When an instance is initialized, its `self.get_joint_realization` attribute
    is dynamically assigned to a method according to its `self.transition_type`
    attribute. This enables all instances to use the same method during
    simulation.

    Attributes:
        origin (Compartment):
            specifies origin of `TransitionVariableGroup` --
            corresponding populations leave this compartment.
        _transition_type (str):
            only values defined in `JointTransitionTypes` are valid,
            specifies joint probability distribution of all outflows
            from origin.
        transition_variables (list[`TransitionVariable`]):
            specifying `TransitionVariable` instances that outflow from origin --
            order does not matter.
        get_joint_realization (function):
            assigned at initialization, generates realizations according
            to probability distribution given by `self._transition_type` attribute,
            returns either (M x |A| x |R|) or ((M+1) x |A| x |R|) np.ndarray,
            where M is the length of `self.transition_variables` (i.e., number of
            outflows from origin), |A| is the number of age groups, |R| is number of
            risk groups.
        current_vals_list (list):
            used to store results from `self.get_joint_realization` --
            has either M or M+1 arrays of size |A| x |R|.

    See `__init__` docstring for other attributes.
    """

    def __init__(self,
                 origin: Compartment,
                 transition_type: TransitionTypes,
                 transition_variables: list[TransitionVariable]):
        """
        Args:
            transition_type (str):
                only values defined in `TransitionTypes` are valid, specifying
                probability distribution of transitions between compartments.

        See class docstring for other parameters.
        """

        self.origin = origin
        self.transition_variables = transition_variables

        # If marginal transition type is any kind of binomial transition,
        #   then its joint transition type is a multinomial counterpart
        # For example, if the marginal transition type is TransitionTypes.BINOMIAL_DETERMINISTIC,
        #   then the joint transition type is JointTransitionTypes.MULTINOMIAL_DETERMINISTIC
        transition_type = transition_type.replace("binomial", "multinomial")
        self._transition_type = transition_type

        # Dynamically assign a method to get_joint_realization attribute
        #   based on the value of transition_type
        # getattr fetches a method by name
        self.get_joint_realization = getattr(self, "get_" + transition_type + "_realization")

        self.current_vals_list = []

    @property
    def transition_type(self) -> JointTransitionTypes:
        return self._transition_type

    def get_total_rate(self) -> np.ndarray:
        """
        Return the age-risk-specific total transition rate,
        which is the sum of the current rate of each transition variable
        in this transition variable group.

        Used to properly scale multinomial probabilities vector so
        that elements sum to 1.

        Returns:
            numpy array of positive floats with size equal to number
            of age groups x number of risk groups, and with value
            corresponding to sum of current rates of transition variables in
            transition variable group.
        """

        # axis 0: corresponds to outgoing transition variable
        # axis 1: corresponds to age groups
        # axis 2: corresponds to risk groups
        # --> summing over axis 0 gives the total rate for each age-risk group
        return np.sum(self.get_current_rates_array(), axis=0)

    def get_probabilities_array(self,
                                num_timesteps: int) -> list:
        """
        Returns an array of probabilities used for joint binomial
        (multinomial) transitions (`get_multinomial_realization` method).

        Returns:
            contains positive floats <= 1, size equal to
            ((length of outgoing transition variables list + 1)
            x number of age groups x number of risk groups) --
            note the "+1" corresponds to the multinomial outcome of staying
            in the same epi compartment (not transitioning to any outgoing
            epi compartment).
        """

        total_rate = self.get_total_rate()

        total_outgoing_probability = approx_binomial_probability_from_rate(total_rate,
                                                                           1 / num_timesteps)

        # Create probabilities_list, where element i corresponds to the
        #   transition variable i's current rate divided by the total rate,
        #   multiplized by the total outgoing probability
        # This generates the probabilities array that parameterizes the
        #   multinomial distribution
        probabilities_list = []

        for transition_variable in self.transition_variables:
            probabilities_list.append((transition_variable.current_rate / total_rate) *
                                      total_outgoing_probability)

        # Append the probability that a person stays in the compartment
        probabilities_list.append(1 - total_outgoing_probability)

        return np.asarray(probabilities_list)

    def get_current_rates_array(self) -> np.ndarray:
        """
        Returns an array of current rates of transition variables in
        self.transition_variables -- ith element in array
        corresponds to current rate of ith transition variable.

        Returns:
            array of positive floats, size equal to (length of outgoing
            transition variables list x number of age groups x number of risk groups).
        """

        current_rates_list = []
        for tvar in self.transition_variables:
            current_rates_list.append(tvar.current_rate)

        return np.asarray(current_rates_list)

    def get_joint_realization(self,
                              RNG: np.random.Generator,
                              num_timesteps: int) -> np.ndarray:
        """
        This function is dynamically assigned based on the
        `TransitionVariableGroup`'s `transition_type` -- this function is set to
        one of the following methods: `self.get_multinomial_realization`,
        `self.get_multinomial_taylor_approx_realization`,
        `self.get_poisson_realization`, `self.get_multinomial_deterministic_realization`,
        `self.get_multinomial_taylor_approx_deterministic_realization`,
        `self.get_poisson_deterministic_realization`.

        Parameters:
            RNG (np.random.Generator object):
                 used to generate stochastic transitions in the model and control
                 reproducibility.
            num_timesteps (int):
                number of timesteps per day -- used to determine time interval
                length for discretization.
        """

        pass

    def get_multinomial_realization(self,
                                    RNG: np.random.Generator,
                                    num_timesteps: int) -> np.ndarray:
        """
        Returns an array of transition realizations (number transitioning
        to outgoing compartments) sampled from multinomial distribution.

        Parameters:
            RNG (np.random.Generator object):
                 used to generate stochastic transitions in the model and control
                 reproducibility.
            num_timesteps (int):
                number of timesteps per day -- used to determine time interval
                length for discretization.

        Returns:
            np.ndarray:
                contains positive floats, size equal to
                ((length of outgoing transition variables list + 1)
                x number of age groups x number of risk groups) --
                note the "+1" corresponds to the multinomial outcome of staying
                in the same compartment (not transitioning to any outgoing
                epi compartment).
        """

        probabilities_array = self.get_probabilities_array(num_timesteps)

        num_outflows = len(self.transition_variables)

        num_age_groups, num_risk_groups = np.shape(self.origin.current_val)

        # We use num_outflows + 1 because for the multinomial distribution we explicitly model
        #   the number who stay/remain in the compartment
        realizations_array = np.zeros((num_outflows + 1, num_age_groups, num_risk_groups))

        for age_group in range(num_age_groups):
            for risk_group in range(num_risk_groups):
                realizations_array[:, age_group, risk_group] = RNG.multinomial(
                    np.asarray(self.origin.current_val[age_group, risk_group], dtype=int),
                    probabilities_array[:, age_group, risk_group])

        return realizations_array

    def get_multinomial_taylor_approx_realization(self,
                                                  RNG: np.random.Generator,
                                                  num_timesteps: int) -> np.ndarray:
        """
        Returns an array of transition realizations (number transitioning
        to outgoing compartments) sampled from multinomial distribution
        using Taylor Series approximation for probability parameter.

        Parameters:
            RNG (np.random.Generator object):
                 used to generate stochastic transitions in the model and control
                 reproducibility.
            num_timesteps (int):
                number of timesteps per day -- used to determine time interval
                length for discretization.

        Returns:
            np.ndarray:
                size equal to (length of outgoing transition variables list + 1)
                x number of age groups x number of risk groups --
                note the "+1" corresponds to the multinomial outcome of staying
                in the same compartment (not transitioning to any outgoing
                compartment).
        """

        num_outflows = len(self.transition_variables)

        current_rates_array = self.get_current_rates_array()

        total_rate = self.get_total_rate()

        # Multiply current rates array by length of time interval (1 / num_timesteps)
        # Also append additional value corresponding to probability of
        #   remaining in current epi compartment (not transitioning at all)
        # Note: "vstack" function here works better than append function because append
        #   automatically flattens the resulting array, resulting in dimension issues
        current_scaled_rates_array = np.vstack((current_rates_array / num_timesteps,
                                                np.expand_dims(1 - total_rate / num_timesteps, axis=0)))

        num_age_groups, num_risk_groups = np.shape(self.origin.current_val)

        # We use num_outflows + 1 because for the multinomial distribution we explicitly model
        #   the number who stay/remain in the compartment
        realizations_array = np.zeros((num_outflows + 1, num_age_groups, num_risk_groups))

        for age_group in range(num_age_groups):
            for risk_group in range(num_risk_groups):
                realizations_array[:, age_group, risk_group] = RNG.multinomial(
                    np.asarray(self.origin.current_val[age_group, risk_group], dtype=int),
                    current_scaled_rates_array[:, age_group, risk_group])

        return realizations_array

    def get_poisson_realization(self,
                                RNG: np.random.Generator,
                                num_timesteps: int) -> np.ndarray:
        """
        Returns an array of transition realizations (number transitioning
        to outgoing compartments) sampled from Poisson distribution.

        Parameters:
            RNG (np.random.Generator object):
                 used to generate stochastic transitions in the model and control
                 reproducibility.
            num_timesteps (int):
                number of timesteps per day -- used to determine time interval
                length for discretization.

        Returns:
            np.ndarray:
                contains positive floats, size equal to length of
                (outgoing transition variables list x
                number of age groups x number of risk groups).
        """

        num_outflows = len(self.transition_variables)

        num_age_groups, num_risk_groups = np.shape(self.origin.current_val)

        realizations_array = np.zeros((num_outflows, num_age_groups, num_risk_groups))

        transition_variables = self.transition_variables

        for age_group in range(num_age_groups):
            for risk_group in range(num_risk_groups):
                for outflow_ix in range(num_outflows):
                    realizations_array[outflow_ix, age_group, risk_group] = RNG.poisson(
                        self.origin.current_val[age_group, risk_group] *
                        transition_variables[outflow_ix].current_rate[
                            age_group, risk_group] / num_timesteps)

        return realizations_array

    def get_multinomial_deterministic_realization(self,
                                                  RNG: np.random.Generator,
                                                  num_timesteps: int) -> np.ndarray:
        """
        Deterministic counterpart to `self.get_multinomial_realization` --
        uses mean (n x p, i.e. total counts x probability array) as realization
        rather than randomly sampling.

        Parameters:
            RNG (np.random.Generator object):
                NOT USED -- only included so that get_realization has
                the same function arguments regardless of transition type.
            num_timesteps (int):
                number of timesteps per day -- used to determine time interval
                length for discretization.

        Returns:
            np.ndarray:
                contains positive floats, size equal to
                (length of outgoing transition variables list + 1)
                x number of age groups x number of risk groups --
                note the "+1" corresponds to the multinomial outcome of staying
                in the same compartment (not transitioning to any outgoing
                compartment).
        """

        probabilities_array = self.get_probabilities_array(num_timesteps)
        return np.asarray(self.origin.current_val * probabilities_array, dtype=int)

    def get_multinomial_taylor_approx_deterministic_realization(self,
                                                                RNG: np.random.Generator,
                                                                num_timesteps: int) -> np.ndarray:
        """
        Deterministic counterpart to `self.get_multinomial_taylor_approx_realization` --
        uses mean (n x p, i.e. total counts x probability array) as realization
        rather than randomly sampling.

        Parameters:
            RNG (np.random.Generator object):
                NOT USED -- only included so that get_realization has
                the same function arguments regardless of transition type.
            num_timesteps (int):
                number of timesteps per day -- used to determine time interval
                length for discretization.

        Returns:
            np.ndarray:
                contains positive floats, size equal to
                (length of outgoing transition variables list + 1)
                x number of age groups x number of risk groups --
                note the "+1" corresponds to the multinomial outcome of staying
                in the same compartment (not transitioning to any outgoing
                compartment).
        """

        current_rates_array = self.get_current_rates_array()
        return np.asarray(self.origin.current_val * current_rates_array / num_timesteps, dtype=int)

    def get_poisson_deterministic_realization(self,
                                              RNG: np.random.Generator,
                                              num_timesteps: int) -> np.ndarray:
        """
        Deterministic counterpart to `self.get_poisson_realization` --
        uses mean (rate array) as realization rather than randomly sampling.

        Parameters:
            RNG (np.random.Generator object):
                NOT USED -- only included so that get_realization has
                the same function arguments regardless of transition type.
            num_timesteps (int):
                number of timesteps per day -- used to determine time interval
                length for discretization.

        Returns:
            np.ndarray:
                contains positive floats, size equal to
                (length of outgoing transition variables list
                x number of age groups x number of risk groups).
        """

        return np.asarray(self.origin.current_val *
                          self.get_current_rates_array() / num_timesteps, dtype=int)

    def reset(self) -> None:
        self.current_vals_list = []

    def update_transition_variable_realizations(self) -> None:
        """
        Updates current_val attribute on all `TransitionVariable`
        instances contained in this `TransitionVariableGroup`.
        """

        # Since the ith element in probabilities_array corresponds to the ith transition variable
        #   in transition_variables, the ith element in multinomial_realizations_list
        #   also corresponds to the ith transition variable in transition_variables
        # Update the current realization of the transition variables contained in this group
        for ix in range(len(self.transition_variables)):
            self.transition_variables[ix].current_val = \
                self.current_vals_list[ix, :, :]

__init__(origin: Compartment, transition_type: TransitionTypes, transition_variables: list[TransitionVariable])

Parameters:

Name	Type	Description	Default
transition_type	str	only values defined in TransitionTypes are valid, specifying probability distribution of transitions between compartments.	required

See class docstring for other parameters.

Source code in CLT_BaseModel/clt_base/base_components.py

def __init__(self,
             origin: Compartment,
             transition_type: TransitionTypes,
             transition_variables: list[TransitionVariable]):
    """
    Args:
        transition_type (str):
            only values defined in `TransitionTypes` are valid, specifying
            probability distribution of transitions between compartments.

    See class docstring for other parameters.
    """

    self.origin = origin
    self.transition_variables = transition_variables

    # If marginal transition type is any kind of binomial transition,
    #   then its joint transition type is a multinomial counterpart
    # For example, if the marginal transition type is TransitionTypes.BINOMIAL_DETERMINISTIC,
    #   then the joint transition type is JointTransitionTypes.MULTINOMIAL_DETERMINISTIC
    transition_type = transition_type.replace("binomial", "multinomial")
    self._transition_type = transition_type

    # Dynamically assign a method to get_joint_realization attribute
    #   based on the value of transition_type
    # getattr fetches a method by name
    self.get_joint_realization = getattr(self, "get_" + transition_type + "_realization")

    self.current_vals_list = []

get_current_rates_array() -> np.ndarray

Returns an array of current rates of transition variables in self.transition_variables -- ith element in array corresponds to current rate of ith transition variable.

Returns:

Type	Description
ndarray	array of positive floats, size equal to (length of outgoing
ndarray	transition variables list x number of age groups x number of risk groups).

Source code in CLT_BaseModel/clt_base/base_components.py

def get_current_rates_array(self) -> np.ndarray:
    """
    Returns an array of current rates of transition variables in
    self.transition_variables -- ith element in array
    corresponds to current rate of ith transition variable.

    Returns:
        array of positive floats, size equal to (length of outgoing
        transition variables list x number of age groups x number of risk groups).
    """

    current_rates_list = []
    for tvar in self.transition_variables:
        current_rates_list.append(tvar.current_rate)

    return np.asarray(current_rates_list)

get_joint_realization(RNG: np.random.Generator, num_timesteps: int) -> np.ndarray

This function is dynamically assigned based on the TransitionVariableGroup's transition_type -- this function is set to one of the following methods: self.get_multinomial_realization, self.get_multinomial_taylor_approx_realization, self.get_poisson_realization, self.get_multinomial_deterministic_realization, self.get_multinomial_taylor_approx_deterministic_realization, self.get_poisson_deterministic_realization.

Parameters:

Name	Type	Description	Default
RNG	np.random.Generator object	used to generate stochastic transitions in the model and control reproducibility.	required
num_timesteps	int	number of timesteps per day -- used to determine time interval length for discretization.	required

Source code in CLT_BaseModel/clt_base/base_components.py

def get_joint_realization(self,
                          RNG: np.random.Generator,
                          num_timesteps: int) -> np.ndarray:
    """
    This function is dynamically assigned based on the
    `TransitionVariableGroup`'s `transition_type` -- this function is set to
    one of the following methods: `self.get_multinomial_realization`,
    `self.get_multinomial_taylor_approx_realization`,
    `self.get_poisson_realization`, `self.get_multinomial_deterministic_realization`,
    `self.get_multinomial_taylor_approx_deterministic_realization`,
    `self.get_poisson_deterministic_realization`.

    Parameters:
        RNG (np.random.Generator object):
             used to generate stochastic transitions in the model and control
             reproducibility.
        num_timesteps (int):
            number of timesteps per day -- used to determine time interval
            length for discretization.
    """

    pass

get_multinomial_deterministic_realization(RNG: np.random.Generator, num_timesteps: int) -> np.ndarray

Deterministic counterpart to self.get_multinomial_realization -- uses mean (n x p, i.e. total counts x probability array) as realization rather than randomly sampling.

Parameters:

Name	Type	Description	Default
RNG	np.random.Generator object	NOT USED -- only included so that get_realization has the same function arguments regardless of transition type.	required
num_timesteps	int	number of timesteps per day -- used to determine time interval length for discretization.	required

Returns:

Type	Description
ndarray	np.ndarray: contains positive floats, size equal to (length of outgoing transition variables list + 1) x number of age groups x number of risk groups -- note the "+1" corresponds to the multinomial outcome of staying in the same compartment (not transitioning to any outgoing compartment).

Source code in CLT_BaseModel/clt_base/base_components.py

def get_multinomial_deterministic_realization(self,
                                              RNG: np.random.Generator,
                                              num_timesteps: int) -> np.ndarray:
    """
    Deterministic counterpart to `self.get_multinomial_realization` --
    uses mean (n x p, i.e. total counts x probability array) as realization
    rather than randomly sampling.

    Parameters:
        RNG (np.random.Generator object):
            NOT USED -- only included so that get_realization has
            the same function arguments regardless of transition type.
        num_timesteps (int):
            number of timesteps per day -- used to determine time interval
            length for discretization.

    Returns:
        np.ndarray:
            contains positive floats, size equal to
            (length of outgoing transition variables list + 1)
            x number of age groups x number of risk groups --
            note the "+1" corresponds to the multinomial outcome of staying
            in the same compartment (not transitioning to any outgoing
            compartment).
    """

    probabilities_array = self.get_probabilities_array(num_timesteps)
    return np.asarray(self.origin.current_val * probabilities_array, dtype=int)

get_multinomial_realization(RNG: np.random.Generator, num_timesteps: int) -> np.ndarray

Returns an array of transition realizations (number transitioning to outgoing compartments) sampled from multinomial distribution.

Parameters:

Name	Type	Description	Default
RNG	np.random.Generator object	used to generate stochastic transitions in the model and control reproducibility.	required
num_timesteps	int	number of timesteps per day -- used to determine time interval length for discretization.	required

Returns:

Type	Description
ndarray	np.ndarray: contains positive floats, size equal to ((length of outgoing transition variables list + 1) x number of age groups x number of risk groups) -- note the "+1" corresponds to the multinomial outcome of staying in the same compartment (not transitioning to any outgoing epi compartment).

Source code in CLT_BaseModel/clt_base/base_components.py

def get_multinomial_realization(self,
                                RNG: np.random.Generator,
                                num_timesteps: int) -> np.ndarray:
    """
    Returns an array of transition realizations (number transitioning
    to outgoing compartments) sampled from multinomial distribution.

    Parameters:
        RNG (np.random.Generator object):
             used to generate stochastic transitions in the model and control
             reproducibility.
        num_timesteps (int):
            number of timesteps per day -- used to determine time interval
            length for discretization.

    Returns:
        np.ndarray:
            contains positive floats, size equal to
            ((length of outgoing transition variables list + 1)
            x number of age groups x number of risk groups) --
            note the "+1" corresponds to the multinomial outcome of staying
            in the same compartment (not transitioning to any outgoing
            epi compartment).
    """

    probabilities_array = self.get_probabilities_array(num_timesteps)

    num_outflows = len(self.transition_variables)

    num_age_groups, num_risk_groups = np.shape(self.origin.current_val)

    # We use num_outflows + 1 because for the multinomial distribution we explicitly model
    #   the number who stay/remain in the compartment
    realizations_array = np.zeros((num_outflows + 1, num_age_groups, num_risk_groups))

    for age_group in range(num_age_groups):
        for risk_group in range(num_risk_groups):
            realizations_array[:, age_group, risk_group] = RNG.multinomial(
                np.asarray(self.origin.current_val[age_group, risk_group], dtype=int),
                probabilities_array[:, age_group, risk_group])

    return realizations_array

get_multinomial_taylor_approx_deterministic_realization(RNG: np.random.Generator, num_timesteps: int) -> np.ndarray

Deterministic counterpart to self.get_multinomial_taylor_approx_realization -- uses mean (n x p, i.e. total counts x probability array) as realization rather than randomly sampling.

Parameters:

Name	Type	Description	Default
RNG	np.random.Generator object	NOT USED -- only included so that get_realization has the same function arguments regardless of transition type.	required
num_timesteps	int	number of timesteps per day -- used to determine time interval length for discretization.	required

Returns:

Type	Description
ndarray	np.ndarray: contains positive floats, size equal to (length of outgoing transition variables list + 1) x number of age groups x number of risk groups -- note the "+1" corresponds to the multinomial outcome of staying in the same compartment (not transitioning to any outgoing compartment).

Source code in CLT_BaseModel/clt_base/base_components.py

def get_multinomial_taylor_approx_deterministic_realization(self,
                                                            RNG: np.random.Generator,
                                                            num_timesteps: int) -> np.ndarray:
    """
    Deterministic counterpart to `self.get_multinomial_taylor_approx_realization` --
    uses mean (n x p, i.e. total counts x probability array) as realization
    rather than randomly sampling.

    Parameters:
        RNG (np.random.Generator object):
            NOT USED -- only included so that get_realization has
            the same function arguments regardless of transition type.
        num_timesteps (int):
            number of timesteps per day -- used to determine time interval
            length for discretization.

    Returns:
        np.ndarray:
            contains positive floats, size equal to
            (length of outgoing transition variables list + 1)
            x number of age groups x number of risk groups --
            note the "+1" corresponds to the multinomial outcome of staying
            in the same compartment (not transitioning to any outgoing
            compartment).
    """

    current_rates_array = self.get_current_rates_array()
    return np.asarray(self.origin.current_val * current_rates_array / num_timesteps, dtype=int)

get_multinomial_taylor_approx_realization(RNG: np.random.Generator, num_timesteps: int) -> np.ndarray

Returns an array of transition realizations (number transitioning to outgoing compartments) sampled from multinomial distribution using Taylor Series approximation for probability parameter.

Parameters:

Name	Type	Description	Default
RNG	np.random.Generator object	used to generate stochastic transitions in the model and control reproducibility.	required
num_timesteps	int	number of timesteps per day -- used to determine time interval length for discretization.	required

Returns:

Type	Description
ndarray	np.ndarray: size equal to (length of outgoing transition variables list + 1) x number of age groups x number of risk groups -- note the "+1" corresponds to the multinomial outcome of staying in the same compartment (not transitioning to any outgoing compartment).

Source code in CLT_BaseModel/clt_base/base_components.py

def get_multinomial_taylor_approx_realization(self,
                                              RNG: np.random.Generator,
                                              num_timesteps: int) -> np.ndarray:
    """
    Returns an array of transition realizations (number transitioning
    to outgoing compartments) sampled from multinomial distribution
    using Taylor Series approximation for probability parameter.

    Parameters:
        RNG (np.random.Generator object):
             used to generate stochastic transitions in the model and control
             reproducibility.
        num_timesteps (int):
            number of timesteps per day -- used to determine time interval
            length for discretization.

    Returns:
        np.ndarray:
            size equal to (length of outgoing transition variables list + 1)
            x number of age groups x number of risk groups --
            note the "+1" corresponds to the multinomial outcome of staying
            in the same compartment (not transitioning to any outgoing
            compartment).
    """

    num_outflows = len(self.transition_variables)

    current_rates_array = self.get_current_rates_array()

    total_rate = self.get_total_rate()

    # Multiply current rates array by length of time interval (1 / num_timesteps)
    # Also append additional value corresponding to probability of
    #   remaining in current epi compartment (not transitioning at all)
    # Note: "vstack" function here works better than append function because append
    #   automatically flattens the resulting array, resulting in dimension issues
    current_scaled_rates_array = np.vstack((current_rates_array / num_timesteps,
                                            np.expand_dims(1 - total_rate / num_timesteps, axis=0)))

    num_age_groups, num_risk_groups = np.shape(self.origin.current_val)

    # We use num_outflows + 1 because for the multinomial distribution we explicitly model
    #   the number who stay/remain in the compartment
    realizations_array = np.zeros((num_outflows + 1, num_age_groups, num_risk_groups))

    for age_group in range(num_age_groups):
        for risk_group in range(num_risk_groups):
            realizations_array[:, age_group, risk_group] = RNG.multinomial(
                np.asarray(self.origin.current_val[age_group, risk_group], dtype=int),
                current_scaled_rates_array[:, age_group, risk_group])

    return realizations_array

get_poisson_deterministic_realization(RNG: np.random.Generator, num_timesteps: int) -> np.ndarray

Deterministic counterpart to self.get_poisson_realization -- uses mean (rate array) as realization rather than randomly sampling.

Parameters:

Name	Type	Description	Default
RNG	np.random.Generator object	NOT USED -- only included so that get_realization has the same function arguments regardless of transition type.	required
num_timesteps	int	number of timesteps per day -- used to determine time interval length for discretization.	required

Returns:

Type	Description
ndarray	np.ndarray: contains positive floats, size equal to (length of outgoing transition variables list x number of age groups x number of risk groups).

Source code in CLT_BaseModel/clt_base/base_components.py

def get_poisson_deterministic_realization(self,
                                          RNG: np.random.Generator,
                                          num_timesteps: int) -> np.ndarray:
    """
    Deterministic counterpart to `self.get_poisson_realization` --
    uses mean (rate array) as realization rather than randomly sampling.

    Parameters:
        RNG (np.random.Generator object):
            NOT USED -- only included so that get_realization has
            the same function arguments regardless of transition type.
        num_timesteps (int):
            number of timesteps per day -- used to determine time interval
            length for discretization.

    Returns:
        np.ndarray:
            contains positive floats, size equal to
            (length of outgoing transition variables list
            x number of age groups x number of risk groups).
    """

    return np.asarray(self.origin.current_val *
                      self.get_current_rates_array() / num_timesteps, dtype=int)

get_poisson_realization(RNG: np.random.Generator, num_timesteps: int) -> np.ndarray

Returns an array of transition realizations (number transitioning to outgoing compartments) sampled from Poisson distribution.

Parameters:

Name	Type	Description	Default
RNG	np.random.Generator object	used to generate stochastic transitions in the model and control reproducibility.	required
num_timesteps	int	number of timesteps per day -- used to determine time interval length for discretization.	required

Returns:

Type	Description
ndarray	np.ndarray: contains positive floats, size equal to length of (outgoing transition variables list x number of age groups x number of risk groups).

Source code in CLT_BaseModel/clt_base/base_components.py

def get_poisson_realization(self,
                            RNG: np.random.Generator,
                            num_timesteps: int) -> np.ndarray:
    """
    Returns an array of transition realizations (number transitioning
    to outgoing compartments) sampled from Poisson distribution.

    Parameters:
        RNG (np.random.Generator object):
             used to generate stochastic transitions in the model and control
             reproducibility.
        num_timesteps (int):
            number of timesteps per day -- used to determine time interval
            length for discretization.

    Returns:
        np.ndarray:
            contains positive floats, size equal to length of
            (outgoing transition variables list x
            number of age groups x number of risk groups).
    """

    num_outflows = len(self.transition_variables)

    num_age_groups, num_risk_groups = np.shape(self.origin.current_val)

    realizations_array = np.zeros((num_outflows, num_age_groups, num_risk_groups))

    transition_variables = self.transition_variables

    for age_group in range(num_age_groups):
        for risk_group in range(num_risk_groups):
            for outflow_ix in range(num_outflows):
                realizations_array[outflow_ix, age_group, risk_group] = RNG.poisson(
                    self.origin.current_val[age_group, risk_group] *
                    transition_variables[outflow_ix].current_rate[
                        age_group, risk_group] / num_timesteps)

    return realizations_array

get_probabilities_array(num_timesteps: int) -> list

Returns an array of probabilities used for joint binomial (multinomial) transitions (get_multinomial_realization method).

Returns:

Type	Description
list	contains positive floats <= 1, size equal to
list	((length of outgoing transition variables list + 1)
list	x number of age groups x number of risk groups) --
list	note the "+1" corresponds to the multinomial outcome of staying
list	in the same epi compartment (not transitioning to any outgoing
list	epi compartment).

Source code in CLT_BaseModel/clt_base/base_components.py

def get_probabilities_array(self,
                            num_timesteps: int) -> list:
    """
    Returns an array of probabilities used for joint binomial
    (multinomial) transitions (`get_multinomial_realization` method).

    Returns:
        contains positive floats <= 1, size equal to
        ((length of outgoing transition variables list + 1)
        x number of age groups x number of risk groups) --
        note the "+1" corresponds to the multinomial outcome of staying
        in the same epi compartment (not transitioning to any outgoing
        epi compartment).
    """

    total_rate = self.get_total_rate()

    total_outgoing_probability = approx_binomial_probability_from_rate(total_rate,
                                                                       1 / num_timesteps)

    # Create probabilities_list, where element i corresponds to the
    #   transition variable i's current rate divided by the total rate,
    #   multiplized by the total outgoing probability
    # This generates the probabilities array that parameterizes the
    #   multinomial distribution
    probabilities_list = []

    for transition_variable in self.transition_variables:
        probabilities_list.append((transition_variable.current_rate / total_rate) *
                                  total_outgoing_probability)

    # Append the probability that a person stays in the compartment
    probabilities_list.append(1 - total_outgoing_probability)

    return np.asarray(probabilities_list)

get_total_rate() -> np.ndarray

Return the age-risk-specific total transition rate, which is the sum of the current rate of each transition variable in this transition variable group.

Used to properly scale multinomial probabilities vector so that elements sum to 1.

Returns:

Type	Description
ndarray	numpy array of positive floats with size equal to number
ndarray	of age groups x number of risk groups, and with value
ndarray	corresponding to sum of current rates of transition variables in
ndarray	transition variable group.

Source code in CLT_BaseModel/clt_base/base_components.py

def get_total_rate(self) -> np.ndarray:
    """
    Return the age-risk-specific total transition rate,
    which is the sum of the current rate of each transition variable
    in this transition variable group.

    Used to properly scale multinomial probabilities vector so
    that elements sum to 1.

    Returns:
        numpy array of positive floats with size equal to number
        of age groups x number of risk groups, and with value
        corresponding to sum of current rates of transition variables in
        transition variable group.
    """

    # axis 0: corresponds to outgoing transition variable
    # axis 1: corresponds to age groups
    # axis 2: corresponds to risk groups
    # --> summing over axis 0 gives the total rate for each age-risk group
    return np.sum(self.get_current_rates_array(), axis=0)

update_transition_variable_realizations() -> None

Updates current_val attribute on all TransitionVariable instances contained in this TransitionVariableGroup.

Source code in CLT_BaseModel/clt_base/base_components.py

def update_transition_variable_realizations(self) -> None:
    """
    Updates current_val attribute on all `TransitionVariable`
    instances contained in this `TransitionVariableGroup`.
    """

    # Since the ith element in probabilities_array corresponds to the ith transition variable
    #   in transition_variables, the ith element in multinomial_realizations_list
    #   also corresponds to the ith transition variable in transition_variables
    # Update the current realization of the transition variables contained in this group
    for ix in range(len(self.transition_variables)):
        self.transition_variables[ix].current_val = \
            self.current_vals_list[ix, :, :]

approx_binomial_probability_from_rate(rate: np.ndarray, interval_length: int) -> np.ndarray

Converts a rate (events per time) to the probability of any event occurring in the next time interval of length interval_length, assuming the number of events occurring in time interval follows a Poisson distribution with given rate parameter.

The probability of 0 events in interval_length is e^(-rate * interval_length), so the probability of any event in interval_length is 1 - e^(-rate * interval_length).

Rate must be |A| x |R| np.ndarray, where |A| is the number of age groups and |R| is the number of risk groups. Rate is transformed to |A| x |R| np.ndarray corresponding to probabilities.

Parameters:

Name	Type	Description	Default
rate	ndarray	dimension |A| x |R| (number of age groups x number of risk groups), rate parameters in a Poisson distribution.	required
interval_length	positive int	length of time interval in simulation days.	required

Returns:

Type	Description
ndarray	np.ndarray: array of positive scalars, dimension |A| x |R|

Source code in CLT_BaseModel/clt_base/base_components.py

def approx_binomial_probability_from_rate(rate: np.ndarray,
                                          interval_length: int) -> np.ndarray:
    """
    Converts a rate (events per time) to the probability of any event
    occurring in the next time interval of length `interval_length`,
    assuming the number of events occurring in time interval
    follows a Poisson distribution with given rate parameter.

    The probability of 0 events in `interval_length` is
    e^(-`rate` * `interval_length`), so the probability of any event
    in `interval_length` is 1 - e^(-`rate` * `interval_length`).

    Rate must be |A| x |R| `np.ndarray`, where |A| is the number of
    age groups and |R| is the number of risk groups. Rate is transformed to
    |A| x |R| `np.ndarray` corresponding to probabilities.

    Parameters:
        rate (np.ndarray):
            dimension |A| x |R| (number of age groups x number of risk groups),
            rate parameters in a Poisson distribution.
        interval_length (positive int):
            length of time interval in simulation days.

    Returns:
        np.ndarray:
            array of positive scalars, dimension |A| x |R|
    """

    return 1 - np.exp(-rate * interval_length)

check_is_subset_list(listA: list, listB: list) -> bool

Parameters:

Name	Type	Description	Default
listA	list	list-like of elements to check if subset of listB.	required
listB	list	list-like of elements.	required

Returns:

Type	Description
bool	True if listA is a subset of listB, and False otherwise.

Source code in CLT_BaseModel/clt_base/experiments.py

def check_is_subset_list(listA: list,
                         listB: list) -> bool:
    """
    Params:
        listA (list):
            list-like of elements to check if subset of listB.
        listB (list):
            list-like of elements.

    Returns:
        True if listA is a subset of listB, and False otherwise.
    """

    return all(item in listB for item in listA)

convert_dict_vals_lists_to_arrays(d: dict) -> dict

Source code in CLT_BaseModel/clt_base/input_parsers.py

def convert_dict_vals_lists_to_arrays(d: dict) -> dict:

    # convert lists to numpy arrays to support numpy operations
    for key, val in d.items():
        if type(val) is list:
            d[key] = np.asarray(val)

    return d

format_current_val_for_sql(subpop_model: SubpopModel, state_var_name: str, rep: int) -> list

Processes current_val of given subpop_model's StateVariable specified by state_var_name. Current_val is an |A| x |R| numpy array (for age-risk) -- this function "unpacks" it into an (|A| x |R|, 1) numpy array (a column vector). Converts metadata (subpop_name, state_var_name, rep, and current_simulation_day) into list of |A| x |R| rows, where each row has 7 elements, for consistent row formatting for batch SQL insertion.

Parameters:

Name	Type	Description	Default
subpop_model	SubpopModel	SubpopModel to record.	required
state_var_name	str	StateVariable name to record.	required
rep	int	replication counter to record.	required

Returns:

Name	Type	Description
data	list	list of |A| x |R| rows, where each row is a list of 7 elements corresponding to subpop_name, state_var_name, age_group, risk_group, rep, current_simulation_day, and the scalar element of current_val corresponding to that age-risk group.

Source code in CLT_BaseModel/clt_base/experiments.py

def format_current_val_for_sql(subpop_model: SubpopModel,
                               state_var_name: str,
                               rep: int) -> list:
    """
    Processes current_val of given subpop_model's `StateVariable`
    specified by `state_var_name`. Current_val is an |A| x |R|
    numpy array (for age-risk) -- this function "unpacks" it into an
    (|A| x |R|, 1) numpy array (a column vector). Converts metadata
    (subpop_name, state_var_name, `rep`, and current_simulation_day)
    into list of |A| x |R| rows, where each row has 7 elements, for
    consistent row formatting for batch SQL insertion.

    Params:
        subpop_model (SubpopModel):
            SubpopModel to record.
        state_var_name (str):
            StateVariable name to record.
        rep (int):
            replication counter to record.

    Returns:
        data (list):
            list of |A| x |R| rows, where each row is a list of 7 elements
            corresponding to subpop_name, state_var_name, age_group, risk_group,
            rep, current_simulation_day, and the scalar element of current_val
            corresponding to that age-risk group.
    """

    current_val = subpop_model.all_state_variables[state_var_name].current_val

    A, R = np.shape(current_val)

    # numpy's default is row-major / C-style order
    # This means the elements are unpacked ROW BY ROW
    current_val_reshaped = current_val.reshape(-1, 1)

    # (AxR, 1) column vector of row indices, indicating the original row in current_val
    #   before reshaping
    # Each integer in np.arange(A) repeated R times
    age_group_indices = np.repeat(np.arange(A), R).reshape(-1, 1)

    # (AxR, 1) column vector of column indices, indicating the original column
    #   each element belonged to in current_val before reshaping
    # Repeat np.arange(R) A times
    risk_group_indices = np.tile(np.arange(R), A).reshape(-1, 1)

    # (subpop_name, state_var_name, age_group, risk_group, rep, timepoint)
    data = np.column_stack(
        (np.full((A * R, 1), subpop_model.name),
         np.full((A * R, 1), state_var_name),
         age_group_indices,
         risk_group_indices,
         np.full((A * R, 1), rep),
         np.full((A * R, 1), subpop_model.current_simulation_day),
         current_val_reshaped)).tolist()

    return data

get_sql_table_as_df(conn: sqlite3.Connection, sql_query: str, sql_query_params: tuple[str] = None, chunk_size: int = int(10000.0)) -> pd.DataFrame

Returns a pandas DataFrame containing data from specified SQL table, retrieved using the provided database connection. Reads in SQL rows in batches of size chunk_size to avoid memory issues for very large tables.

Parameters:

Name	Type	Description	Default
conn	Connection	connection to SQL database.	required
sql_query	str	SQL query/statement to execute on database.	required
sql_query_params	tuple[str]	tuple of strings to pass as parameters to SQL query -- used to avoid SQL injections.	None
chunk_size	positive int	number of rows to read in at a time.	int(10000.0)

Returns:

Type	Description
DataFrame	DataFrame containing data from specified SQL table,
DataFrame	or empty DataFrame if table does not exist.

Source code in CLT_BaseModel/clt_base/experiments.py

def get_sql_table_as_df(conn: sqlite3.Connection,
                        sql_query: str,
                        sql_query_params: tuple[str] = None,
                        chunk_size: int = int(1e4)) -> pd.DataFrame:
    """
    Returns a pandas DataFrame containing data from specified SQL table,
    retrieved using the provided database connection. Reads in SQL rows
    in batches of size `chunk_size` to avoid memory issues for very large
    tables.

    Params:
        conn (sqlite3.Connection):
            connection to SQL database.
        sql_query (str):
            SQL query/statement to execute on database.
        sql_query_params (tuple[str]):
            tuple of strings to pass as parameters to
            SQL query -- used to avoid SQL injections.
        chunk_size (positive int):
            number of rows to read in at a time.

    Returns:
        DataFrame containing data from specified SQL table,
        or empty DataFrame if table does not exist.
    """

    chunks = []

    try:
        for chunk in pd.read_sql_query(sql_query,
                                       conn,
                                       chunksize=chunk_size,
                                       params=sql_query_params):
            chunks.append(chunk)
            df = pd.concat(chunks, ignore_index=True)

    # Handle exception gracefully -- print a warning and
    #   return an empty DataFrame if table given by sql_query
    #   does not exist
    except sqlite3.OperationalError as e:
        if "no such table" in str(e).lower():
            print(f"Warning: table does not exist for query: {sql_query}. "
                  f"Returning empty DataFrame.")
            df = pd.DataFrame()

    return df

load_json_augment_dict(json_filepath: str, d: dict) -> dict

Source code in CLT_BaseModel/clt_base/input_parsers.py

def load_json_augment_dict(json_filepath: str, d: dict) -> dict:

    with open(json_filepath, 'r') as file:
        data = json.load(file)

    data = convert_dict_vals_lists_to_arrays(data)

    for key, val in data.items():
        d[key] = val

    return d

load_json_new_dict(json_filepath: str) -> dict

Source code in CLT_BaseModel/clt_base/input_parsers.py

def load_json_new_dict(json_filepath: str) -> dict:

    # Note: the "with open" is important for file handling
    #   and avoiding resource leaks -- otherwise,
    #   we have to manually close the file, which is a bit
    #   more cumbersome
    with open(json_filepath, 'r') as file:
        data = json.load(file)

    # json does not support numpy, so we must convert
    #   lists to numpy arrays
    return convert_dict_vals_lists_to_arrays(data)

make_dataclass_from_dict(dataclass_ref: Type[DataClassProtocol], d: dict) -> DataClassProtocol

Create instance of class dataclass_ref, based on information in dictionary.

Parameters:

Name	Type	Description	Default
dataclass_ref	Type[DataClassProtocol]	(class, not instance) from which to create instance -- must have dataclass decorator.	required
d	dict	all keys and values respectively must match name and datatype of dataclass_ref instance attributes.	required

Returns:

Name	Type	Description
DataClassProtocol	DataClassProtocol	instance of dataclass_ref with attributes dynamically assigned by json_filepath file contents.

Source code in CLT_BaseModel/clt_base/input_parsers.py

def make_dataclass_from_dict(dataclass_ref: Type[DataClassProtocol],
                             d: dict) -> DataClassProtocol:
    """
    Create instance of class dataclass_ref,
    based on information in dictionary.

    Args:
        dataclass_ref (Type[DataClassProtocol]):
            (class, not instance) from which to create instance --
            must have dataclass decorator.
        d (dict):
            all keys and values respectively must match name and datatype
            of dataclass_ref instance attributes.

    Returns:
        DataClassProtocol:
            instance of dataclass_ref with attributes dynamically
            assigned by json_filepath file contents.
    """

    d = convert_dict_vals_lists_to_arrays(d)

    return dataclass_ref(**d)

make_dataclass_from_json(dataclass_ref: Type[DataClassProtocol], json_filepath: str) -> DataClassProtocol

Create instance of class dataclass_ref, based on information in json_filepath.

Parameters:

Name	Type	Description	Default
dataclass_ref	Type[DataClassProtocol]	(class, not instance) from which to create instance -- must have dataclass decorator.	required
json_filepath	str	path to json file (path includes actual filename with suffix ".json") -- all json fields must match name and datatype of dataclass_ref instance attributes.	required

Returns:

Name	Type	Description
DataClassProtocol	DataClassProtocol	instance of dataclass_ref with attributes dynamically assigned by json_filepath file contents.

Source code in CLT_BaseModel/clt_base/input_parsers.py

def make_dataclass_from_json(dataclass_ref: Type[DataClassProtocol],
                             json_filepath: str) -> DataClassProtocol:
    """
    Create instance of class dataclass_ref,
    based on information in json_filepath.

    Args:
        dataclass_ref (Type[DataClassProtocol]):
            (class, not instance) from which to create instance --
            must have dataclass decorator.
        json_filepath (str):
            path to json file (path includes actual filename
            with suffix ".json") -- all json fields must
            match name and datatype of dataclass_ref instance
            attributes.

    Returns:
        DataClassProtocol:
            instance of dataclass_ref with attributes dynamically
            assigned by json_filepath file contents.
    """

    d = load_json_new_dict(json_filepath)

    return make_dataclass_from_dict(dataclass_ref, d)

plot_metapop_basic_compartment_history(metapop_model: MetapopModel, axes: matplotlib.axes.Axes = None)

Plots the compartment data for a metapopulation model.

Parameters:

Name	Type	Description	Default
metapop_model	MetapopModel	Metapopulation model containing compartments.	required
axes	Axes	Matplotlib axes to plot on.	None

Source code in CLT_BaseModel/clt_base/plotting.py

@plot_metapop_decorator
def plot_metapop_basic_compartment_history(metapop_model: MetapopModel,
                                           axes: matplotlib.axes.Axes = None):
    """
    Plots the compartment data for a metapopulation model.

    Args:
        metapop_model (MetapopModel):
            Metapopulation model containing compartments.
        axes (matplotlib.axes.Axes):
            Matplotlib axes to plot on.
    """

    # Iterate over subpop models and plot
    for ix, (subpop_name, subpop_model) in enumerate(metapop_model.subpop_models.items()):
        plot_subpop_basic_compartment_history(subpop_model, axes[ix])

plot_metapop_decorator(plot_func)

Decorator to handle common metapopulation plotting tasks.

Source code in CLT_BaseModel/clt_base/plotting.py

def plot_metapop_decorator(plot_func):
    """
    Decorator to handle common metapopulation plotting tasks.
    """

    @functools.wraps(plot_func)
    def wrapper(metapop_model: MetapopModel,
                savefig_filename = None):

        num_plots = len(metapop_model.subpop_models)
        num_cols = 2
        num_rows = (num_plots + num_cols - 1) // num_cols

        # Create figure and axes
        fig, axes = plt.subplots(num_rows, num_cols, figsize=(5 * num_cols, 4 * num_rows))
        axes = axes.flatten()

        plot_func(metapop_model=metapop_model, axes=axes)

        # Turn off any unused subplots
        for j in range(num_plots, len(axes)):
            fig.delaxes(axes[j])  # Remove empty subplot

        # Adjust layout and save/show the figure
        plt.tight_layout()

        if savefig_filename:
            plt.savefig(savefig_filename, dpi=1200)

        plt.show()

    return wrapper

plot_metapop_epi_metrics(metapop_model: MetapopModel, axes: matplotlib.axes.Axes)

Plots the EpiMetric data for a metapopulation model.

Parameters:

Name	Type	Description	Default
metapop_model	MetapopModel	Metapopulation model containing compartments.	required
axes	Axes	Matplotlib axes to plot on.	required

Source code in CLT_BaseModel/clt_base/plotting.py

@plot_metapop_decorator
def plot_metapop_epi_metrics(metapop_model: MetapopModel,
                             axes: matplotlib.axes.Axes):
    """
    Plots the EpiMetric data for a metapopulation model.

    Args:
        metapop_model (MetapopModel):
            Metapopulation model containing compartments.
        axes (matplotlib.axes.Axes):
            Matplotlib axes to plot on.
    """

    for ix, (subpop_name, subpop_model) in enumerate(metapop_model.subpop_models.items()):
        plot_subpop_epi_metrics(subpop_model, axes[ix])

plot_metapop_total_infected_deaths(metapop_model: MetapopModel, axes: matplotlib.axes.Axes)

Plots the total infected (IP+IS+IA) and deaths data for a metapopulation model.

Parameters:

Name	Type	Description	Default
metapop_model	MetapopModel	Metapopulation model containing compartments.	required
axes	Axes	Matplotlib axes to plot on.	required

Source code in CLT_BaseModel/clt_base/plotting.py

@plot_metapop_decorator
def plot_metapop_total_infected_deaths(metapop_model: MetapopModel,
                                       axes: matplotlib.axes.Axes):
    """
    Plots the total infected (IP+IS+IA) and deaths data for a metapopulation model.

    Args:
        metapop_model (MetapopModel):
            Metapopulation model containing compartments.
        axes (matplotlib.axes.Axes):
            Matplotlib axes to plot on.
    """

    # Iterate over subpop models and plot
    for ix, (subpop_name, subpop_model) in enumerate(metapop_model.subpop_models.items()):
        plot_subpop_total_infected_deaths(subpop_model, axes[ix])

plot_subpop_basic_compartment_history(subpop_model: SubpopModel, ax: matplotlib.axes.Axes = None)

Plots data for a single subpopulation model on the given axis.

Parameters:

Name	Type	Description	Default
subpop_model	SubpopModel	Subpopulation model containing compartments.	required
ax	Axes	Matplotlib axis to plot on.	None

Source code in CLT_BaseModel/clt_base/plotting.py

@plot_subpop_decorator
def plot_subpop_basic_compartment_history(subpop_model: SubpopModel,
                                          ax: matplotlib.axes.Axes = None):
    """
    Plots data for a single subpopulation model on the given axis.

    Args:
        subpop_model (SubpopModel):
            Subpopulation model containing compartments.
        ax (matplotlib.axes.Axes):
            Matplotlib axis to plot on.
    """

    for name, compartment in subpop_model.compartments.items():
        # Compute summed history values for each age-risk group
        history_vals_list = [np.sum(age_risk_group_entry) for
                             age_risk_group_entry in compartment.history_vals_list]

        # Plot data with a label
        ax.plot(history_vals_list, label=name, alpha=0.6)

    # Set axis title and labels
    ax.set_title(f"{subpop_model.name}")
    ax.set_xlabel("Days")
    ax.set_ylabel("Number of individuals")
    ax.legend()

plot_subpop_decorator(plot_func)

Decorator to handle common subpopulation plotting tasks.

Source code in CLT_BaseModel/clt_base/plotting.py

def plot_subpop_decorator(plot_func):
    """
    Decorator to handle common subpopulation plotting tasks.
    """

    @functools.wraps(plot_func)
    def wrapper(subpop_model: SubpopModel,
                ax: matplotlib.axes.Axes = None,
                savefig_filename: str = None):
        """
        Args:
            subpop_model (SubpopModel):
                SubpopModel to plot.
            ax (matplotlib.axes.Axes):
                Matplotlib axis to plot on.
            savefig_filename (str):
                Optional filename to save the figure.
        """

        ax_provided = ax

        # If no axis is provided, create own axis
        if ax is None:
            fig, ax = plt.subplots()

        plot_func(subpop_model=subpop_model, ax=ax)

        if savefig_filename:
            plt.savefig(savefig_filename, dpi=1200)

        if ax_provided is None:
            plt.show()

    return wrapper

plot_subpop_epi_metrics(subpop_model: SubpopModel, ax: matplotlib.axes.Axes = None)

Plots EpiMetric history for a single subpopulation model on the given axis.

Parameters:

Name	Type	Description	Default
subpop_model	SubpopModel	Subpopulation model containing compartments.	required
ax	Axes	Matplotlib axis to plot on.	None

Source code in CLT_BaseModel/clt_base/plotting.py

@plot_subpop_decorator
def plot_subpop_epi_metrics(subpop_model: SubpopModel,
                            ax: matplotlib.axes.Axes = None):
    """
    Plots EpiMetric history for a single subpopulation model on the given axis.

    Args:
        subpop_model (SubpopModel):
            Subpopulation model containing compartments.
        ax (matplotlib.axes.Axes):
            Matplotlib axis to plot on.
    """

    for name, epi_metric in subpop_model.epi_metrics.items():

        # Compute summed history values for each age-risk group
        history_vals_list = [np.average(age_risk_group_entry) for
                             age_risk_group_entry in epi_metric.history_vals_list]

        # Plot data with a label
        ax.plot(history_vals_list, label=name, alpha=0.6)

    # Set axis title and labels
    ax.set_title(f"{subpop_model.name}")
    ax.set_xlabel("Days")
    ax.set_ylabel("Epi Metric Value")
    ax.legend()

plot_subpop_total_infected_deaths(subpop_model: SubpopModel, ax: matplotlib.axes.Axes = None)

Plots data for a single subpopulation model on the given axis.

Parameters:

Name	Type	Description	Default
subpop_model	SubpopModel	Subpopulation model containing compartments.	required
ax	Axes	Matplotlib axis to plot on.	None

Source code in CLT_BaseModel/clt_base/plotting.py

@plot_subpop_decorator
def plot_subpop_total_infected_deaths(subpop_model: SubpopModel,
                                      ax: matplotlib.axes.Axes = None):
    """
    Plots data for a single subpopulation model on the given axis.

    Args:
        subpop_model (SubpopModel):
            Subpopulation model containing compartments.
        ax (matplotlib.axes.Axes):
            Matplotlib axis to plot on.
    """

    infected_compartment_names = [name for name in subpop_model.compartments.keys() if
                                  "I" in name or "H" in name]

    infected_compartments_history = [subpop_model.compartments[compartment_name].history_vals_list
                                     for compartment_name in infected_compartment_names]

    total_infected = np.sum(np.asarray(infected_compartments_history), axis=(0, 2, 3))

    ax.plot(total_infected, label="Total infected", alpha=0.6)

    if "D" in subpop_model.compartments.keys():
        deaths = [np.sum(age_risk_group_entry)
                  for age_risk_group_entry
                  in subpop_model.compartments.D.history_vals_list]

        ax.plot(deaths, label="D", alpha=0.6)

    ax.set_title(f"{subpop_model.name}")
    ax.set_xlabel("Days")
    ax.set_ylabel("Number of individuals")
    ax.legend()