pingu_98/cara
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

added "occupancy_format" input that controls the definition of dynamic activities

Browse source
This commit is contained in:
lrdossan 2024-07-19 11:39:49 +02:00
parent e3bd714834
commit 2f8b053bb1
5 changed files with 73 additions and 62 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

30
caimira/src/caimira/calculator/report/virus_report_data.py
View file

@ -173,15 +173,16 @@ def calculate_report_data(form: VirusFormData, executor_factory: typing.Callable
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()]
@ -415,7 +416,8 @@ 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'])
@ -424,10 +426,15 @@ def scenario_statistics(
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
@ -453,17 +460,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(),
[report_data['times']] * len(scenarios),
[compute_prob_exposure] * len(scenarios),
[compute_expected_new_cases] * len(scenarios),
timeout=60,
)

5
caimira/src/caimira/calculator/validators/defaults.py
View file

@ -50,13 +50,14 @@ DEFAULTS = {
'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)',
@ -73,7 +74,7 @@ DEFAULTS = {
'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,
}

11
caimira/src/caimira/calculator/validators/form_validator.py
View file

@ -26,16 +26,13 @@ class FormData:
exposed_lunch_option: bool
exposed_lunch_start: minutes_since_midnight
exposed_start: minutes_since_midnight
# Used if infected_dont_have_breaks_with_exposed
infected_coffee_break_option: str
infected_coffee_duration: int # Used if infected_dont_have_breaks_with_exposed
infected_coffee_duration: int
infected_dont_have_breaks_with_exposed: bool
infected_finish: minutes_since_midnight
# Used if infected_dont_have_breaks_with_exposed
infected_lunch_finish: minutes_since_midnight
infected_lunch_option: bool # Used if infected_dont_have_breaks_with_exposed
# Used if infected_dont_have_breaks_with_exposed
infected_lunch_start: minutes_since_midnight
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

88
caimira/src/caimira/calculator/validators/virus/virus_validator.py
View file

@ -202,7 +202,8 @@ class VirusFormData(FormData):
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}).'
@ -494,36 +495,33 @@ class VirusFormData(FormData):
# Initializes the virus
virus = virus_distributions(self.data_registry)[self.virus_type]
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}
elif (self.activity_type == 'precise'):
activity_defn, expiration_defn = self.generate_precise_activity_expiration() # TODO: what to do here?
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.')
# 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.
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()
activity = activity_distributions(self.data_registry)[activity_defn]
expiration = build_expiration(self.data_registry, expiration_defn)
infected = mc.InfectedPopulation(
data_registry=self.data_registry,
number=infected_occupancy,
@ -542,11 +540,14 @@ class VirusFormData(FormData):
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.
@ -597,8 +598,12 @@ def baseline_raw_form_data() -> typing.Dict[str, typing.Union[str, float]]:
'activity_type': 'office',
'air_changes': '',
'air_supply': '',
'ascertainment_bias': 'confidence_low',
'ceiling_height': '',
'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',
@ -607,6 +612,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',
@ -618,43 +625,38 @@ 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': '[]',
}

1
cern_caimira/src/cern_caimira/apps/templates/base/calculator.form.html.j2
View file

@ -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">