Skip to content

Infection selector

InfectionSelector

Source code in june/epidemiology/infection/infection_selector.py 
 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
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
class InfectionSelector:
    """ """
    def __init__(
        self,
        disease_config: DiseaseConfig,
        infection_class: Optional[Infection] = None,
        trajectory_maker: Optional[TrajectoryMakers] = None,
        health_index_generator: Optional[HealthIndexGenerator] = None,
    ):
        """
        Selects the type of infection a person is given.

        Parameters
        ----------
        disease_config : DiseaseConfig
            Configuration object for the disease.
        infection_class : Infection, optional
            Infection class for the disease (e.g., Measles, Covid19).
        trajectory_maker : TrajectoryMakers, optional
            Object to manage symptom trajectories.
        health_index_generator : HealthIndexGenerator, optional
            Generator for health index based on infection outcomes.
        """
        self.disease_config = disease_config
        self.disease_name = disease_config.disease_name
        self.infection_class = infection_class or self._resolve_infection_class(self.disease_name)
        self.trajectory_maker = trajectory_maker or TrajectoryMakers.from_disease_config(self.disease_config)

        # Retrieve the rates file from DiseaseConfig
        self.health_index_generator = health_index_generator or HealthIndexGenerator.from_disease_config(
            disease_config=disease_config
        )

        self.transmission_type = None  # Will be initialised in `_load_transmission`
        self._load_transmission()


    @classmethod
    def from_disease_config(cls, disease_config: DiseaseConfig) -> "InfectionSelector":
        """Generate infection selector from default config file.

        Args:
            disease_config (DiseaseConfig): 

        Returns:
            InfectionSelector: An instance of the infection selector.

        """
        # Dynamically select the infection class based on the disease name
        infection_class = disease_to_infection_class.get(disease_config.disease_name.lower())

        # Initialise the InfectionSelector with all required components
        return cls(
            disease_config=disease_config,
            infection_class=infection_class,
            trajectory_maker=TrajectoryMakers.from_disease_config(disease_config),
            health_index_generator=HealthIndexGenerator.from_disease_config(
                disease_config=disease_config
            )
        )

    @staticmethod
    def _get_rates_file_path(disease_name: str) -> str:
        """Construct the rates file path for the given disease.

        Args:
            disease_name (str): Name of the disease.

        Returns:
            str: Path to the rates file.

        """
        return paths.data_path / f"input/health_index/infection_outcome_rates_{disease_name.lower()}.csv"

    def _resolve_infection_class(self, disease_name: str) -> Infection:
        """Resolves the infection class based on the disease name.

        Args:
            disease_name (str): Name of the disease.

        Returns:
            Infection: Infection class for the disease.

        """
        infection_class = disease_to_infection_class.get(disease_name.lower())
        if not infection_class:
            raise ValueError(f"No infection class defined for disease '{disease_name}'.")
        return infection_class

    def _load_transmission(self):
        """Load transmission config from the disease configuration."""
        transmission_config = self.disease_config.disease_yaml.get("disease", {}).get("transmission", {})
        self.transmission_type = transmission_config.get("type")

        if self.transmission_type == "xnexp":
            self._load_transmission_xnexp(transmission_config)
        elif self.transmission_type == "gamma":
            self._load_transmission_gamma(transmission_config)
        elif self.transmission_type == "constant":
            self._load_transmission_constant(transmission_config)
        else:
            raise NotImplementedError(f"Transmission type '{self.transmission_type}' is not implemented.")

    def infect_person_at_time(self, person: "Person", time: float):
        """Infects a person at a given time.

        Args:
            person ("Person"): The person to be infected.
            time (float): The time at which the infection occurs.

        """
        # Create and assign infection
        person.infection = self._make_infection(person, time)

        # Update immunity
        immunity_ids = person.infection.immunity_ids()
        person.immunity.add_immunity(immunity_ids)

    def _make_infection(self, person: "Person", time: float):
        """Generate symptoms and infectiousness of the infected person.

        Args:
            person ("Person"): The person to be infected.
            time (float): Time at which infection happens.

        Returns:
            Infection: 

        """
        symptoms = self._select_symptoms(person)
        time_to_symptoms_onset = symptoms.time_exposed
        transmission = self._select_transmission(
            time_to_symptoms_onset=time_to_symptoms_onset,
            max_symptoms_tag=symptoms.max_tag,
        )
        return self.infection_class(
            transmission=transmission, symptoms=symptoms, start_time=time
        )

    def _select_symptoms(self, person: "Person") -> "Symptoms":
        """Select symptoms and their evolution for an infected person.

        Args:
            person ("Person"): The person to be infected.

        Returns:
            Symptoms: 

        """
        health_index = self.health_index_generator(person, infection_id=self.infection_id)
        symptoms = Symptoms(disease_config=self.disease_config, health_index=health_index)
        return symptoms

    @property
    def infection_id(self):
        """Retrieve the infection ID from the infection class.


        Returns:
            str: 

        """
        return self.infection_class.infection_id()

    def _load_transmission_xnexp(self, transmission_config: dict):
        """Load parameters for transmission of type `xnexp`.

        Args:
            transmission_config (dict): Dictionary containing configuration for the `xnexp` transmission type.

        """
        self.smearing_time_first_infectious = CompletionTime.from_dict(
            transmission_config["smearing_time_first_infectious"]
        )
        self.smearing_peak_position = CompletionTime.from_dict(
            transmission_config["smearing_peak_position"]
        )
        self.alpha = CompletionTime.from_dict(transmission_config["alpha"])
        self.max_probability = CompletionTime.from_dict(
            transmission_config["max_probability"]
        )
        self.norm_time = CompletionTime.from_dict(transmission_config["norm_time"])
        self.asymptomatic_infectious_factor = CompletionTime.from_dict(
            transmission_config["asymptomatic_infectious_factor"]
        )
        self.mild_infectious_factor = CompletionTime.from_dict(
            transmission_config["mild_infectious_factor"]
        )


    def _load_transmission_gamma(self, transmission_config: dict):
        """Load parameters for transmission of type `gamma`.

        Args:
            transmission_config (dict): Dictionary containing configuration for the `gamma` transmission type.

        """
        self.max_infectiousness = CompletionTime.from_dict(
            transmission_config["max_infectiousness"]
        )
        self.shape = CompletionTime.from_dict(transmission_config["shape"])
        self.rate = CompletionTime.from_dict(transmission_config["rate"])
        self.shift = CompletionTime.from_dict(transmission_config["shift"])
        self.asymptomatic_infectious_factor = CompletionTime.from_dict(
            transmission_config["asymptomatic_infectious_factor"]
        )
        self.mild_infectious_factor = CompletionTime.from_dict(
            transmission_config["mild_infectious_factor"]
        )


    def _load_transmission_constant(self, transmission_config: dict):
        """Load parameters for transmission of type `constant`.

        Args:
            transmission_config (dict): Dictionary containing configuration for the `constant` transmission type.

        """
        self.probability = CompletionTime.from_dict(transmission_config["probability"])


    def _select_transmission(
        self, time_to_symptoms_onset: float, max_symptoms_tag: "SymptomTag"
    ) -> "Transmission":
        """Selects the transmission type specified by the user in the configuration,
        and links its parameters to the symptom onset for the person (incubation
        period).

        Args:
            time_to_symptoms_onset (float): Time from infection to symptom onset for the person.
            max_symptoms_tag ("SymptomTag"): The maximum severity of symptoms for the person.

        Returns:
            Transmission: A transmission object configured for the specified disease and parameters.

        """
        if self.transmission_type == "xnexp":
            time_first_infectious = (
                self.smearing_time_first_infectious() + time_to_symptoms_onset
            )
            peak_position = (
                time_to_symptoms_onset
                - time_first_infectious
                + self.smearing_peak_position()
            )
            return TransmissionXNExp(
                max_probability=self.max_probability(),
                time_first_infectious=time_first_infectious,
                norm_time=self.norm_time(),
                n=peak_position / self.alpha(),
                alpha=self.alpha(),
                max_symptoms=max_symptoms_tag,
                asymptomatic_infectious_factor=self.asymptomatic_infectious_factor(),
                mild_infectious_factor=self.mild_infectious_factor(),
            )
        elif self.transmission_type == "gamma":
            return TransmissionGamma(
                max_infectiousness=self.max_infectiousness(),
                shape=self.shape(),
                rate=self.rate(),
                shift=self.shift() + time_to_symptoms_onset,
                max_symptoms=max_symptoms_tag,
                asymptomatic_infectious_factor=self.asymptomatic_infectious_factor(),
                mild_infectious_factor=self.mild_infectious_factor(),
            )
        elif self.transmission_type == "constant":
            return TransmissionConstant(probability=self.probability())
        else:
            raise NotImplementedError(f"Transmission type {self.transmission_type} is not implemented.")

infection_id property

Retrieve the infection ID from the infection class.

Returns:

Name Type Description
str

__init__(disease_config, infection_class=None, trajectory_maker=None, health_index_generator=None)

Selects the type of infection a person is given.

Parameters

disease_config : DiseaseConfig Configuration object for the disease. infection_class : Infection, optional Infection class for the disease (e.g., Measles, Covid19). trajectory_maker : TrajectoryMakers, optional Object to manage symptom trajectories. health_index_generator : HealthIndexGenerator, optional Generator for health index based on infection outcomes.

Source code in june/epidemiology/infection/infection_selector.py 
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
def __init__(
    self,
    disease_config: DiseaseConfig,
    infection_class: Optional[Infection] = None,
    trajectory_maker: Optional[TrajectoryMakers] = None,
    health_index_generator: Optional[HealthIndexGenerator] = None,
):
    """
    Selects the type of infection a person is given.

    Parameters
    ----------
    disease_config : DiseaseConfig
        Configuration object for the disease.
    infection_class : Infection, optional
        Infection class for the disease (e.g., Measles, Covid19).
    trajectory_maker : TrajectoryMakers, optional
        Object to manage symptom trajectories.
    health_index_generator : HealthIndexGenerator, optional
        Generator for health index based on infection outcomes.
    """
    self.disease_config = disease_config
    self.disease_name = disease_config.disease_name
    self.infection_class = infection_class or self._resolve_infection_class(self.disease_name)
    self.trajectory_maker = trajectory_maker or TrajectoryMakers.from_disease_config(self.disease_config)

    # Retrieve the rates file from DiseaseConfig
    self.health_index_generator = health_index_generator or HealthIndexGenerator.from_disease_config(
        disease_config=disease_config
    )

    self.transmission_type = None  # Will be initialised in `_load_transmission`
    self._load_transmission()

from_disease_config(disease_config) classmethod

Generate infection selector from default config file.

Parameters:

Name Type Description Default
disease_config DiseaseConfig
required

Returns:

Name Type Description
InfectionSelector InfectionSelector

An instance of the infection selector.
Source code in june/epidemiology/infection/infection_selector.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
@classmethod
def from_disease_config(cls, disease_config: DiseaseConfig) -> "InfectionSelector":
    """Generate infection selector from default config file.

    Args:
        disease_config (DiseaseConfig): 

    Returns:
        InfectionSelector: An instance of the infection selector.

    """
    # Dynamically select the infection class based on the disease name
    infection_class = disease_to_infection_class.get(disease_config.disease_name.lower())

    # Initialise the InfectionSelector with all required components
    return cls(
        disease_config=disease_config,
        infection_class=infection_class,
        trajectory_maker=TrajectoryMakers.from_disease_config(disease_config),
        health_index_generator=HealthIndexGenerator.from_disease_config(
            disease_config=disease_config
        )
    )

infect_person_at_time(person, time)

Infects a person at a given time.

Parameters:

Name Type Description Default
person Person

The person to be infected.

 required
time float

The time at which the infection occurs.

 required
Source code in june/epidemiology/infection/infection_selector.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def infect_person_at_time(self, person: "Person", time: float):
    """Infects a person at a given time.

    Args:
        person ("Person"): The person to be infected.
        time (float): The time at which the infection occurs.

    """
    # Create and assign infection
    person.infection = self._make_infection(person, time)

    # Update immunity
    immunity_ids = person.infection.immunity_ids()
    person.immunity.add_immunity(immunity_ids)

InfectionSelectors

Source code in june/epidemiology/infection/infection_selector.py 
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
class InfectionSelectors:
    """ """
    def __init__(self, infection_selectors: list = None):
        self._infection_selectors = infection_selectors
        self.infection_id_to_selector = self.make_dict()

    def make_dict(self):
        """Makes two dicts:
        infection_type_id -> infection_class (needed for easier MPI comms)
        infection_class -> infection_selector (needed to map infection to
                            the class that creates infections)

        """
        if not self._infection_selectors:
            return {Covid19.infection_id(): InfectionSelector.from_disease_config()}
        ret = {}
        for i, selector in enumerate(self._infection_selectors):
            ret[selector.infection_class.infection_id()] = selector
        return ret

    def infect_person_at_time(
        self, person: "Person", time: float, infection_id: int = Covid19.infection_id()
    ):
        """Infects a person at a given time with the given infection_class.

        Args:
            person ("Person"): person that will be infected
            time (float): time at which infection happens
            infection_id (int, optional): (Default value = Covid19.infection_id())

        """
        selector = self.infection_id_to_selector[infection_id]
        selector.infect_person_at_time(person=person, time=time)

    def __iter__(self):
        return iter(self._infection_selectors)

    def __getitem__(self, item):
        return self._infection_selectors[item]

infect_person_at_time(person, time, infection_id=Covid19.infection_id())

Infects a person at a given time with the given infection_class.

Parameters:

Name Type Description Default
person Person

person that will be infected

 required
time float

time at which infection happens

 required
infection_id int

(Default value = Covid19.infection_id())

 infection_id()
Source code in june/epidemiology/infection/infection_selector.py 
322
323
324
325
326
327
328
329
330
331
332
333
334
def infect_person_at_time(
    self, person: "Person", time: float, infection_id: int = Covid19.infection_id()
):
    """Infects a person at a given time with the given infection_class.

    Args:
        person ("Person"): person that will be infected
        time (float): time at which infection happens
        infection_id (int, optional): (Default value = Covid19.infection_id())

    """
    selector = self.infection_id_to_selector[infection_id]
    selector.infect_person_at_time(person=person, time=time)

make_dict()

Makes two dicts: infection_type_id -> infection_class (needed for easier MPI comms) infection_class -> infection_selector (needed to map infection to the class that creates infections)

Source code in june/epidemiology/infection/infection_selector.py 
308
309
310
311
312
313
314
315
316
317
318
319
320
def make_dict(self):
    """Makes two dicts:
    infection_type_id -> infection_class (needed for easier MPI comms)
    infection_class -> infection_selector (needed to map infection to
                        the class that creates infections)

    """
    if not self._infection_selectors:
        return {Covid19.infection_id(): InfectionSelector.from_disease_config()}
    ret = {}
    for i, selector in enumerate(self._infection_selectors):
        ret[selector.infection_class.infection_id()] = selector
    return ret