Skip to content

Interactive

InteractiveGroup

Extracts the necessary information about a group to perform an interaction time step over it.

This step is necessary, since all the information is stored in numpy arrays that allow for efficient computation.

Source code in june/groups/group/interactive.py 
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
class InteractiveGroup:
    """Extracts the necessary information about a group to perform an interaction time
    step over it.

    This step is necessary, since all the information is stored in numpy
    arrays that allow for efficient computation.

    """

    def __init__(self, group: "Group", people_from_abroad=None):
        """
        This function is very long to avoid function calls for performance reasons.
        InteractiveGroups are created millions of times. Given a group, we need to extract:
        - ids of the people that can infect (infector).
        - ids of the people that can be infected (susceptible).
        - probabilities of transmission of the infectors.
        - susceptibilities of the susceptible.
        - indices of the subgroups that contain infectors.
        - sizes of the subgroups that contain infectors.
        - indices of the subgroups that contain susceptible.
        - spec of the group
        - super area of the group (for geo attributes like local regional compliances)

        Args:
            group ("Group"): the group to make interact.
            people_from_abroad (bool, optional): (Default = None).

        """
        people_from_abroad = people_from_abroad or {}
        self.group = group
        self.infectors_per_infection_per_subgroup = defaultdict(
            lambda: defaultdict(lambda: defaultdict(list))
        )  # maps virus variant -> subgroup -> infectors -> {infector ids, transmission probs}
        self.susceptibles_per_subgroup = defaultdict(
            dict
        )  # maps subgroup -> susceptible id -> {variant -> susceptibility}
        self.subgroup_sizes = {}
        group_size = 0

        for subgroup_index, subgroup in enumerate(group.subgroups):
            subgroup_size = len(subgroup.people)
            if subgroup.subgroup_type in people_from_abroad:
                people_abroad_data = people_from_abroad[subgroup.subgroup_type]
                people_abroad_ids = people_abroad_data.keys()
                subgroup_size += len(people_abroad_ids)
            else:
                people_abroad_data = None
                people_abroad_ids = []
            if subgroup_size == 0:
                continue
            self.subgroup_sizes[subgroup_index] = subgroup_size
            group_size += subgroup_size

            # Get susceptible people
            # local
            for person in subgroup:
                if not person.infected:
                    self.susceptibles_per_subgroup[subgroup_index][
                        person.id
                    ] = person.immunity.susceptibility_dict
            # from abroad
            for id in people_abroad_ids:
                if people_abroad_data[id]["susc"]:
                    dd = {
                        key: value
                        for key, value in zip(
                            people_abroad_data[id]["immunity_inf_ids"],
                            people_abroad_data[id]["immunity_suscs"],
                        )
                    }
                    self.susceptibles_per_subgroup[subgroup_index][id] = dd

            # Get infectors
            for person in subgroup:
                if person.infection is not None:
                    infection_id = person.infection.infection_id()
                    self.infectors_per_infection_per_subgroup[infection_id][
                        subgroup_index
                    ]["ids"].append(person.id)
                    self.infectors_per_infection_per_subgroup[infection_id][
                        subgroup_index
                    ]["trans_probs"].append(person.infection.transmission.probability)
            for id in people_abroad_ids:
                if people_abroad_data[id]["inf_id"] != 0:
                    infection_id = people_abroad_data[id]["inf_id"]
                    self.infectors_per_infection_per_subgroup[infection_id][
                        subgroup_index
                    ]["ids"].append(id)
                    self.infectors_per_infection_per_subgroup[infection_id][
                        subgroup_index
                    ]["trans_probs"].append(people_abroad_data[id]["inf_prob"])
        self.must_timestep = self.has_susceptible and self.has_infectors
        self.size = group_size


    @classmethod
    def get_raw_contact_matrix(
        cls, contact_matrix, alpha_physical, proportion_physical, characteristic_time
    ):
        """Returns the processed contact matrix.

        By default it returns the input, but children of this class will interact differently.

        Args:
            contact_matrix:
            alpha_physical: 
            proportion_physical: 
            characteristic_time:

        Returns:
            processed_contact_matrix (npt.NDArray):
                contact matrix after adjustment for if the contact is physical,
                and rescaled for the time of contact. 

        """
        processed_contact_matrix = contact_matrix * (
            1.0 + (alpha_physical - 1.0) * proportion_physical
        )
        processed_contact_matrix *= 24 / characteristic_time
        return processed_contact_matrix

    def get_processed_beta(self, betas, beta_reductions):
        """Returns the processed contact intensity, by taking into account the policies
        beta reductions and regional compliance. This is a group method as different interactive
        groups may choose to treat this differently.

        Args:
            betas: 
            beta_reductions:

        Returns:
            Modified betas

        """
        beta = betas[self.spec]
        beta_reduction = beta_reductions.get(self.spec, 1.0)
        try:
            regional_compliance = self.super_area.region.regional_compliance
        except AttributeError:
            regional_compliance = 1
        try:
            lockdown_tier = self.super_area.region.policy["lockdown_tier"]
            if lockdown_tier is None:
                lockdown_tier = 1
        except Exception:
            lockdown_tier = 1
        if int(lockdown_tier) == 4:
            tier_reduction = 0.5
        else:
            tier_reduction = 1.0

        return beta * (1 + regional_compliance * tier_reduction * (beta_reduction - 1))

    def get_processed_contact_matrix(self, contact_matrix):
        """ """
        return contact_matrix

    @property
    def spec(self):
        """ """
        return self.group.spec

    @property
    def super_area(self):
        """ """
        return self.group.super_area

    @property
    def regional_compliance(self):
        """ """
        return self.group.super_area.region.regional_compliance

    @property
    def has_susceptible(self):
        """ """
        return bool(self.susceptibles_per_subgroup)

    @property
    def has_infectors(self):
        """ """
        return bool(self.infectors_per_infection_per_subgroup)

has_infectors property

has_susceptible property

regional_compliance property

spec property

super_area property

__init__(group, people_from_abroad=None)

This function is very long to avoid function calls for performance reasons. InteractiveGroups are created millions of times. Given a group, we need to extract: - ids of the people that can infect (infector). - ids of the people that can be infected (susceptible). - probabilities of transmission of the infectors. - susceptibilities of the susceptible. - indices of the subgroups that contain infectors. - sizes of the subgroups that contain infectors. - indices of the subgroups that contain susceptible. - spec of the group - super area of the group (for geo attributes like local regional compliances)

Parameters:

Name Type Description Default
group Group

the group to make interact.

 required
people_from_abroad bool

(Default = None).

 None
Source code in june/groups/group/interactive.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def __init__(self, group: "Group", people_from_abroad=None):
    """
    This function is very long to avoid function calls for performance reasons.
    InteractiveGroups are created millions of times. Given a group, we need to extract:
    - ids of the people that can infect (infector).
    - ids of the people that can be infected (susceptible).
    - probabilities of transmission of the infectors.
    - susceptibilities of the susceptible.
    - indices of the subgroups that contain infectors.
    - sizes of the subgroups that contain infectors.
    - indices of the subgroups that contain susceptible.
    - spec of the group
    - super area of the group (for geo attributes like local regional compliances)

    Args:
        group ("Group"): the group to make interact.
        people_from_abroad (bool, optional): (Default = None).

    """
    people_from_abroad = people_from_abroad or {}
    self.group = group
    self.infectors_per_infection_per_subgroup = defaultdict(
        lambda: defaultdict(lambda: defaultdict(list))
    )  # maps virus variant -> subgroup -> infectors -> {infector ids, transmission probs}
    self.susceptibles_per_subgroup = defaultdict(
        dict
    )  # maps subgroup -> susceptible id -> {variant -> susceptibility}
    self.subgroup_sizes = {}
    group_size = 0

    for subgroup_index, subgroup in enumerate(group.subgroups):
        subgroup_size = len(subgroup.people)
        if subgroup.subgroup_type in people_from_abroad:
            people_abroad_data = people_from_abroad[subgroup.subgroup_type]
            people_abroad_ids = people_abroad_data.keys()
            subgroup_size += len(people_abroad_ids)
        else:
            people_abroad_data = None
            people_abroad_ids = []
        if subgroup_size == 0:
            continue
        self.subgroup_sizes[subgroup_index] = subgroup_size
        group_size += subgroup_size

        # Get susceptible people
        # local
        for person in subgroup:
            if not person.infected:
                self.susceptibles_per_subgroup[subgroup_index][
                    person.id
                ] = person.immunity.susceptibility_dict
        # from abroad
        for id in people_abroad_ids:
            if people_abroad_data[id]["susc"]:
                dd = {
                    key: value
                    for key, value in zip(
                        people_abroad_data[id]["immunity_inf_ids"],
                        people_abroad_data[id]["immunity_suscs"],
                    )
                }
                self.susceptibles_per_subgroup[subgroup_index][id] = dd

        # Get infectors
        for person in subgroup:
            if person.infection is not None:
                infection_id = person.infection.infection_id()
                self.infectors_per_infection_per_subgroup[infection_id][
                    subgroup_index
                ]["ids"].append(person.id)
                self.infectors_per_infection_per_subgroup[infection_id][
                    subgroup_index
                ]["trans_probs"].append(person.infection.transmission.probability)
        for id in people_abroad_ids:
            if people_abroad_data[id]["inf_id"] != 0:
                infection_id = people_abroad_data[id]["inf_id"]
                self.infectors_per_infection_per_subgroup[infection_id][
                    subgroup_index
                ]["ids"].append(id)
                self.infectors_per_infection_per_subgroup[infection_id][
                    subgroup_index
                ]["trans_probs"].append(people_abroad_data[id]["inf_prob"])
    self.must_timestep = self.has_susceptible and self.has_infectors
    self.size = group_size

get_processed_beta(betas, beta_reductions)

Returns the processed contact intensity, by taking into account the policies beta reductions and regional compliance. This is a group method as different interactive groups may choose to treat this differently.

Parameters:

Name Type Description Default
betas
required
beta_reductions
required

Returns:

Type Description

Modified betas
Source code in june/groups/group/interactive.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def get_processed_beta(self, betas, beta_reductions):
    """Returns the processed contact intensity, by taking into account the policies
    beta reductions and regional compliance. This is a group method as different interactive
    groups may choose to treat this differently.

    Args:
        betas: 
        beta_reductions:

    Returns:
        Modified betas

    """
    beta = betas[self.spec]
    beta_reduction = beta_reductions.get(self.spec, 1.0)
    try:
        regional_compliance = self.super_area.region.regional_compliance
    except AttributeError:
        regional_compliance = 1
    try:
        lockdown_tier = self.super_area.region.policy["lockdown_tier"]
        if lockdown_tier is None:
            lockdown_tier = 1
    except Exception:
        lockdown_tier = 1
    if int(lockdown_tier) == 4:
        tier_reduction = 0.5
    else:
        tier_reduction = 1.0

    return beta * (1 + regional_compliance * tier_reduction * (beta_reduction - 1))

get_processed_contact_matrix(contact_matrix)

Source code in june/groups/group/interactive.py 
178
179
180
def get_processed_contact_matrix(self, contact_matrix):
    """ """
    return contact_matrix

get_raw_contact_matrix(contact_matrix, alpha_physical, proportion_physical, characteristic_time) classmethod

Returns the processed contact matrix.

By default it returns the input, but children of this class will interact differently.

Parameters:

Name Type Description Default
contact_matrix
required
alpha_physical
required
proportion_physical
required
characteristic_time
required

Returns:

Name Type Description
processed_contact_matrix NDArray

contact matrix after adjustment for if the contact is physical, and rescaled for the time of contact.
Source code in june/groups/group/interactive.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
@classmethod
def get_raw_contact_matrix(
    cls, contact_matrix, alpha_physical, proportion_physical, characteristic_time
):
    """Returns the processed contact matrix.

    By default it returns the input, but children of this class will interact differently.

    Args:
        contact_matrix:
        alpha_physical: 
        proportion_physical: 
        characteristic_time:

    Returns:
        processed_contact_matrix (npt.NDArray):
            contact matrix after adjustment for if the contact is physical,
            and rescaled for the time of contact. 

    """
    processed_contact_matrix = contact_matrix * (
        1.0 + (alpha_physical - 1.0) * proportion_physical
    )
    processed_contact_matrix *= 24 / characteristic_time
    return processed_contact_matrix