Flu Core API Reference

Docstrings and references for flu_core module.

base_path = clt.utils.PROJECT_ROOT / 'flu_instances' / 'texas_input_files' module-attribute

AbsoluteHumidity

Bases: Schedule

Source code in CLT_BaseModel/flu_core/flu_components.py

class AbsoluteHumidity(clt.Schedule):

    def __init__(self,
                 init_val: Optional[np.ndarray | float] = None,
                 timeseries_df: pd.DataFrame = None):
        """
        Args:
            init_val (Optional[np.ndarray | float]):
                starting value(s) at the beginning of the simulation
            timeseries_df (Optional[pd.DataFrame] = None):
                must have columns "date" and "absolute_humidity" --
                "date" entries must correspond to consecutive calendar days
                and must either be strings with `"YYYY-MM-DD"` format or
                `datetime.date` objects -- "value" entries correspond to
                absolute humidity on those days. Identical to
                `FluSubpopSchedules` field of same name.
        """

        super().__init__(init_val)

        self.timeseries_df = timeseries_df

    def update_current_val(self, params, current_date: datetime.date) -> None:
        self.current_val = self.timeseries_df.loc[
            self.timeseries_df["date"] == current_date, "absolute_humidity"].values[0]

__init__(init_val: Optional[np.ndarray | float] = None, timeseries_df: pd.DataFrame = 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	must have columns "date" and "absolute_humidity" -- "date" entries must correspond to consecutive calendar days and must either be strings with "YYYY-MM-DD" format or datetime.date objects -- "value" entries correspond to absolute humidity on those days. Identical to FluSubpopSchedules field of same name.	None

Source code in CLT_BaseModel/flu_core/flu_components.py

def __init__(self,
             init_val: Optional[np.ndarray | float] = None,
             timeseries_df: pd.DataFrame = None):
    """
    Args:
        init_val (Optional[np.ndarray | float]):
            starting value(s) at the beginning of the simulation
        timeseries_df (Optional[pd.DataFrame] = None):
            must have columns "date" and "absolute_humidity" --
            "date" entries must correspond to consecutive calendar days
            and must either be strings with `"YYYY-MM-DD"` format or
            `datetime.date` objects -- "value" entries correspond to
            absolute humidity on those days. Identical to
            `FluSubpopSchedules` field of same name.
    """

    super().__init__(init_val)

    self.timeseries_df = timeseries_df

AsympToRecovered

Bases: TransitionVariable

TransitionVariable-derived class for movement from the "IA" to "R" compartment. The functional form is the same across subpopulations.

Source code in CLT_BaseModel/flu_core/flu_components.py

class AsympToRecovered(clt.TransitionVariable):
    """
    TransitionVariable-derived class for movement from the
    "IA" to "R" compartment. The functional form is the same across
    subpopulations.
    """

    def get_current_rate(self,
                         state: FluSubpopState,
                         params: FluSubpopParams) -> np.ndarray:
        """
        Returns:
            np.ndarray of shape (A, R)
        """

        return np.full((params.num_age_groups, params.num_risk_groups),
                       params.IA_to_R_rate)

get_current_rate(state: FluSubpopState, params: FluSubpopParams) -> np.ndarray

Returns:

Type	Description
ndarray	np.ndarray of shape (A, R)

Source code in CLT_BaseModel/flu_core/flu_components.py

def get_current_rate(self,
                     state: FluSubpopState,
                     params: FluSubpopParams) -> np.ndarray:
    """
    Returns:
        np.ndarray of shape (A, R)
    """

    return np.full((params.num_age_groups, params.num_risk_groups),
                   params.IA_to_R_rate)

BetaReduce

Bases: DynamicVal

"Toy" function representing staged-alert policy that reduces transmission by 50% when more than 5% of the total population is infected. Note: the numbers are completely made up :) The "permanent_lockdown" toggle is to avoid "bang-bang" behavior where the staged-alert policy gets triggered one day and then is off the next, and then is on the day after, and so on... but as the name suggests, it IS permanent. TODO: replace with realistic function.

Source code in CLT_BaseModel/flu_core/flu_components.py

class BetaReduce(clt.DynamicVal):
    """
    "Toy" function representing staged-alert policy
        that reduces transmission by 50% when more than 5%
        of the total population is infected. Note: the
        numbers are completely made up :)
    The "permanent_lockdown" toggle is to avoid "bang-bang"
        behavior where the staged-alert policy gets triggered
        one day and then is off the next, and then is on the
        day after, and so on... but as the name suggests,
        it IS permanent.
    TODO: replace with realistic function.
    """

    def __init__(self, init_val, is_enabled):
        super().__init__(init_val, is_enabled)
        self.permanent_lockdown = False

    def update_current_val(self, state, params):
        if np.sum(state.IS) / np.sum(params.total_pop_age_risk) > 0.05:
            self.current_val = .5
            self.permanent_lockdown = True
        else:
            if not self.permanent_lockdown:
                self.current_val = 0.0

DailyVaccines

Bases: Schedule

Source code in CLT_BaseModel/flu_core/flu_components.py

class DailyVaccines(clt.Schedule):

    def __init__(self,
                 init_val: Optional[np.ndarray | float] = None,
                 timeseries_df: pd.DataFrame = None):
        """
        Args:
            init_val (Optional[np.ndarray | float]):
                starting value(s) at the beginning of the simulation
            timeseries_df (Optional[pd.DataFrame] = None):
                must have "date" and "daily_vaccines" -- "date" entries must
                correspond to consecutive calendar days and must either
                be strings with `"YYYY-MM-DD"` format or `datetime.date`
                objects -- "value" entries correspond to historical
                number vaccinated on those days. Identical to
                `FluSubpopSchedules` field of same name.
        """

        super().__init__(init_val)

        self.timeseries_df = timeseries_df

    def update_current_val(self, params, current_date: datetime.date) -> None:
        self.current_val = self.timeseries_df.loc[
            self.timeseries_df["date"] == current_date, "daily_vaccines"].values[0]

__init__(init_val: Optional[np.ndarray | float] = None, timeseries_df: pd.DataFrame = 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	must have "date" and "daily_vaccines" -- "date" entries must correspond to consecutive calendar days and must either be strings with "YYYY-MM-DD" format or datetime.date objects -- "value" entries correspond to historical number vaccinated on those days. Identical to FluSubpopSchedules field of same name.	None

Source code in CLT_BaseModel/flu_core/flu_components.py

def __init__(self,
             init_val: Optional[np.ndarray | float] = None,
             timeseries_df: pd.DataFrame = None):
    """
    Args:
        init_val (Optional[np.ndarray | float]):
            starting value(s) at the beginning of the simulation
        timeseries_df (Optional[pd.DataFrame] = None):
            must have "date" and "daily_vaccines" -- "date" entries must
            correspond to consecutive calendar days and must either
            be strings with `"YYYY-MM-DD"` format or `datetime.date`
            objects -- "value" entries correspond to historical
            number vaccinated on those days. Identical to
            `FluSubpopSchedules` field of same name.
    """

    super().__init__(init_val)

    self.timeseries_df = timeseries_df

ExposedToAsymp

Bases: TransitionVariable

TransitionVariable-derived class for movement from the "E" to "IA" compartment. The functional form is the same across subpopulations.

Each ExposedToAsymp instance forms a TransitionVariableGroup with a corresponding ExposedToPresymp instance (these two transition variables are jointly distributed).

Source code in CLT_BaseModel/flu_core/flu_components.py

class ExposedToAsymp(clt.TransitionVariable):
    """
    TransitionVariable-derived class for movement from the
    "E" to "IA" compartment. The functional form is the same across
    subpopulations.

    Each ExposedToAsymp instance forms a TransitionVariableGroup with
    a corresponding ExposedToPresymp instance (these two
    transition variables are jointly distributed).
    """

    def get_current_rate(self,
                         state: FluSubpopState,
                         params: FluSubpopParams) -> np.ndarray:
        """
        Returns:
            np.ndarray of shape (A, R)
        """
        return np.full((params.num_age_groups, params.num_risk_groups),
                       params.E_to_I_rate * params.E_to_IA_prop)

get_current_rate(state: FluSubpopState, params: FluSubpopParams) -> np.ndarray

Returns:

Type	Description
ndarray	np.ndarray of shape (A, R)

Source code in CLT_BaseModel/flu_core/flu_components.py

def get_current_rate(self,
                     state: FluSubpopState,
                     params: FluSubpopParams) -> np.ndarray:
    """
    Returns:
        np.ndarray of shape (A, R)
    """
    return np.full((params.num_age_groups, params.num_risk_groups),
                   params.E_to_I_rate * params.E_to_IA_prop)

ExposedToPresymp

Bases: TransitionVariable

TransitionVariable-derived class for movement from the "E" to "IP" compartment. The functional form is the same across subpopulations.

Each ExposedToPresymp instance forms a TransitionVariableGroup with a corresponding ExposedToAsymp instance (these two transition variables are jointly distributed).

Source code in CLT_BaseModel/flu_core/flu_components.py

class ExposedToPresymp(clt.TransitionVariable):
    """
    TransitionVariable-derived class for movement from the
    "E" to "IP" compartment. The functional form is the same across
    subpopulations.

    Each ExposedToPresymp instance forms a TransitionVariableGroup with
    a corresponding ExposedToAsymp instance (these two
    transition variables are jointly distributed).
    """

    def get_current_rate(self,
                         state: FluSubpopState,
                         params: FluSubpopParams) -> np.ndarray:
        """
        Returns:
            np.ndarray of shape (A, R)
        """

        return np.full((params.num_age_groups, params.num_risk_groups),
                       params.E_to_I_rate * (1 - params.E_to_IA_prop))

get_current_rate(state: FluSubpopState, params: FluSubpopParams) -> np.ndarray

Returns:

Type	Description
ndarray	np.ndarray of shape (A, R)

Source code in CLT_BaseModel/flu_core/flu_components.py

def get_current_rate(self,
                     state: FluSubpopState,
                     params: FluSubpopParams) -> np.ndarray:
    """
    Returns:
        np.ndarray of shape (A, R)
    """

    return np.full((params.num_age_groups, params.num_risk_groups),
                   params.E_to_I_rate * (1 - params.E_to_IA_prop))

FluContactMatrix

Bases: Schedule

Flu contact matrix.

Attributes:

Name	Type	Description
timeseries_df	DataFrame	must have columns "date", "is_school_day", and "is_work_day" -- "date" entries must correspond to consecutive calendar days and must either be strings with "YYYY-MM-DD" format or datetime.date object and "is_school_day" and "is_work_day" entries are Booleans indicating if that date is a school day or work day. Identical to FluSubpopSchedules field of same name.

See parent class docstring for other attributes.

Source code in CLT_BaseModel/flu_core/flu_components.py

class FluContactMatrix(clt.Schedule):
    """
    Flu contact matrix.

    Attributes:
        timeseries_df (pd.DataFrame):
            must have columns "date", "is_school_day", and "is_work_day"
            -- "date" entries must correspond to consecutive calendar
            days and must either be strings with `"YYYY-MM-DD"` format
            or `datetime.date` object and "is_school_day" and
            "is_work_day" entries are Booleans indicating if that date is
            a school day or work day. Identical to `FluSubpopSchedules` field
            of same name.

    See parent class docstring for other attributes.
    """

    def __init__(self,
                 init_val: Optional[np.ndarray | float] = None,
                 timeseries_df: pd.DataFrame = None):

        super().__init__(init_val)

        self.timeseries_df = timeseries_df

    def update_current_val(self,
                           subpop_params: FluSubpopParams,
                           current_date: datetime.date) -> None:

        df = self.timeseries_df

        try:
            current_row = df[df["date"] == current_date].iloc[0]
            self.current_val = subpop_params.total_contact_matrix - \
                               (1 - current_row["is_school_day"]) * subpop_params.school_contact_matrix - \
                               (1 - current_row["is_work_day"]) * subpop_params.work_contact_matrix
        except IndexError:
            # print(f"Error: {current_date} is not in `timeseries_df`. Using total contact matrix.")
            self.current_val = subpop_params.total_contact_matrix

FluFullMetapopParamsTensors dataclass

Bases: FluTravelParamsTensors

Data container for tensors for FluMetapopModel -- used to store arrays that contain data across all subpopulations (collected from parameters on each location/subpopulation model, as well as from the metapopulation's associated FluMixingParams instance). Note that in contrast to FluTravelParamsTensors, ALL fields in FluSubpopParams are included -- this is for running the simulation via torch.

Attributes:

Name	Type	Description
num_locations	(Tensor, 0 - dimensional)	number of locations (subpopulations) in the metapopulation model and therefore the travel model.
travel_proportions	Tensor	L x L array, where L is the number of locations or subpopulations, where element i,j corresponds to the proportion of the population in location i who travels to location j (on average).

See FluSubpopParams docstring for other attributes. Other fields are analogous except they are size (L, A, R) tensors or size 0 tensors.

Source code in CLT_BaseModel/flu_core/flu_data_structures.py

@dataclass
class FluFullMetapopParamsTensors(FluTravelParamsTensors):
    """
    Data container for tensors for `FluMetapopModel` -- used to store arrays that
    contain data across all subpopulations (collected from parameters
    on each location/subpopulation model, as well as from the
    metapopulation's associated `FluMixingParams` instance).
    Note that in contrast to `FluTravelParamsTensors`,
    ALL fields in `FluSubpopParams` are included --
    this is for running the simulation via torch.

    Attributes:
        num_locations (torch.Tensor, 0-dimensional):
            number of locations (subpopulations) in the
            metapopulation model and therefore the travel
            model.
        travel_proportions (torch.Tensor):
            L x L array, where L is the number of locations
            or subpopulations, where element i,j corresponds
            to the proportion of the population in location i
            who travels to location j (on average).

    See `FluSubpopParams` docstring for other attributes.
    Other fields are analogous except they are size (L, A, R)
    tensors or size 0 tensors.
    """

    beta_baseline: Optional[torch.Tensor] = None
    total_pop_age_risk: Optional[torch.Tensor] = None
    humidity_impact: Optional[torch.Tensor] = None

    inf_induced_saturation: Optional[torch.Tensor] = None
    inf_induced_immune_wane: Optional[torch.Tensor] = None
    vax_induced_saturation: Optional[torch.Tensor] = None
    vax_induced_immune_wane: Optional[torch.Tensor] = None
    inf_induced_inf_risk_reduce: Optional[torch.Tensor] = None
    inf_induced_hosp_risk_reduce: Optional[torch.Tensor] = None
    inf_induced_death_risk_reduce: Optional[torch.Tensor] = None
    vax_induced_inf_risk_reduce: Optional[torch.Tensor] = None
    vax_induced_hosp_risk_reduce: Optional[torch.Tensor] = None
    vax_induced_death_risk_reduce: Optional[torch.Tensor] = None

    R_to_S_rate: Optional[torch.Tensor] = None
    E_to_I_rate: Optional[torch.Tensor] = None
    IP_to_IS_rate: Optional[torch.Tensor] = None
    IS_to_R_rate: Optional[torch.Tensor] = None
    IA_to_R_rate: Optional[torch.Tensor] = None
    IS_to_H_rate: Optional[torch.Tensor] = None
    H_to_R_rate: Optional[torch.Tensor] = None
    H_to_D_rate: Optional[torch.Tensor] = None
    E_to_IA_prop: Optional[torch.Tensor] = None

    IS_to_H_adjusted_prop: Optional[torch.Tensor] = None
    H_to_D_adjusted_prop: Optional[torch.Tensor] = None

    IP_relative_inf: Optional[torch.Tensor] = None
    IA_relative_inf: Optional[torch.Tensor] = None

    relative_suscept: Optional[torch.Tensor] = None
    mobility_modifier: Optional[torch.Tensor] = None

FluFullMetapopScheduleTensors dataclass

Source code in CLT_BaseModel/flu_core/flu_data_structures.py

@dataclass
class FluFullMetapopScheduleTensors:

    absolute_humidity: Optional[list[torch.tensor]] = None
    is_school_day: Optional[list[torch.tensor]] = None
    is_work_day: Optional[list[torch.tensor]] = None
    daily_vaccines: Optional[list[torch.tensor]] = None

FluFullMetapopStateTensors dataclass

Bases: FluTravelStateTensors

Data container for tensors for FluMetapopModel -- used to store arrays that contain data across all subpopulations (collected from each location/subpopulation model). In contrast to FluTravelStateTensors, ALL fields in FluSubpopState are included -- this is for running the simulation via torch.

Attributes:

Name	Type	Description
flu_contact_matrix	torch.Tensor of nonnegative integers	contact matrix for location-age-risk groups -- the lth element holds current_val of FluContactMatrix Schedule for subpopulation l -- this value is a combination of the total contact matrix, the work contact matrix, and the school contact matrix (and the value is adjusted depending on whether the date is a work or school day)
init_vals	dict	dictionary of torch.Tensor instances, where keys correspond to "IP", "IS", "IA", and "H", and values correspond to their initial values for location-age-risk groups.

See FluSubpopState and FluTravelStateTensors for other attributes -- other attributes here correspond to FluSubpopState, but are size (L, A, R) tensors for location-age-risk or size 0 tensors.

Source code in CLT_BaseModel/flu_core/flu_data_structures.py

@dataclass
class FluFullMetapopStateTensors(FluTravelStateTensors):
    """
    Data container for tensors for `FluMetapopModel` -- used to store arrays that
    contain data across all subpopulations (collected from each
    location/subpopulation model). In contrast to `FluTravelStateTensors`,
    ALL fields in `FluSubpopState` are included -- this is
    for running the simulation via torch.

    Attributes:
        flu_contact_matrix (torch.Tensor of nonnegative integers):
            contact matrix for location-age-risk groups -- the
            lth element holds current_val of `FluContactMatrix`
            `Schedule` for subpopulation l -- this value is a
            combination of the total contact matrix, the
            work contact matrix, and the school contact matrix
            (and the value is adjusted depending on whether
            the date is a work or school day)
        init_vals (dict):
            dictionary of torch.Tensor instances, where keys
            correspond to "IP", "IS", "IA", and "H", and values
            correspond to their initial values for location-age-risk
            groups.

    See `FluSubpopState` and `FluTravelStateTensors` for other
        attributes -- other attributes here correspond to
        `FluSubpopState`, but are size (L, A, R) tensors for
        location-age-risk or size 0 tensors.
    """

    # `IP`, `IS`, `IA`, `H`, `flu_contact_matrix` already in
    #   parent class
    # Same with `init_vals`

    S: Optional[torch.Tensor] = None
    E: Optional[torch.Tensor] = None
    R: Optional[torch.Tensor] = None
    D: Optional[torch.Tensor] = None

    M: Optional[torch.Tensor] = None
    MV: Optional[torch.Tensor] = None

    absolute_humidity: Optional[float] = None
    daily_vaccines: Optional[torch.Tensor] = None

FluMetapopModel

Bases: MetapopModel, ABC

MetapopModel-derived class specific to flu model.

Source code in CLT_BaseModel/flu_core/flu_components.py

class FluMetapopModel(clt.MetapopModel, ABC):
    """
    MetapopModel-derived class specific to flu model.
    """

    def __init__(self,
                 subpop_models: list[dict],
                 mixing_params: FluMixingParams,
                 name: str = ""):

        super().__init__(subpop_models,
                         mixing_params,
                         name)

        # Confirm validity and consistency of `FluMixingParams`
        try:
            num_locations = mixing_params.num_locations
        except KeyError:
            raise FluMetapopModelError("'mixing_params' must contain the key 'num_locations'. \n"
                                       "Please specify it before continuing.")
        if num_locations != len(subpop_models):
            raise FluMetapopModelError("'num_locations' should equal the number of items in \n"
                                       "'subpop_models'. Please amend before continuing.")

        self.travel_state_tensors = FluTravelStateTensors()
        self.update_travel_state_tensors()

        # `FluMixingParams` info is stored on `FluTravelParamsTensors` --
        # this order of operations below is important, because
        # `mixing_params` attribute must be defined before `update_travel_params_tensors()`
        # is called.
        self.mixing_params = mixing_params
        self.travel_params_tensors = FluTravelParamsTensors()
        self.update_travel_params_tensors()

        total_pop_LAR_tensor = self.compute_total_pop_LAR_tensor()

        self.precomputed = FluPrecomputedTensors(total_pop_LAR_tensor,
                                                 self.travel_params_tensors)

        # Generally not used unless using torch version
        self._full_metapop_params_tensors = None
        self._full_metapop_state_tensors = None
        self._full_metapop_schedule_tensors = None

    def modify_subpop_params(self,
                             subpop_name: str,
                             updates_dict: dict):
        """
        This method lets users safely modify a single subpopulation
        parameters field; the metapopulation-wide tensors are updated
        automatically afterward.

        In a `FluMetapopModel`, subpopulation parameters are combined into
        (L, A, R) tensors across L subpopulations.`FluSubpopParams` is a frozen
        dataclass to avoid users naively changing parameter values and getting
        undesirable results -- thus, `FluSubpopParams` on a subpopulation
        model cannot be updated directly.

        Parameters:
            subpop_name (str):
               Value must match the `name` attribute of one of the
               `FluSubpopModel` instances contained in this metapopulation
                model's `subpop_models` attribute.
            updates_dict (dict):
                Dictionary specifying values to update in a
                `FluSubpopParams` instance -- keys must match the
                field names of `FluSubpopParams`.
        """

        # Since `FluSubpopParams` is frozen, we return a new instance
        #   with the reflected updates
        self.subpop_models[subpop_name].params = clt.updated_dataclass(
            self.subpop_models[subpop_name].params, updates_dict
        )

        self.update_travel_params_tensors()

        # Adding this for extra safety in case the user does not
        # call `get_flu_torch_inputs` for accessing the
        # `FullMetapopParams` instance.

        # If this attribute is not `None`, it means we are using
        # the `torch` implementation, and we should update the
        # corresponding `FullMetapopParams` instance with the new
        # `FluMixingParams` values.
        if self._full_metapop_params_tensors:
            self.update_full_metapop_params_tensors()

    def modify_mixing_params(self,
                             updates_dict: dict):
        """
        This method lets users safely modify flu mixing parameters;
        the metapopulation-wide tensors are updated automatically afterward.
        `FluMixingParams` is a frozen dataclass to avoid users
        naively changing parameter values and getting undesirable results --
        thus, `FluMixingParams` cannot be updated directly.

        Parameters:
            updates_dict (dict):
                Dictionary specifying values to update in a
                `FluSubpopParams` instance -- keys must match the
                field names of `FluSubpopParams`. 
        """

        self.mixing_params = clt.updated_dataclass(self.mixing_params, updates_dict)
        self.update_travel_params_tensors()

        nonlocal_travel_prop = self.travel_params_tensors.travel_proportions.clone().fill_diagonal_(0.0)

        self.precomputed.sum_residents_nonlocal_travel_prop = nonlocal_travel_prop.sum(dim=1)

        # Adding this for extra safety in case the user does not
        # call `get_flu_torch_inputs` for accessing the
        # `FullMetapopParams` instance.

        # If this attribute is not `None`, it means we are using
        # the `torch` implementation, and we should update the
        # corresponding `FullMetapopParams` instance with the new
        # `FluMixingParams` values.
        if self._full_metapop_params_tensors:
            self.update_full_metapop_params_tensors()

    def compute_total_pop_LAR_tensor(self) -> torch.tensor:
        """
        For each subpopulation, sum initial values of population
        in each compartment for age-risk groups. Store all information
        as tensor and return tensor.

        Returns:
        --------
        torch.tensor of size (L, A, R):
            Total population (across all compartments) for
            location-age-risk (l, a, r).
        """

        # ORDER MATTERS! USE ORDERED DICTIONARY HERE
        #   to preserve correct index order in tensors!
        #   See `update_travel_params_tensors` for detailed note.
        subpop_models_ordered = self._subpop_models_ordered

        total_pop_LAR_tensor = torch.zeros(self.travel_params_tensors.num_locations,
                                           self.travel_params_tensors.num_age_groups,
                                           self.travel_params_tensors.num_risk_groups)

        # All subpop models should have the same compartments' keys
        for name in subpop_models_ordered[0].compartments.keys():

            metapop_vals = []

            for model in subpop_models_ordered.values():
                compartment = getattr(model.compartments, name)
                metapop_vals.append(compartment.current_val)

            total_pop_LAR_tensor = total_pop_LAR_tensor + torch.tensor(np.asarray(metapop_vals))

        return total_pop_LAR_tensor

    def update_state_tensors(self,
                             target: FluTravelStateTensors) -> None:
        """
        Update `target` instance in-place with current simulation
        values. Each field of `target` corresponds to a field in
        `FluSubpopState`, and contains either a tensor of size
        (L, A, R) or a tensor of size (L), where (l, a, r) refers to
        location-age-risk.
        """

        # ORDER MATTERS! USE ORDERED DICTIONARY HERE
        #   to preserve correct index order in tensors!
        #   See `update_travel_params_tensors` for detailed note.
        subpop_models_ordered = self._subpop_models_ordered

        for field in fields(target):

            name = field.name

            # FluTravelStateTensors has an attribute
            #   that is a dictionary called `init_vals` --
            #   disregard, as this only used to store
            #   initial values for resetting, but is not
            #   used in the travel model computation
            if name == "init_vals":
                continue

            metapop_vals = []

            for model in subpop_models_ordered.values():
                current_val = getattr(model.state, name)
                metapop_vals.append(current_val)

            # Probably want to update this to be cleaner...
            # `SubpopState` fields that correspond to `Schedule` instances
            # have initial values of `None` -- but we cannot build a tensor
            # with `None` values, so we convert values to 0s.
            if any(v is None for v in metapop_vals):
                setattr(target, name, torch.tensor(np.full(np.shape(metapop_vals), 0.0)))
            else:
                setattr(target, name, torch.tensor(np.asarray(metapop_vals)))

            # Only fields corresponding to `Schedule` instances can be
            # size (L) -- this is because the schedule value may be scalar for
            # each subpopulation. Other fields should all be size (L, A, R). 

    def update_travel_state_tensors(self) -> None:
        """
        Update `travel_state_tensors` attribute in-place.
        `FluTravelStateTensors` only has fields corresponding
        to state variables relevant for the travel model.
        Converts subpopulation-specific state to
        tensors of size (L, A, R) for location-age-risk
        (except for a few exceptions that have different dimensions).
        """

        self.update_state_tensors(self.travel_state_tensors)

    def update_full_metapop_state_tensors(self) -> None:
        """
        Update `_full_metapop_state_tensors` attribute in-place.
        `FluFullMetapopStateTensors` has fields corresponding
        to all state variables in the simulation.
        Converts subpopulation-specific state to
        tensors of size (L, A, R) for location-age-risk
        (except for a few exceptions that have different dimensions).
        """

        if self._full_metapop_state_tensors is None:
            self._full_metapop_state_tensors = FluFullMetapopStateTensors()
        self.update_state_tensors(self._full_metapop_state_tensors)

    def update_params_tensors(self,
                              target: FluTravelParamsTensors) -> FluTravelParamsTensors:
        """
        Update `target` in-place. Converts subpopulation-specific
        parameters to tensors of size (L, A, R) for location-age-risk,
        except for `num_locations` and `travel_proportions`, which
        have size 1 and (L, L) respectively.
        """

        # USE THE ORDERED DICTIONARY HERE FOR SAFETY!
        #   AGAIN, ORDER MATTERS BECAUSE ORDER DETERMINES
        #   THE SUBPOPULATION INDEX IN THE METAPOPULATION
        #   TENSOR!
        subpop_models_ordered = self._subpop_models_ordered

        # Subpop models should have the same A, R so grab
        #   from the first subpop model
        A = subpop_models_ordered[0].params.num_age_groups
        R = subpop_models_ordered[0].params.num_risk_groups

        for field in fields(target):

            name = field.name

            metapop_vals = []

            if name == "num_locations" or name == "travel_proportions":
                setattr(target, name, torch.tensor(getattr(self.mixing_params, name)))

            else:
                for model in subpop_models_ordered.values():
                    metapop_vals.append(getattr(model.params, name))

                # If all values are equal to each other, then
                #   simply store the first value (since its value is common
                #   across metapopulations)
                first_val = metapop_vals[0]
                if all(np.allclose(x, first_val) for x in metapop_vals):
                    metapop_vals = first_val

                # Converting list of arrays to tensors is slow --
                #   better to convert to array first
                if isinstance(metapop_vals, list):
                    metapop_vals = np.asarray(metapop_vals)
                    # metapop_vals = np.stack([clt.to_AR_array(x, A, R) for x in metapop_vals])

                setattr(target, name, torch.tensor(metapop_vals))

        # Convert all tensors to correct size!
        target.standardize_shapes()

    def update_travel_params_tensors(self) -> None:
        """
        Update `travel_params_tensors` attribute in-place.
        `FluTravelParamsTensors` only has fields corresponding
        to parameters relevant for the travel model.
        Converts subpopulation-specific parameters to
        tensors of size (L, A, R) for location-age-risk
        (except for a few exceptions that have different dimensions).
        """

        self.update_params_tensors(target=self.travel_params_tensors)

    def update_full_metapop_params_tensors(self) -> None:
        """
        Update `_full_metapop_params_tensors` attribute in-place.
        `FluFullMetapopParamsTensors` has fields corresponding
        to all parameters in the simulation. Converts subpopulation-specific
        parameters to tensors of size (L, A, R) for location-age-risk
        (except for a few exceptions that have different dimensions).
        """

        if self._full_metapop_params_tensors is None:
            self._full_metapop_params_tensors = FluFullMetapopParamsTensors()
        self.update_params_tensors(target=self._full_metapop_params_tensors)

    def apply_inter_subpop_updates(self) -> None:
        """
        Update the `FluTravelStateTensors` according to the simulation state
        and compute the total mixing exposure, which includes across-subpopulation
        mixing/travel. Update the `total_mixing_exposure` attribute on each
        subpopulation's `SusceptibleToExposed` instance accordingly, so each
        of these transition variables can compute its transition rate.

        See `apply_inter_subpop_updates` on `MetapopModel` base class
        for logic of how/when this is called in the simulation.
        """

        self.update_travel_state_tensors()

        total_mixing_exposure = compute_total_mixing_exposure(self.travel_state_tensors,
                                                              self.travel_params_tensors,
                                                              self.precomputed)

        # Again, `self.subpop_models` is an ordered dictionary --
        #   so iterating over the dictionary like this is well-defined
        #   and responsible -- the order is important because it
        #   determines the order (index) in any metapopulation tensors
        subpop_models = self._subpop_models_ordered

        # Updates `total_mixing_exposure` attribute on each `SusceptibleToExposed`
        # instance -- this value captures across-population travel/mixing.
        for i in range(len(subpop_models)):
            subpop_models.values()[i].transition_variables.S_to_E.total_mixing_exposure = \
                total_mixing_exposure[i, :, :]

    def setup_full_metapop_schedule_tensors(self):
        """
        Creates `FluFullMetapopScheduleTensors` instance and assigns to
        `_full_metapop_schedule_tensors` attribute.

        For the metapopulation model's L locations/subpopulations, for each day,
        each value-related column in each schedule is either a float or
        array of size (A, R) for age-risk groups.

        We aggregate and reformat this schedule information and put it
        into a `FluFullMetapopScheduleTensors` instance, where fields
        correspond to a schedule value, and values are lists of tensors of
        size (L, A, R). The ith element of each list corresponds to the
        ith simulation day.
        """

        self._full_metapop_schedule_tensors = FluFullMetapopScheduleTensors()

        L = self.precomputed.L
        A = self.precomputed.A
        R = self.precomputed.R

        # Note: there is probably a more consistent way to do this,
        # because now `flu_contact_matrix` has two values: "is_school_day"
        # and "is_work_day" -- other schedules' dataframes only have one
        # relevant column value rather than two
        for item in [("absolute_humidity", "absolute_humidity"),
                     ("flu_contact_matrix", "is_school_day"),
                     ("flu_contact_matrix", "is_work_day"),
                     ("daily_vaccines", "daily_vaccines")]:

            schedule_name = item[0]
            values_column_name = item[1]

            metapop_vals = []

            for subpop_model in self._subpop_models_ordered.values():
                df = subpop_model.schedules[schedule_name].timeseries_df

                # Using the `start_real_date` specification given in subpop's `SimulationSettings`,
                # extract the relevant part of the dataframe with dates >= the simulation start date.
                # Note that `start_real_date` should be the same for each subpopulation
                start_date = datetime.datetime.strptime(subpop_model.simulation_settings.start_real_date, "%Y-%m-%d")
                df["simulation_day"] = (pd.to_datetime(df["date"], format="%Y-%m-%d") - start_date).dt.days
                df = df[df["simulation_day"] >= 0]

                # Make each day's value an A x R array
                # Pandas complains about `SettingWithCopyWarning` so we work on a copy explicitly to stop it
                #   from complaining...
                df = df.copy()
                df[values_column_name] = df[values_column_name].astype(object)
                df.loc[:, values_column_name] = df[values_column_name].apply(
                    lambda x, A=A, R=R: np.broadcast_to(np.asarray(x).reshape(1, 1), (A, R))
                )

                metapop_vals.append(np.asarray(df[values_column_name]))

            # IMPORTANT: tedious array/tensor shape/size manipulation here
            # metapop_vals: list of L arrays, each shape (num_days, A, R)
            # We need to transpose this... to be a list of num_days tensors, of size L x A x R
            num_items = metapop_vals[0].shape[0]

            # This is ugly and inefficient -- but at least we only do this once, when we get the initial
            #   state of a metapopulation model in tensor form
            transposed_metapop_vals = [torch.tensor(np.array([metapop_vals[l][i] for l in range(L)])) for i in
                                       range(num_items)]

            setattr(self._full_metapop_schedule_tensors, values_column_name, transposed_metapop_vals)

    def get_flu_torch_inputs(self) -> dict:
        """
        Prepares and returns metapopulation simulation data in tensor format
        that can be directly used for `torch` implementation.

        Returns:
             d (dict):
                Has keys "state_tensors", "params_tensors", "schedule_tensors",
                and "precomputed". Corresponds to `FluFullMetapopStateTensors`,
                `FluFullMetapopParamsTensors`, `FluFullMetapopScheduleTensors`,
                and `FluPrecomputedTensors` instances respectively.
        """

        # Note: does not support dynamic variables (yet). If want to
        #   run pytorch with dynamic variables, will need to create
        #   a method similar to `setup_full_metapop_schedule_tensors`
        #   but for dynamic variables. Also note that we cannot differentiate
        #   with respect to dynamic variables that are discontinuous
        #   (e.g. a 0-1 intervention) -- so we cannot optimize discontinuous
        #   dynamic variables.

        self.update_full_metapop_state_tensors()
        self.update_full_metapop_params_tensors()
        self._full_metapop_params_tensors.standardize_shapes()
        self.setup_full_metapop_schedule_tensors()

        d = {}

        d["state_tensors"] = copy.deepcopy(self._full_metapop_state_tensors)
        d["params_tensors"] = copy.deepcopy(self._full_metapop_params_tensors)
        d["schedule_tensors"] = copy.deepcopy(self._full_metapop_schedule_tensors)
        d["precomputed"] = copy.deepcopy(self.precomputed)

        return d

apply_inter_subpop_updates() -> None

Update the FluTravelStateTensors according to the simulation state and compute the total mixing exposure, which includes across-subpopulation mixing/travel. Update the total_mixing_exposure attribute on each subpopulation's SusceptibleToExposed instance accordingly, so each of these transition variables can compute its transition rate.

See apply_inter_subpop_updates on MetapopModel base class for logic of how/when this is called in the simulation.

Source code in CLT_BaseModel/flu_core/flu_components.py

def apply_inter_subpop_updates(self) -> None:
    """
    Update the `FluTravelStateTensors` according to the simulation state
    and compute the total mixing exposure, which includes across-subpopulation
    mixing/travel. Update the `total_mixing_exposure` attribute on each
    subpopulation's `SusceptibleToExposed` instance accordingly, so each
    of these transition variables can compute its transition rate.

    See `apply_inter_subpop_updates` on `MetapopModel` base class
    for logic of how/when this is called in the simulation.
    """

    self.update_travel_state_tensors()

    total_mixing_exposure = compute_total_mixing_exposure(self.travel_state_tensors,
                                                          self.travel_params_tensors,
                                                          self.precomputed)

    # Again, `self.subpop_models` is an ordered dictionary --
    #   so iterating over the dictionary like this is well-defined
    #   and responsible -- the order is important because it
    #   determines the order (index) in any metapopulation tensors
    subpop_models = self._subpop_models_ordered

    # Updates `total_mixing_exposure` attribute on each `SusceptibleToExposed`
    # instance -- this value captures across-population travel/mixing.
    for i in range(len(subpop_models)):
        subpop_models.values()[i].transition_variables.S_to_E.total_mixing_exposure = \
            total_mixing_exposure[i, :, :]

compute_total_pop_LAR_tensor() -> torch.tensor

For each subpopulation, sum initial values of population in each compartment for age-risk groups. Store all information as tensor and return tensor.

Returns:

torch.tensor of size (L, A, R): Total population (across all compartments) for location-age-risk (l, a, r).

Source code in CLT_BaseModel/flu_core/flu_components.py

def compute_total_pop_LAR_tensor(self) -> torch.tensor:
    """
    For each subpopulation, sum initial values of population
    in each compartment for age-risk groups. Store all information
    as tensor and return tensor.

    Returns:
    --------
    torch.tensor of size (L, A, R):
        Total population (across all compartments) for
        location-age-risk (l, a, r).
    """

    # ORDER MATTERS! USE ORDERED DICTIONARY HERE
    #   to preserve correct index order in tensors!
    #   See `update_travel_params_tensors` for detailed note.
    subpop_models_ordered = self._subpop_models_ordered

    total_pop_LAR_tensor = torch.zeros(self.travel_params_tensors.num_locations,
                                       self.travel_params_tensors.num_age_groups,
                                       self.travel_params_tensors.num_risk_groups)

    # All subpop models should have the same compartments' keys
    for name in subpop_models_ordered[0].compartments.keys():

        metapop_vals = []

        for model in subpop_models_ordered.values():
            compartment = getattr(model.compartments, name)
            metapop_vals.append(compartment.current_val)

        total_pop_LAR_tensor = total_pop_LAR_tensor + torch.tensor(np.asarray(metapop_vals))

    return total_pop_LAR_tensor

get_flu_torch_inputs() -> dict

Prepares and returns metapopulation simulation data in tensor format that can be directly used for torch implementation.

Returns:

Name	Type	Description
d	dict	Has keys "state_tensors", "params_tensors", "schedule_tensors", and "precomputed". Corresponds to FluFullMetapopStateTensors, FluFullMetapopParamsTensors, FluFullMetapopScheduleTensors, and FluPrecomputedTensors instances respectively.

Source code in CLT_BaseModel/flu_core/flu_components.py

def get_flu_torch_inputs(self) -> dict:
    """
    Prepares and returns metapopulation simulation data in tensor format
    that can be directly used for `torch` implementation.

    Returns:
         d (dict):
            Has keys "state_tensors", "params_tensors", "schedule_tensors",
            and "precomputed". Corresponds to `FluFullMetapopStateTensors`,
            `FluFullMetapopParamsTensors`, `FluFullMetapopScheduleTensors`,
            and `FluPrecomputedTensors` instances respectively.
    """

    # Note: does not support dynamic variables (yet). If want to
    #   run pytorch with dynamic variables, will need to create
    #   a method similar to `setup_full_metapop_schedule_tensors`
    #   but for dynamic variables. Also note that we cannot differentiate
    #   with respect to dynamic variables that are discontinuous
    #   (e.g. a 0-1 intervention) -- so we cannot optimize discontinuous
    #   dynamic variables.

    self.update_full_metapop_state_tensors()
    self.update_full_metapop_params_tensors()
    self._full_metapop_params_tensors.standardize_shapes()
    self.setup_full_metapop_schedule_tensors()

    d = {}

    d["state_tensors"] = copy.deepcopy(self._full_metapop_state_tensors)
    d["params_tensors"] = copy.deepcopy(self._full_metapop_params_tensors)
    d["schedule_tensors"] = copy.deepcopy(self._full_metapop_schedule_tensors)
    d["precomputed"] = copy.deepcopy(self.precomputed)

    return d

modify_mixing_params(updates_dict: dict)

This method lets users safely modify flu mixing parameters; the metapopulation-wide tensors are updated automatically afterward. FluMixingParams is a frozen dataclass to avoid users naively changing parameter values and getting undesirable results -- thus, FluMixingParams cannot be updated directly.

Parameters:

Name	Type	Description	Default
updates_dict	dict	Dictionary specifying values to update in a FluSubpopParams instance -- keys must match the field names of FluSubpopParams.	required

Source code in CLT_BaseModel/flu_core/flu_components.py

def modify_mixing_params(self,
                         updates_dict: dict):
    """
    This method lets users safely modify flu mixing parameters;
    the metapopulation-wide tensors are updated automatically afterward.
    `FluMixingParams` is a frozen dataclass to avoid users
    naively changing parameter values and getting undesirable results --
    thus, `FluMixingParams` cannot be updated directly.

    Parameters:
        updates_dict (dict):
            Dictionary specifying values to update in a
            `FluSubpopParams` instance -- keys must match the
            field names of `FluSubpopParams`. 
    """

    self.mixing_params = clt.updated_dataclass(self.mixing_params, updates_dict)
    self.update_travel_params_tensors()

    nonlocal_travel_prop = self.travel_params_tensors.travel_proportions.clone().fill_diagonal_(0.0)

    self.precomputed.sum_residents_nonlocal_travel_prop = nonlocal_travel_prop.sum(dim=1)

    # Adding this for extra safety in case the user does not
    # call `get_flu_torch_inputs` for accessing the
    # `FullMetapopParams` instance.

    # If this attribute is not `None`, it means we are using
    # the `torch` implementation, and we should update the
    # corresponding `FullMetapopParams` instance with the new
    # `FluMixingParams` values.
    if self._full_metapop_params_tensors:
        self.update_full_metapop_params_tensors()

modify_subpop_params(subpop_name: str, updates_dict: dict)

This method lets users safely modify a single subpopulation parameters field; the metapopulation-wide tensors are updated automatically afterward.

In a FluMetapopModel, subpopulation parameters are combined into (L, A, R) tensors across L subpopulations.FluSubpopParams is a frozen dataclass to avoid users naively changing parameter values and getting undesirable results -- thus, FluSubpopParams on a subpopulation model cannot be updated directly.

Parameters:

Name	Type	Description	Default
subpop_name	str	Value must match the name attribute of one of the FluSubpopModel instances contained in this metapopulation model's subpop_models attribute.	required
updates_dict	dict	Dictionary specifying values to update in a FluSubpopParams instance -- keys must match the field names of FluSubpopParams.	required

Source code in CLT_BaseModel/flu_core/flu_components.py

def modify_subpop_params(self,
                         subpop_name: str,
                         updates_dict: dict):
    """
    This method lets users safely modify a single subpopulation
    parameters field; the metapopulation-wide tensors are updated
    automatically afterward.

    In a `FluMetapopModel`, subpopulation parameters are combined into
    (L, A, R) tensors across L subpopulations.`FluSubpopParams` is a frozen
    dataclass to avoid users naively changing parameter values and getting
    undesirable results -- thus, `FluSubpopParams` on a subpopulation
    model cannot be updated directly.

    Parameters:
        subpop_name (str):
           Value must match the `name` attribute of one of the
           `FluSubpopModel` instances contained in this metapopulation
            model's `subpop_models` attribute.
        updates_dict (dict):
            Dictionary specifying values to update in a
            `FluSubpopParams` instance -- keys must match the
            field names of `FluSubpopParams`.
    """

    # Since `FluSubpopParams` is frozen, we return a new instance
    #   with the reflected updates
    self.subpop_models[subpop_name].params = clt.updated_dataclass(
        self.subpop_models[subpop_name].params, updates_dict
    )

    self.update_travel_params_tensors()

    # Adding this for extra safety in case the user does not
    # call `get_flu_torch_inputs` for accessing the
    # `FullMetapopParams` instance.

    # If this attribute is not `None`, it means we are using
    # the `torch` implementation, and we should update the
    # corresponding `FullMetapopParams` instance with the new
    # `FluMixingParams` values.
    if self._full_metapop_params_tensors:
        self.update_full_metapop_params_tensors()

setup_full_metapop_schedule_tensors()

Creates FluFullMetapopScheduleTensors instance and assigns to _full_metapop_schedule_tensors attribute.

For the metapopulation model's L locations/subpopulations, for each day, each value-related column in each schedule is either a float or array of size (A, R) for age-risk groups.

We aggregate and reformat this schedule information and put it into a FluFullMetapopScheduleTensors instance, where fields correspond to a schedule value, and values are lists of tensors of size (L, A, R). The ith element of each list corresponds to the ith simulation day.

Source code in CLT_BaseModel/flu_core/flu_components.py

def setup_full_metapop_schedule_tensors(self):
    """
    Creates `FluFullMetapopScheduleTensors` instance and assigns to
    `_full_metapop_schedule_tensors` attribute.

    For the metapopulation model's L locations/subpopulations, for each day,
    each value-related column in each schedule is either a float or
    array of size (A, R) for age-risk groups.

    We aggregate and reformat this schedule information and put it
    into a `FluFullMetapopScheduleTensors` instance, where fields
    correspond to a schedule value, and values are lists of tensors of
    size (L, A, R). The ith element of each list corresponds to the
    ith simulation day.
    """

    self._full_metapop_schedule_tensors = FluFullMetapopScheduleTensors()

    L = self.precomputed.L
    A = self.precomputed.A
    R = self.precomputed.R

    # Note: there is probably a more consistent way to do this,
    # because now `flu_contact_matrix` has two values: "is_school_day"
    # and "is_work_day" -- other schedules' dataframes only have one
    # relevant column value rather than two
    for item in [("absolute_humidity", "absolute_humidity"),
                 ("flu_contact_matrix", "is_school_day"),
                 ("flu_contact_matrix", "is_work_day"),
                 ("daily_vaccines", "daily_vaccines")]:

        schedule_name = item[0]
        values_column_name = item[1]

        metapop_vals = []

        for subpop_model in self._subpop_models_ordered.values():
            df = subpop_model.schedules[schedule_name].timeseries_df

            # Using the `start_real_date` specification given in subpop's `SimulationSettings`,
            # extract the relevant part of the dataframe with dates >= the simulation start date.
            # Note that `start_real_date` should be the same for each subpopulation
            start_date = datetime.datetime.strptime(subpop_model.simulation_settings.start_real_date, "%Y-%m-%d")
            df["simulation_day"] = (pd.to_datetime(df["date"], format="%Y-%m-%d") - start_date).dt.days
            df = df[df["simulation_day"] >= 0]

            # Make each day's value an A x R array
            # Pandas complains about `SettingWithCopyWarning` so we work on a copy explicitly to stop it
            #   from complaining...
            df = df.copy()
            df[values_column_name] = df[values_column_name].astype(object)
            df.loc[:, values_column_name] = df[values_column_name].apply(
                lambda x, A=A, R=R: np.broadcast_to(np.asarray(x).reshape(1, 1), (A, R))
            )

            metapop_vals.append(np.asarray(df[values_column_name]))

        # IMPORTANT: tedious array/tensor shape/size manipulation here
        # metapop_vals: list of L arrays, each shape (num_days, A, R)
        # We need to transpose this... to be a list of num_days tensors, of size L x A x R
        num_items = metapop_vals[0].shape[0]

        # This is ugly and inefficient -- but at least we only do this once, when we get the initial
        #   state of a metapopulation model in tensor form
        transposed_metapop_vals = [torch.tensor(np.array([metapop_vals[l][i] for l in range(L)])) for i in
                                   range(num_items)]

        setattr(self._full_metapop_schedule_tensors, values_column_name, transposed_metapop_vals)

update_full_metapop_params_tensors() -> None

Update _full_metapop_params_tensors attribute in-place. FluFullMetapopParamsTensors has fields corresponding to all parameters in the simulation. Converts subpopulation-specific parameters to tensors of size (L, A, R) for location-age-risk (except for a few exceptions that have different dimensions).

Source code in CLT_BaseModel/flu_core/flu_components.py

def update_full_metapop_params_tensors(self) -> None:
    """
    Update `_full_metapop_params_tensors` attribute in-place.
    `FluFullMetapopParamsTensors` has fields corresponding
    to all parameters in the simulation. Converts subpopulation-specific
    parameters to tensors of size (L, A, R) for location-age-risk
    (except for a few exceptions that have different dimensions).
    """

    if self._full_metapop_params_tensors is None:
        self._full_metapop_params_tensors = FluFullMetapopParamsTensors()
    self.update_params_tensors(target=self._full_metapop_params_tensors)

update_full_metapop_state_tensors() -> None

Update _full_metapop_state_tensors attribute in-place. FluFullMetapopStateTensors has fields corresponding to all state variables in the simulation. Converts subpopulation-specific state to tensors of size (L, A, R) for location-age-risk (except for a few exceptions that have different dimensions).

Source code in CLT_BaseModel/flu_core/flu_components.py

def update_full_metapop_state_tensors(self) -> None:
    """
    Update `_full_metapop_state_tensors` attribute in-place.
    `FluFullMetapopStateTensors` has fields corresponding
    to all state variables in the simulation.
    Converts subpopulation-specific state to
    tensors of size (L, A, R) for location-age-risk
    (except for a few exceptions that have different dimensions).
    """

    if self._full_metapop_state_tensors is None:
        self._full_metapop_state_tensors = FluFullMetapopStateTensors()
    self.update_state_tensors(self._full_metapop_state_tensors)

update_params_tensors(target: FluTravelParamsTensors) -> FluTravelParamsTensors

Update target in-place. Converts subpopulation-specific parameters to tensors of size (L, A, R) for location-age-risk, except for num_locations and travel_proportions, which have size 1 and (L, L) respectively.

Source code in CLT_BaseModel/flu_core/flu_components.py

def update_params_tensors(self,
                          target: FluTravelParamsTensors) -> FluTravelParamsTensors:
    """
    Update `target` in-place. Converts subpopulation-specific
    parameters to tensors of size (L, A, R) for location-age-risk,
    except for `num_locations` and `travel_proportions`, which
    have size 1 and (L, L) respectively.
    """

    # USE THE ORDERED DICTIONARY HERE FOR SAFETY!
    #   AGAIN, ORDER MATTERS BECAUSE ORDER DETERMINES
    #   THE SUBPOPULATION INDEX IN THE METAPOPULATION
    #   TENSOR!
    subpop_models_ordered = self._subpop_models_ordered

    # Subpop models should have the same A, R so grab
    #   from the first subpop model
    A = subpop_models_ordered[0].params.num_age_groups
    R = subpop_models_ordered[0].params.num_risk_groups

    for field in fields(target):

        name = field.name

        metapop_vals = []

        if name == "num_locations" or name == "travel_proportions":
            setattr(target, name, torch.tensor(getattr(self.mixing_params, name)))

        else:
            for model in subpop_models_ordered.values():
                metapop_vals.append(getattr(model.params, name))

            # If all values are equal to each other, then
            #   simply store the first value (since its value is common
            #   across metapopulations)
            first_val = metapop_vals[0]
            if all(np.allclose(x, first_val) for x in metapop_vals):
                metapop_vals = first_val

            # Converting list of arrays to tensors is slow --
            #   better to convert to array first
            if isinstance(metapop_vals, list):
                metapop_vals = np.asarray(metapop_vals)
                # metapop_vals = np.stack([clt.to_AR_array(x, A, R) for x in metapop_vals])

            setattr(target, name, torch.tensor(metapop_vals))

    # Convert all tensors to correct size!
    target.standardize_shapes()

update_state_tensors(target: FluTravelStateTensors) -> None

Update target instance in-place with current simulation values. Each field of target corresponds to a field in FluSubpopState, and contains either a tensor of size (L, A, R) or a tensor of size (L), where (l, a, r) refers to location-age-risk.

Source code in CLT_BaseModel/flu_core/flu_components.py

def update_state_tensors(self,
                         target: FluTravelStateTensors) -> None:
    """
    Update `target` instance in-place with current simulation
    values. Each field of `target` corresponds to a field in
    `FluSubpopState`, and contains either a tensor of size
    (L, A, R) or a tensor of size (L), where (l, a, r) refers to
    location-age-risk.
    """

    # ORDER MATTERS! USE ORDERED DICTIONARY HERE
    #   to preserve correct index order in tensors!
    #   See `update_travel_params_tensors` for detailed note.
    subpop_models_ordered = self._subpop_models_ordered

    for field in fields(target):

        name = field.name

        # FluTravelStateTensors has an attribute
        #   that is a dictionary called `init_vals` --
        #   disregard, as this only used to store
        #   initial values for resetting, but is not
        #   used in the travel model computation
        if name == "init_vals":
            continue

        metapop_vals = []

        for model in subpop_models_ordered.values():
            current_val = getattr(model.state, name)
            metapop_vals.append(current_val)

        # Probably want to update this to be cleaner...
        # `SubpopState` fields that correspond to `Schedule` instances
        # have initial values of `None` -- but we cannot build a tensor
        # with `None` values, so we convert values to 0s.
        if any(v is None for v in metapop_vals):
            setattr(target, name, torch.tensor(np.full(np.shape(metapop_vals), 0.0)))
        else:
            setattr(target, name, torch.tensor(np.asarray(metapop_vals)))

update_travel_params_tensors() -> None

Update travel_params_tensors attribute in-place. FluTravelParamsTensors only has fields corresponding to parameters relevant for the travel model. Converts subpopulation-specific parameters to tensors of size (L, A, R) for location-age-risk (except for a few exceptions that have different dimensions).

Source code in CLT_BaseModel/flu_core/flu_components.py

def update_travel_params_tensors(self) -> None:
    """
    Update `travel_params_tensors` attribute in-place.
    `FluTravelParamsTensors` only has fields corresponding
    to parameters relevant for the travel model.
    Converts subpopulation-specific parameters to
    tensors of size (L, A, R) for location-age-risk
    (except for a few exceptions that have different dimensions).
    """

    self.update_params_tensors(target=self.travel_params_tensors)

update_travel_state_tensors() -> None

Update travel_state_tensors attribute in-place. FluTravelStateTensors only has fields corresponding to state variables relevant for the travel model. Converts subpopulation-specific state to tensors of size (L, A, R) for location-age-risk (except for a few exceptions that have different dimensions).

Source code in CLT_BaseModel/flu_core/flu_components.py

def update_travel_state_tensors(self) -> None:
    """
    Update `travel_state_tensors` attribute in-place.
    `FluTravelStateTensors` only has fields corresponding
    to state variables relevant for the travel model.
    Converts subpopulation-specific state to
    tensors of size (L, A, R) for location-age-risk
    (except for a few exceptions that have different dimensions).
    """

    self.update_state_tensors(self.travel_state_tensors)

FluMetapopModelError

Bases: MetapopModelError

Custom exceptions for flu metapopulation simulation model errors.

Source code in CLT_BaseModel/flu_core/flu_components.py

class FluMetapopModelError(clt.MetapopModelError):
    """Custom exceptions for flu metapopulation simulation model errors."""
    pass

FluMixingParams dataclass

Contains parameters corresponding to inter-subpopulation (metapopulation model) specifications: the number of subpopulations included, and the travel proportions between them.

Parameters:

Name	Type	Description	Default
num_locations	int	Number of locations (subpopulations) in the metapopulation model.	required
travel_proportions	np.ndarray of shape (A, R	L x L array of floats in [0,1], where L is the number of locations (subpopulations), and the i-jth element is the proportion of people in subpopulation i that travel to subpopulation j.	required

Source code in CLT_BaseModel/flu_core/flu_data_structures.py

@dataclass(frozen=True)
class FluMixingParams:
    """
    Contains parameters corresponding to inter-subpopulation
    (metapopulation model) specifications: the number of
    subpopulations included, and the travel proportions between them.

    Params:
        num_locations (int):
            Number of locations (subpopulations) in the
            metapopulation model.
        travel_proportions (np.ndarray of shape (A, R)):
            L x L array of floats in [0,1], where L is the number
            of locations (subpopulations), and the i-jth element
            is the proportion of people in subpopulation i that
            travel to subpopulation j.
    """

    num_locations: Optional[int]
    travel_proportions: Optional[np.ndarray]

FluPrecomputedTensors

Stores precomputed quantities that are repeatedly used, for computational efficiency.

Source code in CLT_BaseModel/flu_core/flu_data_structures.py

class FluPrecomputedTensors:
    """
    Stores precomputed quantities that are repeatedly
    used, for computational efficiency.
    """

    def __init__(self,
                 total_pop_LAR_tensor: torch.Tensor,
                 params: FluTravelParamsTensors) -> None:

        self.total_pop_LAR_tensor = total_pop_LAR_tensor

        self.L = int(params.num_locations.item())
        self.A = int(params.num_age_groups.item())
        self.R = int(params.num_risk_groups.item())

        self.total_pop_LA = torch.sum(self.total_pop_LAR_tensor, dim=2)

        # Remove the diagonal!
        self.nonlocal_travel_prop = params.travel_proportions.clone().fill_diagonal_(0.0)

        # We don't need einsum for residents traveling
        #   -- Dave and Remy helped me check this
        # \sum_{k \not = \ell} v^{\ell \rightarrow k}
        # Note we already have k \not = \ell because we set the diagonal of
        #   nonlocal_travel_prop to 0
        self.sum_residents_nonlocal_travel_prop = self.nonlocal_travel_prop.sum(dim=1)

FluSubpopModel

Bases: SubpopModel

Class for creating ImmunoSEIRS flu model with predetermined fixed structure -- initial values and epidemiological structure are populated by user-specified JSON files.

Key method create_transmission_model returns a SubpopModel instance with S-E-I-H-R-D compartments and M and MV epi metrics.

The update structure is as follows

• S <- S + R_to_S - S_to_E
• E <- E + S_to_E - E_to_IP - E_to_IA
• IA <- IA + E_to_IA - IA_to_R
• IP <- IP + E_to_IP - IP_to_IS
• IS <- IS + IP_to_IS - IS_to_R - IS_to_H
• H <- H + IS_to_H - H_to_R - H_to_D
• R <- R + IS_to_R + H_to_R - R_to_S
• D <- D + H_to_D

The following are TransitionVariable instances

• R_to_S is a RecoveredToSusceptible instance
• S_to_E is a SusceptibleToExposed instance
• IP_to_IS is a PresympToSymp instance
• IS_to_H is a SympToHosp instance
• IS_to_R is a SympToRecovered instance
• H_to_R is a HospToRecovered instance
• H_to_D is a HospToDead instance

There are three TransitionVariableGroups

• E_out (handles E_to_IP and E_to_IA)
• IS_out (handles IS_to_H and IS_to_R)
• H_out (handles H_to_R and H_to_D)

The following are EpiMetric instances

• M is a InfInducedImmunity instance
• MV is a VaxInducedImmunity instance

Transition rates and update formulas are specified in corresponding classes.

See parent class SubpopModel's docstring for additional attributes.

Source code in CLT_BaseModel/flu_core/flu_components.py

class FluSubpopModel(clt.SubpopModel):
    """
    Class for creating ImmunoSEIRS flu model with predetermined fixed
    structure -- initial values and epidemiological structure are
    populated by user-specified `JSON` files.

    Key method create_transmission_model returns a `SubpopModel`
    instance with S-E-I-H-R-D compartments and M
    and MV epi metrics.

    The update structure is as follows:
        - S <- S + R_to_S - S_to_E
        - E <- E + S_to_E - E_to_IP - E_to_IA
        - IA <- IA + E_to_IA - IA_to_R 
        - IP <- IP + E_to_IP - IP_to_IS
        - IS <- IS + IP_to_IS - IS_to_R - IS_to_H
        - H <- H + IS_to_H - H_to_R - H_to_D
        - R <- R + IS_to_R + H_to_R - R_to_S
        - D <- D + H_to_D

    The following are TransitionVariable instances:
        - R_to_S is a RecoveredToSusceptible instance
        - S_to_E is a SusceptibleToExposed instance
        - IP_to_IS is a PresympToSymp instance
        - IS_to_H is a SympToHosp instance
        - IS_to_R is a SympToRecovered instance
        - H_to_R is a HospToRecovered instance
        - H_to_D is a HospToDead instance

    There are three TransitionVariableGroups:
        - E_out (handles E_to_IP and E_to_IA)
        - IS_out (handles IS_to_H and IS_to_R)
        - H_out (handles H_to_R and H_to_D)

    The following are EpiMetric instances:
        - M is a InfInducedImmunity instance
        - MV is a VaxInducedImmunity instance

    Transition rates and update formulas are specified in
    corresponding classes.

    See parent class `SubpopModel`'s docstring for additional attributes.
    """

    def __init__(self,
                 state: FluSubpopState,
                 params: FluSubpopParams,
                 simulation_settings: FluSubpopSchedules,
                 RNG: np.random.Generator,
                 schedules_spec: FluSubpopSchedules,
                 name: str):
        """
        Args:
            state (FluSubpopState):
                holds current simulation state information,
                such as current values of epidemiological compartments
                and epi metrics.
            params (FluSubpopParams):
                holds epidemiological parameter values.
            simulation_settings (SimulationSettings):
                holds simulation settings.
            RNG (np.random.Generator):
                numpy random generator object used to obtain
                random numbers.
            schedules_spec (FluSubpopSchedules):
                holds dataframes that specify `Schedule` instances.
            name (str):
                unique name of MetapopModel instance.
        """

        self.schedules_spec = schedules_spec

        # IMPORTANT NOTE: as always, we must be careful with mutable objects
        # and generally use deep copies to avoid modification of the same
        # object. But in this function call, using deep copies is unnecessary
        # (redundant) because the parent class `SubpopModel`'s `__init__`
        # creates deep copies.
        super().__init__(state, params, simulation_settings, RNG, name)

    def create_compartments(self) -> sc.objdict[str, clt.Compartment]:

        # Create `Compartment` instances S-E-IA-IP-IS-H-R-D (7 compartments total)
        # Save instances in `sc.objdict` and return objdict

        compartments = sc.objdict()

        for name in ("S", "E", "IP", "IS", "IA", "H", "R", "D"):
            compartments[name] = clt.Compartment(getattr(self.state, name))

        return compartments

    def create_dynamic_vals(self) -> sc.objdict[str, clt.DynamicVal]:
        """
        Create all `DynamicVal` instances, save in `sc.objdict`, and return objdict
        """

        dynamic_vals = sc.objdict()

        dynamic_vals["beta_reduce"] = BetaReduce(init_val=0.0,
                                                 is_enabled=False)

        return dynamic_vals

    def create_schedules(self) -> sc.objdict[str, clt.Schedule]:
        """
        Create all `Schedule` instances, save in `sc.objdict`, and return objdict
        """

        schedules = sc.objdict()

        schedules["absolute_humidity"] = AbsoluteHumidity()
        schedules["flu_contact_matrix"] = FluContactMatrix()
        schedules["daily_vaccines"] = DailyVaccines()

        for field, df in asdict(self.schedules_spec).items():

            try:
                df["date"] = pd.to_datetime(df["date"], format='%Y-%m-%d').dt.date
            except ValueError as e:
                raise ValueError("Error: dates should be strings in YYYY-MM-DD format or "
                                 "`date.datetime` objects.") from e

            schedules[field].timeseries_df = df

        return schedules

    def create_transition_variables(self) -> sc.objdict[str, clt.TransitionVariable]:
        """
        Create all `TransitionVariable` instances,
        save in `sc.objdict`, and return objdict
        """

        # NOTE: see the parent class `SubpopModel`'s `__init__` --
        # `create_transition_variables` is called after
        # `simulation_settings` is assigned

        transition_type = self.simulation_settings.transition_type

        transition_variables = sc.objdict()

        S = self.compartments.S
        E = self.compartments.E
        IP = self.compartments.IP
        IS = self.compartments.IS
        IA = self.compartments.IA
        H = self.compartments.H
        R = self.compartments.R
        D = self.compartments.D

        transition_variables.R_to_S = RecoveredToSusceptible(R, S, transition_type)
        transition_variables.S_to_E = SusceptibleToExposed(S, E, transition_type)
        transition_variables.IP_to_IS = PresympToSymp(IP, IS, transition_type)
        transition_variables.IA_to_R = AsympToRecovered(IA, R, transition_type)
        transition_variables.E_to_IP = ExposedToPresymp(E, IP, transition_type, True)
        transition_variables.E_to_IA = ExposedToAsymp(E, IA, transition_type, True)
        transition_variables.IS_to_R = SympToRecovered(IS, R, transition_type, True)
        transition_variables.IS_to_H = SympToHosp(IS, H, transition_type, True)
        transition_variables.H_to_R = HospToRecovered(H, R, transition_type, True)
        transition_variables.H_to_D = HospToDead(H, D, transition_type, True)

        return transition_variables

    def create_transition_variable_groups(self) -> sc.objdict[str, clt.TransitionVariableGroup]:
        """
        Create all transition variable groups described in docstring (2 transition
        variable groups total), save in `sc.objdict`, return objdict
        """

        # Shortcuts for attribute access
        # NOTE: see the parent class `SubpopModel`'s `__init__` --
        # `create_transition_variable_groups` is called after
        # `simulation_settings` is assigned

        transition_type = self.simulation_settings.transition_type

        transition_variable_groups = sc.objdict()

        transition_variable_groups.E_out = clt.TransitionVariableGroup(self.compartments.E,
                                                                       transition_type,
                                                                       (self.transition_variables.E_to_IP,
                                                                        self.transition_variables.E_to_IA))

        transition_variable_groups.IS_out = clt.TransitionVariableGroup(self.compartments.IS,
                                                                        transition_type,
                                                                        (self.transition_variables.IS_to_R,
                                                                         self.transition_variables.IS_to_H))

        transition_variable_groups.H_out = clt.TransitionVariableGroup(self.compartments.H,
                                                                       transition_type,
                                                                       (self.transition_variables.H_to_R,
                                                                        self.transition_variables.H_to_D))

        return transition_variable_groups

    def create_epi_metrics(self) -> sc.objdict[str, clt.EpiMetric]:
        """
        Create all epi metric described in docstring (2 state
        variables total), save in `sc.objdict`, and return objdict
        """

        epi_metrics = sc.objdict()

        epi_metrics.M = \
            InfInducedImmunity(getattr(self.state, "M"),
                               self.transition_variables.R_to_S)

        epi_metrics.MV = \
            VaxInducedImmunity(getattr(self.state, "MV"))

        return epi_metrics

    def modify_subpop_params(self,
                             updates_dict: dict):
        """
        This method lets users safely modify a single subpopulation
        parameters field; if this subpop model is associated with
        a metapop model, the metapopulation-wide tensors are updated
        automatically afterward. See also `modify_subpop_params` method on
        `FluMetapopModel`.

        Parameters:
            updates_dict (dict):
                Dictionary specifying values to update in a
                `FluSubpopParams` instance -- keys must match the
                field names of `FluSubpopParams`.
        """

        # If associated with metapop model, run this method
        #   on the metapop model itself to handle metapopulation-wide
        #   tensor updating
        if self.metapop_model:
            self.metapop_model.modify_subpop_params(self.name,
                                                    updates_dict)
        else:
            # Since `SubpopParams` is frozen, we return a new instance
            #   with the reflected updates
            self.params = clt.updated_dataclass(self.params, updates_dict)

__init__(state: FluSubpopState, params: FluSubpopParams, simulation_settings: FluSubpopSchedules, RNG: np.random.Generator, schedules_spec: FluSubpopSchedules, name: str)

Parameters:

Name	Type	Description	Default
state	FluSubpopState	holds current simulation state information, such as current values of epidemiological compartments and epi metrics.	required
params	FluSubpopParams	holds epidemiological parameter values.	required
simulation_settings	SimulationSettings	holds simulation settings.	required
RNG	Generator	numpy random generator object used to obtain random numbers.	required
schedules_spec	FluSubpopSchedules	holds dataframes that specify Schedule instances.	required
name	str	unique name of MetapopModel instance.	required

Source code in CLT_BaseModel/flu_core/flu_components.py

def __init__(self,
             state: FluSubpopState,
             params: FluSubpopParams,
             simulation_settings: FluSubpopSchedules,
             RNG: np.random.Generator,
             schedules_spec: FluSubpopSchedules,
             name: str):
    """
    Args:
        state (FluSubpopState):
            holds current simulation state information,
            such as current values of epidemiological compartments
            and epi metrics.
        params (FluSubpopParams):
            holds epidemiological parameter values.
        simulation_settings (SimulationSettings):
            holds simulation settings.
        RNG (np.random.Generator):
            numpy random generator object used to obtain
            random numbers.
        schedules_spec (FluSubpopSchedules):
            holds dataframes that specify `Schedule` instances.
        name (str):
            unique name of MetapopModel instance.
    """

    self.schedules_spec = schedules_spec

    # IMPORTANT NOTE: as always, we must be careful with mutable objects
    # and generally use deep copies to avoid modification of the same
    # object. But in this function call, using deep copies is unnecessary
    # (redundant) because the parent class `SubpopModel`'s `__init__`
    # creates deep copies.
    super().__init__(state, params, simulation_settings, RNG, name)

create_dynamic_vals() -> sc.objdict[str, clt.DynamicVal]

Create all DynamicVal instances, save in sc.objdict, and return objdict

Source code in CLT_BaseModel/flu_core/flu_components.py

def create_dynamic_vals(self) -> sc.objdict[str, clt.DynamicVal]:
    """
    Create all `DynamicVal` instances, save in `sc.objdict`, and return objdict
    """

    dynamic_vals = sc.objdict()

    dynamic_vals["beta_reduce"] = BetaReduce(init_val=0.0,
                                             is_enabled=False)

    return dynamic_vals

create_epi_metrics() -> sc.objdict[str, clt.EpiMetric]

Create all epi metric described in docstring (2 state variables total), save in sc.objdict, and return objdict

Source code in CLT_BaseModel/flu_core/flu_components.py

def create_epi_metrics(self) -> sc.objdict[str, clt.EpiMetric]:
    """
    Create all epi metric described in docstring (2 state
    variables total), save in `sc.objdict`, and return objdict
    """

    epi_metrics = sc.objdict()

    epi_metrics.M = \
        InfInducedImmunity(getattr(self.state, "M"),
                           self.transition_variables.R_to_S)

    epi_metrics.MV = \
        VaxInducedImmunity(getattr(self.state, "MV"))

    return epi_metrics

create_schedules() -> sc.objdict[str, clt.Schedule]

Create all Schedule instances, save in sc.objdict, and return objdict

Source code in CLT_BaseModel/flu_core/flu_components.py

def create_schedules(self) -> sc.objdict[str, clt.Schedule]:
    """
    Create all `Schedule` instances, save in `sc.objdict`, and return objdict
    """

    schedules = sc.objdict()

    schedules["absolute_humidity"] = AbsoluteHumidity()
    schedules["flu_contact_matrix"] = FluContactMatrix()
    schedules["daily_vaccines"] = DailyVaccines()

    for field, df in asdict(self.schedules_spec).items():

        try:
            df["date"] = pd.to_datetime(df["date"], format='%Y-%m-%d').dt.date
        except ValueError as e:
            raise ValueError("Error: dates should be strings in YYYY-MM-DD format or "
                             "`date.datetime` objects.") from e

        schedules[field].timeseries_df = df

    return schedules

create_transition_variable_groups() -> sc.objdict[str, clt.TransitionVariableGroup]

Create all transition variable groups described in docstring (2 transition variable groups total), save in sc.objdict, return objdict

Source code in CLT_BaseModel/flu_core/flu_components.py

def create_transition_variable_groups(self) -> sc.objdict[str, clt.TransitionVariableGroup]:
    """
    Create all transition variable groups described in docstring (2 transition
    variable groups total), save in `sc.objdict`, return objdict
    """

    # Shortcuts for attribute access
    # NOTE: see the parent class `SubpopModel`'s `__init__` --
    # `create_transition_variable_groups` is called after
    # `simulation_settings` is assigned

    transition_type = self.simulation_settings.transition_type

    transition_variable_groups = sc.objdict()

    transition_variable_groups.E_out = clt.TransitionVariableGroup(self.compartments.E,
                                                                   transition_type,
                                                                   (self.transition_variables.E_to_IP,
                                                                    self.transition_variables.E_to_IA))

    transition_variable_groups.IS_out = clt.TransitionVariableGroup(self.compartments.IS,
                                                                    transition_type,
                                                                    (self.transition_variables.IS_to_R,
                                                                     self.transition_variables.IS_to_H))

    transition_variable_groups.H_out = clt.TransitionVariableGroup(self.compartments.H,
                                                                   transition_type,
                                                                   (self.transition_variables.H_to_R,
                                                                    self.transition_variables.H_to_D))

    return transition_variable_groups

create_transition_variables() -> sc.objdict[str, clt.TransitionVariable]

Create all TransitionVariable instances, save in sc.objdict, and return objdict

Source code in CLT_BaseModel/flu_core/flu_components.py

def create_transition_variables(self) -> sc.objdict[str, clt.TransitionVariable]:
    """
    Create all `TransitionVariable` instances,
    save in `sc.objdict`, and return objdict
    """

    # NOTE: see the parent class `SubpopModel`'s `__init__` --
    # `create_transition_variables` is called after
    # `simulation_settings` is assigned

    transition_type = self.simulation_settings.transition_type

    transition_variables = sc.objdict()

    S = self.compartments.S
    E = self.compartments.E
    IP = self.compartments.IP
    IS = self.compartments.IS
    IA = self.compartments.IA
    H = self.compartments.H
    R = self.compartments.R
    D = self.compartments.D

    transition_variables.R_to_S = RecoveredToSusceptible(R, S, transition_type)
    transition_variables.S_to_E = SusceptibleToExposed(S, E, transition_type)
    transition_variables.IP_to_IS = PresympToSymp(IP, IS, transition_type)
    transition_variables.IA_to_R = AsympToRecovered(IA, R, transition_type)
    transition_variables.E_to_IP = ExposedToPresymp(E, IP, transition_type, True)
    transition_variables.E_to_IA = ExposedToAsymp(E, IA, transition_type, True)
    transition_variables.IS_to_R = SympToRecovered(IS, R, transition_type, True)
    transition_variables.IS_to_H = SympToHosp(IS, H, transition_type, True)
    transition_variables.H_to_R = HospToRecovered(H, R, transition_type, True)
    transition_variables.H_to_D = HospToDead(H, D, transition_type, True)

    return transition_variables

modify_subpop_params(updates_dict: dict)

This method lets users safely modify a single subpopulation parameters field; if this subpop model is associated with a metapop model, the metapopulation-wide tensors are updated automatically afterward. See also modify_subpop_params method on FluMetapopModel.

Parameters:

Name	Type	Description	Default
updates_dict	dict	Dictionary specifying values to update in a FluSubpopParams instance -- keys must match the field names of FluSubpopParams.	required

Source code in CLT_BaseModel/flu_core/flu_components.py

def modify_subpop_params(self,
                         updates_dict: dict):
    """
    This method lets users safely modify a single subpopulation
    parameters field; if this subpop model is associated with
    a metapop model, the metapopulation-wide tensors are updated
    automatically afterward. See also `modify_subpop_params` method on
    `FluMetapopModel`.

    Parameters:
        updates_dict (dict):
            Dictionary specifying values to update in a
            `FluSubpopParams` instance -- keys must match the
            field names of `FluSubpopParams`.
    """

    # If associated with metapop model, run this method
    #   on the metapop model itself to handle metapopulation-wide
    #   tensor updating
    if self.metapop_model:
        self.metapop_model.modify_subpop_params(self.name,
                                                updates_dict)
    else:
        # Since `SubpopParams` is frozen, we return a new instance
        #   with the reflected updates
        self.params = clt.updated_dataclass(self.params, updates_dict)

FluSubpopModelError

Bases: SubpopModelError

Custom exceptions for flu subpopulation simulation model errors.

Source code in CLT_BaseModel/flu_core/flu_components.py

class FluSubpopModelError(clt.SubpopModelError):
    """Custom exceptions for flu subpopulation simulation model errors."""
    pass

FluSubpopParams dataclass

Bases: SubpopParams

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

Each field of datatype np.ndarray must be A x R, where A is the number of age groups and R is the number of risk groups. Note: this means all arrays should be 2D. See FluSubpopState docstring for important formatting note on 2D arrays.

Note: the user does not have to specify total_pop_age_risk -- this is automatically computed when a FluSubpopModel is instantiated. This is to ensure that the total population (summed across all compartments) actually equals total_pop_age_risk -- and the user doesn't change one without updating the other.

Attributes:

Name	Type	Description
num_age_groups	positive int	number of age groups.
num_risk_groups	positive int	number of risk groups.
beta_baseline	positive float	transmission rate.
total_pop_age_risk	np.ndarray of positive ints	total number in population, summed across all age-risk groups.
humidity_impact	positive float	coefficient that determines how much absolute humidity affects beta_baseline.
inf_induced_saturation	np.ndarray of positive floats	constant(s) modeling saturation of antibody production of infected individuals.
inf_induced_immune_wane	positive float	rate at which infection-induced immunity against infection wanes.
vax_induced_saturation	np.ndarray of positive floats	constant(s) modeling saturation of antibody production of vaccinated individuals.
vax_induced_immune_wane	positive float	rate at which vaccine-induced immunity against infection wanes.
inf_induced_inf_risk_reduce	positive float	reduction in risk of getting infected after getting infected
inf_induced_hosp_risk_reduce	positive float	reduction in risk of hospitalization after getting infected
inf_induced_death_risk_reduce	positive float	reduction in risk of death after getting infected
vax_induced_inf_risk_reduce	positive float	reduction in risk of getting infected after getting vaccinated
vax_induced_hosp_risk_reduce	positive float	reduction in risk of hospitalization after getting vaccinated
vax_induced_death_risk_reduce	positive float	reduction in risk of death after getting vaccinated
R_to_S_rate	positive float	rate at which people in R move to S.
E_to_I_rate	positive float	rate at which people in E move to I (both IP and IA, infected pre-symptomatic and infected asymptomatic)
IP_to_IS_rate	positive float	rate a which people in IP (infected pre-symptomatic) move to IS (infected symptomatic)
IS_to_R_rate	positive float	rate at which people in IS (infected symptomatic) move to R.
IA_to_R_rate	positive float	rate at which people in IA (infected asymptomatic) move to R
IS_to_H_rate	positive float	rate at which people in IS (infected symptomatic) move to H.
H_to_R_rate	positive float	rate at which people in H move to R.
H_to_D_rate	positive float	rate at which people in H move to D.
E_to_IA_prop	np.ndarray of positive floats in [0,1]	proportion exposed who are asymptomatic based on age-risk groups.
IS_to_H_adjusted_prop	np.ndarray of positive floats in [0,1]	rate-adjusted proportion infected who are hospitalized based on age-risk groups.
H_to_D_adjusted_prop	np.ndarray of positive floats in [0,1]	rate-adjusted proportion hospitalized who die based on age-risk groups.
IP_relative_inf	positive float	relative infectiousness of pre-symptomatic to symptomatic people (IP to IS compartment).
IA_relative_inf	positive float	relative infectiousness of asymptomatic to symptomatic people (IA to IS compartment).
relative_suscept	np.ndarray of positive floats in [0,1]	relative susceptibility to infection by age group
mobility_modifier	np.ndarray of positive floats in [0,1]	total proportion of time spent away from home by age group
total_contact_matrix	np.ndarray of positive floats	A x A contact matrix (where A is the number of age groups), where element i,j is the average contacts from age group j that an individual in age group i has
school_contact_matrix	np.ndarray of positive floats	A x A contact matrix (where A is the number of age groups), where element i,j is the average contacts from age group j that an individual in age group i has at school -- this matrix plus the work_contact_matrix must be less than the total_contact_matrix, element-wise
work_contact_matrix	np.ndarray of positive floats	A x A contact matrix (where A is the number of age groups), where element i,j is the average contacts from age group j that an individual in age group i has at work -- this matrix plus the work_contact_matrix must be less than the total_contact_matrix, element-wise

Source code in CLT_BaseModel/flu_core/flu_data_structures.py

@dataclass(frozen=True)
class FluSubpopParams(clt.SubpopParams):
    """
    Data container for pre-specified and fixed epidemiological
    parameters in `FluSubpopModel`.

    Each field of datatype np.ndarray must be A x R,
    where A is the number of age groups and R is the number of
    risk groups. Note: this means all arrays should be 2D.
    See FluSubpopState docstring for important formatting note
    on 2D arrays.

    Note: the user does not have to specify `total_pop_age_risk` --
    this is automatically computed when a `FluSubpopModel` is
    instantiated. This is to ensure that the total population
    (summed across all compartments) actually equals `total_pop_age_risk` --
    and the user doesn't change one without updating the other.

    Attributes:
        num_age_groups (positive int):
            number of age groups.
        num_risk_groups (positive int):
            number of risk groups.
        beta_baseline (positive float): transmission rate.
        total_pop_age_risk (np.ndarray of positive ints):
            total number in population, summed across all
            age-risk groups.
        humidity_impact (positive float):
            coefficient that determines how much absolute
            humidity affects beta_baseline.
        inf_induced_saturation (np.ndarray of positive floats):
            constant(s) modeling saturation of antibody
            production of infected individuals.
        inf_induced_immune_wane (positive float):
            rate at which infection-induced immunity
            against infection wanes.
        vax_induced_saturation (np.ndarray of positive floats):
            constant(s) modeling saturation of antibody
            production of vaccinated individuals.
        vax_induced_immune_wane (positive float):
            rate at which vaccine-induced immunity
            against infection wanes.
        inf_induced_inf_risk_reduce (positive float):
            reduction in risk of getting infected
            after getting infected
        inf_induced_hosp_risk_reduce (positive float):
            reduction in risk of hospitalization
            after getting infected
        inf_induced_death_risk_reduce (positive float):
            reduction in risk of death
            after getting infected
        vax_induced_inf_risk_reduce (positive float):
            reduction in risk of getting infected
            after getting vaccinated
        vax_induced_hosp_risk_reduce (positive float):
            reduction in risk of hospitalization
            after getting vaccinated
        vax_induced_death_risk_reduce (positive float):
            reduction in risk of death
            after getting vaccinated
        R_to_S_rate (positive float):
            rate at which people in R move to S.
        E_to_I_rate (positive float):
            rate at which people in E move to I (both
            IP and IA, infected pre-symptomatic and infected
            asymptomatic)
        IP_to_IS_rate (positive float):
            rate a which people in IP (infected pre-symptomatic)
            move to IS (infected symptomatic)
        IS_to_R_rate (positive float):
            rate at which people in IS (infected symptomatic)
            move to R.
        IA_to_R_rate (positive float):
            rate at which people in IA (infected asymptomatic)
            move to R
        IS_to_H_rate (positive float):
            rate at which people in IS (infected symptomatic)
            move to H.
        H_to_R_rate (positive float):
            rate at which people in H move to R.
        H_to_D_rate (positive float):
            rate at which people in H move to D.
        E_to_IA_prop (np.ndarray of positive floats in [0,1]):
            proportion exposed who are asymptomatic based on
            age-risk groups.
        IS_to_H_adjusted_prop (np.ndarray of positive floats in [0,1]):
            rate-adjusted proportion infected who are hospitalized
            based on age-risk groups.
        H_to_D_adjusted_prop (np.ndarray of positive floats in [0,1]):
            rate-adjusted proportion hospitalized who die based on
            age-risk groups.
        IP_relative_inf (positive float):
            relative infectiousness of pre-symptomatic to symptomatic
            people (IP to IS compartment).
        IA_relative_inf (positive float):
            relative infectiousness of asymptomatic to symptomatic
            people (IA to IS compartment).
        relative_suscept (np.ndarray of positive floats in [0,1]):
            relative susceptibility to infection by age group
        mobility_modifier (np.ndarray of positive floats in [0,1]):
            total proportion of time spent away from home by age group
        total_contact_matrix (np.ndarray of positive floats):
            A x A contact matrix (where A is the number
            of age groups), where element i,j is the average
            contacts from age group j that an individual in
            age group i has
        school_contact_matrix (np.ndarray of positive floats):
            A x A contact matrix (where A is the number
            of age groups), where element i,j is the average
            contacts from age group j that an individual in
            age group i has at school -- this matrix plus the
            work_contact_matrix must be less than the
            total_contact_matrix, element-wise
        work_contact_matrix (np.ndarray of positive floats):
            A x A contact matrix (where A is the number
            of age groups), where element i,j is the average
            contacts from age group j that an individual in
            age group i has at work -- this matrix plus the
            work_contact_matrix must be less than the
            total_contact_matrix, element-wise
    """

    num_age_groups: Optional[int] = None
    num_risk_groups: Optional[int] = None
    beta_baseline: Optional[float] = None
    total_pop_age_risk: Optional[np.ndarray] = None
    humidity_impact: Optional[float] = None

    inf_induced_saturation: Optional[float] = None
    inf_induced_immune_wane: Optional[float] = None
    vax_induced_saturation: Optional[float] = None
    vax_induced_immune_wane: Optional[float] = None
    inf_induced_inf_risk_reduce: Optional[float] = None
    inf_induced_hosp_risk_reduce: Optional[float] = None
    inf_induced_death_risk_reduce: Optional[float] = None
    vax_induced_inf_risk_reduce: Optional[float] = None
    vax_induced_hosp_risk_reduce: Optional[float] = None
    vax_induced_death_risk_reduce: Optional[float] = None

    R_to_S_rate: Optional[float] = None
    E_to_I_rate: Optional[float] = None
    IP_to_IS_rate: Optional[float] = None
    IS_to_R_rate: Optional[float] = None
    IA_to_R_rate: Optional[float] = None
    IS_to_H_rate: Optional[float] = None
    H_to_R_rate: Optional[float] = None
    H_to_D_rate: Optional[float] = None
    E_to_IA_prop: Optional[np.ndarray] = None

    IS_to_H_adjusted_prop: Optional[np.ndarray] = None
    H_to_D_adjusted_prop: Optional[np.ndarray] = None

    IP_relative_inf: Optional[float] = None
    IA_relative_inf: Optional[float] = None

    relative_suscept: Optional[np.ndarray] = None

    mobility_modifier: Optional[np.ndarray] = None

    total_contact_matrix: Optional[np.ndarray] = None
    school_contact_matrix: Optional[np.ndarray] = None
    work_contact_matrix: Optional[np.ndarray] = None

FluSubpopSchedules dataclass

Data container for dataframes used to specify schedules for each FluSubpopModel instance.

THE FORMAT FOR EACH DATAFRAME IS VERY IMPORTANT -- please read and implement carefully.

Attributes:

Name	Type	Description
absolute_humidity	DataFrame	must have columns "date" and "absolute_humidity" -- "date" entries must correspond to consecutive calendar days and must either be strings with "YYYY-MM-DD" format or datetime.date objects -- "value" entries correspond to absolute humidity on those days
flu_contact_matrix	DataFrame	must have columns "date", "is_school_day", and "is_work_day" -- "date" entries must correspond to consecutive calendar days and must either be strings with "YYYY-MM-DD" format or datetime.date object and "is_school_day" and "is_work_day" entries are Booleans indicating if that date is a school day or work day
daily_vaccines	DataFrame	must have "date" and "daily_vaccines" -- "date" entries must correspond to consecutive calendar days and must either be strings with "YYYY-MM-DD" format or datetime.date objects -- "value" entries correspond to historical number vaccinated on those days

Source code in CLT_BaseModel/flu_core/flu_data_structures.py

@dataclass
class FluSubpopSchedules:
    """
    Data container for dataframes used to specify schedules
    for each `FluSubpopModel` instance.

    THE FORMAT FOR EACH DATAFRAME IS VERY IMPORTANT -- please
    read and implement carefully.

    Attributes:
        absolute_humidity (pd.DataFrame):
            must have columns "date" and "absolute_humidity" --
            "date" entries must correspond to consecutive calendar days
            and must either be strings with `"YYYY-MM-DD"` format or
            `datetime.date` objects -- "value" entries correspond to
            absolute humidity on those days
        flu_contact_matrix (pd.DataFrame):
            must have columns "date", "is_school_day", and "is_work_day"
            -- "date" entries must correspond to consecutive calendar
            days and must either be strings with `"YYYY-MM-DD"` format
            or `datetime.date` object and "is_school_day" and
            "is_work_day" entries are Booleans indicating if that date is
            a school day or work day
        daily_vaccines (pd.DataFrame):
            must have "date" and "daily_vaccines" -- "date" entries must
            correspond to consecutive calendar days and must either
            be strings with `"YYYY-MM-DD"` format or `datetime.date`
            objects -- "value" entries correspond to historical
            number vaccinated on those days
    """

    absolute_humidity: Optional[pd.DataFrame] = None
    flu_contact_matrix: Optional[pd.DataFrame] = None
    daily_vaccines: Optional[pd.DataFrame] = None

FluSubpopState dataclass

Bases: SubpopState

Data container for pre-specified and fixed set of Compartment initial values and EpiMetric initial values for FluSubpopModel.

Each field below should be A x R np.ndarray, where A is the number of age groups and R is the number of risk groups. Note: this means all arrays should be 2D. Even if there is 1 age group and 1 risk group (no group stratification), each array should be 1x1, which is two-dimensional. For example, np.array([[100]]) is correct -- np.array([100]) is wrong.

Attributes:

Name	Type	Description
S	np.ndarray of nonnegative integers	susceptible compartment for age-risk groups -- (holds current_val of Compartment "S").
E	np.ndarray of nonnegative integers	exposed compartment for age-risk groups -- (holds current_val of Compartment "E").
IP	np.ndarray of nonnegative integers	infected pre-symptomatic compartment for age-risk groups (holds current_val of Compartment "IP").
IS	np.ndarray of nonnegative integers	infected symptomatic compartment for age-risk groups (holds current_val of Compartment "IS").
IA	np.ndarray of nonnegative integers	infected asymptomatic compartment for age-risk groups (holds current_val of Compartment "IA").
H	np.ndarray of nonnegative integers	hospital compartment for age-risk groups (holds current_val of Compartment "H").
R	np.ndarray of nonnegative integers	recovered compartment for age-risk groups (holds current_val of Compartment "R").
D	np.ndarray of nonnegative integers	dead compartment for age-risk groups (holds current_val of Compartment "D").
M	np.ndarray of nonnegative floats	infection-induced population-level immunity for age-risk groups (holds current_val of EpiMetric "M").
MV	np.ndarray of nonnegative floats	vaccine-induced population-level immunity for age-risk groups (holds current_val of EpiMetric "MV").
absolute_humidity	positive float	grams of water vapor per cubic meter g/m^3, used as seasonality parameter that influences transmission rate beta_baseline.
flu_contact_matrix	np.ndarray of positive floats	A x A array, where A is the number of age groups -- element (a, a') corresponds to the number of contacts that a person in age group a has with people in age-risk group a'.
beta_reduce	float in [0, 1]	starting value of DynamicVal "beta_reduce" on starting day of simulation -- this DynamicVal emulates a simple staged-alert policy
daily_vaccines	np.ndarray of positive ints	holds current value of DailyVaccines instance, corresponding number of individuals who received influenza vaccine on that day, for given age-risk group (generally derived from historical data)

Source code in CLT_BaseModel/flu_core/flu_data_structures.py

@dataclass
class FluSubpopState(clt.SubpopState):
    """
    Data container for pre-specified and fixed set of
    Compartment initial values and EpiMetric initial values
    for `FluSubpopModel`.

    Each field below should be A x R np.ndarray, where
    A is the number of age groups and R is the number of risk groups.
    Note: this means all arrays should be 2D. Even if there is
    1 age group and 1 risk group (no group stratification),
    each array should be 1x1, which is two-dimensional.
    For example, np.array([[100]]) is correct --
    np.array([100]) is wrong.

    Attributes:
        S (np.ndarray of nonnegative integers):
            susceptible compartment for age-risk groups --
            (holds current_val of Compartment "S").
        E (np.ndarray of nonnegative integers):
            exposed compartment for age-risk groups --
            (holds current_val of Compartment "E").
        IP (np.ndarray of nonnegative integers):
            infected pre-symptomatic compartment for age-risk groups
            (holds current_val of Compartment "IP").
        IS (np.ndarray of nonnegative integers):
            infected symptomatic compartment for age-risk groups
            (holds current_val of Compartment "IS").
        IA (np.ndarray of nonnegative integers):
            infected asymptomatic compartment for age-risk groups
            (holds current_val of Compartment "IA").
        H (np.ndarray of nonnegative integers):
            hospital compartment for age-risk groups
            (holds current_val of Compartment "H").
        R (np.ndarray of nonnegative integers):
            recovered compartment for age-risk groups
            (holds current_val of Compartment "R").
        D (np.ndarray of nonnegative integers):
            dead compartment for age-risk groups
            (holds current_val of Compartment "D").
        M (np.ndarray of nonnegative floats):
            infection-induced population-level immunity
            for age-risk groups (holds current_val
            of EpiMetric "M").
        MV (np.ndarray of nonnegative floats):
            vaccine-induced population-level immunity
            for age-risk groups (holds current_val
            of EpiMetric "MV").
        absolute_humidity (positive float):
            grams of water vapor per cubic meter g/m^3,
            used as seasonality parameter that influences
            transmission rate beta_baseline.
        flu_contact_matrix (np.ndarray of positive floats):
            A x A array, where A is the number of age
            groups -- element (a, a') corresponds to the number
            of contacts that a person in age group a
            has with people in age-risk group a'.
        beta_reduce (float in [0,1]):
            starting value of DynamicVal "beta_reduce" on
            starting day of simulation -- this DynamicVal
            emulates a simple staged-alert policy
        daily_vaccines (np.ndarray of positive ints):
            holds current value of DailyVaccines instance,
            corresponding number of individuals who received influenza
            vaccine on that day, for given age-risk group
            (generally derived from historical data)
    """

    S: Optional[np.ndarray] = None
    E: Optional[np.ndarray] = None
    IP: Optional[np.ndarray] = None
    IS: Optional[np.ndarray] = None
    IA: Optional[np.ndarray] = None
    H: Optional[np.ndarray] = None
    R: Optional[np.ndarray] = None
    D: Optional[np.ndarray] = None

    M: Optional[np.ndarray] = None
    MV: Optional[np.ndarray] = None

    absolute_humidity: Optional[float] = None
    flu_contact_matrix: Optional[np.ndarray] = None
    beta_reduce: Optional[float] = 0.0

    daily_vaccines: Optional[np.ndarray] = None

FluTravelParamsTensors dataclass

Data container for tensors for FluMetapopModel -- used to store arrays that contain data across all subpopulations (collected from parameters on each location/subpopulation model, as well as from the metapopulation's associated FluMixingParams instance). Note that not all fields in FluSubpopParams are included -- we only include parameters needed for the travel model computation, for efficiency.

Attributes:

Name	Type	Description
num_locations	(Tensor, 0 - dimensional)	number of locations (subpopulations) in the metapopulation model and therefore the travel model.
travel_proportions	Tensor	L x L array, where L is the number of locations or subpopulations, where element i,j corresponds to the proportion of the population in location i who travels to location j (on average).

See FluSubpopParams docstring for other attributes.

Fields are analogous -- but (most) are size (L, A, R) for location-age-risk or size 0 tensors. Exceptions are travel_proportions, which is size (L, L), and any of the contact matrices, which are size (L, A, A).

Source code in CLT_BaseModel/flu_core/flu_data_structures.py

@dataclass
class FluTravelParamsTensors:
    """
    Data container for tensors for `FluMetapopModel` -- used to store arrays
    that contain data across all subpopulations (collected from parameters
    on each location/subpopulation model, as well as from the
    metapopulation's associated `FluMixingParams` instance).
    Note that not all fields in `FluSubpopParams` are included
    -- we only include parameters needed for the travel model
    computation, for efficiency.

    Attributes:
        num_locations (torch.Tensor, 0-dimensional):
            number of locations (subpopulations) in the
            metapopulation model and therefore the travel
            model.
        travel_proportions (torch.Tensor):
            L x L array, where L is the number of locations
            or subpopulations, where element i,j corresponds
            to the proportion of the population in location i
            who travels to location j (on average).

    See `FluSubpopParams` docstring for other attributes.

    Fields are analogous -- but (most) are size (L, A, R) for
    location-age-risk or size 0 tensors. Exceptions are
    `travel_proportions`, which is size (L, L),
    and any of the contact matrices, which are size (L, A, A).
    """

    num_locations: Optional[torch.tensor] = None
    num_age_groups: Optional[torch.tensor] = None
    num_risk_groups: Optional[torch.tensor] = None

    travel_proportions: torch.Tensor = None

    IP_relative_inf: torch.Tensor = None
    IA_relative_inf: torch.Tensor = None

    relative_suscept: torch.Tensor = None
    mobility_modifier: torch.Tensor = None

    total_contact_matrix: Optional[torch.Tensor] = None
    school_contact_matrix: Optional[torch.Tensor] = None
    work_contact_matrix: Optional[torch.Tensor] = None

    def standardize_shapes(self) -> None:
        """
        If field is size (L, A, R) for location-age-risk or size 0 tensors,
            or is not a special variable listed below, then apply dimension
            expansion so that fields are size (L, A, R) tensors for tensor multiplication.

        Exceptions are `travel_proportions`, which is size (L, L),
        and any of the contact matrices, which are size (L, A, A).

        Not all dimension combinations are considered not all make sense --
        we assume that we only have risk IF we have age, for example.
        """

        L = int(self.num_locations.item())
        A = int(self.num_age_groups.item())
        R = int(self.num_risk_groups.item())

        error_str = "Each SubpopParams field must have size (L, A, R) " \
                    "(for location-age-risk groups) or size 0 -- please check files " \
                    "and inputs, then try again."

        for name, value in vars(self).items():

            # Ignore the field that corresponds to a dictionary
            if name == "init_vals":
                continue

            elif name == "travel_proportions":
                if value.size() != torch.Size([L, L]):
                    raise Exception(str(name) + error_str)

            # `total_contact_matrix`, `school_contact_matrix`, `work_contact_matrix`
            elif "contact_matrix" in name:
                if value.size() == torch.Size([L, A, A]):
                    continue
                elif value.size() != torch.Size([A, A]):
                    raise Exception(str(name) + error_str)
                else:
                    setattr(self, name, value.view(1, A, A).expand(L, A, A))

            # If scalar or already L x A x R, do not need to adjust
            #   dimensions
            elif value.size() == torch.Size([]):
                continue

            elif value.size() == torch.Size([L, A, R]):
                continue

            elif value.size() == torch.Size([L]):
                setattr(self, name, value.view(L, 1, 1).expand(L, A, R))

            elif value.size() == torch.Size([A, R]):
                setattr(self, name, value.view(1, A, R).expand(L, A, R))

standardize_shapes() -> None

If field is size (L, A, R) for location-age-risk or size 0 tensors, or is not a special variable listed below, then apply dimension expansion so that fields are size (L, A, R) tensors for tensor multiplication.

Exceptions are travel_proportions, which is size (L, L), and any of the contact matrices, which are size (L, A, A).

Not all dimension combinations are considered not all make sense -- we assume that we only have risk IF we have age, for example.

Source code in CLT_BaseModel/flu_core/flu_data_structures.py

def standardize_shapes(self) -> None:
    """
    If field is size (L, A, R) for location-age-risk or size 0 tensors,
        or is not a special variable listed below, then apply dimension
        expansion so that fields are size (L, A, R) tensors for tensor multiplication.

    Exceptions are `travel_proportions`, which is size (L, L),
    and any of the contact matrices, which are size (L, A, A).

    Not all dimension combinations are considered not all make sense --
    we assume that we only have risk IF we have age, for example.
    """

    L = int(self.num_locations.item())
    A = int(self.num_age_groups.item())
    R = int(self.num_risk_groups.item())

    error_str = "Each SubpopParams field must have size (L, A, R) " \
                "(for location-age-risk groups) or size 0 -- please check files " \
                "and inputs, then try again."

    for name, value in vars(self).items():

        # Ignore the field that corresponds to a dictionary
        if name == "init_vals":
            continue

        elif name == "travel_proportions":
            if value.size() != torch.Size([L, L]):
                raise Exception(str(name) + error_str)

        # `total_contact_matrix`, `school_contact_matrix`, `work_contact_matrix`
        elif "contact_matrix" in name:
            if value.size() == torch.Size([L, A, A]):
                continue
            elif value.size() != torch.Size([A, A]):
                raise Exception(str(name) + error_str)
            else:
                setattr(self, name, value.view(1, A, A).expand(L, A, A))

        # If scalar or already L x A x R, do not need to adjust
        #   dimensions
        elif value.size() == torch.Size([]):
            continue

        elif value.size() == torch.Size([L, A, R]):
            continue

        elif value.size() == torch.Size([L]):
            setattr(self, name, value.view(L, 1, 1).expand(L, A, R))

        elif value.size() == torch.Size([A, R]):
            setattr(self, name, value.view(1, A, R).expand(L, A, R))

FluTravelStateTensors dataclass

Data container for tensors for FluMetapopModel -- used to store arrays that contain data across all subpopulations (collected from each location/subpopulation model). Note that not all fields in FluSubpopState are included -- we only include compartments needed for the travel model computation, for efficiency.

Attributes:

Name	Type	Description
IP	torch.Tensor of nonnegative integers	presymptomatic infected compartment for location-age-risk groups -- the lth element holds current_val of Compartment "IP" on the lth location / subpopulation on the associated MetapopModel.
IS	torch.Tensor of nonnegative integers	symptomatic infected compartment for location-age-risk groups -- the lth element holds current_val of Compartment "IS" on the lth location / subpopulation on the associated MetapopModel.
IA	torch.Tensor of nonnegative integers	asymptomatic infected compartment for location-age-risk groups -- the lth element holds current_val of Compartment "IA" on the lth location / subpopulation on the associated MetapopModel.
H	torch.Tensor of nonnegative integers	hospital compartment for location-age-risk groups -- the lth element holds current_val of Compartment "H" on the lth location / subpopulation on the associated MetapopModel.
flu_contact_matrix	torch.Tensor of nonnegative integers	contact matrix for location-age-risk groups -- the lth element holds current_val of FluContactMatrix Schedule for subpopulation l -- this value is a combination of the total contact matrix, the work contact matrix, and the school contact matrix (and the value is adjusted depending on whether the date is a work or school day)
init_vals	dict	dictionary of torch.Tensor instances, where keys correspond to "IP", "IS", "IA", and "H", and values correspond to their initial values for location-age-risk groups.

Source code in CLT_BaseModel/flu_core/flu_data_structures.py

@dataclass
class FluTravelStateTensors:
    """
    Data container for tensors for `FluMetapopModel` -- used to store arrays
    that contain data across all subpopulations (collected from each
    location/subpopulation model). Note that not all fields in
    `FluSubpopState` are included -- we only include compartments
    needed for the travel model computation, for efficiency.

    Attributes:
        IP (torch.Tensor of nonnegative integers):
            presymptomatic infected compartment for location-age-risk
            groups -- the lth element holds current_val of
            Compartment "IP" on the lth location / subpopulation
            on the associated `MetapopModel`.
        IS (torch.Tensor of nonnegative integers):
            symptomatic infected compartment for location-age-risk
            groups -- the lth element holds current_val of
            Compartment "IS" on the lth location / subpopulation
            on the associated `MetapopModel`.
        IA (torch.Tensor of nonnegative integers):
            asymptomatic infected compartment for location-age-risk
            groups -- the lth element holds current_val of
            Compartment "IA" on the lth location / subpopulation
            on the associated `MetapopModel`.
        H (torch.Tensor of nonnegative integers):
            hospital compartment for location-age-risk
            groups -- the lth element holds current_val of
            Compartment "H" on the lth location / subpopulation
            on the associated `MetapopModel`.
        flu_contact_matrix (torch.Tensor of nonnegative integers):
            contact matrix for location-age-risk groups -- the
            lth element holds current_val of `FluContactMatrix`
            `Schedule` for subpopulation l -- this value is a
            combination of the total contact matrix, the
            work contact matrix, and the school contact matrix
            (and the value is adjusted depending on whether
            the date is a work or school day)
        init_vals (dict):
            dictionary of torch.Tensor instances, where keys
            correspond to "IP", "IS", "IA", and "H", and values
            correspond to their initial values for location-age-risk
            groups.
    """

    IP: torch.Tensor = None
    IS: torch.Tensor = None
    IA: torch.Tensor = None
    H: torch.Tensor = None

    flu_contact_matrix: torch.Tensor = None

    init_vals: dict = field(default_factory=dict)

    # Note: `init_vals: dict = {}` does NOT work --
    #   gives "mutable default" argument

    def save_current_vals_as_init_vals(self):

        for field in fields(self):
            if field.name == "init_vals":
                continue
            self.init_vals[field.name] = getattr(self, field.name).clone()

    def reset_to_init_vals(self):

        for name, val in self.init_vals.items():
            setattr(self, name, val.clone())

HospToDead

Bases: TransitionVariable

TransitionVariable-derived class for movement from the "H" to "D" compartment. The functional form is the same across subpopulations.

Each HospToDead instance forms a TransitionVariableGroup with a corresponding HospToRecovered instance (these two transition variables are jointly distributed).

The rate of HospToDead decreases as population-level immunity against hospitalization increases.

Source code in CLT_BaseModel/flu_core/flu_components.py

class HospToDead(clt.TransitionVariable):
    """
    TransitionVariable-derived class for movement from the
    "H" to "D" compartment. The functional form is the same across
    subpopulations.

    Each HospToDead instance forms a TransitionVariableGroup with
    a corresponding HospToRecovered instance (these two
    transition variables are jointly distributed).

    The rate of HospToDead decreases as population-level immunity
    against hospitalization increases.
    """

    def get_current_rate(self,
                         state: FluSubpopState,
                         params: FluSubpopParams) -> np.ndarray:
        """
        Returns:
            np.ndarray of shape (A, R)
        """

        inf_induced_death_risk_reduce = params.inf_induced_death_risk_reduce
        vax_induced_death_risk_reduce = params.vax_induced_death_risk_reduce

        inf_induced_proportional_risk_reduce = \
            inf_induced_death_risk_reduce / (1 - inf_induced_death_risk_reduce)

        vax_induced_proportional_risk_reduce = \
            vax_induced_death_risk_reduce / (1 - vax_induced_death_risk_reduce)

        immunity_force = (1 + inf_induced_proportional_risk_reduce * state.M +
                          vax_induced_proportional_risk_reduce * state.MV)

        return np.asarray(params.H_to_D_adjusted_prop * params.H_to_D_rate / immunity_force)

get_current_rate(state: FluSubpopState, params: FluSubpopParams) -> np.ndarray

Returns:

Type	Description
ndarray	np.ndarray of shape (A, R)

Source code in CLT_BaseModel/flu_core/flu_components.py

def get_current_rate(self,
                     state: FluSubpopState,
                     params: FluSubpopParams) -> np.ndarray:
    """
    Returns:
        np.ndarray of shape (A, R)
    """

    inf_induced_death_risk_reduce = params.inf_induced_death_risk_reduce
    vax_induced_death_risk_reduce = params.vax_induced_death_risk_reduce

    inf_induced_proportional_risk_reduce = \
        inf_induced_death_risk_reduce / (1 - inf_induced_death_risk_reduce)

    vax_induced_proportional_risk_reduce = \
        vax_induced_death_risk_reduce / (1 - vax_induced_death_risk_reduce)

    immunity_force = (1 + inf_induced_proportional_risk_reduce * state.M +
                      vax_induced_proportional_risk_reduce * state.MV)

    return np.asarray(params.H_to_D_adjusted_prop * params.H_to_D_rate / immunity_force)

HospToRecovered

Bases: TransitionVariable

TransitionVariable-derived class for movement from the "H" to "R" compartment. The functional form is the same across subpopulations.

Each HospToRecovered instance forms a TransitionVariableGroup with a corresponding HospToDead instance (these two transition variables are jointly distributed).

Source code in CLT_BaseModel/flu_core/flu_components.py

class HospToRecovered(clt.TransitionVariable):
    """
    TransitionVariable-derived class for movement from the
    "H" to "R" compartment. The functional form is the same across
    subpopulations.

    Each HospToRecovered instance forms a TransitionVariableGroup with
    a corresponding HospToDead instance (these two
    transition variables are jointly distributed).
    """

    def get_current_rate(self,
                         state: FluSubpopState,
                         params: FluSubpopParams) -> np.ndarray:
        """
        Returns:
            np.ndarray of shape (A, R)
        """

        inf_induced_death_risk_reduce = params.inf_induced_death_risk_reduce
        vax_induced_death_risk_reduce = params.vax_induced_death_risk_reduce

        inf_induced_proportional_risk_reduce = \
            inf_induced_death_risk_reduce / (1 - inf_induced_death_risk_reduce)

        vax_induced_proportional_risk_reduce = \
            vax_induced_death_risk_reduce / (1 - vax_induced_death_risk_reduce)

        immunity_force = (1 + inf_induced_proportional_risk_reduce * state.M +
                          vax_induced_proportional_risk_reduce * state.MV)

        return np.full((params.num_age_groups, params.num_risk_groups),
                       (1 - params.H_to_D_adjusted_prop / immunity_force) * params.H_to_R_rate)

get_current_rate(state: FluSubpopState, params: FluSubpopParams) -> np.ndarray

Returns:

Type	Description
ndarray	np.ndarray of shape (A, R)

Source code in CLT_BaseModel/flu_core/flu_components.py

def get_current_rate(self,
                     state: FluSubpopState,
                     params: FluSubpopParams) -> np.ndarray:
    """
    Returns:
        np.ndarray of shape (A, R)
    """

    inf_induced_death_risk_reduce = params.inf_induced_death_risk_reduce
    vax_induced_death_risk_reduce = params.vax_induced_death_risk_reduce

    inf_induced_proportional_risk_reduce = \
        inf_induced_death_risk_reduce / (1 - inf_induced_death_risk_reduce)

    vax_induced_proportional_risk_reduce = \
        vax_induced_death_risk_reduce / (1 - vax_induced_death_risk_reduce)

    immunity_force = (1 + inf_induced_proportional_risk_reduce * state.M +
                      vax_induced_proportional_risk_reduce * state.MV)

    return np.full((params.num_age_groups, params.num_risk_groups),
                   (1 - params.H_to_D_adjusted_prop / immunity_force) * params.H_to_R_rate)

InfInducedImmunity

Bases: EpiMetric

EpiMetric-derived class for infection-induced population-level immunity.

Population-level immunity increases as people move from "R" to "S" -- this is a design choice intended to avoid "double-counting." People in "R" cannot be infected at all. People who move from "R" to "S" are susceptible again, but these recently-recovered people should have partial immunity. To handle this phenomenon, this epi metric increases as people move from "R" to "S."

Parameters:

Name	Type	Description	Default
R_to_S	RecoveredToSusceptible	RecoveredToSusceptible TransitionVariable in the SubpopModel -- it is an attribute because the population-level immunity increases as people move from "R" to "S".	required

See parent class docstring for other attributes.

Source code in CLT_BaseModel/flu_core/flu_components.py

class InfInducedImmunity(clt.EpiMetric):
    """
    EpiMetric-derived class for infection-induced
    population-level immunity.

    Population-level immunity increases as people move
    from "R" to "S" -- this is a design choice intended
    to avoid "double-counting." People in "R" cannot be
    infected at all. People who move from "R" to "S"
    are susceptible again, but these recently-recovered people
    should have partial immunity. To handle this phenomenon,
    this epi metric increases as people move from "R" to "S."

    Params:
        R_to_S (RecoveredToSusceptible):
            RecoveredToSusceptible TransitionVariable
            in the SubpopModel -- it is an attribute
            because the population-level immunity
            increases as people move from "R" to "S".

    See parent class docstring for other attributes.
    """

    def __init__(self, init_val, R_to_S):
        super().__init__(init_val)
        self.R_to_S = R_to_S

    def get_change_in_current_val(self,
                                  state: FluSubpopState,
                                  params: FluSubpopParams,
                                  num_timesteps: int) -> np.ndarray:
        """
        Returns:
            np.ndarray of shape (A, R)
        """

        # Note: the current values of transition variables already include
        #   discretization (division by the number of timesteps) -- therefore,
        #   we do not divide the first part of this equation by the number of
        #   timesteps -- see `TransitionVariable` class's methods for getting
        #   various realizations for more information

        return (self.R_to_S.current_val / params.total_pop_age_risk) * \
               (1 - params.inf_induced_saturation * state.M - params.vax_induced_saturation * state.MV) - \
               params.inf_induced_immune_wane * state.M / num_timesteps

get_change_in_current_val(state: FluSubpopState, params: FluSubpopParams, num_timesteps: int) -> np.ndarray

Returns:

Type	Description
ndarray	np.ndarray of shape (A, R)

Source code in CLT_BaseModel/flu_core/flu_components.py

def get_change_in_current_val(self,
                              state: FluSubpopState,
                              params: FluSubpopParams,
                              num_timesteps: int) -> np.ndarray:
    """
    Returns:
        np.ndarray of shape (A, R)
    """

    # Note: the current values of transition variables already include
    #   discretization (division by the number of timesteps) -- therefore,
    #   we do not divide the first part of this equation by the number of
    #   timesteps -- see `TransitionVariable` class's methods for getting
    #   various realizations for more information

    return (self.R_to_S.current_val / params.total_pop_age_risk) * \
           (1 - params.inf_induced_saturation * state.M - params.vax_induced_saturation * state.MV) - \
           params.inf_induced_immune_wane * state.M / num_timesteps

PresympToSymp

Bases: TransitionVariable

TransitionVariable-derived class for movement from the "IP" to "IS" compartment. The functional form is the same across subpopulations.

Source code in CLT_BaseModel/flu_core/flu_components.py

class PresympToSymp(clt.TransitionVariable):
    """
    TransitionVariable-derived class for movement from the
    "IP" to "IS" compartment. The functional form is the same across
    subpopulations.
    """

    def get_current_rate(self,
                         state: FluSubpopState,
                         params: FluSubpopParams) -> np.ndarray:
        """
        Returns:
            np.ndarray of shape (A, R)
        """
        return np.full((params.num_age_groups, params.num_risk_groups),
                       params.IP_to_IS_rate)

get_current_rate(state: FluSubpopState, params: FluSubpopParams) -> np.ndarray

Returns:

Type	Description
ndarray	np.ndarray of shape (A, R)

Source code in CLT_BaseModel/flu_core/flu_components.py

def get_current_rate(self,
                     state: FluSubpopState,
                     params: FluSubpopParams) -> np.ndarray:
    """
    Returns:
        np.ndarray of shape (A, R)
    """
    return np.full((params.num_age_groups, params.num_risk_groups),
                   params.IP_to_IS_rate)

RecoveredToSusceptible

Bases: TransitionVariable

TransitionVariable-derived class for movement from the "R" to "S" compartment. The functional form is the same across subpopulations.

Source code in CLT_BaseModel/flu_core/flu_components.py

class RecoveredToSusceptible(clt.TransitionVariable):
    """
    TransitionVariable-derived class for movement from the
    "R" to "S" compartment. The functional form is the same across
    subpopulations.
    """

    def get_current_rate(self,
                         state: FluSubpopState,
                         params: FluSubpopParams) -> np.ndarray:
        """
        Returns:
            np.ndarray of shape (A, R)
        """
        return np.full((params.num_age_groups, params.num_risk_groups),
                       params.R_to_S_rate)

get_current_rate(state: FluSubpopState, params: FluSubpopParams) -> np.ndarray

Returns:

Type	Description
ndarray	np.ndarray of shape (A, R)

Source code in CLT_BaseModel/flu_core/flu_components.py

def get_current_rate(self,
                     state: FluSubpopState,
                     params: FluSubpopParams) -> np.ndarray:
    """
    Returns:
        np.ndarray of shape (A, R)
    """
    return np.full((params.num_age_groups, params.num_risk_groups),
                   params.R_to_S_rate)

SusceptibleToExposed

Bases: TransitionVariable

TransitionVariable-derived class for movement from the "S" to "E" compartment. The functional form is the same across subpopulations.

The rate depends on the corresponding subpopulation's contact matrix, transmission rate beta, number infected (symptomatic, asymptomatic, and pre-symptomatic), and population-level immunity against infection, among other parameters.

This is the most complicated transition variable in the flu model. If using metapopulation model (travel model), then the rate depends on the total_mixing_exposure attribute, which is a function of other subpopulations' states and parameters, and travel between subpopulations.

If there is no metapopulation model, the rate is much simpler.

Attributes:

Name	Type	Description
total_mixing_exposure	np.ndarray of positive floats	weighted infectious count (exposure) from movement within home location, travel to other locations, and visitors from other locations

See parent class docstring for other attributes.

Source code in CLT_BaseModel/flu_core/flu_components.py

class SusceptibleToExposed(clt.TransitionVariable):
    """
    TransitionVariable-derived class for movement from the
    "S" to "E" compartment. The functional form is the same across
    subpopulations.

    The rate depends on the corresponding subpopulation's
    contact matrix, transmission rate beta, number
    infected (symptomatic, asymptomatic, and pre-symptomatic),
    and population-level immunity against infection,
    among other parameters.

    This is the most complicated transition variable in the
    flu model. If using metapopulation model (travel model), then
    the rate depends on the `total_mixing_exposure` attribute,
    which is a function of other subpopulations' states and
    parameters, and travel between subpopulations.

    If there is no metapopulation model, the rate
    is much simpler.

    Attributes:
        total_mixing_exposure (np.ndarray of positive floats):
            weighted infectious count (exposure) from movement
            within home location, travel to other locations,
            and visitors from other locations

    See parent class docstring for other attributes.
    """

    def __init__(self,
                 origin: clt.Compartment,
                 destination: clt.Compartment,
                 transition_type: clt.TransitionTypes,
                 is_jointly_distributed: str = False):

        super().__init__(origin,
                         destination,
                         transition_type,
                         is_jointly_distributed)

        self.total_mixing_exposure = None

    def get_current_rate(self,
                         state: FluSubpopState,
                         params: FluSubpopParams) -> np.ndarray:
        """
        Returns:
            np.ndarray of shape (A, R)
        """

        # If `total_mixing_exposure` has not been updated,
        #   then there is no travel model -- so, simulate
        #   this subpopulation entirely independently and
        #   use the simplified transition rate that does not
        #   depend on travel dynamics

        beta_adjusted = compute_beta_adjusted(state, params)

        inf_induced_inf_risk_reduce = params.inf_induced_inf_risk_reduce
        inf_induced_proportional_risk_reduce = inf_induced_inf_risk_reduce / (1 - inf_induced_inf_risk_reduce)

        vax_induced_inf_risk_reduce = params.vax_induced_inf_risk_reduce
        vax_induced_proportional_risk_reduce = vax_induced_inf_risk_reduce / (1 - vax_induced_inf_risk_reduce)

        immune_force = (1 + inf_induced_proportional_risk_reduce * state.M +
                        vax_induced_proportional_risk_reduce * state.MV)

        if self.total_mixing_exposure is not None:

            # Note here `self.total_mixing_exposure` includes
            #   `suscept_by_age` -- see `compute_total_mixing_exposure_prop`
            #   in `flu_travel_functions`

            # Need to convert tensor into array because combining np.ndarrays and
            #   tensors doesn't work, and everything else is an array
            return np.asarray(beta_adjusted * self.total_mixing_exposure / immune_force)

        else:
            wtd_presymp_asymp_by_age = compute_wtd_presymp_asymp_by_age(state, params)

            # Super confusing syntax... but this is the pain of having A x R,
            #   but having the contact matrix (contact patterns) be for
            #   ONLY age groups
            wtd_infectious_prop = np.divide(np.sum(state.IS, axis=1, keepdims=True) + wtd_presymp_asymp_by_age,
                                            compute_pop_by_age(params))

            raw_total_exposure = np.matmul(state.flu_contact_matrix, wtd_infectious_prop)

            # The total rate is only age-dependent -- it's the same rate across age groups
            return params.relative_suscept * (beta_adjusted * raw_total_exposure / immune_force)

get_current_rate(state: FluSubpopState, params: FluSubpopParams) -> np.ndarray

Returns:

Type	Description
ndarray	np.ndarray of shape (A, R)

Source code in CLT_BaseModel/flu_core/flu_components.py

def get_current_rate(self,
                     state: FluSubpopState,
                     params: FluSubpopParams) -> np.ndarray:
    """
    Returns:
        np.ndarray of shape (A, R)
    """

    # If `total_mixing_exposure` has not been updated,
    #   then there is no travel model -- so, simulate
    #   this subpopulation entirely independently and
    #   use the simplified transition rate that does not
    #   depend on travel dynamics

    beta_adjusted = compute_beta_adjusted(state, params)

    inf_induced_inf_risk_reduce = params.inf_induced_inf_risk_reduce
    inf_induced_proportional_risk_reduce = inf_induced_inf_risk_reduce / (1 - inf_induced_inf_risk_reduce)

    vax_induced_inf_risk_reduce = params.vax_induced_inf_risk_reduce
    vax_induced_proportional_risk_reduce = vax_induced_inf_risk_reduce / (1 - vax_induced_inf_risk_reduce)

    immune_force = (1 + inf_induced_proportional_risk_reduce * state.M +
                    vax_induced_proportional_risk_reduce * state.MV)

    if self.total_mixing_exposure is not None:

        # Note here `self.total_mixing_exposure` includes
        #   `suscept_by_age` -- see `compute_total_mixing_exposure_prop`
        #   in `flu_travel_functions`

        # Need to convert tensor into array because combining np.ndarrays and
        #   tensors doesn't work, and everything else is an array
        return np.asarray(beta_adjusted * self.total_mixing_exposure / immune_force)

    else:
        wtd_presymp_asymp_by_age = compute_wtd_presymp_asymp_by_age(state, params)

        # Super confusing syntax... but this is the pain of having A x R,
        #   but having the contact matrix (contact patterns) be for
        #   ONLY age groups
        wtd_infectious_prop = np.divide(np.sum(state.IS, axis=1, keepdims=True) + wtd_presymp_asymp_by_age,
                                        compute_pop_by_age(params))

        raw_total_exposure = np.matmul(state.flu_contact_matrix, wtd_infectious_prop)

        # The total rate is only age-dependent -- it's the same rate across age groups
        return params.relative_suscept * (beta_adjusted * raw_total_exposure / immune_force)

SympToHosp

Bases: TransitionVariable

TransitionVariable-derived class for movement from the "IS" to "H" compartment. The functional form is the same across subpopulations.

Each SympToHosp instance forms a TransitionVariableGroup with a corresponding SympToRecovered instance (these two transition variables are jointly distributed).

The rate of SympToHosp decreases as population-level immunity against hospitalization increases.

Source code in CLT_BaseModel/flu_core/flu_components.py

class SympToHosp(clt.TransitionVariable):
    """
    TransitionVariable-derived class for movement from the
    "IS" to "H" compartment. The functional form is the same across
    subpopulations.

    Each SympToHosp instance forms a TransitionVariableGroup with
    a corresponding SympToRecovered instance (these two
    transition variables are jointly distributed).

    The rate of SympToHosp decreases as population-level immunity
    against hospitalization increases.
    """

    def get_current_rate(self,
                         state: FluSubpopState,
                         params: FluSubpopParams) -> np.ndarray:
        """
        Returns:
            np.ndarray of shape (A, R)
        """

        inf_induced_hosp_risk_reduce = params.inf_induced_hosp_risk_reduce
        inf_induced_proportional_risk_reduce = inf_induced_hosp_risk_reduce / (1 - inf_induced_hosp_risk_reduce)

        vax_induced_hosp_risk_reduce = params.vax_induced_hosp_risk_reduce
        vax_induced_proportional_risk_reduce = vax_induced_hosp_risk_reduce / (1 - vax_induced_hosp_risk_reduce)

        immunity_force = (1 + inf_induced_proportional_risk_reduce * state.M +
                          vax_induced_proportional_risk_reduce * state.MV)

        return np.asarray(params.IS_to_H_rate * params.IS_to_H_adjusted_prop / immunity_force)

get_current_rate(state: FluSubpopState, params: FluSubpopParams) -> np.ndarray

Returns:

Type	Description
ndarray	np.ndarray of shape (A, R)

Source code in CLT_BaseModel/flu_core/flu_components.py

def get_current_rate(self,
                     state: FluSubpopState,
                     params: FluSubpopParams) -> np.ndarray:
    """
    Returns:
        np.ndarray of shape (A, R)
    """

    inf_induced_hosp_risk_reduce = params.inf_induced_hosp_risk_reduce
    inf_induced_proportional_risk_reduce = inf_induced_hosp_risk_reduce / (1 - inf_induced_hosp_risk_reduce)

    vax_induced_hosp_risk_reduce = params.vax_induced_hosp_risk_reduce
    vax_induced_proportional_risk_reduce = vax_induced_hosp_risk_reduce / (1 - vax_induced_hosp_risk_reduce)

    immunity_force = (1 + inf_induced_proportional_risk_reduce * state.M +
                      vax_induced_proportional_risk_reduce * state.MV)

    return np.asarray(params.IS_to_H_rate * params.IS_to_H_adjusted_prop / immunity_force)

SympToRecovered

Bases: TransitionVariable

TransitionVariable-derived class for movement from the "IS" to "R" compartment. The functional form is the same across subpopulations.

Each SympToRecovered instance forms a TransitionVariableGroup with a corresponding SympToHosp instance (these two transition variables are jointly distributed).

Source code in CLT_BaseModel/flu_core/flu_components.py

class SympToRecovered(clt.TransitionVariable):
    """
    TransitionVariable-derived class for movement from the
    "IS" to "R" compartment. The functional form is the same across
    subpopulations.

    Each SympToRecovered instance forms a TransitionVariableGroup with
    a corresponding SympToHosp instance (these two
    transition variables are jointly distributed).
    """

    def get_current_rate(self,
                         state: FluSubpopState,
                         params: FluSubpopParams) -> np.ndarray:
        """
        Returns:
            np.ndarray of shape (A, R)
        """
        inf_induced_hosp_risk_reduce = params.inf_induced_hosp_risk_reduce
        inf_induced_proportional_risk_reduce = inf_induced_hosp_risk_reduce / (1 - inf_induced_hosp_risk_reduce)

        vax_induced_hosp_risk_reduce = params.vax_induced_hosp_risk_reduce
        vax_induced_proportional_risk_reduce = vax_induced_hosp_risk_reduce / (1 - vax_induced_hosp_risk_reduce)

        immunity_force = (1 + inf_induced_proportional_risk_reduce * state.M +
                          vax_induced_proportional_risk_reduce * state.MV)

        return np.asarray((1 - params.IS_to_H_adjusted_prop / immunity_force) * params.IS_to_R_rate)

get_current_rate(state: FluSubpopState, params: FluSubpopParams) -> np.ndarray

Returns:

Type	Description
ndarray	np.ndarray of shape (A, R)

Source code in CLT_BaseModel/flu_core/flu_components.py

def get_current_rate(self,
                     state: FluSubpopState,
                     params: FluSubpopParams) -> np.ndarray:
    """
    Returns:
        np.ndarray of shape (A, R)
    """
    inf_induced_hosp_risk_reduce = params.inf_induced_hosp_risk_reduce
    inf_induced_proportional_risk_reduce = inf_induced_hosp_risk_reduce / (1 - inf_induced_hosp_risk_reduce)

    vax_induced_hosp_risk_reduce = params.vax_induced_hosp_risk_reduce
    vax_induced_proportional_risk_reduce = vax_induced_hosp_risk_reduce / (1 - vax_induced_hosp_risk_reduce)

    immunity_force = (1 + inf_induced_proportional_risk_reduce * state.M +
                      vax_induced_proportional_risk_reduce * state.MV)

    return np.asarray((1 - params.IS_to_H_adjusted_prop / immunity_force) * params.IS_to_R_rate)

VaxInducedImmunity

Bases: EpiMetric

EpiMetric-derived class for vaccine-induced population-level immunity.

Source code in CLT_BaseModel/flu_core/flu_components.py

class VaxInducedImmunity(clt.EpiMetric):
    """
    EpiMetric-derived class for vaccine-induced
    population-level immunity.
    """

    def __init__(self, init_val):
        super().__init__(init_val)

    def get_change_in_current_val(self,
                                  state: FluSubpopState,
                                  params: FluSubpopParams,
                                  num_timesteps: int) -> np.ndarray:
        """
        Returns:
            np.ndarray of shape (A, R)
        """

        # Note: `state.daily_vaccines` (based on the value of the `DailyVaccines`
        #   `Schedule` is NOT divided by the number of timesteps -- so we need to
        #   do this division in the equation here.

        return state.daily_vaccines / (params.total_pop_age_risk * num_timesteps) - \
               params.vax_induced_immune_wane * state.MV / num_timesteps

get_change_in_current_val(state: FluSubpopState, params: FluSubpopParams, num_timesteps: int) -> np.ndarray

Returns:

Type	Description
ndarray	np.ndarray of shape (A, R)

Source code in CLT_BaseModel/flu_core/flu_components.py

def get_change_in_current_val(self,
                              state: FluSubpopState,
                              params: FluSubpopParams,
                              num_timesteps: int) -> np.ndarray:
    """
    Returns:
        np.ndarray of shape (A, R)
    """

    # Note: `state.daily_vaccines` (based on the value of the `DailyVaccines`
    #   `Schedule` is NOT divided by the number of timesteps -- so we need to
    #   do this division in the equation here.

    return state.daily_vaccines / (params.total_pop_age_risk * num_timesteps) - \
           params.vax_induced_immune_wane * state.MV / num_timesteps

accept_reject_admits(metapop_model: FluMetapopModel, sampling_RNG: np.random.Generator, sampling_info: dict[str, dict[str, clt.UniformSamplingSpec]], total_daily_target_admits: list[np.ndarray], num_days: int = 50, target_accepted_reps: int = int(100.0), max_reps: int = int(1000.0), early_stop_percent: float = 0.5, target_rsquared: float = 0.75)

Accept-reject sampler for a metapopulation model.

This function repeatedly samples parameters from uniform distributions (as specified in spec) and simulates the model until the R-squared between simulated total admits and reference data exceeds target_rsquared. Accepted parameter sets and simulation states are saved as JSON files.

Parameters:

Name	Type	Description	Default
metapop_model	FluMetapopModel	The metapopulation model to simulate and sample parameters for.	required
sampling_RNG	Generator	Random number generator used for uniform sampling.	required
sampling_info	dict[str, dict[str, UniformSamplingSpec]]	See clt_toolkit / sampling / sample_uniform_metapop_params / sampling_info parameter for description.	required
total_daily_target_admits	list[ndarray]	"Target" time series of total admits (across subpopulations) for computing R-squared -- we would like parameters and sample paths that give simulated admits close to total_daily_target_admits. Must have length equal to num_days.	required
num_days	int, default=50	Total number of days to simulate for accepted parameter sets.	50
target_accepted_reps	int, default=100	Target number of accepted parameter sets (replicates) to collect.	int(100.0)
max_reps	int, default=1000	Maximum number of sampling attempts before stopping.	int(1000.0)
early_stop_percent	float, default=0.5	Fraction of num_days to simulate initially for early R-squared check.	0.5
target_rsquared	float, default=0.75	Minimum R-squared required between simulated and reference admits for acceptance.	0.75

Notes: - Early stopping is performed at num_days * early_stop_percent to quickly reject poor parameter samples. - Accepted samples (and the state of the simulation at day num_days) are saved to JSON files per subpopulation. Note that for efficiency, NOT ALL PARAMETERS ARE SAVED! Only the parameters that are randomly sampled (and thus are different between replications). - Running this function can be slow -- test this function with a small number of replications or simulation days to start.

Source code in CLT_BaseModel/flu_core/flu_accept_reject.py

def accept_reject_admits(metapop_model: FluMetapopModel,
                         sampling_RNG: np.random.Generator,
                         sampling_info: dict[str, dict[str, clt.UniformSamplingSpec]],
                         total_daily_target_admits: list[np.ndarray],
                         num_days: int = 50,
                         target_accepted_reps: int = int(1e2),
                         max_reps: int = int(1e3),
                         early_stop_percent: float = 0.5,
                         target_rsquared: float = 0.75):
    """
    Accept-reject sampler for a metapopulation model.

    This function repeatedly samples parameters from uniform distributions
    (as specified in `spec`) and simulates the model until the R-squared between
    simulated total admits and reference data exceeds `target_rsquared`.
    Accepted parameter sets and simulation states are saved as JSON files.

    Parameters:
        metapop_model (flu.FluMetapopModel):
            The metapopulation model to simulate and sample parameters for.
        sampling_RNG (np.random.Generator):
            Random number generator used for uniform sampling.
        sampling_info (dict[str, dict[str, clt.UniformSamplingSpec]]):
            See `clt_toolkit / sampling / sample_uniform_metapop_params / sampling_info`
            parameter for description.
        total_daily_target_admits (list[np.ndarray]):
            "Target" time series of total admits (across subpopulations)
            for computing R-squared -- we would like parameters and
            sample paths that give simulated admits close to
            `total_daily_target_admits`. Must have length equal to `num_days`.
        num_days (int, default=50):
            Total number of days to simulate for accepted parameter sets.
        target_accepted_reps (int, default=100):
            Target number of accepted parameter sets (replicates) to collect.
        max_reps (int, default=1000):
            Maximum number of sampling attempts before stopping.
        early_stop_percent (float, default=0.5):
            Fraction of `num_days` to simulate initially for early R-squared check.
        target_rsquared (float, default=0.75):
            Minimum R-squared required between simulated and reference admits for acceptance.

    Notes:
    - Early stopping is performed at `num_days * early_stop_percent` to
        quickly reject poor parameter samples.
    - Accepted samples (and the state of the simulation at day
        `num_days`) are saved to JSON files per subpopulation.
        Note that for efficiency, NOT ALL PARAMETERS ARE SAVED!
        Only the parameters that are randomly sampled (and thus are
        different between replications).
    - Running this function can be slow -- test this function with a small
        number of replications or simulation days to start.
    """

    if target_accepted_reps > max_reps:
        max_reps = 10 * target_accepted_reps

    num_days_early_stop = int(num_days * early_stop_percent)

    reps_counter = 0
    accepted_reps_counter = 0

    while reps_counter < max_reps and accepted_reps_counter < target_accepted_reps:

        reps_counter += 1

        metapop_model.reset_simulation()

        param_samples = clt.sample_uniform_metapop_params(metapop_model,
                                                          sampling_RNG,
                                                          sampling_info)

        # Save IS to H transition variable history
        # But do not save daily (compartment) history for efficiency
        for subpop_name, updates_dict in param_samples.items():
            metapop_model.modify_subpop_params(subpop_name, updates_dict)
            metapop_model.modify_simulation_settings({"transition_variables_to_save": ["IS_to_H"],
                                                      "save_daily_history": False})

        metapop_model.simulate_until_day(num_days_early_stop)
        total_simulated_admits = clt.aggregate_daily_tvar_history(metapop_model, "IS_to_H")
        current_rsquared = compute_rsquared(reference_timeseries=total_daily_target_admits[:num_days_early_stop],
                                            simulated_timeseries=total_simulated_admits)
        if current_rsquared < target_rsquared:
            continue

        else:
            metapop_model.simulate_until_day(num_days)
            total_simulated_admits = clt.aggregate_daily_tvar_history(metapop_model, "IS_to_H")
            current_rsquared = compute_rsquared(reference_timeseries=total_daily_target_admits,
                                                simulated_timeseries=total_simulated_admits)
            if current_rsquared < target_rsquared:
                continue
            else:
                accepted_reps_counter += 1

                for subpop_name, subpop in metapop_model.subpop_models.items():
                    with open("subpop_" + str(subpop_name) + "_rep_" + str(accepted_reps_counter) +
                              "_accepted_sample_params.json", "w") as f:
                        json.dump(clt.serialize_dataclass(param_samples[subpop_name]), f, indent=4)
                    with open("subpop_" + str(subpop_name) + "_rep_" + str(accepted_reps_counter) +
                              "_accepted_state.json", "w") as f:
                        json.dump(clt.serialize_dataclass(subpop.state), f, indent=4)

advance_timestep(state: FluFullMetapopStateTensors, params: FluFullMetapopParamsTensors, precomputed: FluPrecomputedTensors, dt: float, save_calibration_targets: bool = False, save_tvar_history: bool = False) -> Tuple[FluFullMetapopStateTensors, dict, dict]

Advance the simulation one timestep, with length dt. Updates state corresponding to compartments and epidemiological metrics after computing transition variables and metric changes.

Note that in this torch "mean" deterministic implementation... - We compute rates in the same way as the get_binom_deterministic_no_round transition type in the OOP code -- see TransitionVariables class in clt_toolkit / base_components for more details. - We also implement a "mean" deterministic analog of the multinomial distribution to handle multiple outflows from the same compartment - We do not round the transition variables - We also use softplus, a smooth approximation to the ReLU function, to ensure that compartments are nonnegative (which is not guaranteed using the mean of a binomial/multinomial random variable rather than sampling from those distributions).

Returns:

Type	Description
Tuple[FluFullMetapopStateTensors, dict, dict]	New FluFullMetapopStateTensors with updated state, dict of calibration targets corresponding to state values or transition variable values used for calibration, and dict of transition variable values to save this history. If save_calibration_targets is False, then the corresponding dict is empty, and similarly with save_tvar_history.

Source code in CLT_BaseModel/flu_core/flu_torch_det_components.py

def advance_timestep(state: FluFullMetapopStateTensors,
                     params: FluFullMetapopParamsTensors,
                     precomputed: FluPrecomputedTensors,
                     dt: float,
                     save_calibration_targets: bool=False,
                     save_tvar_history: bool=False) -> Tuple[FluFullMetapopStateTensors, dict, dict]:
    """
    Advance the simulation one timestep, with length `dt`.
    Updates state corresponding to compartments and
    epidemiological metrics after computing transition variables
    and metric changes.

    Note that in this torch "mean" deterministic implementation...
    - We compute rates in the same way as the
        `get_binom_deterministic_no_round`
        transition type in the OOP code -- see
        `TransitionVariables` class in
        `clt_toolkit / base_components` for more details.
    - We also implement a "mean" deterministic analog
        of the multinomial distribution to handle
        multiple outflows from the same compartment
    - We do not round the transition variables
    - We also use `softplus`, a smooth approximation to the
        ReLU function, to ensure that compartments are
        nonnegative (which is not guaranteed using
        the mean of a binomial/multinomial random variable
        rather than sampling from those distributions).

    Returns:
        (Tuple[FluFullMetapopStateTensors, dict, dict]):
            New `FluFullMetapopStateTensors` with updated state,
            `dict` of calibration targets corresponding to state
            values or transition variable values used for calibration,
            and `dict` of transition variable values to save this
            history. If `save_calibration_targets` is `False`,
            then the corresponding `dict` is empty, and similarly with
            `save_tvar_history`.
    """

    S_to_E = compute_S_to_E(state, params, precomputed, dt)

    # Deterministic multinomial implementation to match
    #   object-oriented version
    E_to_IP_rate = compute_E_to_IP_rate(params)
    E_to_IA_rate = compute_E_to_IA_rate(params)
    E_outgoing_total_rate = E_to_IP_rate + E_to_IA_rate
    E_to_IA = state.E * (E_to_IA_rate / E_outgoing_total_rate) * \
              torch_approx_binom_probability_from_rate(E_outgoing_total_rate, dt)
    E_to_IP = state.E * (E_to_IP_rate / E_outgoing_total_rate) * \
              torch_approx_binom_probability_from_rate(E_outgoing_total_rate, dt)

    IA_to_R = compute_IA_to_R(state, params, dt)

    IP_to_IS = compute_IP_to_IS(state, params, dt)

    # Deterministic multinomial implementation to match
    #   object-oriented version
    IS_to_R_rate = compute_IS_to_R_rate(state, params)
    IS_to_H_rate = compute_IS_to_H_rate(state, params)
    IS_outgoing_total_rate = IS_to_R_rate + IS_to_H_rate
    IS_to_R = state.IS * (IS_to_R_rate / IS_outgoing_total_rate) * \
              torch_approx_binom_probability_from_rate(IS_outgoing_total_rate, dt)
    IS_to_H = state.IS * (IS_to_H_rate / IS_outgoing_total_rate) * \
              torch_approx_binom_probability_from_rate(IS_outgoing_total_rate, dt)

    # Deterministic multinomial implementation to match
    #   object-oriented version
    H_to_R_rate = compute_H_to_R_rate(state, params)
    H_to_D_rate = compute_H_to_D_rate(state, params)
    H_outgoing_total_rate = H_to_R_rate + H_to_D_rate
    H_to_R = state.H * (H_to_R_rate / H_outgoing_total_rate) * torch_approx_binom_probability_from_rate(
        H_outgoing_total_rate, dt)
    H_to_D = state.H * (H_to_D_rate / H_outgoing_total_rate) * torch_approx_binom_probability_from_rate(
        H_outgoing_total_rate, dt)

    R_to_S = compute_R_to_S(state, params, dt)

    # Make sure compartments are nonnegative
    S_new = torch.nn.functional.softplus(state.S + R_to_S - S_to_E)
    E_new = torch.nn.functional.softplus(state.E + S_to_E - E_to_IP - E_to_IA)
    IP_new = torch.nn.functional.softplus(state.IP + E_to_IP - IP_to_IS)
    IS_new = torch.nn.functional.softplus(state.IS + IP_to_IS - IS_to_R - IS_to_H)
    IA_new = torch.nn.functional.softplus(state.IA + E_to_IA - IA_to_R)
    H_new = torch.nn.functional.softplus(state.H + IS_to_H - H_to_R - H_to_D)
    R_new = torch.nn.functional.softplus(state.R + IS_to_R + IA_to_R + H_to_R - R_to_S)
    D_new = torch.nn.functional.softplus(state.D + H_to_D)

    # Update immunity variables
    M_change = compute_M_change(state, params, precomputed, dt)
    MV_change = compute_MV_change(state, params, precomputed, dt)
    M_new = state.M + M_change
    MV_new = state.MV + MV_change

    state_new = FluFullMetapopStateTensors(S=S_new,
                                           E=E_new,
                                           IP=IP_new,
                                           IS=IS_new,
                                           IA=IA_new,
                                           H=H_new,
                                           R=R_new,
                                           D=D_new,
                                           M=M_new,
                                           MV=MV_new,
                                           absolute_humidity=state.absolute_humidity,
                                           daily_vaccines=state.daily_vaccines,
                                           flu_contact_matrix=state.flu_contact_matrix)

    calibration_targets = {}
    if save_calibration_targets:
        calibration_targets["IS_to_H"] = IS_to_H

    transition_variables = {}
    if save_tvar_history:
        transition_variables["S_to_E"] = S_to_E
        transition_variables["E_to_IP"] = E_to_IP
        transition_variables["E_to_IA"] = E_to_IA
        transition_variables["IA_to_R"] = IA_to_R
        transition_variables["IP_to_IS"] = IP_to_IS
        transition_variables["IS_to_R"] = IS_to_R
        transition_variables["IS_to_H"] = IS_to_H
        transition_variables["H_to_R"] = H_to_R
        transition_variables["H_to_D"] = H_to_D
        transition_variables["R_to_S"] = R_to_S
        transition_variables["M_change"] = M_change
        transition_variables["MV_change"] = MV_change

    return state_new, calibration_targets, transition_variables

compute_E_to_IA_rate(params: FluFullMetapopParamsTensors) -> torch.Tensor

Returns:

Type	Description
Tensor	(torch.Tensor of size (L, A, R))

Source code in CLT_BaseModel/flu_core/flu_torch_det_components.py

def compute_E_to_IA_rate(params: FluFullMetapopParamsTensors) -> torch.Tensor:
    """
    Returns:
        (torch.Tensor of size (L, A, R))
    """

    return params.E_to_I_rate * params.E_to_IA_prop

compute_E_to_IP_rate(params: FluFullMetapopParamsTensors) -> torch.Tensor

Returns:

Type	Description
Tensor	(torch.Tensor of size (L, A, R))

Source code in CLT_BaseModel/flu_core/flu_torch_det_components.py

def compute_E_to_IP_rate(params: FluFullMetapopParamsTensors) -> torch.Tensor:
    """
    Returns:
        (torch.Tensor of size (L, A, R))
    """

    return params.E_to_I_rate * (1 - params.E_to_IA_prop)

compute_H_to_D_rate(state: FluFullMetapopStateTensors, params: FluFullMetapopParamsTensors) -> torch.Tensor

Returns:

Type	Description
Tensor	(torch.Tensor of size (L, A, R))

Source code in CLT_BaseModel/flu_core/flu_torch_det_components.py

def compute_H_to_D_rate(state: FluFullMetapopStateTensors,
                        params: FluFullMetapopParamsTensors) -> torch.Tensor:
    """
    Returns:
        (torch.Tensor of size (L, A, R))
    """

    inf_induced_death_risk_reduce = params.inf_induced_death_risk_reduce
    vax_induced_death_risk_reduce = params.vax_induced_death_risk_reduce

    inf_induced_proportional_risk_reduce = \
        inf_induced_death_risk_reduce / (1 - inf_induced_death_risk_reduce)

    vax_induced_proportional_risk_reduce = \
        vax_induced_death_risk_reduce / (1 - vax_induced_death_risk_reduce)

    immunity_force = (1 + inf_induced_proportional_risk_reduce * state.M +
                      vax_induced_proportional_risk_reduce * state.MV)

    rate = params.H_to_D_rate * params.H_to_D_adjusted_prop / immunity_force

    return rate

compute_H_to_R_rate(state: FluFullMetapopStateTensors, params: FluFullMetapopParamsTensors) -> torch.Tensor

Returns:

Type	Description
Tensor	(torch.Tensor of size (L, A, R))

Source code in CLT_BaseModel/flu_core/flu_torch_det_components.py

def compute_H_to_R_rate(state: FluFullMetapopStateTensors,
                        params: FluFullMetapopParamsTensors) -> torch.Tensor:
    """
    Returns:
        (torch.Tensor of size (L, A, R))
    """

    inf_induced_death_risk_reduce = params.inf_induced_death_risk_reduce
    vax_induced_death_risk_reduce = params.vax_induced_death_risk_reduce

    inf_induced_proportional_risk_reduce = \
        inf_induced_death_risk_reduce / (1 - inf_induced_death_risk_reduce)

    vax_induced_proportional_risk_reduce = \
        vax_induced_death_risk_reduce / (1 - vax_induced_death_risk_reduce)

    immunity_force = (1 + inf_induced_proportional_risk_reduce * state.M +
                      vax_induced_proportional_risk_reduce * state.MV)

    rate = params.H_to_R_rate * (1 - params.H_to_D_adjusted_prop / immunity_force)

    return rate

compute_IA_to_R(state: FluFullMetapopStateTensors, params: FluFullMetapopParamsTensors, dt: float) -> torch.Tensor

Returns:

Type	Description
Tensor	(torch.Tensor of size (L, A, R))

Source code in CLT_BaseModel/flu_core/flu_torch_det_components.py

def compute_IA_to_R(state: FluFullMetapopStateTensors,
                    params: FluFullMetapopParamsTensors,
                    dt: float) -> torch.Tensor:
    """
    Returns:
        (torch.Tensor of size (L, A, R))
    """

    rate = params.IA_to_R_rate

    IA_to_R = state.IA * torch_approx_binom_probability_from_rate(rate, dt)

    return IA_to_R

compute_IP_to_IS(state: FluFullMetapopStateTensors, params: FluFullMetapopParamsTensors, dt: float) -> torch.Tensor

Returns:

Type	Description
Tensor	(torch.Tensor of size (L, A, R))

Source code in CLT_BaseModel/flu_core/flu_torch_det_components.py

def compute_IP_to_IS(state: FluFullMetapopStateTensors,
                     params: FluFullMetapopParamsTensors,
                     dt: float) -> torch.Tensor:
    """
    Returns:
        (torch.Tensor of size (L, A, R))
    """

    rate = params.IP_to_IS_rate

    IP_to_IS = state.IP * torch_approx_binom_probability_from_rate(rate, dt)

    return IP_to_IS

compute_IS_to_H_rate(state: FluFullMetapopStateTensors, params: FluFullMetapopParamsTensors) -> torch.Tensor

Returns:

Type	Description
Tensor	(torch.Tensor of size (L, A, R))

Source code in CLT_BaseModel/flu_core/flu_torch_det_components.py

def compute_IS_to_H_rate(state: FluFullMetapopStateTensors,
                         params: FluFullMetapopParamsTensors) -> torch.Tensor:
    """
    Returns:
        (torch.Tensor of size (L, A, R))
    """

    inf_induced_hosp_risk_reduce = params.inf_induced_hosp_risk_reduce
    inf_induced_proportional_risk_reduce = inf_induced_hosp_risk_reduce / (1 - inf_induced_hosp_risk_reduce)

    vax_induced_hosp_risk_reduce = params.vax_induced_hosp_risk_reduce
    vax_induced_proportional_risk_reduce = vax_induced_hosp_risk_reduce / (1 - vax_induced_hosp_risk_reduce)

    immunity_force = (1 + inf_induced_proportional_risk_reduce * state.M +
                      vax_induced_proportional_risk_reduce * state.MV)

    rate = params.IS_to_H_rate * params.IS_to_H_adjusted_prop / immunity_force

    return rate

compute_IS_to_R_rate(state: FluFullMetapopStateTensors, params: FluFullMetapopParamsTensors) -> torch.Tensor

Returns:

Type	Description
Tensor	(torch.Tensor of size (L, A, R))

Source code in CLT_BaseModel/flu_core/flu_torch_det_components.py

def compute_IS_to_R_rate(state: FluFullMetapopStateTensors,
                         params: FluFullMetapopParamsTensors) -> torch.Tensor:
    """
    Returns:
        (torch.Tensor of size (L, A, R))
    """

    inf_induced_hosp_risk_reduce = params.inf_induced_hosp_risk_reduce
    inf_induced_proportional_risk_reduce = inf_induced_hosp_risk_reduce / (1 - inf_induced_hosp_risk_reduce)

    vax_induced_hosp_risk_reduce = params.vax_induced_hosp_risk_reduce
    vax_induced_proportional_risk_reduce = vax_induced_hosp_risk_reduce / (1 - vax_induced_hosp_risk_reduce)

    immunity_force = (1 + inf_induced_proportional_risk_reduce * state.M +
                      vax_induced_proportional_risk_reduce * state.MV)

    rate = params.IS_to_R_rate * (1 - params.IS_to_H_adjusted_prop / immunity_force)

    return rate

compute_MV_change(state: FluFullMetapopStateTensors, params: FluFullMetapopParamsTensors, precomputed: FluPrecomputedTensors, dt: float) -> torch.Tensor

Returns:

Type	Description
Tensor	(torch.Tensor of size (L, A, R))

Source code in CLT_BaseModel/flu_core/flu_torch_det_components.py

def compute_MV_change(state: FluFullMetapopStateTensors,
                      params: FluFullMetapopParamsTensors,
                      precomputed: FluPrecomputedTensors,
                      dt: float) -> torch.Tensor:
    """
    Returns:
        (torch.Tensor of size (L, A, R))
    """

    MV_change = state.daily_vaccines / precomputed.total_pop_LAR_tensor - \
                params.vax_induced_immune_wane * state.MV

    return MV_change * dt

compute_M_change(state: FluFullMetapopStateTensors, params: FluFullMetapopParamsTensors, precomputed: FluPrecomputedTensors, dt: float) -> torch.Tensor

Returns:

Type	Description
Tensor	(torch.Tensor of size (L, A, R))

Source code in CLT_BaseModel/flu_core/flu_torch_det_components.py

def compute_M_change(state: FluFullMetapopStateTensors, params: FluFullMetapopParamsTensors,
                     precomputed: FluPrecomputedTensors,
                     dt: float) -> torch.Tensor:
    """
    Returns:
        (torch.Tensor of size (L, A, R))
    """

    # Note: already includes dt
    R_to_S = state.R * torch_approx_binom_probability_from_rate(params.R_to_S_rate, dt)

    M_change = (R_to_S / precomputed.total_pop_LAR_tensor) * \
               (1 - params.inf_induced_saturation * state.M - params.vax_induced_saturation * state.MV) - \
               params.inf_induced_immune_wane * state.M * dt

    # Because R_to_S includes dt already, we do not return M_change * dt -- we only multiply
    #   the last term in the expression above by dt
    return M_change

compute_R_to_S(state: FluFullMetapopStateTensors, params: FluFullMetapopParamsTensors, dt: float) -> torch.Tensor

Returns:

Type	Description
Tensor	(torch.Tensor of size (L, A, R))

Source code in CLT_BaseModel/flu_core/flu_torch_det_components.py

def compute_R_to_S(state: FluFullMetapopStateTensors,
                   params: FluFullMetapopParamsTensors,
                   dt: float) -> torch.Tensor:
    """
    Returns:
        (torch.Tensor of size (L, A, R))
    """

    rate = params.R_to_S_rate

    R_to_S = state.R * torch_approx_binom_probability_from_rate(rate, dt)

    return R_to_S

compute_S_to_E(state: FluFullMetapopStateTensors, params: FluFullMetapopParamsTensors, precomputed: FluPrecomputedTensors, dt: float) -> torch.Tensor

Returns:

Type	Description
Tensor	(torch.Tensor of size (L, A, R))

Source code in CLT_BaseModel/flu_core/flu_torch_det_components.py

def compute_S_to_E(state: FluFullMetapopStateTensors,
                   params: FluFullMetapopParamsTensors,
                   precomputed: FluPrecomputedTensors,
                   dt: float) -> torch.Tensor:
    """
    Returns:
        (torch.Tensor of size (L, A, R))
    """

    # Needs flu_contact_matrix to be in state for this
    total_mixing_exposure = compute_total_mixing_exposure(state, params, precomputed)

    if total_mixing_exposure.size() != torch.Size([precomputed.L,
                                                   precomputed.A,
                                                   precomputed.R]):
        raise Exception("force_of_infection must be L x A x R corresponding \n"
                        "to number of locations (subpopulations), age groups, \n"
                        "and risk groups.")

    beta_adjusted = compute_beta_adjusted(state, params)

    inf_induced_inf_risk_reduce = params.inf_induced_inf_risk_reduce
    inf_induced_proportional_risk_reduce = inf_induced_inf_risk_reduce / (1 - inf_induced_inf_risk_reduce)

    vax_induced_inf_risk_reduce = params.vax_induced_inf_risk_reduce
    vax_induced_proportional_risk_reduce = vax_induced_inf_risk_reduce / (1 - vax_induced_inf_risk_reduce)

    immune_force = (1 + inf_induced_proportional_risk_reduce * state.M +
                    vax_induced_proportional_risk_reduce * state.MV)

    rate = beta_adjusted * total_mixing_exposure / immune_force

    S_to_E = state.S * torch_approx_binom_probability_from_rate(rate, dt)

    return S_to_E

compute_active_pop_LAR(state: FluTravelStateTensors, _params: FluTravelParamsTensors, precomputed: FluPrecomputedTensors) -> torch.Tensor

Compute the active population for location-age-risk (l, a, r) as a tensor. Used to compute the effective population in the travel model, which is the population size adjusted for incoming visitors, residents traveling, and assuming hospitalized individuals are not mobile enough to infect others.

Returns:

Type	Description
Tensor	torch.Tensor of size (L, A, R): Active population: those who are not hospitalized (i.e. those who are not too sick to move and travel regularly)

Source code in CLT_BaseModel/flu_core/flu_travel_functions.py

def compute_active_pop_LAR(state: FluTravelStateTensors,
                           _params: FluTravelParamsTensors,
                           precomputed: FluPrecomputedTensors) -> torch.Tensor:
    """
    Compute the active population for location-age-risk
    (l, a, r) as a tensor. Used to compute the
    effective population in the travel model, which is
    the population size adjusted for incoming visitors,
    residents traveling, and assuming hospitalized
    individuals are not mobile enough to infect others.

    Returns:
        torch.Tensor of size (L, A, R):
            Active population: those who are not
            hospitalized (i.e. those who are not too sick
            to move and travel regularly)
    """

    # _params is not used now -- but this is included for
    #   function signature consistency with other
    #   similar computation functions

    return precomputed.total_pop_LAR_tensor - state.H

compute_beta_adjusted(state: FluFullMetapopStateTensors, params: FluFullMetapopParamsTensors) -> torch.Tensor

Computes beta-adjusted humidity.

Returns:

Type	Description
Tensor	(torch.Tensor of size (L, A, R))

Source code in CLT_BaseModel/flu_core/flu_torch_det_components.py

def compute_beta_adjusted(state: FluFullMetapopStateTensors,
                          params: FluFullMetapopParamsTensors) -> torch.Tensor:
    """
    Computes beta-adjusted humidity.

    Returns:
        (torch.Tensor of size (L, A, R))
    """

    absolute_humidity = state.absolute_humidity
    beta_adjusted = params.beta_baseline * (1 + params.humidity_impact * np.exp(-180 * absolute_humidity))

    return beta_adjusted

compute_effective_pop_LA(state: FluTravelStateTensors, params: FluTravelParamsTensors, precomputed: FluPrecomputedTensors) -> torch.Tensor

Returns:

Type	Description
Tensor	torch.Tensor of size (L, A): Effective population, summed over risk groups. See compute_active_pop_LAR docstring for more information.

Source code in CLT_BaseModel/flu_core/flu_travel_functions.py

def compute_effective_pop_LA(state: FluTravelStateTensors,
                             params: FluTravelParamsTensors,
                             precomputed: FluPrecomputedTensors) -> torch.Tensor:
    """
    Returns:
        torch.Tensor of size (L, A):
            Effective population, summed over risk groups.
            See `compute_active_pop_LAR` docstring for more
            information.
    """

    active_pop_LAR = compute_active_pop_LAR(state, params, precomputed)

    # Nonlocal travel proportions is L x L
    # Active population LAR is L x A x R
    outside_visitors_LAR = torch.einsum("kl,kar->lar",
                                        precomputed.nonlocal_travel_prop,
                                        active_pop_LAR)

    # This is correct -- Dave checked in meeting -- we don't need Einstein
    #   notation here!
    # In computation, broadcast sum_residents_nonlocal_travel_prop to be L x 1 x 1
    traveling_residents_LAR = precomputed.sum_residents_nonlocal_travel_prop[:, None, None] * \
                              active_pop_LAR

    mobility_modifier = params.mobility_modifier[0, :, 0]

    effective_pop_LA = precomputed.total_pop_LA + mobility_modifier * \
                       torch.sum(outside_visitors_LAR + traveling_residents_LAR, dim=2)

    return effective_pop_LA

compute_flu_contact_matrix(params: FluFullMetapopParamsTensors, schedules: FluFullMetapopScheduleTensors, day_counter: int) -> torch.Tensor

Computes flu model contact matrix in tensor format -- makes adjustments based on whether day is school day or work day.

Returns:

Type	Description
Tensor	(torch.Tensor of size (L, A, A))

Source code in CLT_BaseModel/flu_core/flu_torch_det_components.py

def compute_flu_contact_matrix(params: FluFullMetapopParamsTensors,
                               schedules: FluFullMetapopScheduleTensors,
                               day_counter: int) -> torch.Tensor:
    """
    Computes flu model contact matrix in tensor format -- makes
    adjustments based on whether day is school day or work day.

    Returns:
        (torch.Tensor of size (L, A, A))
    """

    # Here, using schedules.is_school_day[day_counter][:,:,0] and similarly for
    #   is_work_day because each contact matrix (as a metapop tensor) is L x A x A --
    #   we don't use risk -- assume here that we do not have a different school/work-day
    #   schedule based on risk, so just grab the first risk group
    # But then we have to take (1 - schedules.is_school_day[day_counter][:, :, 0]), which is
    #   L x A, and then make it L x A x 1 (unsqueeze the last dimension) to make the
    #   broadcasting work (because this gets element-wise multiplied by params.school_contact_matrix)
    flu_contact_matrix = \
        params.total_contact_matrix - \
        params.school_contact_matrix * (1 - schedules.is_school_day[day_counter][:, :, 0]).unsqueeze(dim=2) - \
        params.work_contact_matrix * (1 - schedules.is_work_day[day_counter][:, :, 0]).unsqueeze(dim=2)

    return flu_contact_matrix

compute_local_to_local_exposure(flu_contact_matrix: torch.Tensor, mobility_modifier: torch.Tensor, sum_residents_nonlocal_travel_prop: torch.Tensor, wtd_infectious_ratio_LLA: torch.Tensor, location_ix: int) -> torch.Tensor

Raw means that this is unnormalized by relative_suscept. Excludes beta and population-level immunity adjustments -- those are factored in later.

Returns:

Type	Description
Tensor	torch.Tensor of size (A): For a given location (specified by location_ix), compute local transmission caused by residents traveling within their home location, summed over risk groups.

Source code in CLT_BaseModel/flu_core/flu_travel_functions.py

def compute_local_to_local_exposure(flu_contact_matrix: torch.Tensor,
                                    mobility_modifier: torch.Tensor,
                                    sum_residents_nonlocal_travel_prop: torch.Tensor,
                                    wtd_infectious_ratio_LLA: torch.Tensor,
                                    location_ix: int) -> torch.Tensor:
    """
    Raw means that this is unnormalized by `relative_suscept`.
    Excludes beta and population-level immunity adjustments --
    those are factored in later.

    Returns:
        torch.Tensor of size (A):
            For a given location (specified by `location_ix`), compute
            local transmission caused by residents traveling within their
            home location, summed over risk groups.
    """

    # WARNING: we assume `mobility_modifier` is input as (A, 1) --
    # if this changes, we have to change the implementation.
    # The risk dimension does not have unique values, so we just
    # grab the first element of the risk dimension.
    mobility_modifier = mobility_modifier[location_ix, :, 0]

    result = (1 - mobility_modifier * sum_residents_nonlocal_travel_prop[location_ix]) * \
             torch.matmul(flu_contact_matrix[location_ix, :, :],
                          wtd_infectious_ratio_LLA[location_ix, location_ix, :])

    return result

compute_outside_visitors_exposure(flu_contact_matrix: torch.Tensor, mobility_modifier: torch.Tensor, travel_proportions: torch.Tensor, wtd_infectious_ratio_LLA: torch.Tensor, local_ix: int, visitors_ix: int) -> torch.Tensor

Computes raw (unnormalized by relative_suscept) transmission to local_ix due to outside visitors from visitors_ix. Excludes beta and population-level immunity adjustments -- those are factored in later.

Returns:

Type	Description
Tensor	torch.Tensor of size (A)

Source code in CLT_BaseModel/flu_core/flu_travel_functions.py

def compute_outside_visitors_exposure(flu_contact_matrix: torch.Tensor,
                                      mobility_modifier: torch.Tensor,
                                      travel_proportions: torch.Tensor,
                                      wtd_infectious_ratio_LLA: torch.Tensor,
                                      local_ix: int,
                                      visitors_ix: int) -> torch.Tensor:
    """
    Computes raw (unnormalized by `relative_suscept`) transmission
    to `local_ix` due to outside visitors from `visitors_ix`.
    Excludes beta and population-level immunity adjustments --
    those are factored in later.

    Returns:
        torch.Tensor of size (A)
    """

    # In location `local_ix`, we are looking at the visitors from
    #   `visitors_ix` who come to `local_ix` (and infect folks in `local_ix`)

    # See WARNING in `compute_local_to_local_exposure()`
    mobility_modifier = mobility_modifier[visitors_ix, :, 0]

    result = travel_proportions[visitors_ix, local_ix] * \
             torch.matmul(mobility_modifier * flu_contact_matrix[local_ix, :, :],
                          wtd_infectious_ratio_LLA[visitors_ix, local_ix, :])

    return result

compute_pop_by_age(subpop_params: FluSubpopParams) -> np.ndarray

Returns:

Type	Description
ndarray	np.ndarray: A x 1 array -- where A is the number of age groups -- where ith element corresponds to total population (across all compartments, including "D", and across all risk groups) in age group i

Source code in CLT_BaseModel/flu_core/flu_components.py

def compute_pop_by_age(subpop_params: FluSubpopParams) -> np.ndarray:
    """
    Returns:
        np.ndarray:
            A x 1 array -- where A is the number of age groups --
            where ith element corresponds to total population
            (across all compartments, including "D", and across all risk groups)
            in age group i
    """

    return np.sum(subpop_params.total_pop_age_risk, axis=1, keepdims=True)

compute_residents_traveling_exposure(flu_contact_matrix: torch.Tensor, mobility_modifier: torch.Tensor, travel_proportions: torch.Tensor, wtd_infectious_ratio_LLA: torch.Tensor, local_ix: int, dest_ix: int) -> torch.Tensor

Computes raw (unnormalized by relative_suscept) transmission to local_ix, due to residents of local_ix traveling to dest_ix and getting infected in dest_ix. Excludes beta and population-level immunity adjustments -- those are factored in later.

Returns:

Type	Description
Tensor	torch.Tensor of size (A)

Source code in CLT_BaseModel/flu_core/flu_travel_functions.py

def compute_residents_traveling_exposure(flu_contact_matrix: torch.Tensor,
                                         mobility_modifier: torch.Tensor,
                                         travel_proportions: torch.Tensor,
                                         wtd_infectious_ratio_LLA: torch.Tensor,
                                         local_ix: int,
                                         dest_ix: int) -> torch.Tensor:
    """
    Computes raw (unnormalized by `relative_suscept`) transmission
    to `local_ix`, due to residents of `local_ix` traveling to `dest_ix`
    and getting infected in `dest_ix`. Excludes beta and population-level
    immunity adjustments -- those are factored in later.

    Returns:
        torch.Tensor of size (A)
    """

    # See WARNING in `compute_local_to_local_exposure()`
    mobility_modifier = mobility_modifier[local_ix, :, 0]

    result = mobility_modifier * travel_proportions[local_ix, dest_ix] * \
             torch.matmul(flu_contact_matrix[local_ix, :, :],
                          wtd_infectious_ratio_LLA[dest_ix, dest_ix, :])

    return result

compute_rsquared(reference_timeseries: list[np.ndarray], simulated_timeseries: list[np.ndarray]) -> float

Source code in CLT_BaseModel/flu_core/flu_accept_reject.py

def compute_rsquared(reference_timeseries: list[np.ndarray],
                     simulated_timeseries: list[np.ndarray]) -> float:
    if len(reference_timeseries) != len(simulated_timeseries):
        raise ValueError("Reference time series and simulated time series \n"
                         "must have same length.")

    reference_timeseries = np.asarray(reference_timeseries)
    simulated_timeseries = np.asarray(simulated_timeseries)

    ybar = reference_timeseries.mean(axis=0)

    ss_residual = np.sum(np.square(simulated_timeseries - reference_timeseries))
    ss_total = np.sum(np.square(reference_timeseries - ybar))

    return 1 - ss_residual / ss_total

compute_total_mixing_exposure(state: FluTravelStateTensors, params: FluTravelParamsTensors, precomputed: FluPrecomputedTensors) -> torch.Tensor

Computes "total mixing exposure" for location-age-risk (l, a, r) -- the rate of exposure to infectious individuals, accounting for both local transmission, incoming visitors, and residents traveling. Normalized by relative_suscept!

Combines subroutines compute_local_to_local_exposure(), compute_outside_visitors_exposure(), and compute_residents_traveling_exposure(). Note that these subroutines do not include relative susceptibility -- but this function includes relative susceptibility -- this is to avoid unnecessary repeated multiplication by relative susceptible in each subroutine.

Returns:

Type	Description
Tensor	torch.Tensor of size (L, A, R)

Source code in CLT_BaseModel/flu_core/flu_travel_functions.py

def compute_total_mixing_exposure(state: FluTravelStateTensors,
                                  params: FluTravelParamsTensors,
                                  precomputed: FluPrecomputedTensors) -> torch.Tensor:
    """
    Computes "total mixing exposure" for location-age-risk
    (l, a, r) -- the rate of exposure to infectious individuals,
    accounting for both local transmission, incoming visitors, and
    residents traveling. **Normalized by `relative_suscept`!**

    Combines subroutines `compute_local_to_local_exposure()`,
    `compute_outside_visitors_exposure()`, and `compute_residents_traveling_exposure()`.
    Note that these subroutines do not include relative susceptibility --
    but this function includes relative susceptibility -- this is to avoid
    unnecessary repeated multiplication by relative susceptible in each subroutine.

    Returns:
        torch.Tensor of size (L, A, R)
    """

    L, A, R = precomputed.L, precomputed.A, precomputed.R

    mobility_modifier = params.mobility_modifier
    flu_contact_matrix = state.flu_contact_matrix
    travel_proportions = params.travel_proportions
    sum_residents_nonlocal_travel_prop = precomputed.sum_residents_nonlocal_travel_prop
    wtd_infectious_ratio_LLA = compute_wtd_infectious_ratio_LLA(state, params, precomputed)

    relative_suscept = params.relative_suscept[0, :, 0]

    total_mixing_exposure = torch.tensor(np.zeros((L, A, R)))

    # Couldn't figure out how to do this without two for-loops ;)
    # Welcoming any efficiency improvements!
    for l in np.arange(L):

        raw_total_mixing_exposure = torch.tensor(np.zeros(A))

        raw_total_mixing_exposure = raw_total_mixing_exposure + \
                                    compute_local_to_local_exposure(flu_contact_matrix,
                                                                    mobility_modifier,
                                                                    sum_residents_nonlocal_travel_prop,
                                                                    wtd_infectious_ratio_LLA,
                                                                    l)

        for k in np.arange(L):
            raw_total_mixing_exposure = raw_total_mixing_exposure + \
                                        compute_outside_visitors_exposure(
                                            flu_contact_matrix,
                                            mobility_modifier,
                                            travel_proportions,
                                            wtd_infectious_ratio_LLA,
                                            l,
                                            k)

            raw_total_mixing_exposure = raw_total_mixing_exposure + \
                                        compute_residents_traveling_exposure(
                                            flu_contact_matrix,
                                            mobility_modifier,
                                            travel_proportions,
                                            wtd_infectious_ratio_LLA,
                                            l,
                                            k)

        normalized_total_mixing_exposure = relative_suscept * raw_total_mixing_exposure

        total_mixing_exposure[l, :, :] = normalized_total_mixing_exposure.view(A, 1).expand((A, R))

    return total_mixing_exposure

compute_wtd_infectious_LA(state: FluTravelStateTensors, params: FluTravelParamsTensors) -> torch.Tensor

Returns:

Type	Description
Tensor	torch.Tensor of size (L, A): Weighted infectious, summed over risk groups: includes presymptomatic, asymptomatic, and symptomatic, weighted by relative infectiousness

Source code in CLT_BaseModel/flu_core/flu_travel_functions.py

def compute_wtd_infectious_LA(state: FluTravelStateTensors,
                              params: FluTravelParamsTensors) -> torch.Tensor:
    """
    Returns:
        torch.Tensor of size (L, A):
            Weighted infectious, summed over risk groups:
            includes presymptomatic, asymptomatic, and symptomatic,
            weighted by relative infectiousness
    """

    # Einstein notation here means sum over risk groups
    IS = torch.einsum("lar->la", state.IS)
    wtd_IP = \
        params.IP_relative_inf * torch.einsum("lar->la", state.IP)
    wtd_IA = \
        params.IA_relative_inf * torch.einsum("lar->la", state.IA)

    return IS + wtd_IP + wtd_IA

compute_wtd_infectious_ratio_LLA(state: FluTravelStateTensors, params: FluTravelParamsTensors, precomputed: FluPrecomputedTensors) -> torch.Tensor

Returns:

Type	Description
Tensor	torch.Tensor of size (L, L, A): Element i,j,a corresponds to ratio of weighted infectious people in location i, age group a (summed over risk groups) to the effective population in location j (summed over risk groups)

Source code in CLT_BaseModel/flu_core/flu_travel_functions.py

def compute_wtd_infectious_ratio_LLA(state: FluTravelStateTensors,
                                     params: FluTravelParamsTensors,
                                     precomputed: FluPrecomputedTensors) -> torch.Tensor:
    """
    Returns:
        torch.Tensor of size (L, L, A):
            Element i,j,a corresponds to ratio of weighted infectious people
            in location i, age group a (summed over risk groups) to the effective
            population in location j (summed over risk groups)
    """

    wtd_infectious_LA = compute_wtd_infectious_LA(state, params)

    effective_pop_LA = compute_effective_pop_LA(state, params, precomputed)

    prop_wtd_infectious = torch.einsum("ka,la->kla",
                                       wtd_infectious_LA,
                                       1 / effective_pop_LA)

    return prop_wtd_infectious

compute_wtd_presymp_asymp_by_age(subpop_state: FluSubpopState, subpop_params: FluSubpopParams) -> np.ndarray

Returns weighted sum of IP and IA compartment for subpopulation with given state and parameters. IP and IA are weighted by their relative infectiousness respectively, and then summed over risk groups.

Returns:

Type	Description
ndarray	np.ndarray of shape (A, R)

Source code in CLT_BaseModel/flu_core/flu_components.py

def compute_wtd_presymp_asymp_by_age(subpop_state: FluSubpopState,
                                     subpop_params: FluSubpopParams) -> np.ndarray:
    """
    Returns weighted sum of IP and IA compartment for
        subpopulation with given state and parameters.
        IP and IA are weighted by their relative infectiousness
        respectively, and then summed over risk groups.

    Returns:
        np.ndarray of shape (A, R)
    """

    # sum over risk groups
    wtd_IP = \
        subpop_params.IP_relative_inf * np.sum(subpop_state.IP, axis=1, keepdims=True)
    wtd_IA = \
        subpop_params.IA_relative_inf * np.sum(subpop_state.IA, axis=1, keepdims=True)

    return wtd_IP + wtd_IA

create_dict_of_tensors(d: dict, requires_grad: bool = True) -> dict

Converts dictionary entries to tensor (of type torch.float32) and if requires_grad is True, turns on gradient tracking for each entry -- returns new dictionary.

Source code in CLT_BaseModel/flu_core/flu_torch_det_components.py

def create_dict_of_tensors(d: dict,
                           requires_grad: bool = True) -> dict:
    """
    Converts dictionary entries to `tensor` (of type `torch.float32`)
    and if `requires_grad` is `True`, turns on gradient tracking for
    each entry -- returns new dictionary.
    """

    def to_tensor(k, v):
        if v is None:
            return None
        else:
            return torch.tensor(v, dtype=torch.float32, requires_grad=requires_grad)

    return {k: to_tensor(k, v) for k, v in d.items()}

torch_approx_binom_probability_from_rate(rate, dt)

Torch-compatible implementation of converting a rate into a probability. See analogous numpy implementation base_components/approx_binom_probability_from_rate() docstring for details.

Source code in CLT_BaseModel/flu_core/flu_torch_det_components.py

def torch_approx_binom_probability_from_rate(rate, dt):
    """
    Torch-compatible implementation of converting a
    rate into a probability. See analogous numpy implementation
    `base_components/approx_binom_probability_from_rate()` docstring
    for details.
    """

    return 1 - torch.exp(-rate * dt)

torch_simulate_full_history(state: FluFullMetapopStateTensors, params: FluFullMetapopParamsTensors, precomputed: FluPrecomputedTensors, schedules: FluFullMetapopScheduleTensors, num_days: int, timesteps_per_day: int) -> Tuple[dict, dict]

Simulates the flu model with a differentiable torch implementation that carries out binom_deterministic_no_round transition types -- returns hospital admits for calibration use.

See subroutine advance_timestep for additional details.

Returns:

Type	Description
Tuple[dict, dict]	Returns hospital admits (the IS to H transition variable value) for day, location, age, risk, in tensor format.

Source code in CLT_BaseModel/flu_core/flu_torch_det_components.py

def torch_simulate_full_history(state: FluFullMetapopStateTensors,
                                params: FluFullMetapopParamsTensors,
                                precomputed: FluPrecomputedTensors,
                                schedules: FluFullMetapopScheduleTensors,
                                num_days: int,
                                timesteps_per_day: int) -> Tuple[dict, dict]:
    """
    Simulates the flu model with a differentiable torch implementation
    that carries out `binom_deterministic_no_round` transition types --
    returns hospital admits for calibration use.

    See subroutine `advance_timestep` for additional details.

    Returns:
        (Tuple[dict, dict]):
            Returns hospital admits (the IS to H transition variable value)
            for day, location, age, risk, in tensor format.
    """

    dt = 1 / float(timesteps_per_day)

    state_history_dict = defaultdict(list)
    tvar_history_dict = defaultdict(list)

    # This could probably be written better so we don't have
    #   unused variables "_" that grab `advance_timestep` output?

    for day in range(num_days):
        state = update_state_with_schedules(state, params, schedules, day)

        for timestep in range(timesteps_per_day):
            if timestep == timesteps_per_day-1:
                state, _, tvar_history = \
                    advance_timestep(state, params, precomputed, dt, save_tvar_history=True)
                for key in tvar_history:
                    tvar_history_dict[key].append(tvar_history[key])
            else:
                state, _, _ = \
                    advance_timestep(state, params, precomputed, dt, save_tvar_history=False)

        for field in fields(state):
            if field.name == "init_vals":
                continue
            state_history_dict[str(field.name)].append(getattr(state, field.name).clone())

    return state_history_dict, tvar_history_dict

torch_simulate_hospital_admits(state: FluFullMetapopStateTensors, params: FluFullMetapopParamsTensors, precomputed: FluPrecomputedTensors, schedules: FluFullMetapopScheduleTensors, num_days: int, timesteps_per_day: int) -> torch.Tensor

Analogous to torch_simulate_full_history but only saves and returns hospital admits for calibration use.

Returns:

Type	Description
torch.Tensor of size (num_days, L, A, R)	Returns hospital admits (the IS to H transition variable value) for day, location, age, risk, in tensor format.

Source code in CLT_BaseModel/flu_core/flu_torch_det_components.py

def torch_simulate_hospital_admits(state: FluFullMetapopStateTensors,
                                     params: FluFullMetapopParamsTensors,
                                     precomputed: FluPrecomputedTensors,
                                     schedules: FluFullMetapopScheduleTensors,
                                     num_days: int,
                                     timesteps_per_day: int) -> torch.Tensor:
    """
    Analogous to `torch_simulate_full_history` but only saves and
    returns hospital admits for calibration use.

    Returns:
        (torch.Tensor of size (num_days, L, A, R)):
            Returns hospital admits (the IS to H transition variable value)
            for day, location, age, risk, in tensor format.
    """

    hospital_admits_history = []

    dt = 1 / float(timesteps_per_day)

    for day in range(num_days):
        state = update_state_with_schedules(state, params, schedules, day)
        for timestep in range(timesteps_per_day):
            state, calibration_targets, _ = advance_timestep(state, params, precomputed, day, dt)
        hospital_admits_history.append(calibration_targets["IS_to_H"].clone())

    return torch.stack(hospital_admits_history)

update_state_with_schedules(state: FluFullMetapopStateTensors, params: FluFullMetapopParamsTensors, schedules: FluFullMetapopScheduleTensors, day_counter: int) -> FluFullMetapopStateTensors

Returns new dataclass formed by copying the current state and updating specific values according to schedules and the simulation's current day_counter.

Returns:

Type	Description
FluFullMetapopStateTensors	New state with updated schedule-related values: - flu_contact_matrix - absolute_humidity - daily_vaccines All other fields remain unchanged from the input state.

Source code in CLT_BaseModel/flu_core/flu_torch_det_components.py

def update_state_with_schedules(state: FluFullMetapopStateTensors,
                                params: FluFullMetapopParamsTensors,
                                schedules: FluFullMetapopScheduleTensors,
                                day_counter: int) -> FluFullMetapopStateTensors:
    """
    Returns new dataclass formed by copying the current `state`
    and updating specific values according to `schedules` and
    the simulation's current `day_counter`.

    Returns:
        (FluFullMetapopStateTensors):
            New state with updated schedule-related values:
              - `flu_contact_matrix`
              - `absolute_humidity`
              - `daily_vaccines`
            All other fields remain unchanged from the input `state`.
    """

    flu_contact_matrix = compute_flu_contact_matrix(params, schedules, day_counter)
    absolute_humidity = schedules.absolute_humidity[day_counter]
    daily_vaccines = schedules.daily_vaccines[day_counter]

    state_new = FluFullMetapopStateTensors(
        S=state.S,
        E=state.E,
        IP=state.IP,
        IS=state.IS,
        IA=state.IA,
        H=state.H,
        R=state.R,
        D=state.D,
        M=state.M,
        MV=state.MV,
        absolute_humidity=absolute_humidity,
        daily_vaccines=daily_vaccines,
        flu_contact_matrix=flu_contact_matrix
    )

    return state_new