MultiplePopulations and dynamic occupancy in _ConcentrationModelBase

Author: mheie

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

@@ -842,6 +842,9 @@ def people_present(self, time: float):
             return self.number * self.person_present(time)
         else:
             return int(self.number.value(time))
+        
+    def transition_times(self):
+        return self.presence_interval().transition_times()
 
 
 @dataclass(frozen=True)
@@ -989,6 +992,22 @@ def particle(self) -> Particle:
         The Particle object representing the aerosol - here the default one
         """
         return self.expiration.particle
+    
+@dataclass(frozen=True)
+class MultiplePopulations:
+    """
+    Store the information of multiple populations to be used for dynamic emitters/infected.
+    """
+    populations: list[SimplePopulation]
+
+    def people_present(self, time: float):
+        return np.sum([population.people_present(time) for population in self.populations])
+    
+    def transition_times(self):
+        state_change_times = {0.}
+        for population in self.populations:
+            state_change_times.update(population.transition_times())
+        return sorted(state_change_times)
 
 
 @dataclass(frozen=True)
@@ -1029,7 +1048,7 @@ class _ConcentrationModelBase:
     ventilation: _VentilationBase
 
     @property
-    def population(self) -> SimplePopulation:
+    def population(self) -> typing.Union[MultiplePopulations, SimplePopulation]:
         """
         Population in the room (the emitters of what we compute the
         concentration of)
@@ -1089,7 +1108,7 @@ def state_change_times(self) -> typing.List[float]:
         the times at which their state changes.
         """
         state_change_times = {0.}
-        state_change_times.update(self.population.presence_interval().transition_times())
+        state_change_times.update(self.population.transition_times())
         state_change_times.update(self.ventilation.transition_times(self.room))
         return sorted(state_change_times)
 
@@ -1098,7 +1117,10 @@ def _first_presence_time(self) -> float:
         """
         First presence time. Before that, the concentration is zero.
         """
-        return self.population.presence_interval().boundaries()[0][0]
+        if isinstance(self.population, MultiplePopulations):
+            return np.min([simple_population.presence_interval().boundaries()[0][0] for simple_population in self.population.populations])
+        else:
+            return self.population.presence_interval().boundaries()[0][0]
 
     def last_state_change(self, time: float) -> float:
         """

caimira/tests/dynamic_emitters/test_dynamic_concentration_model.py

@@ -0,0 +1,100 @@
+import re
+
+import numpy as np
+import numpy.testing as npt
+import pytest
+from dataclasses import dataclass
+
+from caimira.calculator.models import models
+from caimira.calculator.store.data_registry import DataRegistry
+
+known_min_background_concentration = 440.44
+
+@dataclass(frozen=True)
+class KnownConcentrationModelBase(models._ConcentrationModelBase):
+    """
+    A _ConcentrationModelBase class where all the class methods are
+    redefined with a value taken from new parameters. Useful for testing.
+
+    """
+    known_multiple_populations: models.MultiplePopulations
+
+    known_removal_rate: float
+
+    known_min_background_concentration: float
+
+    known_normalization_factor: float
+
+    @property
+    def population(self) -> models.MultiplePopulations:
+        return self.known_multiple_populations
+
+    def removal_rate(self, time: float) -> float:
+        return self.known_removal_rate
+
+    def min_background_concentration(self) -> float:
+        return self.known_min_background_concentration
+
+    def normalization_factor(self) -> float:
+        return self.known_normalization_factor
+
+    
+@pytest.fixture
+def dummy_multiple_populations() -> models.MultiplePopulations:
+    interesting_times1 = models.SpecificInterval(([0.5, 1.], [1.1, 2], [2., 3.]), )
+    population1 = models.Population(
+        number=1,
+        presence=interesting_times1,
+        mask=models.Mask.types['No mask'],
+        activity=models.Activity.types['Seated'],
+        host_immunity=0.,
+    )
+
+    interesting_times2 = models.SpecificInterval(([0.4, 1.], [1.1, 2]), )
+    population2 = models.Population(
+        number=10,
+        presence=interesting_times2,
+        mask=models.Mask.types['Cloth'],
+        activity=models.Activity.types['Heavy exercise'],
+        host_immunity=0.,
+    )
+
+    interesting_times3 = models.SpecificInterval(([5., 6.],), )
+    population3 = models.Population(
+        number=5,
+        presence=interesting_times3,
+        mask=models.Mask.types['Type I'],
+        activity=models.Activity.types['Light activity'],
+        host_immunity=0.5,
+    )
+    return models.MultiplePopulations([population1, population2, population3])
+
+@pytest.fixture
+def simple_dynamic_conc_model(data_registry, dummy_multiple_populations):
+    interesting_times = models.SpecificInterval(([0.5, 1.], [1.1, 2], [2., 3.]), )
+    return KnownConcentrationModelBase(
+        data_registry,
+        room = models.Room(75, models.PiecewiseConstant((0., 24.), (293,))),
+        ventilation = models.AirChange(interesting_times, 100),
+        known_multiple_populations = dummy_multiple_populations,
+        known_removal_rate = 10.,
+        known_min_background_concentration = known_min_background_concentration,
+        known_normalization_factor = 10.)
+
+def test_multiple_populations(simple_dynamic_conc_model):
+    assert isinstance(simple_dynamic_conc_model.population, models.MultiplePopulations)
+    assert np.all([isinstance(population, models.SimplePopulation) for population in simple_dynamic_conc_model.population.populations])
+
+def test_state_change_times(simple_dynamic_conc_model):
+    expected_transition_times = sorted(set((0.0, 0.4, 0.5, 1., 1.1, 2, 3., 5, 6)))
+    result = simple_dynamic_conc_model.state_change_times()
+    npt.assert_array_equal(result, expected_transition_times)
+
+def test_first_presence_time(simple_dynamic_conc_model):
+    npt.assert_array_equal(simple_dynamic_conc_model._first_presence_time(), 0.4)
+
+def test_concentration(simple_dynamic_conc_model):
+    assert isinstance(simple_dynamic_conc_model.concentration(0), float)
+    npt.assert_almost_equal(simple_dynamic_conc_model.concentration(0),known_min_background_concentration)
+    assert simple_dynamic_conc_model.concentration(0.5) > known_min_background_concentration
+

caimira/tests/dynamic_emitters/test_multiple_population.py

@@ -0,0 +1,70 @@
+import re
+
+import numpy as np
+import numpy.testing as npt
+import pytest
+from dataclasses import dataclass
+
+from caimira.calculator.models import models
+from caimira.calculator.store.data_registry import DataRegistry
+
+@pytest.fixture
+def dummy_multiple_populations() -> models.MultiplePopulations:
+    interesting_times1 = models.SpecificInterval(([0.5, 1.], [1.1, 2], [2., 3.]), )
+    population1 = models.Population(
+        number=1,
+        presence=interesting_times1,
+        mask=models.Mask.types['No mask'],
+        activity=models.Activity.types['Seated'],
+        host_immunity=0.,
+    )
+
+    interesting_times2 = models.SpecificInterval(([0.4, 1.], [1.1, 2]), )
+    population2 = models.Population(
+        number=10,
+        presence=interesting_times2,
+        mask=models.Mask.types['Cloth'],
+        activity=models.Activity.types['Heavy exercise'],
+        host_immunity=0.,
+    )
+
+    interesting_times3 = models.SpecificInterval(([5., 6.],), )
+    population3 = models.Population(
+        number=5,
+        presence=interesting_times3,
+        mask=models.Mask.types['Type I'],
+        activity=models.Activity.types['Light activity'],
+        host_immunity=0.5,
+    )
+    return models.MultiplePopulations([population1, population2, population3])
+
+
+def test_multiple_populations(dummy_multiple_populations):
+    assert isinstance(dummy_multiple_populations, models.MultiplePopulations)
+    assert isinstance(dummy_multiple_populations.populations, list)
+    assert np.all([isinstance(population, models.SimplePopulation) for population in dummy_multiple_populations.populations])
+
+
+@pytest.mark.parametrize(
+    "time, expected_people_present", [
+        [0.5, 10],  # Out of range goes to the first state.
+        [1., 11],
+        [1.1, 0],
+        [2., 11],
+        [2, 11],
+        [2.5, 1],
+        [5., 0],
+        [6., 5],
+        [4., 0],
+        [7., 0],
+        [0.1, 0],
+    ]
+)
+def test_people_present(time, expected_people_present, dummy_multiple_populations):
+    result = dummy_multiple_populations.people_present(time)
+    npt.assert_array_equal(result, expected_people_present)
+
+def test_transition_times(dummy_multiple_populations):
+    expected_transition_times = sorted(set((0.0, 0.4, 0.5, 1., 1.1, 2, 3., 5, 6)))
+    result = dummy_multiple_populations.transition_times()
+    npt.assert_array_equal(result, expected_transition_times)