Merge branch 'feature/dynamic_groups' into 'master'

Dynamic groups in backend

See merge request caimira/caimira!510

Author: andrejhenriques

CHANGELOG.md

@@ -1,3 +1,19 @@
+# 4.18.0 (July 24, 2025)
+
+## Feature Added
+- Dynamic occupancy: groups of exposed + infected may now be defined.
+Model and calculator updated, and tests added. Documentation still to
+be updated. User interface mostly unchanged (the new feature is not
+visible there).
+
+## Bug Fixes
+- Fix in profiler to adapt to a non back-compatible change in pyinstrument
+package.
+- Type ignored in the expert app.
+
+## Other
+- Update of mypy, pytest-mypy and pyinstrument dependencies (version).
+
 # 4.17.8 (March 13, 2025)
 
 ## Bug Fixes

caimira/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "caimira"
-version = "4.17.8"
+version = "4.18.0"
 description = "CAiMIRA - CERN Airborne Model for Indoor Risk Assessment"
 readme = "README.md"
 license = { text = "Apache-2.0" }
@@ -23,7 +23,7 @@ dependencies = [
     "mistune",
     "numpy",
     "pandas",
-    "pyinstrument",
+    "pyinstrument >= 5.0.3",
     "python-dateutil",
     "requests",
     "retry",
@@ -39,8 +39,8 @@ dependencies = [
 dev = []
 test = [
     "pytest",
-    "pytest-mypy >= 0.10.3",
-    "mypy >= 1.0.0",
+    "pytest-mypy >= 1.0.1",
+    "mypy >= 1.17.0",
     "pytest-tornasync",
     "types-dataclasses",
     "types-python-dateutil",

caimira/src/caimira/api/controller/virus_report_controller.py

@@ -31,6 +31,9 @@ def submit_virus_form(form_data: typing.Dict, report_generation_parallelism: typ
     report_data: typing.Dict = generate_report(form_obj=form_obj, report_generation_parallelism=report_generation_parallelism)
 
     # Handle model representation
-    if report_data['model']: report_data['model'] = repr(report_data['model'])
+    if report_data['model']:
+        report_data['model'] = repr(report_data['model'])
+    for single_group_output in report_data['groups'].values():
+        del single_group_output['model'] # Model representation per group not needed
 
     return report_data

caimira/src/caimira/calculator/models/models.py

@@ -666,7 +666,7 @@ def fraction_deposited(self, evaporation_factor: float=0.3) -> _VectorisedFloat:
             # deposition fraction depends on aerosol particle diameter.
             d = (self.diameter * evaporation_factor)
             IFrac = 1 - 0.5 * (1 - (1 / (1 + (0.00076*(d**2.8)))))
-            fdep = IFrac * (0.0587
+            fdep = IFrac * (0.0587 # type: ignore
                     + (0.911/(1 + np.exp(4.77 + 1.485 * np.log(d))))
                     + (0.943/(1 + np.exp(0.508 - 2.58 * np.log(d))))) # type: ignore
         return fdep
@@ -711,6 +711,9 @@ class Expiration(_ExpirationBase):
     # to c_n,i in Eq. (4) of https://doi.org/10.1101/2021.10.14.21264988)
     cn: float = 1.
 
+    #: Expiration name
+    name: typing.Optional[str] = None
+
     @property
     def particle(self) -> Particle:
         """
@@ -799,7 +802,7 @@ class Activity:
 @dataclass(frozen=True)
 class SimplePopulation:
     """
-    Represents a group of people all with exactly the same behaviour and
+    Represents a group of people all with exactly the same behavior and
     situation.
 
     """
@@ -844,7 +847,7 @@ def people_present(self, time: float):
 @dataclass(frozen=True)
 class Population(SimplePopulation):
     """
-    Represents a group of people all with exactly the same behaviour and
+    Represents a group of people all with exactly the same behavior and
     situation, considering the usage of mask and a certain host immunity.
 
     """
@@ -1313,7 +1316,7 @@ class ShortRangeModel:
     data_registry: DataRegistry
 
     #: Expiration type
-    expiration: _ExpirationBase
+    expiration: Expiration
 
     #: Activity type
     activity: Activity
@@ -1639,6 +1642,9 @@ class ExposureModel:
     #: Total people with short-range interactions
     exposed_to_short_range: int = 0
 
+    #: Unique group identifier
+    identifier: str = 'group_1'
+
     #: The number of times the exposure event is repeated (default 1).
     @property
     def repeats(self) -> int:
@@ -1653,6 +1659,9 @@ def __post_init__(self):
         In other words, the air exchange rate from the
         ventilation, and the virus decay constant, must
         not be given as arrays.
+
+        It also checks that the number of exposed is
+        static during the simulation time.
         """
         c_model = self.concentration_model
         # Check if the diameter is vectorised.
@@ -1663,6 +1672,11 @@ def __post_init__(self):
                 c_model.ventilation.air_exchange(c_model.room, time)) for time in c_model.state_change_times()))):
             raise ValueError("If the diameter is an array, none of the ventilation parameters "
                              "or virus decay constant can be arrays at the same time.")
+        
+        # Check if exposed population is static
+        if not isinstance(self.exposed.number, int) or not isinstance(self.exposed.presence, Interval):
+            raise TypeError("The exposed number must be an int and presence an Interval. "
+                            f"Got {type(self.exposed.number)} and {type(self.exposed.presence)}.")
 
     @method_cache
     def population_state_change_times(self) -> typing.List[float]:
@@ -1809,11 +1823,9 @@ def _deposited_exposure_list(self):
         The number of virus per m^3 deposited on the respiratory tract.
         """
         population_change_times = self.population_state_change_times()
-
         deposited_exposure = []
         for start, stop in zip(population_change_times[:-1], population_change_times[1:]):
             deposited_exposure.append(self.deposited_exposure_between_bounds(start, stop))
-
         return deposited_exposure
 
     def deposited_exposure(self) -> _VectorisedFloat:
@@ -1838,18 +1850,17 @@ def infection_probability(self) -> _VectorisedFloat:
         return (1 - np.prod([1 - prob for prob in self._infection_probability_list()], axis = 0)) * 100
 
     def total_probability_rule(self) -> _VectorisedFloat:
-        if (isinstance(self.concentration_model.infected.number, IntPiecewiseConstant) or
-                isinstance(self.exposed.number, IntPiecewiseConstant)):
+        if (isinstance(self.concentration_model.infected.number, IntPiecewiseConstant)):
                 raise NotImplementedError("Cannot compute total probability "
                         "(including incidence rate) with dynamic occupancy")
 
         if (self.geographical_data.geographic_population != 0 and self.geographical_data.geographic_cases != 0):
             sum_probability = 0.0
 
             # Create an equivalent exposure model but changing the number of infected cases.
-            total_people = self.concentration_model.infected.number + self.exposed.number
+            total_people = self.concentration_model.infected.number + self.exposed.number # type: ignore
             max_num_infected = (total_people if total_people < 10 else 10)
-            # The influence of a higher number of simultainious infected people (> 4 - 5) yields an almost negligible contirbution to the total probability.
+            # The influence of a higher number of simultaneous infected people (> 4 - 5) yields an almost negligible contribution to the total probability.
             # To be on the safe side, a hard coded limit with a safety margin of 2x was set.
             # Therefore we decided a hard limit of 10 infected people.
             for num_infected in range(1, max_num_infected + 1):
@@ -1872,43 +1883,81 @@ def expected_new_cases(self) -> _VectorisedFloat:
             1) Long-range exposure: take the infection_probability and multiply by the occupants exposed to long-range. 
             2) Short- and long-range exposure: take the infection_probability of long-range multiplied by the occupants exposed to long-range only, 
                plus the infection_probability of short- and long-range multiplied by the occupants exposed to short-range only.
-
-        Currently disabled when dynamic occupancy is defined for the exposed population.
         """
-
-        if (isinstance(self.concentration_model.infected.number, IntPiecewiseConstant) or
-            isinstance(self.exposed.number, IntPiecewiseConstant)):
-            raise NotImplementedError("Cannot compute expected new cases "
-                    "with dynamic occupancy")
-
+        number = self.exposed.number
         if self.short_range != ():
-            new_cases_long_range = nested_replace(self, {'short_range': [],}).infection_probability() * (self.exposed.number - self.exposed_to_short_range)
+            new_cases_long_range = nested_replace(self, {'short_range': [],}).infection_probability() * (number - self.exposed_to_short_range) # type: ignore
             return (new_cases_long_range + (self.infection_probability() * self.exposed_to_short_range)) / 100
 
-        return self.infection_probability() * self.exposed.number / 100
+        return self.infection_probability() * number / 100
 
     def reproduction_number(self) -> _VectorisedFloat:
         """
         The reproduction number can be thought of as the expected number of
         cases directly generated by one infected case in a population.
-
-        Currently disabled when dynamic occupancy is defined for both the infected and exposed population.
         """
-
-        if (isinstance(self.concentration_model.infected.number, IntPiecewiseConstant) or
-            isinstance(self.exposed.number, IntPiecewiseConstant)):
-            raise NotImplementedError("Cannot compute reproduction number "
-                    "with dynamic occupancy")
-
-        if self.concentration_model.infected.number == 1:
+        infected_population: InfectedPopulation = self.concentration_model.infected
+        if isinstance(infected_population.number, int) and infected_population.number == 1:
             return self.expected_new_cases()
 
         # Create an equivalent exposure model but with precisely
-        # one infected case.
+        # one infected case, respecting the presence interval.
         single_exposure_model = nested_replace(
             self, {
-                'concentration_model.infected.number': 1}
+                'concentration_model.infected.number': 1,
+                'concentration_model.infected.presence': infected_population.presence_interval(),
+            }
         )
-
         return single_exposure_model.expected_new_cases()
-    
+    
+
+@dataclass(frozen=True)
+class ExposureModelGroup:
+    """
+    Represents a group of exposure models. This is to handle the case
+    when different groups of people come and go in the room at different
+    times. These groups are then handled fully independently, with
+    exposure dose and probability of infection defined for each of them.
+    """
+    data_registry: DataRegistry
+
+    #: The set of exposure models for each exposed population
+    exposure_models: typing.Tuple[ExposureModel, ...]
+
+    def __post_init__(self):
+        """
+        Validate that all ExposureModels have the same ConcentrationModel.
+        """
+        first_concentration_model = self.exposure_models[0].concentration_model
+        for model in self.exposure_models[1:]:
+            # Check that the number of infected people and their presence is the same
+            if (model.concentration_model.infected.number != first_concentration_model.infected.number or
+                model.concentration_model.infected.presence != first_concentration_model.infected.presence):
+                raise ValueError("All ExposureModels must have the same infected number and presence in the ConcentrationModel.")
+
+    @method_cache
+    def _deposited_exposure_list(self) -> typing.List[_VectorisedFloat]:
+        """
+        List of doses absorbed by each member of the groups.
+        """
+        return [model.deposited_exposure() for model in self.exposure_models]
+    
+    @method_cache
+    def _infection_probability_list(self):
+        """
+        List of the probability of infection for each group.
+        """
+        return [model.infection_probability() for model in self.exposure_models] # type: ignore
+
+    def expected_new_cases(self) -> _VectorisedFloat:
+        """
+        Final expected number of new cases considering the
+        contribution of each individual probability of infection.
+        """
+        return np.sum([model.expected_new_cases() for model in self.exposure_models], axis=0) # type: ignore
+    
+    def reproduction_number(self) -> _VectorisedFloat:
+        """
+        Expected number of cases when there is only one infected case.
+        """
+        return np.sum([model.reproduction_number() for model in self.exposure_models], axis=0) # type: ignore

caimira/src/caimira/calculator/models/monte_carlo/data.py

@@ -365,6 +365,7 @@ def expiration_distribution(
         BLO_factors,
         d_min=0.1,
         d_max=30.,
+        exp_type=None,
 ):
     """
     Returns an Expiration with an aerosol diameter distribution, defined
@@ -382,6 +383,7 @@ def expiration_distribution(
             kernel_bandwidth=0.1,
         ),
         cn=BLOmodel(data_registry, BLO_factors).integrate(d_min, d_max),
+        name=exp_type,
     )
 
 
@@ -432,7 +434,8 @@ def short_range_expiration_distributions(data_registry):
             data_registry=data_registry,
             BLO_factors=BLO_factors,
             d_min=param_evaluation(data_registry.expiration_particle['particle_size_range']['short_range'], 'minimum_diameter'),
-            d_max=param_evaluation(data_registry.expiration_particle['particle_size_range']['short_range'], 'maximum_diameter')
+            d_max=param_evaluation(data_registry.expiration_particle['particle_size_range']['short_range'], 'maximum_diameter'),
+            exp_type=exp_type,
         )
         for exp_type, BLO_factors in expiration_BLO_factors(data_registry).items()
     }

caimira/src/caimira/calculator/models/monte_carlo/models.py

@@ -74,7 +74,6 @@ def _build_mc_model(model: dataclass_instance) -> typing.Type[MCModelBase[_Model
             elif new_field.type == typing.Tuple[models.SpecificInterval, ...]:
                 SI = getattr(sys.modules[__name__], "SpecificInterval")
                 field_type = typing.Tuple[typing.Union[models.SpecificInterval, SI], ...]
-
             elif new_field.type == typing.Union[int, models.IntPiecewiseConstant]:
                 IPC = getattr(sys.modules[__name__], "IntPiecewiseConstant")
                 field_type = typing.Union[int, models.IntPiecewiseConstant, IPC]

caimira/src/caimira/calculator/models/profiler.py

@@ -31,7 +31,7 @@ def from_str(value):
 
 
 class PyInstrumentWrapper:
-    profiler = PyInstrumentProfiler(async_mode=True)
+    profiler = PyInstrumentProfiler(async_mode='enabled')
 
     @property
     def is_running(self):