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

Merge branch 'feature/dynamic_occupancy_UI' into 'master'

Browse source 
Dynamic occupancy feature

See merge request caimira/caimira!505
This commit is contained in:
Luis Aleixo 2024-09-19 11:01:56 +02:00
commit eef7bd422f
16 changed files with 445 additions and 186 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
caimira/pyproject.toml
View file

@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "caimira"
version = "4.17.0a1"
version = "4.17.0a2"
description = "CAiMIRA - CERN Airborne Model for Indoor Risk Assessment"
license = { text = "Apache-2.0" }
authors = [

17
caimira/src/caimira/calculator/models/models.py
View file

@ -1867,20 +1867,22 @@ class ExposureModel:
return 0
def expected_new_cases(self) -> _VectorisedFloat:
if (isinstance(self.concentration_model.infected.number, IntPiecewiseConstant) or
isinstance(self.exposed.number, IntPiecewiseConstant)):
raise NotImplementedError("Cannot compute expected new cases "
"with dynamic occupancy")
"""
The expected_new_cases may provide one or two different outputs:
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")
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() * (self.exposed.number - self.exposed_to_short_range)
return (new_cases_long_range + (self.infection_probability() * self.exposed_to_short_range)) / 100
return self.infection_probability() * self.exposed.number / 100
@ -1890,7 +1892,9 @@ class ExposureModel:
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 "
@ -1907,3 +1911,4 @@ class ExposureModel:
)
return single_exposure_model.expected_new_cases()

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

@ -171,13 +171,17 @@ def calculate_report_data(form: VirusFormData, executor_factory: typing.Callable
long_range_cumulative_doses = np.cumsum(long_range_deposited_exposures)
prob = np.array(model.infection_probability())
prob_dist_count, prob_dist_bins = np.histogram(
prob/100, bins=100, density=True)
prob_probabilistic_exposure = np.array(
model.total_probability_rule()).mean()
expected_new_cases = np.array(model.expected_new_cases()).mean()
exposed_presence_intervals = [
list(interval) for interval in model.exposed.presence_interval().boundaries()]
prob_dist_count, prob_dist_bins = np.histogram(prob/100, bins=100, density=True)
# Probabilistic exposure and expected new cases (only for static occupancy)
prob_probabilistic_exposure = None
expected_new_cases = None
if form.occupancy_format == "static":
if form.exposure_option == "p_probabilistic_exposure":
prob_probabilistic_exposure = np.array(model.total_probability_rule()).mean()
expected_new_cases = np.array(model.expected_new_cases()).mean()
exposed_presence_intervals = [list(interval) for interval in model.exposed.presence_interval().boundaries()]
conditional_probability_data = None
uncertainties_plot_src = None
@ -207,9 +211,9 @@ def calculate_report_data(form: VirusFormData, executor_factory: typing.Callable
"prob_hist_bins": list(prob_dist_bins),
"prob_probabilistic_exposure": prob_probabilistic_exposure,
"expected_new_cases": expected_new_cases,
"uncertainties_plot_src": uncertainties_plot_src,
"CO2_concentrations": CO2_concentrations,
"conditional_probability_data": conditional_probability_data,
"uncertainties_plot_src": uncertainties_plot_src,
}
@ -399,8 +403,14 @@ def manufacture_alternative_scenarios(form: VirusFormData) -> typing.Dict[str, m
scenarios['Neither ventilation nor masks'] = without_mask_or_vent.build_mc_model()
else:
no_short_range_alternative = dataclass_utils.replace(form, short_range_interactions=[],
total_people=form.total_people - form.short_range_occupants)
# When dynamic occupancy is defined, the replace of total people is useless - the expected number of new cases is not calculated.
if form.occupancy_format == 'static':
no_short_range_alternative = dataclass_utils.replace(form, short_range_interactions=[], total_people=form.total_people - form.short_range_occupants)
elif form.occupancy_format == 'dynamic':
for occ in form.dynamic_exposed_occupancy: # Update the number of exposed people with long-range exposure
if occ['total_people'] > form.short_range_occupants: occ['total_people'] = max(0, occ['total_people'] - form.short_range_occupants)
no_short_range_alternative = dataclass_utils.replace(form, short_range_interactions=[], dynamic_exposed_occupancy=form.dynamic_exposed_occupancy)
scenarios['Base scenario without short-range interactions'] = no_short_range_alternative.build_mc_model()
return scenarios
@ -409,24 +419,20 @@ 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
static_occupancy: bool,
compute_prob_exposure: 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.
return {
'probability_of_infection': np.mean(model.infection_probability()),
'expected_new_cases': np.mean(model.expected_new_cases()),
'expected_new_cases': np.mean(model.expected_new_cases()) if static_occupancy else None,
'concentrations': [
np.mean(model.concentration(time))
for time in sample_times
],
'prob_probabilistic_exposure': prob_probabilistic_exposure,
'prob_probabilistic_exposure': model.total_probability_rule() if compute_prob_exposure else None,
}
@ -447,16 +453,15 @@ def comparison_report(
else:
statistics = {}
if (form.short_range_option == "short_range_yes" and form.exposure_option == "p_probabilistic_exposure"):
compute_prob_exposure = True
else:
compute_prob_exposure = False
static_occupancy = form.occupancy_format == "static"
compute_prob_exposure = form.short_range_option == "short_range_yes" and form.exposure_option == "p_probabilistic_exposure" and static_occupancy
with executor_factory() as executor:
results = executor.map(
scenario_statistics,
scenarios.values(),
[report_data['times']] * len(scenarios),
[static_occupancy] * len(scenarios),
[compute_prob_exposure] * len(scenarios),
timeout=60,
)

33
caimira/src/caimira/calculator/validators/co2/co2_validator.py
View file

@ -44,11 +44,14 @@ class CO2FormData(FormData):
'infected_lunch_option': True,
'infected_lunch_start': '12:30',
'infected_people': 1,
'dynamic_infected_occupancy': '[]',
'infected_start': '08:30',
'room_capacity': None,
'room_volume': NO_DEFAULT,
'specific_breaks': '{}',
'total_people': NO_DEFAULT,
'dynamic_exposed_occupancy': '[]',
'occupancy_format': 'static',
}
def __init__(self, **kwargs):
@ -60,7 +63,7 @@ class CO2FormData(FormData):
self.data_registry = DataRegistry()
def validate(self):
# Validate population parameters
# Validate population parameters
self.validate_population_parameters()
# Validate room capacity
@ -194,20 +197,36 @@ class CO2FormData(FormData):
size = size or self.data_registry.monte_carlo['sample_size']
# Build a simple infected and exposed population for the case when presence
# intervals and number of people are dynamic. Activity type is not needed.
infected_presence = self.infected_present_interval()
if self.occupancy_format == 'dynamic':
if isinstance(self.dynamic_infected_occupancy, typing.List) and len(self.dynamic_infected_occupancy) > 0:
infected_people = self.generate_dynamic_occupancy(self.dynamic_infected_occupancy)
infected_presence = None
else:
raise TypeError(f'If dynamic occupancy is selected, a populated list of occupancy intervals is expected. Got "{self.dynamic_infected_occupancy}".')
if isinstance(self.dynamic_exposed_occupancy, typing.List) and len(self.dynamic_exposed_occupancy) > 0:
exposed_people = self.generate_dynamic_occupancy(self.dynamic_exposed_occupancy)
exposed_presence = None
else:
raise TypeError(f'If dynamic occupancy is selected, a populated list of occupancy intervals is expected. Got "{self.dynamic_exposed_occupancy}".')
else:
infected_people = self.infected_people
exposed_people = self.total_people - self.infected_people
infected_presence = self.infected_present_interval()
exposed_presence = self.exposed_present_interval()
infected_population = models.SimplePopulation(
number=self.infected_people,
number=infected_people,
presence=infected_presence,
activity=None, # type: ignore
)
exposed_presence = self.exposed_present_interval()
exposed_population=models.SimplePopulation(
number=self.total_people - self.infected_people,
number=exposed_people,
presence=exposed_presence,
activity=None, # type: ignore
)
all_state_changes = self.population_present_changes(infected_presence, exposed_presence)
all_state_changes=self.population_present_changes(infected_population.presence_interval(),
exposed_population.presence_interval())
total_people = [infected_population.people_present(stop) + exposed_population.people_present(stop)
for _, stop in zip(all_state_changes[:-1], all_state_changes[1:])]

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

@ -37,6 +37,7 @@ DEFAULTS = {
'infected_lunch_option': True,
'infected_lunch_start': '12:30',
'infected_people': 1,
'dynamic_infected_occupancy': NO_DEFAULT,
'infected_start': '08:30',
'inside_temp': NO_DEFAULT,
'location_latitude': NO_DEFAULT,
@ -49,12 +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': NO_DEFAULT,
'vaccine_option': False,
'vaccine_booster_option': False,
'vaccine_type': 'AZD1222_(AstraZeneca)',
@ -71,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,
}

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

@ -4,6 +4,7 @@ import logging
import typing
import ast
import json
import re
import numpy as np
@ -18,7 +19,7 @@ minutes_since_midnight = typing.NewType('minutes_since_midnight', int)
@dataclasses.dataclass
class FormData:
specific_breaks: dict
# Static occupancy inputs
exposed_coffee_break_option: str
exposed_coffee_duration: int
exposed_finish: minutes_since_midnight
@ -26,21 +27,24 @@ 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_break_option: str #Used if infected_dont_have_breaks_with_exposed
infected_coffee_duration: int #Used if infected_dont_have_breaks_with_exposed
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_people: int
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_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
@ -94,76 +98,110 @@ class FormData:
return form_dict
def validate_population_parameters(self):
# Validate number of infected <= number of total people
if self.infected_people >= self.total_people:
raise ValueError(
'Number of infected people cannot be greater or equal to the number of total people.')
# Validate time intervals selected by user
time_intervals = [
['exposed_start', 'exposed_finish'],
['infected_start', 'infected_finish'],
]
if self.exposed_lunch_option:
time_intervals.append(
['exposed_lunch_start', 'exposed_lunch_finish'])
if self.infected_dont_have_breaks_with_exposed and self.infected_lunch_option:
time_intervals.append(
['infected_lunch_start', 'infected_lunch_finish'])
for start_name, end_name in time_intervals:
start = getattr(self, start_name)
end = getattr(self, end_name)
if start > end:
# Static occupancy is defined.
if self.occupancy_format == 'static':
# Validate number of infected <= number of total people
if self.infected_people >= self.total_people:
raise ValueError(
f"{start_name} must be less than {end_name}. Got {start} and {end}.")
'Number of infected people cannot be greater or equal to the number of total people.')
def validate_lunch(start, finish):
lunch_start = getattr(self, f'{population}_lunch_start')
lunch_finish = getattr(self, f'{population}_lunch_finish')
return (start <= lunch_start <= finish and
start <= lunch_finish <= finish)
# Validate time intervals selected by user
time_intervals = [
['exposed_start', 'exposed_finish'],
['infected_start', 'infected_finish'],
]
if self.exposed_lunch_option:
time_intervals.append(
['exposed_lunch_start', 'exposed_lunch_finish'])
if self.infected_dont_have_breaks_with_exposed and self.infected_lunch_option:
time_intervals.append(
['infected_lunch_start', 'infected_lunch_finish'])
def get_lunch_mins(population):
lunch_mins = 0
if getattr(self, f'{population}_lunch_option'):
lunch_mins = getattr(
self, f'{population}_lunch_finish') - getattr(self, f'{population}_lunch_start')
return lunch_mins
def get_coffee_mins(population):
coffee_mins = 0
if getattr(self, f'{population}_coffee_break_option') != 'coffee_break_0':
coffee_mins = COFFEE_OPTIONS_INT[getattr(
self, f'{population}_coffee_break_option')] * getattr(self, f'{population}_coffee_duration')
return coffee_mins
def get_activity_mins(population):
return getattr(self, f'{population}_finish') - getattr(self, f'{population}_start')
populations = [
'exposed', 'infected'] if self.infected_dont_have_breaks_with_exposed else ['exposed']
for population in populations:
# Validate lunch time within the activity times.
if (getattr(self, f'{population}_lunch_option') and
not validate_lunch(getattr(self, f'{population}_start'), getattr(
self, f'{population}_finish'))
):
raise ValueError(
f"{population} lunch break must be within presence times."
)
# Length of breaks < length of activity
if (get_lunch_mins(population) + get_coffee_mins(population)) >= get_activity_mins(population):
raise ValueError(
f"Length of breaks >= Length of {population} presence."
)
for attr_name, valid_set in [('exposed_coffee_break_option', COFFEE_OPTIONS_INT),
('infected_coffee_break_option', COFFEE_OPTIONS_INT)]:
if getattr(self, attr_name) not in valid_set:
for start_name, end_name in time_intervals:
start = getattr(self, start_name)
end = getattr(self, end_name)
if start > end:
raise ValueError(
f"{getattr(self, attr_name)} is not a valid value for {attr_name}")
f"{start_name} must be less than {end_name}. Got {start} and {end}.")
def validate_lunch(start, finish):
lunch_start = getattr(self, f'{population}_lunch_start')
lunch_finish = getattr(self, f'{population}_lunch_finish')
return (start <= lunch_start <= finish and
start <= lunch_finish <= finish)
def get_lunch_mins(population):
lunch_mins = 0
if getattr(self, f'{population}_lunch_option'):
lunch_mins = getattr(
self, f'{population}_lunch_finish') - getattr(self, f'{population}_lunch_start')
return lunch_mins
def get_coffee_mins(population):
coffee_mins = 0
if getattr(self, f'{population}_coffee_break_option') != 'coffee_break_0':
coffee_mins = COFFEE_OPTIONS_INT[getattr(
self, f'{population}_coffee_break_option')] * getattr(self, f'{population}_coffee_duration')
return coffee_mins
def get_activity_mins(population):
return getattr(self, f'{population}_finish') - getattr(self, f'{population}_start')
populations = [
'exposed', 'infected'] if self.infected_dont_have_breaks_with_exposed else ['exposed']
for population in populations:
# Validate lunch time within the activity times.
if (getattr(self, f'{population}_lunch_option') and
not validate_lunch(getattr(self, f'{population}_start'), getattr(
self, f'{population}_finish'))
):
raise ValueError(
f"{population} lunch break must be within presence times."
)
# Length of breaks < length of activity
if (get_lunch_mins(population) + get_coffee_mins(population)) >= get_activity_mins(population):
raise ValueError(
f"Length of breaks >= Length of {population} presence."
)
for attr_name, valid_set in [('exposed_coffee_break_option', COFFEE_OPTIONS_INT),
('infected_coffee_break_option', COFFEE_OPTIONS_INT)]:
if getattr(self, attr_name) not in valid_set:
raise ValueError(
f"{getattr(self, attr_name)} is not a valid value for {attr_name}")
# Dynamic occupancy is defined.
elif self.occupancy_format == 'dynamic':
for dynamic_format in (self.dynamic_infected_occupancy, self.dynamic_exposed_occupancy):
for occupancy in dynamic_format:
# Check if each occupancy entry is a dictionary
if not isinstance(occupancy, typing.Dict):
raise TypeError(f'Each occupancy entry should be in a dictionary format. Got "{type(occupancy)}".')
# Check for required keys in each occupancy entry
dict_keys = list(occupancy.keys())
if "total_people" not in dict_keys:
raise TypeError(f'Unable to fetch "total_people" key. Got "{dict_keys}".')
else:
value = occupancy["total_people"]
# Check if the value is a non-negative integer
if not isinstance(value, int):
raise ValueError(f'Total number of people should be integer. Got "{type(value)}".')
elif not value >= 0:
raise ValueError(f'Total number of people should be non-negative. Got "{value}".')
if "start_time" not in dict_keys:
raise TypeError(f'Unable to fetch "start_time" key. Got "{dict_keys}".')
if "finish_time" not in dict_keys:
raise TypeError(f'Unable to fetch "finish_time" key. Got "{dict_keys}".')
# Validate time format for start_time and finish_time
for time_key in ["start_time", "finish_time"]:
time = occupancy[time_key]
if not re.compile("^(2[0-3]|[01]?[0-9]):([0-5]?[0-9])$").match(time):
raise TypeError(f'Wrong time format - "HH:MM". Got "{time}".')
else:
raise ValueError(f"'{self.occupancy_format}' is not a valid value for 'self.occupancy_format'. Accepted values are 'static' or 'dynamic'.")
def validate(self):
raise NotImplementedError("Subclass must implement")
@ -369,14 +407,6 @@ class FormData:
breaks=breaks,
)
def population_present_interval(self) -> models.Interval:
state_change_times = set(
self.infected_present_interval().transition_times())
state_change_times.update(
self.exposed_present_interval().transition_times())
all_state_changes = sorted(state_change_times)
return models.SpecificInterval(tuple(zip(all_state_changes[:-1], all_state_changes[1:])))
def exposed_present_interval(self) -> models.Interval:
if self.specific_breaks != {}: # It means the breaks are specific and not predefined
breaks = self.generate_specific_break_times(
@ -387,6 +417,26 @@ class FormData:
self.exposed_start, self.exposed_finish,
breaks=breaks,
)
def generate_dynamic_occupancy(self, dynamic_occupancy: typing.List[typing.Dict[str, typing.Any]]):
transition_times = []
values = []
for occupancy in dynamic_occupancy:
start_time = time_string_to_minutes(occupancy['start_time'])/60
finish_time = time_string_to_minutes(occupancy['finish_time'])/60
transition_times.extend([start_time, finish_time])
values.append(occupancy['total_people'])
unique_transition_times_sorted = np.array(sorted(set(transition_times)))
if len(values) != len(unique_transition_times_sorted) - 1:
raise ValueError("Cannot compute dynamic occupancy with the provided inputs.")
population_occupancy: models.IntPiecewiseConstant = models.IntPiecewiseConstant(
transition_times=tuple(unique_transition_times_sorted),
values=tuple(values)
)
return population_occupancy
def _hours2timestring(hours: float):

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

@ -73,6 +73,7 @@ class VirusFormData(FormData):
_DEFAULTS: typing.ClassVar[typing.Dict[str, typing.Any]] = DEFAULTS
def validate(self):
# Validate population parameters
self.validate_population_parameters()
validation_tuples = [('activity_type', self.data_registry.population_scenario_activity.keys()),
@ -202,11 +203,29 @@ 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}).'
)
# Validate short-range interactions interval
if self.short_range_option == "short_range_yes":
for interaction in self.short_range_interactions:
# Check if presence is within long-range exposure
presence = self.short_range_interval(interaction)
if (self.occupancy_format == 'dynamic'):
long_range_start = min(time_string_to_minutes(self.dynamic_infected_occupancy[0]['start_time']),
time_string_to_minutes(self.dynamic_exposed_occupancy[0]['start_time']))
long_range_stop = max(time_string_to_minutes(self.dynamic_infected_occupancy[-1]['finish_time']),
time_string_to_minutes(self.dynamic_exposed_occupancy[-1]['finish_time']))
else:
long_range_start = min(self.infected_start, self.exposed_start)
long_range_stop = max(self.infected_finish, self.exposed_finish)
if not (long_range_start/60 <= presence.present_times[0][0] <= long_range_stop/60 and
long_range_start/60 <= presence.present_times[0][-1] <= long_range_stop/60):
raise ValueError(f"Short-range interactions should be defined during simulation time. Got {interaction}")
def initialize_room(self) -> models.Room:
# Initializes room with volume either given directly or as product of area and height
@ -230,7 +249,7 @@ class VirusFormData(FormData):
def build_mc_model(self) -> mc.ExposureModel:
room = self.initialize_room()
ventilation: models._VentilationBase = self.ventilation()
infected_population = self.infected_population()
infected_population: models.InfectedPopulation = self.infected_population()
short_range = []
if self.short_range_option == "short_range_yes":
for interaction in self.short_range_interactions:
@ -444,27 +463,39 @@ 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']
# 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 = self.generate_dynamic_occupancy(self.dynamic_infected_occupancy)
infected_presence = None
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()
activity = activity_distributions(self.data_registry)[activity_defn]
expiration = build_expiration(self.data_registry, expiration_defn)
infected_occupants = self.infected_people
infected = mc.InfectedPopulation(
data_registry=self.data_registry,
number=infected_occupants,
number=infected_occupancy,
presence=infected_presence,
virus=virus,
presence=self.infected_present_interval(),
mask=self.mask(),
activity=activity,
expiration=expiration,
@ -479,10 +510,19 @@ class VirusFormData(FormData):
else str(self.data_registry.population_scenario_activity[self.activity_type]['activity']))
activity = activity_distributions(self.data_registry)[activity_defn]
infected_occupants = self.infected_people
# The number of exposed occupants is the total number of occupants
# minus the number of infected occupants.
exposed_occupants = self.total_people - infected_occupants
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 = self.generate_dynamic_occupancy(self.dynamic_exposed_occupancy)
exposed_presence = None
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.
exposed_occupancy = self.total_people - self.infected_people
exposed_presence = self.exposed_present_interval()
if (self.vaccine_option):
if (self.vaccine_booster_option and self.vaccine_booster_type != 'Other'):
@ -495,8 +535,8 @@ class VirusFormData(FormData):
host_immunity = 0.
exposed = mc.Population(
number=exposed_occupants,
presence=self.exposed_present_interval(),
number=exposed_occupancy,
presence=exposed_presence,
activity=activity,
mask=self.mask(),
host_immunity=host_immunity,
@ -528,8 +568,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',
@ -538,6 +582,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',
@ -554,36 +600,33 @@ def baseline_raw_form_data() -> typing.Dict[str, typing.Union[str, float]]:
'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',
'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': '[]',
}

52
caimira/tests/apps/calculator/test_model_generator.py
View file

@ -1,5 +1,5 @@
import dataclasses
import typing
import re
import numpy as np
import numpy.testing as npt
@ -588,3 +588,53 @@ def test_form_timezone(baseline_form_data, data_registry, longitude, latitude, m
name, offset = form.tz_name_and_utc_offset()
assert name == expected_tz_name
assert offset == expected_offset
@pytest.mark.parametrize(
["occupancy_format_input", "error"],
[
['dynamc', "'dynamc' is not a valid value for 'self.occupancy_format'. Accepted values are 'static' or 'dynamic'.",],
['stact', "'stact' is not a valid value for 'self.occupancy_format'. Accepted values are 'static' or 'dynamic'.",],
['random', "'random' is not a valid value for 'self.occupancy_format'. Accepted values are 'static' or 'dynamic'.",]
]
)
def test_dynamic_format_input(occupancy_format_input, error, baseline_form: virus_validator.VirusFormData):
baseline_form.occupancy_format = occupancy_format_input
with pytest.raises(ValueError, match=re.escape(error)):
baseline_form.validate()
@pytest.mark.parametrize(
["dynamic_occupancy_input", "error"],
[
[[["total_people", 10, "start_time", "10:00", "finish_time", "11:00"]], "Each occupancy entry should be in a dictionary format. Got \"<class 'list'>\"."],
[[{"tal_people": 10, "start_time": "10:00", "finish_time": "11:00"}], "Unable to fetch \"total_people\" key. Got \"['tal_people', 'start_time', 'finish_time']\"."],
[[{"total_people": 10, "art_time": "10:00", "finish_time": "11:00"}], "Unable to fetch \"start_time\" key. Got \"['total_people', 'art_time', 'finish_time']\"."],
[[{"total_people": 10, "start_time": "10:00", "ish_time": "11:00"}], "Unable to fetch \"finish_time\" key. Got \"['total_people', 'start_time', 'ish_time']\"."],
[[{"total_people": 10, "start_time": "10", "finish_time": "11:00"}], "Wrong time format - \"HH:MM\". Got \"10\"."],
[[{"total_people": 10, "start_time": "10:00", "finish_time": "11"}], "Wrong time format - \"HH:MM\". Got \"11\"."],
]
)
def test_dynamic_occupancy_structure(dynamic_occupancy_input, error, baseline_form: virus_validator.VirusFormData):
baseline_form.occupancy_format = "dynamic"
baseline_form.dynamic_infected_occupancy = dynamic_occupancy_input
baseline_form.dynamic_exposed_occupancy = dynamic_occupancy_input
with pytest.raises(TypeError, match=re.escape(error)):
baseline_form.validate()
@pytest.mark.parametrize(
["dynamic_occupancy_input", "error"],
[
[[{"total_people": "10", "start_time": "10:00", "finish_time": "11:00"}], "Total number of people should be integer. Got \"<class 'str'>\"."],
[[{"total_people": 9.8, "start_time": "10:00", "finish_time": "11:00"}], "Total number of people should be integer. Got \"<class 'float'>\"."],
[[{"total_people": [10], "start_time": "10:00", "finish_time": "11:00"}], "Total number of people should be integer. Got \"<class 'list'>\"."],
[[{"total_people": -1, "start_time": "10:00", "finish_time": "11:00"}], "Total number of people should be non-negative. Got \"-1\"."],
]
)
def test_dynamic_occupancy_total_people(dynamic_occupancy_input, error, baseline_form: virus_validator.VirusFormData):
baseline_form.occupancy_format = "dynamic"
baseline_form.dynamic_infected_occupancy = dynamic_occupancy_input
baseline_form.dynamic_exposed_occupancy = dynamic_occupancy_input
with pytest.raises(ValueError, match=re.escape(error)):
baseline_form.validate()

2
caimira/tests/apps/calculator/test_specific_model_generator.py
View file

@ -1,5 +1,3 @@
from typing import Type
import numpy as np
import pytest
from caimira.calculator.validators.virus import virus_validator

4
caimira/tests/models/test_dynamic_population.py
View file

@ -230,6 +230,7 @@ def test_dynamic_total_probability_rule(
"(including incidence rate) with dynamic occupancy")):
dynamic_population_exposure_model.total_probability_rule()
def test_dynamic_expected_new_cases(
dynamic_infected_single_exposure_model: models.ExposureModel,
dynamic_exposed_single_exposure_model: models.ExposureModel,
@ -245,11 +246,12 @@ def test_dynamic_expected_new_cases(
"with dynamic occupancy")):
dynamic_population_exposure_model.expected_new_cases()
def test_dynamic_reproduction_number(
dynamic_infected_single_exposure_model: models.ExposureModel,
dynamic_exposed_single_exposure_model: models.ExposureModel,
dynamic_population_exposure_model: models.ExposureModel):
with pytest.raises(NotImplementedError, match=re.escape("Cannot compute reproduction number "
"with dynamic occupancy")):
dynamic_infected_single_exposure_model.reproduction_number()

2
cern_caimira/pyproject.toml
View file

@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "cern-caimira"
version = "4.17.0a1"
version = "4.17.1a1"
description = "CAiMIRA - CERN Airborne Model for Indoor Risk Assessment"
license = { text = "Apache-2.0" }
authors = [

3
cern_caimira/src/cern_caimira/apps/calculator/static/js/co2_form.js
View file

@ -17,11 +17,14 @@ const CO2_data_form = [
"infected_lunch_option",
"infected_lunch_start",
"infected_people",
"dynamic_infected_occupancy",
"infected_start",
"room_capacity",
"room_volume",
"specific_breaks",
"total_people",
"dynamic_exposed_occupancy",
"occupancy_format",
];
// Method to upload a valid data file (accepted formats: .xls and .xlsx)

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

@ -447,6 +447,10 @@
<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" value="[]">
<input type="text" class="form-control d-none" name="dynamic_infected_occupancy" value="[]">
<div class="form-group row">
<div class="col-sm-4"><label class="col-form-label">Total number of occupants:</label></div>
<div class="col-sm-6 align-self-center"><input type="number" id="total_people" class="form-control" name="total_people" placeholder="Number" min=1 required></div>

58
cern_caimira/src/cern_caimira/apps/templates/base/calculator.report.html.j2
View file

@ -104,7 +104,7 @@
</div>
{% endblock long_range_warning_animation %}
</div>
<h6><b>Expected new cases:</b> {{ long_range_expected_cases | float_format }}</h6>
{% if form.occupancy_format == "static" %}<h6><b>Expected new cases:</b> {{ long_range_expected_cases | float_format }}</h6>{% endif %}
</div>
<br>
{% if form.short_range_option == "short_range_yes" %}
@ -126,38 +126,50 @@
</div>
{% endblock warning_animation %}
</div>
<h6><b>Expected new cases:</b> {{ expected_new_cases | float_format }}</h6>
{% if form.occupancy_format == "static" %}
<h6><b>Expected new cases:</b> {{ expected_new_cases | float_format }}</h6>
{% endif %}
</div>
{% endif %}
<div class="d-flex">
{% block report_summary %}
<div class="flex-row align-self-center">
<div class="align-self-center alert alert-dark mb-0" role="alert">
Taking into account the uncertainties tied to the model variables, in this scenario and assuming all occupants are exposed equally (i.e. without short-range interactions), the <b>probability of one exposed occupant getting infected is {{ long_range_prob_inf | non_zero_percentage }}</b> and the <b>expected number of new cases is {{ long_range_expected_cases | float_format }}</b>*.
Taking into account the uncertainties tied to the model variables, in this scenario and assuming all occupants are exposed equally (i.e. without short-range interactions), the <b>probability of one exposed occupant getting infected is {{ long_range_prob_inf | non_zero_percentage }}</b>
{% if form.occupancy_format == "static" %}
and the <b>expected number of new cases is {{ long_range_expected_cases | float_format }}</b>
{% endif %}*.
</div>
{% if form.short_range_option == "short_range_yes" %}
<br>
<div class="align-self-center alert alert-dark mb-0" role="alert">
In this scenario, the <b>probability the occupant(s) exposed to short-range interactions get infected can go as high as {{ prob_inf | non_zero_percentage }}</b> and the <b>expected number of new cases increases to {{ expected_new_cases | float_format }}</b>.
In this scenario, the <b>probability the occupant(s) exposed to short-range interactions get infected can go as high as {{ prob_inf | non_zero_percentage }}</b>
{% if form.occupancy_format == "static" %}
and the <b>expected number of new cases increases to {{ expected_new_cases | float_format }}</b>
{% endif %}.
</div>
{% endif %}
{% block probabilistic_exposure_probability %}
{% if form.exposure_option == "p_probabilistic_exposure" %}
<br>
<div class="align-self-center alert alert-dark mb-0" role="alert">
The above {{ "result assumes" if form.short_range_option == "short_range_no" else "results assume" }} that <b>{{ form.infected_people }}
{{ "occupant is infected" if form.infected_people == 1 else "occupants are infected" }}
</b> in the room.
By taking into account the estimate of cases currently circulating in <b>{{ form.location_name }}</b>,
the probability of on-site transmission, having at least 1 new infection in an <b>event
with {{ form.total_people }} occupants</b>, is
{% if form.short_range_option == 'short_range_yes' %}:
<ul>
<li><b>{{ long_range_prob_probabilistic_exposure | non_zero_percentage }}</b>, assuming all occupants are exposed equally (i.e. without short-range interactions).</li>
<li><b>{{ prob_probabilistic_exposure | non_zero_percentage }}</b>, assuming short-range interactions occur with the infector(s).</li>
</ul>
{% if form.occupancy_format == "static" %}
The above {{ "result assumes" if form.short_range_option == "short_range_no" else "results assume" }} that <b>{{ form.infected_people }}
{{ "occupant is infected" if form.infected_people == 1 else "occupants are infected" }}
</b> in the room.
By taking into account the estimate of cases currently circulating in <b>{{ form.location_name }}</b>,
the probability of on-site transmission, having at least 1 new infection in an <b>event
with {{ form.total_people }} occupants</b>, is
{% if form.short_range_option == 'short_range_yes' %}:
<ul>
<li><b>{{ long_range_prob_probabilistic_exposure | non_zero_percentage }}</b>, assuming all occupants are exposed equally (i.e. without short-range interactions).</li>
<li><b>{{ prob_probabilistic_exposure | non_zero_percentage }}</b>, assuming short-range interactions occur with the infector(s).</li>
</ul>
{% else %}
<b>{{ prob_probabilistic_exposure | non_zero_percentage }}</b>.
{% endif %}
{% else %}
<b>{{ prob_probabilistic_exposure | non_zero_percentage }}</b>.
<p><strong>Warning: </strong>Since dynamic occupancy was defined, the results for probabilistic exposure with incidence rates have not been computed.</p>
{% endif %}
</div>
{% endif %}
@ -298,7 +310,7 @@
<tr>
<th>Scenario</th>
<th>P(I)</th>
<th>Expected new cases</th>
{% if form.occupancy_format == "static" %}<th>Expected new cases</th>{% endif %}
</tr>
</thead>
<tbody>
@ -306,7 +318,7 @@
<tr>
<td> {{ scenario_name }}</td>
<td> {{ scenario_stats.probability_of_infection | non_zero_percentage }}</td>
<td style="text-align:right">{{ scenario_stats.expected_new_cases | float_format }}</td>
{% if form.occupancy_format == "static" %}<td style="text-align:right">{{ scenario_stats.expected_new_cases | float_format }}</td>{% endif %}
</tr>
{% endfor %}
</tbody>
@ -626,11 +638,11 @@
<ul>
{% for interaction in form.short_range_interactions %}
<li>Interaction no. {{ loop.index }}:
<ul>
<li>Expiratory activity: {{ "Shouting/Singing" if interaction.expiration == "Shouting" else interaction.expiration }} </li>
<li>Start time: {{ interaction.start_time }} </li>
<li>Duration: {{ interaction.duration }} {{ "minutes" if interaction.duration|float > 1 else "minute" }}</li>
</ul>
<ul>
<li>Expiratory activity: {{ "Shouting/Singing" if interaction.expiration == "Shouting" else interaction.expiration }} </li>
<li>Start time: {{ interaction.start_time }} </li>
<li>Duration: {{ interaction.duration }} {{ "minutes" if interaction.duration|float > 1 else "minute" }}</li>
</ul>
</li>
{% endfor %}
</ul>

20
cern_caimira/src/cern_caimira/apps/templates/cern/calculator.report.html.j2
View file

@ -11,7 +11,7 @@
{% set long_range_prob_inf = prob_inf %}
{% endif %}
{% if ((long_range_prob_inf > red_prob_lim) or (expected_new_cases >= 1)) %}
{% if ((long_range_prob_inf > red_prob_lim) or (form.occupancy_format == "static" and expected_new_cases >= 1)) %}
{% set long_range_scale_warning = 'red' %}
{% set long_range_warning_color= 'bg-danger' %}
{% elif (orange_prob_lim <= long_range_prob_inf <= red_prob_lim) %}
@ -22,7 +22,7 @@
{% set long_range_warning_color = 'bg-success' %}
{% endif %}
{% if ((prob_inf > red_prob_lim) or (expected_new_cases >= 1)) %} {% set scale_warning = 'red' %}
{% if ((prob_inf > red_prob_lim) or (form.occupancy_format == "static" and expected_new_cases >= 1)) %} {% set scale_warning = 'red' %}
{% elif (orange_prob_lim <= prob_inf <= red_prob_lim) %} {% set scale_warning = 'orange' %}
{% else %} {% set scale_warning = 'green' %}
{% endif %}
@ -70,7 +70,10 @@
<div class="alert alert-success mb-0" role="alert">
<strong>Acceptable:</strong>
{% endif %}
Taking into account the uncertainties tied to the model variables, in this scenario and assuming all occupants are exposed equally (i.e. without short-range interactions), the <b>probability of one exposed occupant getting infected is {{long_range_prob_inf | non_zero_percentage}}</b> and the <b>expected number of new cases is {{ long_range_expected_cases | float_format }}</b>*.
Taking into account the uncertainties tied to the model variables, in this scenario and assuming all occupants are exposed equally (i.e. without short-range interactions), the <b>probability of one exposed occupant getting infected is {{long_range_prob_inf | non_zero_percentage}}</b>
{% if form.occupancy_format == "static" %}
and the <b>expected number of new cases is {{ long_range_expected_cases | float_format }}</b>
{% endif %}*.
</div>
{% if form.short_range_option == "short_range_yes" %}
<br>
@ -84,7 +87,10 @@
<div class="alert alert-success mb-0" role="alert">
<strong>Acceptable:</strong>
{% endif %}
In this scenario, the <b>probability the occupant(s) exposed to short-range interactions get infected can go as high as {{ prob_inf | non_zero_percentage }}</b> and the <b>expected number of new cases increases to {{ expected_new_cases | float_format }}</b>.
In this scenario, the <b>probability the occupant(s) exposed to short-range interactions get infected can go as high as {{ prob_inf | non_zero_percentage }}</b>
{% if form.occupancy_format == "static" %}
and the <b>expected number of new cases increases to {{ expected_new_cases | float_format }}</b>
{% endif %}.
</div>
{% endif %}
@ -137,12 +143,12 @@
<tr>
<th>Scenario</th>
<th>P(i)</th>
<th>Expected new cases</th>
{% if form.occupancy_format == "static" %}<th>Expected new cases</th>{% endif %}
</tr>
</thead>
<tbody>
{% for scenario_name, scenario_stats in alternative_scenarios.stats.items() %}
{%if (( scenario_stats.probability_of_infection > red_prob_lim) or (scenario_stats.expected_new_cases >= 1)) %}
{%if (( scenario_stats.probability_of_infection > red_prob_lim) or (form.occupancy_format == "static" and scenario_stats.expected_new_cases >= 1)) %}
<tr class="alert-danger">
{% elif (orange_prob_lim <= scenario_stats.probability_of_infection <= red_prob_lim) %}
<tr class="alert-warning">
@ -151,7 +157,7 @@
{% endif%}
<td> {{ scenario_name }}</td>
<td> {{ scenario_stats.probability_of_infection | non_zero_percentage }}</td>
<td style="text-align:right">{{ scenario_stats.expected_new_cases | float_format }}</td>
{% if form.occupancy_format == "static" %}<td style="text-align:right">{{ scenario_stats.expected_new_cases | float_format }}</td>{% endif %}
</tr>
{% endfor %}
</tbody>

59
cern_caimira/tests/test_report_generator.py
View file

@ -2,6 +2,7 @@ import concurrent.futures
from functools import partial
import os
import time
import json
import numpy as np
import pytest
@ -123,4 +124,62 @@ def test_expected_new_cases(baseline_form_with_sr: VirusFormData):
lr_expected_new_cases = alternative_statistics['stats']['Base scenario without short-range interactions']['expected_new_cases']
np.testing.assert_almost_equal(sr_lr_expected_new_cases, lr_expected_new_cases + sr_lr_prob_inf * baseline_form_with_sr.short_range_occupants, 2)
def test_static_vs_dynamic_occupancy_from_form(baseline_form_data, data_registry):
"""
Assert that the results between a static and dynamic occupancy model (from form inputs) are similar.
"""
executor_factory = partial(
concurrent.futures.ThreadPoolExecutor, 1,
)
# By default the baseline form accepts static occupancy
static_occupancy_baseline_form: VirusFormData = VirusFormData.from_dict(baseline_form_data, data_registry)
static_occupancy_model = static_occupancy_baseline_form.build_model()
static_occupancy_report_data = rep_gen.calculate_report_data(static_occupancy_baseline_form, executor_factory)
# Update the initial form data to include dynamic occupancy (please note the 4 coffee and 1 lunch breaks)
baseline_form_data['occupancy_format'] = 'dynamic'
baseline_form_data['dynamic_infected_occupancy'] = json.dumps([
{'total_people': 1, 'start_time': '09:00', 'finish_time': '10:03'},
{'total_people': 0, 'start_time': '10:03', 'finish_time': '10:13'},
{'total_people': 1, 'start_time': '10:13', 'finish_time': '11:16'},
{'total_people': 0, 'start_time': '11:16', 'finish_time': '11:26'},
{'total_people': 1, 'start_time': '11:26', 'finish_time': '12:30'},
{'total_people': 0, 'start_time': '12:30', 'finish_time': '13:30'},
{'total_people': 1, 'start_time': '13:30', 'finish_time': '14:53'},
{'total_people': 0, 'start_time': '14:53', 'finish_time': '15:03'},
{'total_people': 1, 'start_time': '15:03', 'finish_time': '16:26'},
{'total_people': 0, 'start_time': '16:26', 'finish_time': '16:36'},
{'total_people': 1, 'start_time': '16:36', 'finish_time': '18:00'},
])
baseline_form_data['dynamic_exposed_occupancy'] = json.dumps([
{'total_people': 9, 'start_time': '09:00', 'finish_time': '10:03'},
{'total_people': 0, 'start_time': '10:03', 'finish_time': '10:13'},
{'total_people': 9, 'start_time': '10:13', 'finish_time': '11:16'},
{'total_people': 0, 'start_time': '11:16', 'finish_time': '11:26'},
{'total_people': 9, 'start_time': '11:26', 'finish_time': '12:30'},
{'total_people': 0, 'start_time': '12:30', 'finish_time': '13:30'},
{'total_people': 9, 'start_time': '13:30', 'finish_time': '14:53'},
{'total_people': 0, 'start_time': '14:53', 'finish_time': '15:03'},
{'total_people': 9, 'start_time': '15:03', 'finish_time': '16:26'},
{'total_people': 0, 'start_time': '16:26', 'finish_time': '16:36'},
{'total_people': 9, 'start_time': '16:36', 'finish_time': '18:00'},
])
baseline_form_data['total_people'] = 0
baseline_form_data['infected_people'] = 0
dynamic_occupancy_baseline_form: VirusFormData = VirusFormData.from_dict(baseline_form_data, data_registry)
dynamic_occupancy_model = dynamic_occupancy_baseline_form.build_model()
dynamic_occupancy_report_data = rep_gen.calculate_report_data(dynamic_occupancy_baseline_form, executor_factory)
assert (list(sorted(static_occupancy_model.concentration_model.infected.presence.transition_times())) ==
list(dynamic_occupancy_model.concentration_model.infected.number.transition_times))
assert (list(sorted(static_occupancy_model.exposed.presence.transition_times())) ==
list(dynamic_occupancy_model.exposed.number.transition_times))
np.testing.assert_almost_equal(static_occupancy_report_data['prob_inf'], dynamic_occupancy_report_data['prob_inf'], 1)
assert dynamic_occupancy_report_data['expected_new_cases'] == None
assert dynamic_occupancy_report_data['prob_probabilistic_exposure'] == None