added "occupancy_format" input that controls the definition of dynami…

…c activities

caimira/apps/calculator/defaults.py

@@ -51,13 +51,14 @@
     'mask_type': 'Type I',
     'mask_wearing_option': 'mask_off',
     'mechanical_ventilation_type': 'not-applicable',
+    'occupancy_format': 'static',
     'opening_distance': 0.,
     'room_heating_option': False,
     'room_number': NO_DEFAULT,
     'room_volume': 0.,
     'simulation_name': NO_DEFAULT,
     'total_people': NO_DEFAULT,
-    'dynamic_exposed_occupancy': '[]',
+    'dynamic_exposed_occupancy': NO_DEFAULT,
     'vaccine_option': False,
     'vaccine_booster_option': False,
     'vaccine_type': 'AZD1222_(AstraZeneca)',
@@ -74,7 +75,7 @@
     'window_opening_regime': 'windows_open_permanently',
     'sensor_in_use': '',
     'short_range_option': 'short_range_no',
-    'short_range_interactions': '[]',
+    'short_range_interactions': NO_DEFAULT,
     'short_range_occupants': 0,
 }
 

caimira/apps/calculator/form_data.py

@@ -18,7 +18,7 @@
 
 @dataclasses.dataclass
 class FormData:
-    specific_breaks: dict
+    # Static occupancy inputs
     exposed_coffee_break_option: str
     exposed_coffee_duration: int
     exposed_finish: minutes_since_midnight
@@ -33,15 +33,18 @@ class FormData:
     infected_lunch_finish: minutes_since_midnight   #Used if infected_dont_have_breaks_with_exposed
     infected_lunch_option: bool                     #Used if infected_dont_have_breaks_with_exposed
     infected_lunch_start: minutes_since_midnight    #Used if infected_dont_have_breaks_with_exposed
-    infected_people: int
-    dynamic_infected_occupancy: list
     infected_start: minutes_since_midnight
+    infected_people: int
+    occupancy_format: str
     room_volume: float
+    specific_breaks: dict
     total_people: int
+
+    # Dynamic occupancy inputs
     dynamic_exposed_occupancy: list
+    dynamic_infected_occupancy: list
 
     data_registry: DataRegistry
-
     _DEFAULTS: typing.ClassVar[typing.Dict[str, typing.Any]] = DEFAULTS
 
     @classmethod

caimira/apps/calculator/model_generator.py

@@ -185,7 +185,8 @@ def validate(self):
                 raise ValueError(f'The sum of all respiratory activities should be 100. Got {total_percentage}.')
 
         # Validate number of people with short-range interactions
-        max_occupants_for_sr = self.total_people - self.infected_people
+        if self.occupancy_format == "static": max_occupants_for_sr = self.total_people - self.infected_people
+        else: max_occupants_for_sr = np.max(np.array([entry["total_people"] for entry in self.dynamic_exposed_occupancy]))
         if self.short_range_occupants > max_occupants_for_sr:
             raise ValueError(
                 f'The total number of occupants having short-range interactions ({self.short_range_occupants}) should be lower than the exposed population ({max_occupants_for_sr}).'
@@ -467,33 +468,33 @@ def infected_population(self) -> mc.InfectedPopulation:
         # Initializes the virus
         virus = virus_distributions(self.data_registry)[self.virus_type]
 
+        # Occupancy
+        if self.occupancy_format == 'dynamic':
+            if isinstance(self.dynamic_infected_occupancy, typing.List) and len(self.dynamic_infected_occupancy) > 0:
+                # If dynamic occupancy is defined, the generator will parse and validate the
+                # respective input to a format readable by the model - `IntPiecewiseConstant`.
+                infected_occupancy, infected_presence = self.generate_dynamic_occupancy(self.dynamic_infected_occupancy)
+            else:
+                raise TypeError(f'If dynamic occupancy is selected, a populated list of occupancy intervals is expected. Got "{self.dynamic_infected_occupancy}".')
+        else:
+            # The number of exposed occupants is the total number of occupants
+            # minus the number of infected occupants.
+            infected_occupancy = self.infected_people
+            infected_presence = self.infected_present_interval()
+
+        # Activity and expiration
         activity_defn = self.data_registry.population_scenario_activity[self.activity_type]['activity']
         expiration_defn = self.data_registry.population_scenario_activity[self.activity_type]['expiration']
         if (self.activity_type == 'smallmeeting'):
             # Conversation of N people is approximately 1/N% of the time speaking.
-            expiration_defn = {'Speaking': 1, 'Breathing': self.total_people - 1}
+            total_people: int = max(infected_occupancy.values) if self.occupancy_format == 'dynamic' else self.total_people
+            expiration_defn = {'Speaking': 1, 'Breathing': total_people - 1}
         elif (self.activity_type == 'precise'):
-            activity_defn, expiration_defn = self.generate_precise_activity_expiration() # TODO: what to do here?
+            activity_defn, expiration_defn = self.generate_precise_activity_expiration()
 
         activity = activity_distributions(self.data_registry)[activity_defn]
         expiration = build_expiration(self.data_registry, expiration_defn)
 
-        if isinstance(self.dynamic_infected_occupancy, typing.List) and len(self.dynamic_infected_occupancy) > 0:
-            # If dynamic occupancy is defined, the generator will parse and validate the
-            # respective input to a format readable by the model - IntPiecewiseConstant.
-            infected_occupancy, infected_presence = self.generate_dynamic_occupancy(self.dynamic_infected_occupancy)
-            # If exposed population is static, defined from the "total_people" input, validate
-            # if every occurency of infected population is less or equal than it.
-            if isinstance(self.dynamic_exposed_occupancy, typing.List) and len(self.dynamic_exposed_occupancy) == 0:
-                for infected_people in infected_occupancy.values:
-                    if infected_people >= self.total_people:
-                        raise ValueError('Number of infected people cannot be greater or equal to the number of total people.')
-        else:
-            # The number of exposed occupants is the total number of occupants
-            # minus the number of infected occupants.
-            infected_occupancy = self.infected_people
-            infected_presence = self.infected_present_interval()
-
         infected = mc.InfectedPopulation(
             data_registry=self.data_registry,
             number=infected_occupancy,
@@ -511,11 +512,14 @@ def exposed_population(self) -> mc.Population:
                          if self.activity_type == 'precise'
                          else str(self.data_registry.population_scenario_activity[self.activity_type]['activity']))
         activity = activity_distributions(self.data_registry)[activity_defn]
-
-        if isinstance(self.dynamic_exposed_occupancy, typing.List) and len(self.dynamic_exposed_occupancy) > 0:
-            # If dynamic occupancy is defined, the generator will parse and validate the
-            # respective input to a format readable by the model - IntPiecewiseConstant.
-            exposed_occupancy, exposed_presence = self.generate_dynamic_occupancy(self.dynamic_exposed_occupancy)
+
+        if self.occupancy_format == 'dynamic':
+            if isinstance(self.dynamic_exposed_occupancy, typing.List) and len(self.dynamic_exposed_occupancy) > 0:
+                # If dynamic occupancy is defined, the generator will parse and validate the
+                # respective input to a format readable by the model - IntPiecewiseConstant.
+                exposed_occupancy, exposed_presence = self.generate_dynamic_occupancy(self.dynamic_exposed_occupancy)
+            else:
+                raise TypeError(f'If dynamic occupancy is selected, a populated list of occupancy intervals is expected. Got "{self.dynamic_exposed_occupancy}".')
         else:
             # The number of exposed occupants is the total number of occupants
             # minus the number of infected occupants.
@@ -565,9 +569,14 @@ def baseline_raw_form_data() -> typing.Dict[str, typing.Union[str, float]]:
         'activity_type': 'office',
         'air_changes': '',
         'air_supply': '',
+        'ascertainment_bias': 'confidence_low',
+        'calculator_version': calculator.__version__,
         'ceiling_height': '',
         'conditional_probability_plot': '0',
         'conditional_probability_viral_loads': '0',
+        'dynamic_exposed_occupancy': '[]',
+        'dynamic_infected_occupancy': '[]',
+        'event_month': 'January',
         'exposed_coffee_break_option': 'coffee_break_4',
         'exposed_coffee_duration': '10',
         'exposed_finish': '18:00',
@@ -576,6 +585,8 @@ def baseline_raw_form_data() -> typing.Dict[str, typing.Union[str, float]]:
         'exposed_lunch_start': '12:30',
         'exposed_start': '09:00',
         'floor_area': '',
+        'geographic_cases': 0,
+        'geographic_population': 0,
         'hepa_amount': '250',
         'hepa_option': '0',
         'humidity': '0.5',
@@ -587,43 +598,37 @@ def baseline_raw_form_data() -> typing.Dict[str, typing.Union[str, float]]:
         'infected_lunch_option': '1',
         'infected_lunch_start': '12:30',
         'infected_people': '1',
-        'dynamic_infected_occupancy': '[]',
         'infected_start': '09:00',
         'inside_temp': '293.',
         'location_latitude': 46.20833,
         'location_longitude': 6.14275,
         'location_name': 'Geneva',
-        'geographic_population': 0,
-        'geographic_cases': 0,
-        'ascertainment_bias': 'confidence_low',
         'mask_type': 'Type I',
         'mask_wearing_option': 'mask_off',
         'mechanical_ventilation_type': '',
-        'calculator_version': calculator.__version__,
         'opening_distance': '0.2',
-        'event_month': 'January',
+        'occupancy_format': 'static',
         'room_heating_option': '0',
         'room_number': '123',
         'room_volume': '75',
+        'short_range_interactions': '[]',
+        'short_range_option': 'short_range_no',
         'simulation_name': 'Test',
         'total_people': '10',
-        'dynamic_exposed_occupancy': '[]',
-        'vaccine_option': '0',
         'vaccine_booster_option': '0',
-        'vaccine_type': 'Ad26.COV2.S_(Janssen)',
         'vaccine_booster_type': 'AZD1222_(AstraZeneca)',
+        'vaccine_option': '0',
+        'vaccine_type': 'Ad26.COV2.S_(Janssen)',
         'ventilation_type': 'natural_ventilation',
         'virus_type': 'SARS_CoV_2',
         'volume_type': 'room_volume_explicit',
+        'window_height': '2',
+        'window_opening_regime': 'windows_open_permanently',
         'windows_duration': '10',
         'windows_frequency': '60',
-        'window_height': '2',
+        'windows_number': '1',
         'window_type': 'window_sliding',
         'window_width': '2',
-        'windows_number': '1',
-        'window_opening_regime': 'windows_open_permanently',
-        'short_range_option': 'short_range_no',
-        'short_range_interactions': '[]',
     }
 
 cast_class_fields(VirusFormData)

caimira/apps/calculator/report_generator.py

@@ -168,15 +168,16 @@ def calculate_report_data(form: VirusFormData, model: models.ExposureModel, exec
     prob = np.array(model.infection_probability())
     prob_dist_count, prob_dist_bins = np.histogram(prob/100, bins=100, density=True)
 
-    if form.exposure_option == "p_probabilistic_exposure":
+    # Probabilistic exposure
+    if form.exposure_option == "p_probabilistic_exposure" and form.occupancy_format == "static":
         prob_probabilistic_exposure = np.array(model.total_probability_rule()).mean()
-    else: prob_probabilistic_exposure = None
-
-    if ((isinstance(form.dynamic_infected_occupancy, typing.List) and len(form.dynamic_infected_occupancy) > 0) or 
-        (isinstance(form.dynamic_exposed_occupancy, typing.List) and len(form.dynamic_exposed_occupancy) > 0)):
-        expected_new_cases = None
-    else:
+    else: prob_probabilistic_exposure = 0
+    # Expected new cases
+    if (form.occupancy_format == "static"):
         expected_new_cases = np.array(model.expected_new_cases()).mean()
+    else:
+        # With dynamic occupancy, the expected number of new cases feature is disabled.
+        expected_new_cases = -1
 
     exposed_presence_intervals = [list(interval) for interval in model.exposed.presence_interval().boundaries()]
 
@@ -460,6 +461,7 @@ def manufacture_alternative_scenarios(form: VirusFormData) -> typing.Dict[str, m
             scenarios['Neither ventilation nor masks'] = without_mask_or_vent.build_mc_model()
 
     else:
+        # When dynamic occupancy is defined, the replace of total people is useless - the expected number of new cases is not calculated.
         no_short_range_alternative = dataclass_utils.replace(form, short_range_interactions=[], total_people=form.total_people - form.short_range_occupants)
         scenarios['Base scenario without short-range interactions'] = no_short_range_alternative.build_mc_model()
 
@@ -469,18 +471,24 @@ def manufacture_alternative_scenarios(form: VirusFormData) -> typing.Dict[str, m
 def scenario_statistics(
     mc_model: mc.ExposureModel,
     sample_times: typing.List[float],
-    compute_prob_exposure: bool
+    compute_prob_exposure: bool,
+    compute_expected_new_cases: bool,
 ):
     model = mc_model.build_model(size=mc_model.data_registry.monte_carlo['sample_size'])
     if (compute_prob_exposure):
         # It means we have data to calculate the total_probability_rule
         prob_probabilistic_exposure = model.total_probability_rule()
     else:
         prob_probabilistic_exposure = 0.
+
+    if (compute_expected_new_cases):
+        expected_new_cases = np.mean(model.expected_new_cases())
+    else:
+        expected_new_cases = -1
 
     return {
         'probability_of_infection': np.mean(model.infection_probability()),
-        'expected_new_cases': np.mean(model.expected_new_cases()),
+        'expected_new_cases': expected_new_cases,
         'concentrations': [
             np.mean(model.concentration(time))
             for time in sample_times
@@ -507,17 +515,20 @@ def comparison_report(
     else:
         statistics = {}
 
-    if (form.short_range_option == "short_range_yes" and form.exposure_option == "p_probabilistic_exposure"):
+    if (form.short_range_option == "short_range_yes" and form.exposure_option == "p_probabilistic_exposure" and form.occupancy_format == "static"):
         compute_prob_exposure = True
     else:
         compute_prob_exposure = False
 
+    compute_expected_new_cases = True if (form.occupancy_format == "static") else False
+
     with executor_factory() as executor:
         results = executor.map(
             scenario_statistics,
             scenarios.values(),
             [sample_times] * len(scenarios),
             [compute_prob_exposure] * len(scenarios),
+            [compute_expected_new_cases] * len(scenarios),
             timeout=60,
         )
 

caimira/apps/templates/base/calculator.form.html.j2

@@ -447,6 +447,7 @@
         <span class="tooltip_text">?</span>
     </div><br>
 
+    <input type="text" class="form-control d-none" name="occupancy_format" value="static" required> {# "static" vs. "dynamic" #}
     <input type="text" class="form-control d-none" name="dynamic_exposed_occupancy">
     <input type="text" class="form-control d-none" name="dynamic_infected_occupancy">