added short-range expirations per group

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

@@ -1324,6 +1324,9 @@ class ShortRangeModel:
     #: Interpersonal distances
     distance: _VectorisedFloat
 
+    #: Expiration definition
+    expiration_def: typing.Optional[str] = None
+
     def dilution_factor(self) -> _VectorisedFloat:
         '''
         The dilution factor for the respective expiratory activity type.

caimira/src/caimira/calculator/report/virus_report_data.py

@@ -105,29 +105,47 @@ def interesting_times(model: typing.Union[models.ExposureModel, models.ExposureM
     return nice_times
 
 
-def concentrations_with_sr_breathing(form: VirusFormData, 
-                                     model: typing.Union[models.ExposureModel, models.ExposureModelGroup], 
-                                     times: typing.List[float], 
-                                     short_range_intervals: typing.List) -> typing.List[float]:
+def process_short_range_interactions(model: typing.Union[models.ExposureModel, models.ExposureModelGroup], 
+                                     times: typing.List[float]):
+    """
+    Process both ExposureModel and ExposureModelGroup for short-range 
+    expirations, intervals and concentrations. Returns a tuple containing 
+    lower concentrations, short-range expirations, and short-range intervals.
+    """
+    if isinstance(model, models.ExposureModelGroup):
+        model_list = model.exposure_models
+    elif isinstance(model, models.ExposureModel):
+        model_list = (model,)
+    else:
+        raise TypeError(f"Model should be either an instance of ExposureModel or ExposureModelGroup. Got '{type(model)}.'")
+
+    # Collect short-range expirations and intervals
+    short_range_expirations, short_range_intervals = [], []
+    for nth_model in model_list:
+        for nth_sr_model in nth_model.short_range:
+            short_range_expirations.append(nth_sr_model.expiration_def)
+            short_range_intervals.append(nth_sr_model.presence.boundaries()[0])
+
+    # Collect lower concentrations (including Breathing)
     lower_concentrations = []
     for time in times:
-        for index, (start, stop) in enumerate(short_range_intervals):
-            # For visualization issues, add short-range breathing activity to the initial long-range concentrations
-            if start <= time <= stop and form.short_range_interactions[index]['expiration'] == 'Breathing':
-                if isinstance(model, models.ExposureModelGroup):
-                    lower_concentrations.append(np.sum([np.array(nth_model.concentration(float(time))).mean() for nth_model in model.exposure_models]))
-                elif isinstance(model, models.ExposureModel):
-                    lower_concentrations.append(np.array(model.concentration(float(time))).mean())
-                else:
-                    raise TypeError(f"Model should be either an ExposureModel or ExposureModelGroup. Got '{type(model)}.'")
+        breathing_found = False
+        for nth_model in model_list:
+            for nth_sr_model in nth_model.short_range:
+                start, stop = nth_sr_model.presence.boundaries()[0]
+
+                # Check if the expiration is "Breathing" and the if time is within boundaries
+                if nth_sr_model.expiration_def == 'Breathing' and (start <= time <= stop):
+                    lower_concentrations.append(np.sum([np.array(nth_model.concentration(float(time))).mean() for nth_model in model_list]))
+                    breathing_found = True
+                    break
+
+            if breathing_found:
                 break
-        if isinstance(model, models.ExposureModelGroup):
-            lower_concentrations.append(np.sum([np.array(nth_model.concentration(float(time))).mean() for nth_model in model.exposure_models]))
-        elif isinstance(model, models.ExposureModel):
-            lower_concentrations.append(np.array(model.concentration_model.concentration(float(time))).mean())
-        else:
-            raise TypeError(f"Model should be either an instance of ExposureModel or ExposureModelGroup. Got '{type(model)}.'")
-    return lower_concentrations
+
+        lower_concentrations.append(np.sum([np.array(nth_model.concentration_model.concentration(float(time))).mean() for nth_model in model_list]))
+
+    return lower_concentrations, short_range_expirations, short_range_intervals
 
 
 def _calculate_deposited_exposure(model: typing.Union[models.ExposureModel, models.ExposureModelGroup], 
@@ -167,27 +185,17 @@ def calculate_report_data(form: VirusFormData, executor_factory: typing.Callable
         exposed_presence_intervals = []
         for nth_model in model.exposure_models:
             exposed_presence_intervals.extend(list(nth_model.exposed.presence_interval().boundaries()))
-
-        # Short-range related inputs
-        short_range_intervals = []
-        for nth_model in model.exposure_models:
-            short_range_intervals.extend([interaction.presence.boundaries()[0]
-                                    for interaction in nth_model.short_range])
-        short_range_intervals = list(set(short_range_intervals))
-    else:
+    elif isinstance(model, models.ExposureModel):
         exposed_presence_intervals = [list(interval) for interval in model.exposed.presence_interval().boundaries()]
-        short_range_intervals = [interaction.presence.boundaries()[0] for interaction in model.short_range]
+    else:
+        raise TypeError(f"Model should be either an instance of ExposureModel or ExposureModelGroup. Got '{type(model)}.'")
 
-    short_range_expirations = []
-    if form.short_range_option == "short_range_yes":
-        short_range_expirations.extend([interaction['expiration']
-                            for interaction in form.short_range_interactions])
-    short_range_expirations = list(set(short_range_expirations))
+    # Handle short-range related outputs
+    lower_concentrations, short_range_expirations, short_range_intervals = None, None, None
+     # Short-range related data:
+    if (form.short_range_option == "short_range_yes"):
+       lower_concentrations, short_range_expirations, short_range_intervals = process_short_range_interactions(model, times)
 
-    # Concentration profile
-    lower_concentrations = concentrations_with_sr_breathing(
-        form, model, times, short_range_intervals)
-
     # Probability of infection
     prob = np.array(model.infection_probability())
     prob_dist_count, prob_dist_bins = np.histogram(prob/100, bins=100, density=True)
@@ -234,7 +242,7 @@ def calculate_report_data(form: VirusFormData, executor_factory: typing.Callable
             long_range_deposited_exposures.append(result)
         elif fn_name == "cn":
             concentrations.append(result)
-        elif fn_name == "co2_concentration":
+        elif fn_name == "co2":
             CO2_concentrations.append(result)
 
     cumulative_doses = np.cumsum(deposited_exposures)
@@ -287,7 +295,6 @@ def calculate_report_data(form: VirusFormData, executor_factory: typing.Callable
         "expected_new_cases": expected_new_cases,
         "uncertainties_plot_src": uncertainties_plot_src,
         "CO2_concentrations": CO2_concentrations,
-        "conditional_probability_data": [],
         "conditional_probability_data": conditional_probability_data,
     }
 

caimira/src/caimira/calculator/validators/virus/virus_validator.py

@@ -252,6 +252,14 @@ def initialize_room(self) -> models.Room:
 
         return models.Room(volume=volume, inside_temp=models.PiecewiseConstant((0, 24), (inside_temp,)), humidity=humidity) # type: ignore
 
+    def find_corresponding_short_range_models(self, sr_model: models.ShortRangeModel, start: float, finish: float) -> bool:
+        """
+        Filter function to check which short-range interactions
+        belong to a certain exposure model within the group.
+        """
+        sr_start, sr_finish = sr_model.presence.boundaries()[0]
+        return start <= sr_start and sr_finish <= finish
+
     def build_mc_model(self) -> typing.Union[mc.ExposureModel, mc.ExposureModelGroup]:
         size = self.data_registry.monte_carlo['sample_size']
 
@@ -271,6 +279,7 @@ def build_mc_model(self) -> typing.Union[mc.ExposureModel, mc.ExposureModelGroup
                     activity=infected_population.activity,
                     presence=presence,
                     distance=distances,
+                    expiration_def=interaction['expiration']
                 ).build_model(size))
 
         concentration_model: models.ConcentrationModel = mc.ConcentrationModel(
@@ -297,12 +306,12 @@ def build_mc_model(self) -> typing.Union[mc.ExposureModel, mc.ExposureModelGroup
                 if total_people > 0:
                     start_time: float = float(time_string_to_minutes(exposed_group['start_time'])/60)
                     finish_time: float = float(time_string_to_minutes(exposed_group['finish_time'])/60)
-                    exposed_population: mc.Population = self.exposed_population(
-                        total_people, start_time, finish_time).build_model(size)
+                    sr_models: typing.Tuple[models.ShortRangeModel, ...] = tuple(filter(lambda x: self.find_corresponding_short_range_models(x, start_time, finish_time), short_range))
+                    exposed_population: mc.Population = self.exposed_population(total_people, start_time, finish_time).build_model(size)
                     exposure_model: models.ExposureModel = mc.ExposureModel(
                         data_registry=self.data_registry,
                         concentration_model=concentration_model,
-                        short_range=tuple(short_range),
+                        short_range=sr_models,
                         exposed=exposed_population,
                         geographical_data=geographical_data,
                         exposed_to_short_range=self.short_range_occupants,
@@ -332,7 +341,7 @@ def build_mc_model(self) -> typing.Union[mc.ExposureModel, mc.ExposureModelGroup
                 f'Undefined exposure type. Got "{self.occupancy_format}", accepted formats are "dynamic" or "exposed".')
 
     def build_model(self, sample_size=None) -> typing.Union[models.ExposureModel, models.ExposureModelGroup]:
-        size: int = self.data_registry.monte_carlo['sample_size'] if not sample_size else sample_size
+        size = self.data_registry.monte_carlo['sample_size'] if not sample_size else sample_size
         return self.build_mc_model().build_model(size=size)
 
     def build_CO2_model(self, sample_size=None) -> models.CO2ConcentrationModel:

cern_caimira/src/cern_caimira/apps/calculator/static/js/report.js

@@ -10,7 +10,7 @@ function draw_plot(svg_id) {
     let button_full_exposure = document.getElementById("button_full_exposure");
     let button_hide_high_concentration = document.getElementById("button_hide_high_concentration");
     let long_range_checkbox = document.getElementById('long_range_cumulative_checkbox')
-    let show_sr_legend = short_range_expirations.length > 0;
+    let show_sr_legend = short_range_expirations?.length > 0;
 
     var data_for_graphs = {
         'concentrations': [],
@@ -192,7 +192,7 @@ function draw_plot(svg_id) {
     // Area representing the short-range interaction(s).
     var shortRangeArea = {};
     var drawShortRangeArea = {};
-    short_range_intervals.forEach((b, index) => {
+    short_range_intervals?.forEach((b, index) => {
         shortRangeArea[index] = d3.area();
         drawShortRangeArea[index] = draw_area.append('svg:path');