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

added data_service attr where needed

Browse source
This commit is contained in:
Luis Aleixo 2023-12-11 18:00:31 +01:00
parent 2abdd130cf
commit 20e8bf1df7
15 changed files with 146 additions and 119 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

22
caimira/apps/calculator/__init__.py
View file

@ -103,8 +103,8 @@ class ConcentrationModel(BaseRequestHandler):
async def post(self) -> None:
debug = self.settings.get("debug", False)
data_registry = self.settings.get("data_registry")
data_service = self.settings.get("data_service")
data_registry: DataRegistry = self.settings["data_registry"]
data_service: typing.Optional[DataService] = self.settings.get("data_service", None)
if data_service:
data_service.update_registry(data_registry)
@ -159,10 +159,10 @@ class ConcentrationModelJsonResponse(BaseRequestHandler):
"""
debug = self.settings.get("debug", False)
data_registry = self.settings.get("data_registry")
data_service = self.settings.get("data_service")
data_registry: DataRegistry = self.settings["data_registry"]
data_service: typing.Optional[DataService] = self.settings.get("data_service", None)
if data_service:
data_service.update_configuration(data_registry)
data_service.update_registry(data_registry)
requested_model_config = json.loads(self.request.body)
LOG.debug(pformat(requested_model_config))
@ -190,10 +190,10 @@ class StaticModel(BaseRequestHandler):
async def get(self) -> None:
debug = self.settings.get("debug", False)
data_registry = self.settings.get("data_registry")
data_service = self.settings.get("data_service")
data_registry: DataRegistry = self.settings["data_registry"]
data_service: typing.Optional[DataService] = self.settings.get("data_service", None)
if data_service:
data_service.update_configuration(data_registry)
data_service.update_registry(data_registry)
form = model_generator.VirusFormData.from_dict(model_generator.baseline_raw_form_data(), data_registry)
base_url = self.request.protocol + "://" + self.request.host
@ -368,10 +368,10 @@ class CO2ModelResponse(BaseRequestHandler):
pass
async def post(self, endpoint: str) -> None:
data_registry = self.settings.get("data_registry")
data_service = self.settings.get("data_service")
data_registry: DataRegistry = self.settings["data_registry"]
data_service: typing.Optional[DataService] = self.settings.get("data_service", None)
if data_service:
data_service.update_configuration(data_registry)
data_service.update_registry(data_registry)
requested_model_config = tornado.escape.json_decode(self.request.body)
try:

1
caimira/apps/calculator/form_data.py
View file

@ -1,5 +1,4 @@
import dataclasses
import datetime
import html
import logging
import typing

9
caimira/apps/calculator/report_generator.py
View file

@ -216,17 +216,17 @@ def conditional_prob_inf_given_vl_dist(
def manufacture_conditional_probability_data(
data_registry: DataRegistry,
exposure_model: models.ExposureModel,
infection_probability: models._VectorisedFloat
):
data_registry: DataRegistry = exposure_model.data_registry
min_vl = data_registry.conditional_prob_inf_given_viral_load['min_vl']
max_vl = data_registry.conditional_prob_inf_given_viral_load['max_vl']
step = (max_vl - min_vl)/100
viral_loads = np.arange(min_vl, max_vl, step)
specific_vl = np.log10(exposure_model.concentration_model.virus.viral_load_in_sputum)
pi_means, lower_percentiles, upper_percentiles = conditional_prob_inf_given_vl_dist(infection_probability, viral_loads,
pi_means, lower_percentiles, upper_percentiles = conditional_prob_inf_given_vl_dist(data_registry, infection_probability, viral_loads,
specific_vl, step)
return list(viral_loads), list(pi_means), list(lower_percentiles), list(upper_percentiles)
@ -414,12 +414,11 @@ def manufacture_alternative_scenarios(form: VirusFormData) -> typing.Dict[str, m
def scenario_statistics(
data_registry: DataRegistry,
mc_model: mc.ExposureModel,
sample_times: typing.List[float],
compute_prob_exposure: bool
):
model = mc_model.build_model(size=data_registry.monte_carlo_sample_size)
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()

12
caimira/monte_carlo/data.py
View file

@ -291,7 +291,7 @@ def covid_overal_vl_data(data_registry):
function=lambda d: np.interp(
d,
viral_load(data_registry),
frequencies_pdf,
frequencies_pdf(data_registry),
data_registry.covid_overal_vl_data['interpolation_fp_left'],
data_registry.covid_overal_vl_data['interpolation_fp_right']
),
@ -441,22 +441,24 @@ def expiration_BLO_factors(data_registry):
def expiration_distributions(data_registry):
return {
exp_type: expiration_distribution(
BLO_factors,
data_registry=data_registry,
BLO_factors=BLO_factors,
d_min=param_evaluation(data_registry.long_range_expiration_distributions, 'minimum_diameter'),
d_max=param_evaluation(data_registry.long_range_expiration_distributions, 'maximum_diameter')
)
for exp_type, BLO_factors in expiration_BLO_factors.items()
for exp_type, BLO_factors in expiration_BLO_factors(data_registry).items()
}
def short_range_expiration_distributions(data_registry):
return {
exp_type: expiration_distribution(
BLO_factors,
data_registry=data_registry,
BLO_factors=BLO_factors,
d_min=param_evaluation(data_registry.short_range_expiration_distributions, 'minimum_diameter'),
d_max=param_evaluation(data_registry.short_range_expiration_distributions, 'maximum_diameter')
)
for exp_type, BLO_factors in expiration_BLO_factors.items()
for exp_type, BLO_factors in expiration_BLO_factors(data_registry).items()
}

4
caimira/store/data_service.py
View file

@ -18,14 +18,14 @@ class DataService:
def __init__(
self,
credentials: typing.Dict[str, str],
credentials: typing.Dict[str, typing.Optional[str]],
host: str,
):
self._credentials = credentials
self._host = host
@classmethod
def create(cls, credentials: typing.Dict[str, str], host: str = "https://caimira-data-api.app.cern.ch"):
def create(cls, credentials: typing.Dict[str, typing.Optional[str]], host: str = "https://caimira-data-api.app.cern.ch"):
"""Factory."""
return cls(credentials, host)

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

@ -187,7 +187,7 @@ def test_infected_less_than_total_people(activity, total_people, infected_people
baseline_form.total_people = total_people
baseline_form.infected_people = infected_people
with pytest.raises(ValueError, match=error):
baseline_form.validate(data_registry)
baseline_form.validate()
def present_times(interval: models.Interval) -> models.BoundarySequence_t:
@ -275,7 +275,7 @@ def test_exposed_present_lunch_end_before_beginning(baseline_form: model_generat
baseline_form.exposed_lunch_start = minutes_since_midnight(14 * 60)
baseline_form.exposed_lunch_finish = minutes_since_midnight(13 * 60)
with pytest.raises(ValueError):
baseline_form.validate(data_registry)
baseline_form.validate()
@pytest.mark.parametrize(
@ -291,7 +291,7 @@ def test_exposed_presence_lunch_break(baseline_form: model_generator.VirusFormDa
baseline_form.exposed_lunch_start = minutes_since_midnight(exposed_lunch_start * 60)
baseline_form.exposed_lunch_finish = minutes_since_midnight(exposed_lunch_finish * 60)
with pytest.raises(ValueError, match='exposed lunch break must be within presence times.'):
baseline_form.validate(data_registry)
baseline_form.validate()
@pytest.mark.parametrize(
@ -307,7 +307,7 @@ def test_infected_presence_lunch_break(baseline_form: model_generator.VirusFormD
baseline_form.infected_lunch_start = minutes_since_midnight(infected_lunch_start * 60)
baseline_form.infected_lunch_finish = minutes_since_midnight(infected_lunch_finish * 60)
with pytest.raises(ValueError, match='infected lunch break must be within presence times.'):
baseline_form.validate(data_registry)
baseline_form.validate()
def test_exposed_breaks_length(baseline_form: model_generator.VirusFormData, data_registry: DataRegistry):
@ -317,7 +317,7 @@ def test_exposed_breaks_length(baseline_form: model_generator.VirusFormData, dat
baseline_form.exposed_finish = minutes_since_midnight(11 * 60)
baseline_form.exposed_lunch_option = False
with pytest.raises(ValueError, match='Length of breaks >= Length of exposed presence.'):
baseline_form.validate(data_registry)
baseline_form.validate()
def test_infected_breaks_length(baseline_form: model_generator.VirusFormData, data_registry: DataRegistry):
@ -328,7 +328,7 @@ def test_infected_breaks_length(baseline_form: model_generator.VirusFormData, da
baseline_form.infected_coffee_break_option = 'coffee_break_4'
baseline_form.infected_coffee_duration = 30
with pytest.raises(ValueError, match='Length of breaks >= Length of infected presence.'):
baseline_form.validate(data_registry)
baseline_form.validate()
@pytest.fixture
@ -440,7 +440,7 @@ def test_valid_no_lunch(baseline_form: model_generator.VirusFormData, data_regis
baseline_form.exposed_lunch_option = False
baseline_form.exposed_lunch_start = minutes_since_midnight(0)
baseline_form.exposed_lunch_finish = minutes_since_midnight(0)
assert baseline_form.validate(data_registry) is None
assert baseline_form.validate() is None
def test_no_breaks(baseline_form: model_generator.VirusFormData):
@ -516,7 +516,7 @@ def test_natural_ventilation_window_opening_periodically(baseline_form: model_ge
baseline_form.windows_duration = 20
baseline_form.windows_frequency = 10
with pytest.raises(ValueError, match='Duration cannot be bigger than frequency.'):
baseline_form.validate(data_registry)
baseline_form.validate()
def test_key_validation_mech_ventilation_type_na(baseline_form_data, data_registry):

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

@ -17,7 +17,7 @@ from caimira.store.data_registry import DataRegistry
def test_specific_break_structure(break_input, error, baseline_form: model_generator.VirusFormData, data_registry: DataRegistry):
baseline_form.specific_breaks = break_input
with pytest.raises(TypeError, match=error):
baseline_form.validate(data_registry)
baseline_form.validate()
@pytest.mark.parametrize(
@ -34,7 +34,7 @@ def test_specific_break_structure(break_input, error, baseline_form: model_gener
def test_specific_population_break_data_structure(population_break_input, error, baseline_form: model_generator.VirusFormData, data_registry: DataRegistry):
baseline_form.specific_breaks = {'exposed_breaks': population_break_input, 'infected_breaks': population_break_input}
with pytest.raises(TypeError, match=error):
baseline_form.validate(data_registry)
baseline_form.validate()
@pytest.mark.parametrize(
@ -68,7 +68,7 @@ def test_specific_break_time(break_input, error, baseline_form: model_generator.
def test_precise_activity_structure(precise_activity_input, error, baseline_form: model_generator.VirusFormData, data_registry: DataRegistry):
baseline_form.precise_activity = precise_activity_input
with pytest.raises(TypeError, match=error):
baseline_form.validate(data_registry)
baseline_form.validate()
@pytest.mark.parametrize(
@ -80,7 +80,7 @@ def test_precise_activity_structure(precise_activity_input, error, baseline_form
[{"physical_activity": "Light activity", "respiratory_activity": [{"type": "Breathing", "percentage": 50}]}, 'The sum of all respiratory activities should be 100. Got 50.'],
]
)
def test_sum_precise_activity(precise_activity_input, error, baseline_form: model_generator.VirusFormData, data_registry: DataRegistry):
def test_sum_precise_activity(precise_activity_input, error, baseline_form: model_generator.VirusFormData):
baseline_form.precise_activity = precise_activity_input
with pytest.raises(ValueError, match=error):
baseline_form.validate(data_registry)
baseline_form.validate()

8
caimira/tests/models/test_concentration_model.py
View file

@ -4,9 +4,9 @@ import numpy as np
import numpy.testing as npt
import pytest
from dataclasses import dataclass
import typing
from caimira import models
from caimira.store.data_registry import DataRegistry
@dataclass(frozen=True)
class KnownConcentrationModelBase(models._ConcentrationModelBase):
@ -198,12 +198,14 @@ def test_integrated_concentration(simple_conc_model):
]
)
def test_normed_integrated_concentration_with_background_concentration(
data_registry: DataRegistry,
simple_conc_model: models.ConcentrationModel,
dummy_population: models.Population,
known_min_background_concentration: float,
expected_normed_integrated_concentration: float):
known_conc_model = KnownConcentrationModelBase(
data_registry,
room = simple_conc_model.room,
ventilation = simple_conc_model.ventilation,
known_population = dummy_population,
@ -229,6 +231,7 @@ def test_normed_integrated_concentration_with_background_concentration(
]
)
def test_normed_integrated_concentration_vectorisation(
data_registry: DataRegistry,
simple_conc_model: models.ConcentrationModel,
dummy_population: models.Population,
known_removal_rate: float,
@ -237,6 +240,7 @@ def test_normed_integrated_concentration_vectorisation(
expected_normed_integrated_concentration: float):
known_conc_model = KnownConcentrationModelBase(
data_registry = data_registry,
room = simple_conc_model.room,
ventilation = simple_conc_model.ventilation,
known_population = dummy_population,
@ -264,6 +268,7 @@ def test_normed_integrated_concentration_vectorisation(
]
)
def test_zero_ventilation_rate(
data_registry: DataRegistry,
simple_conc_model: models.ConcentrationModel,
dummy_population: models.Population,
known_removal_rate: float,
@ -271,6 +276,7 @@ def test_zero_ventilation_rate(
expected_concentration: float):
known_conc_model = KnownConcentrationModelBase(
data_registry = data_registry,
room = simple_conc_model.room,
ventilation = simple_conc_model.ventilation,
known_population = dummy_population,

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

@ -10,6 +10,7 @@ import caimira.dataclass_utils as dc_utils
@pytest.fixture
def full_exposure_model(data_registry):
return models.ExposureModel(
data_registry=data_registry,
concentration_model=models.ConcentrationModel(
data_registry=data_registry,
room=models.Room(volume=100),
@ -25,6 +26,7 @@ def full_exposure_model(data_registry):
virus=models.Virus.types['SARS_CoV_2'],
host_immunity=0.
),
evaporation_factor=0.3,
),
short_range=(),
exposed=models.Population(
@ -148,12 +150,13 @@ def test_linearity_with_number_of_infected(full_exposure_model: models.ExposureM
@pytest.mark.parametrize(
"time", (8., 9., 10., 11., 12., 13., 14.),
)
def test_dynamic_dose(full_exposure_model: models.ExposureModel, time: float):
def test_dynamic_dose(data_registry, full_exposure_model: models.ExposureModel, time: float):
dynamic_infected: models.ExposureModel = dc_utils.nested_replace(
full_exposure_model,
{
'concentration_model.infected': models.InfectedPopulation(
data_registry=data_registry,
number=models.IntPiecewiseConstant(
(8, 10, 12, 13, 17), (1, 2, 0, 3)),
presence=None,

52
caimira/tests/models/test_exposure_model.py
View file

@ -71,7 +71,7 @@ def known_concentrations(func, data_registry=DataRegistry()):
)
normed_func = lambda x: (func(x) /
dummy_infected_population.emission_rate_per_person_when_present())
return KnownNormedconcentration(dummy_room, dummy_ventilation,
return KnownNormedconcentration(data_registry, dummy_room, dummy_ventilation,
dummy_infected_population, 0.3, normed_func)
@ -92,9 +92,9 @@ def known_concentrations(func, data_registry=DataRegistry()):
[populations[2], known_concentrations(lambda t: np.array([18., 36.])),
np.array([40.91708675, 91.46172332]), np.array([51.6749232285, 80.3196524031])],
])
def test_exposure_model_ndarray(population, cm,
def test_exposure_model_ndarray(data_registry, population, cm,
expected_exposure, expected_probability, sr_model, cases_model):
model = ExposureModel(cm, sr_model, population, cases_model)
model = ExposureModel(data_registry, cm, sr_model, population, cases_model)
np.testing.assert_almost_equal(
model.deposited_exposure(), expected_exposure
)
@ -113,10 +113,10 @@ def test_exposure_model_ndarray(population, cm,
[populations[1], np.array([2.13410688, 1.98167067])],
[populations[2], np.array([1.36390289, 1.52436206])],
])
def test_exposure_model_ndarray_and_float_mix(population, expected_deposited_exposure, sr_model, cases_model):
def test_exposure_model_ndarray_and_float_mix(data_registry, population, expected_deposited_exposure, sr_model, cases_model):
cm = known_concentrations(
lambda t: 0. if np.floor(t) % 2 else np.array([0.6, 0.6]))
model = ExposureModel(cm, sr_model, population, cases_model)
model = ExposureModel(data_registry, cm, sr_model, population, cases_model)
np.testing.assert_almost_equal(
model.deposited_exposure(), expected_deposited_exposure
@ -131,17 +131,17 @@ def test_exposure_model_ndarray_and_float_mix(population, expected_deposited_exp
[populations[1], np.array([2.13410688, 1.98167067])],
[populations[2], np.array([1.36390289, 1.52436206])],
])
def test_exposure_model_vector(population, expected_deposited_exposure, sr_model, cases_model):
def test_exposure_model_vector(data_registry, population, expected_deposited_exposure, sr_model, cases_model):
cm_array = known_concentrations(lambda t: np.array([0.6, 0.6]))
model_array = ExposureModel(cm_array, sr_model, population, cases_model)
model_array = ExposureModel(data_registry, cm_array, sr_model, population, cases_model)
np.testing.assert_almost_equal(
model_array.deposited_exposure(), np.array(expected_deposited_exposure)
)
def test_exposure_model_scalar(sr_model, cases_model):
def test_exposure_model_scalar(data_registry, sr_model, cases_model):
cm_scalar = known_concentrations(lambda t: 0.6)
model_scalar = ExposureModel(cm_scalar, sr_model, populations[0], cases_model)
model_scalar = ExposureModel(data_registry, cm_scalar, sr_model, populations[0], cases_model)
expected_deposited_exposure = 1.52436206
np.testing.assert_almost_equal(
model_scalar.deposited_exposure(), expected_deposited_exposure
@ -185,7 +185,7 @@ def diameter_dependent_model(conc_model, data_registry) -> models.InfectedPopula
virus=models.Virus.types['SARS_CoV_2_DELTA'],
mask=models.Mask.types['No mask'],
activity=models.Activity.types['Seated'],
expiration=expiration_distributions['Breathing'],
expiration=expiration_distributions(data_registry)['Breathing'],
host_immunity=0.,
))
@ -213,14 +213,14 @@ def cases_model():
[(0., 24.), 645.8401125684933],
]
)
def test_exposure_model_integral_accuracy(exposed_time_interval,
def test_exposure_model_integral_accuracy(data_registry, exposed_time_interval,
expected_deposited_exposure, conc_model, sr_model, cases_model):
presence_interval = models.SpecificInterval((exposed_time_interval,))
population = models.Population(
10, presence_interval, models.Activity.types['Standing'],
models.Mask.types['Type I'], 0.,
)
model = ExposureModel(conc_model, sr_model, population, cases_model)
model = ExposureModel(data_registry, conc_model, sr_model, population, cases_model)
np.testing.assert_allclose(model.deposited_exposure(), expected_deposited_exposure)
@ -248,7 +248,7 @@ def test_infectious_dose_vectorisation(sr_model, cases_model, data_registry):
10, presence_interval, models.Activity.types['Standing'],
models.Mask.types['Type I'], 0.,
)
model = ExposureModel(cm, sr_model, population, cases_model)
model = ExposureModel(data_registry, cm, sr_model, population, cases_model)
inf_probability = model.infection_probability()
assert isinstance(inf_probability, np.ndarray)
assert inf_probability.shape == (3, )
@ -304,13 +304,13 @@ def test_prob_meet_infected_person(pop, cases, AB, exposed, infected, prob_meet_
[30, known_concentrations(lambda t: 0.6),
100000, 68, 5, 55.93154502],
])
def test_probabilistic_exposure_probability(sr_model, exposed_population, cm,
def test_probabilistic_exposure_probability(data_registry, sr_model, exposed_population, cm,
pop, AB, cases, probabilistic_exposure_probability):
population = models.Population(
exposed_population, models.PeriodicInterval(120, 60), models.Activity.types['Standing'],
models.Mask.types['Type I'], host_immunity=0.,)
model = ExposureModel(cm, sr_model, population, models.Cases(geographic_population=pop,
model = ExposureModel(data_registry, cm, sr_model, population, models.Cases(geographic_population=pop,
geographic_cases=cases, ascertainment_bias=AB),)
np.testing.assert_allclose(
model.total_probability_rule(), probabilistic_exposure_probability, rtol=0.05
@ -342,10 +342,10 @@ def test_diameter_vectorisation_window_opening(data_registry, diameter_dependent
)
with pytest.raises(ValueError, match="If the diameter is an array, none of the ventilation parameters "
"or virus decay constant can be arrays at the same time."):
models.ExposureModel(concentration, sr_model, populations[0], cases_model)
models.ExposureModel(data_registry, concentration, sr_model, populations[0], cases_model)
def test_diameter_vectorisation_hinged_window(diameter_dependent_model, sr_model, cases_model):
def test_diameter_vectorisation_hinged_window(data_registry, diameter_dependent_model, sr_model, cases_model):
# Verify (ventilation) window_width vectorisation.
concentration = replace(diameter_dependent_model,
ventilation = models.HingedWindow(active=models.PeriodicInterval(period=120, duration=120),
@ -356,10 +356,10 @@ def test_diameter_vectorisation_hinged_window(diameter_dependent_model, sr_model
)
with pytest.raises(ValueError, match="If the diameter is an array, none of the ventilation parameters "
"or virus decay constant can be arrays at the same time."):
models.ExposureModel(concentration, sr_model, populations[0], cases_model)
models.ExposureModel(data_registry, concentration, sr_model, populations[0], cases_model)
def test_diameter_vectorisation_HEPA_filter(diameter_dependent_model, sr_model, cases_model):
def test_diameter_vectorisation_HEPA_filter(data_registry, diameter_dependent_model, sr_model, cases_model):
# Verify (ventilation) q_air_mech vectorisation.
concentration = replace(diameter_dependent_model,
ventilation = models.HEPAFilter(active=models.PeriodicInterval(period=120, duration=120),
@ -367,10 +367,10 @@ def test_diameter_vectorisation_HEPA_filter(diameter_dependent_model, sr_model,
)
with pytest.raises(ValueError, match="If the diameter is an array, none of the ventilation parameters "
"or virus decay constant can be arrays at the same time."):
models.ExposureModel(concentration, sr_model, populations[1], cases_model)
models.ExposureModel(data_registry, concentration, sr_model, populations[1], cases_model)
def test_diameter_vectorisation_air_change(diameter_dependent_model, sr_model, cases_model):
def test_diameter_vectorisation_air_change(data_registry, diameter_dependent_model, sr_model, cases_model):
# Verify (ventilation) air_exch vectorisation.
concentration = replace(diameter_dependent_model,
ventilation = models.AirChange(active=models.PeriodicInterval(period=120, duration=120),
@ -378,7 +378,7 @@ def test_diameter_vectorisation_air_change(diameter_dependent_model, sr_model, c
)
with pytest.raises(ValueError, match="If the diameter is an array, none of the ventilation parameters "
"or virus decay constant can be arrays at the same time."):
models.ExposureModel(concentration, sr_model, populations[2], cases_model)
models.ExposureModel(data_registry, concentration, sr_model, populations[2], cases_model)
@pytest.mark.parametrize(
@ -394,12 +394,12 @@ def test_diameter_vectorisation_air_change(diameter_dependent_model, sr_model, c
"can be arrays at the same time."], # Verify room humidity vectorisation
]
)
def test_diameter_vectorisation_room(diameter_dependent_model, sr_model, cases_model, volume, inside_temp, humidity, error_message):
def test_diameter_vectorisation_room(data_registry, diameter_dependent_model, sr_model, cases_model, volume, inside_temp, humidity, error_message):
concentration = replace(diameter_dependent_model,
room = models.Room(volume=volume, inside_temp=inside_temp, humidity=humidity),
ventilation = models.HVACMechanical(active=models.SpecificInterval(((0., 24.), )), q_air_mech=100.))
with pytest.raises(ValueError, match=error_message):
models.ExposureModel(concentration, sr_model, populations[0], cases_model)
models.ExposureModel(data_registry, concentration, sr_model, populations[0], cases_model)
@pytest.mark.parametrize(
@ -409,12 +409,12 @@ def test_diameter_vectorisation_room(diameter_dependent_model, sr_model, cases_m
[known_concentrations(lambda t: 18.), np.array([0., 1.]), np.array([67.95037626, 0.])],
]
)
def test_host_immunity_vectorisation(sr_model, cases_model, cm, host_immunity, expected_probability):
def test_host_immunity_vectorisation(data_registry, sr_model, cases_model, cm, host_immunity, expected_probability):
population = models.Population(
10, halftime, models.Activity.types['Standing'],
models.Mask(np.array([0.3, 0.35])), host_immunity=host_immunity
)
model = ExposureModel(cm, sr_model, population, cases_model)
model = ExposureModel(data_registry, cm, sr_model, population, cases_model)
inf_probability = model.infection_probability()
np.testing.assert_almost_equal(

3
caimira/tests/models/test_fitting_algorithm.py
View file

@ -16,8 +16,9 @@ from caimira import models
['Standing', [8, 17], [2.45]],
]
)
def test_fitting_algorithm(activity_type, ventilation_active, air_exch):
def test_fitting_algorithm(data_registry, activity_type, ventilation_active, air_exch):
conc_model = models.CO2ConcentrationModel(
data_registry = data_registry,
room=models.Room(
volume=75, inside_temp=models.PiecewiseConstant((0., 24.), (293,))),
ventilation=models.CustomVentilation(models.PiecewiseConstant(

11
caimira/tests/models/test_short_range_model.py
View file

@ -38,10 +38,10 @@ def concentration_model(data_registry) -> mc_models.ConcentrationModel:
@pytest.fixture
def short_range_model(data_registry):
return mc_models.ShortRangeModel(data_registry=data_registry,
expiration=short_range_expiration_distributions['Breathing'],
expiration=short_range_expiration_distributions(data_registry)['Breathing'],
activity=activity_distributions(data_registry)['Seated'],
presence=models.SpecificInterval(present_times=((10.5, 11.0),)),
distance=short_range_distances)
distance=short_range_distances(data_registry))
def test_short_range_model_ndarray(concentration_model, short_range_model):
@ -66,7 +66,7 @@ def test_short_range_model_ndarray(concentration_model, short_range_model):
def test_dilution_factor(data_registry, activity, expected_dilution):
model = mc_models.ShortRangeModel(
data_registry=data_registry,
expiration=short_range_expiration_distributions['Breathing'],
expiration=short_range_expiration_distributions(data_registry)['Breathing'],
activity=models.Activity.types[activity],
presence=models.SpecificInterval(present_times=((10.5, 11.0),)),
distance=0.854
@ -135,17 +135,18 @@ def test_short_range_exposure_with_ndarray_mask(data_registry):
virus=models.Virus.types['SARS_CoV_2_DELTA'],
mask=models.Mask.types['No mask'],
activity=models.Activity.types['Seated'],
expiration=expiration_distributions['Breathing'],
expiration=expiration_distributions(data_registry)['Breathing'],
host_immunity=0.,
),
evaporation_factor=0.3,
)
sr_model = mc_models.ShortRangeModel(data_registry=data_registry,
expiration=short_range_expiration_distributions['Shouting'],
expiration=short_range_expiration_distributions(data_registry)['Shouting'],
activity=models.Activity.types['Heavy exercise'],
presence=models.SpecificInterval(present_times=((10.5, 11.0),)),
distance=0.854)
e_model = mc_models.ExposureModel(
data_registry = data_registry,
concentration_model = c_model,
short_range = (sr_model,),
exposed = mc_models.Population(

61
caimira/tests/test_full_algorithm.py
View file

@ -9,10 +9,9 @@ import pytest
from retry import retry
import caimira.monte_carlo as mc
from caimira import models,data
from caimira import models
from caimira.utils import method_cache
from caimira.models import _VectorisedFloat,Interval,SpecificInterval
from caimira.monte_carlo.sampleable import LogNormal
from caimira.monte_carlo.data import (expiration_distributions,
expiration_BLO_factors,short_range_expiration_distributions,
short_range_distances,virus_distributions,activity_distributions)
@ -479,7 +478,7 @@ def c_model(data_registry) -> mc.ConcentrationModel:
virus=models.Virus.types['SARS_CoV_2_DELTA'],
mask=models.Mask.types['No mask'],
activity=models.Activity.types['Seated'],
expiration=expiration_distributions['Breathing'],
expiration=expiration_distributions(data_registry)['Breathing'],
host_immunity=0.,
),
evaporation_factor=0.3,
@ -512,14 +511,14 @@ def sr_models(data_registry) -> typing.Tuple[mc.ShortRangeModel, ...]:
return (
mc.ShortRangeModel(
data_registry = data_registry,
expiration = short_range_expiration_distributions['Speaking'],
expiration = short_range_expiration_distributions(data_registry)['Speaking'],
activity = models.Activity.types['Seated'],
presence = interaction_intervals[0],
distance = 0.854,
),
mc.ShortRangeModel(
data_registry = data_registry,
expiration = short_range_expiration_distributions['Breathing'],
expiration = short_range_expiration_distributions(data_registry)['Breathing'],
activity = models.Activity.types['Heavy exercise'],
presence = interaction_intervals[1],
distance = 0.854,
@ -528,40 +527,41 @@ def sr_models(data_registry) -> typing.Tuple[mc.ShortRangeModel, ...]:
@pytest.fixture
def simple_c_model() -> SimpleConcentrationModel:
def simple_c_model(data_registry) -> SimpleConcentrationModel:
return SimpleConcentrationModel(
infected_presence = presence,
viral_load = models.Virus.types['SARS_CoV_2_DELTA'].viral_load_in_sputum,
breathing_rate = models.Activity.types['Seated'].exhalation_rate,
room_volume = 50.,
lambda_ventilation= 1.,
BLO_factors = expiration_BLO_factors['Breathing'],
BLO_factors = expiration_BLO_factors(data_registry)['Breathing'],
viable_to_RNA = models.Virus.types['SARS_CoV_2_DELTA'].viable_to_RNA_ratio,
HI = 0.,
)
@pytest.fixture
def simple_sr_models() -> typing.Tuple[SimpleShortRangeModel, ...]:
def simple_sr_models(data_registry) -> typing.Tuple[SimpleShortRangeModel, ...]:
return (
SimpleShortRangeModel(
interaction_interval = interaction_intervals[0],
distance = 0.854,
breathing_rate = models.Activity.types['Seated'].exhalation_rate,
BLO_factors = expiration_BLO_factors['Speaking'],
BLO_factors = expiration_BLO_factors(data_registry)['Speaking'],
),
SimpleShortRangeModel(
interaction_interval = interaction_intervals[1],
distance = 0.854,
breathing_rate = models.Activity.types['Heavy exercise'].exhalation_rate,
BLO_factors = expiration_BLO_factors['Breathing'],
BLO_factors = expiration_BLO_factors(data_registry)['Breathing'],
),
)
@pytest.fixture
def expo_sr_model(c_model,sr_models) -> mc.ExposureModel:
def expo_sr_model(data_registry, c_model,sr_models) -> mc.ExposureModel:
return mc.ExposureModel(
data_registry=data_registry,
concentration_model=c_model,
short_range=sr_models,
exposed=mc.Population(
@ -576,14 +576,14 @@ def expo_sr_model(c_model,sr_models) -> mc.ExposureModel:
@pytest.fixture
def simple_expo_sr_model(simple_sr_models) -> SimpleExposureModel:
def simple_expo_sr_model(data_registry, simple_sr_models) -> SimpleExposureModel:
return SimpleExposureModel(
infected_presence = presence,
viral_load = models.Virus.types['SARS_CoV_2_DELTA'].viral_load_in_sputum,
breathing_rate = models.Activity.types['Seated'].exhalation_rate,
room_volume = 50.,
lambda_ventilation= 1.,
BLO_factors = expiration_BLO_factors['Breathing'],
BLO_factors = expiration_BLO_factors(data_registry)['Breathing'],
viable_to_RNA = models.Virus.types['SARS_CoV_2_DELTA'].viable_to_RNA_ratio,
HI = 0.,
ID50 = models.Virus.types['SARS_CoV_2_DELTA'].infectious_dose,
@ -595,21 +595,22 @@ def simple_expo_sr_model(simple_sr_models) -> SimpleExposureModel:
@pytest.fixture
def expo_sr_model_distr(data_registry, c_model_distr) -> mc.ExposureModel:
return mc.ExposureModel(
data_registry=data_registry,
concentration_model=c_model_distr,
short_range=(
mc.ShortRangeModel(
data_registry = data_registry,
expiration = short_range_expiration_distributions['Breathing'],
expiration = short_range_expiration_distributions(data_registry)['Breathing'],
activity = activity_distributions(data_registry)['Seated'],
presence = interaction_intervals[0],
distance = short_range_distances,
distance = short_range_distances(data_registry),
),
mc.ShortRangeModel(
data_registry = data_registry,
expiration = short_range_expiration_distributions['Speaking'],
expiration = short_range_expiration_distributions(data_registry)['Speaking'],
activity = activity_distributions(data_registry)['Seated'],
presence = interaction_intervals[1],
distance = short_range_distances,
distance = short_range_distances(data_registry),
),
),
exposed=mc.Population(
@ -633,7 +634,7 @@ def simple_expo_sr_model_distr(data_registry) -> SimpleExposureModel:
SAMPLE_SIZE).exhalation_rate,
room_volume = 50.,
lambda_ventilation= 1.,
BLO_factors = expiration_BLO_factors['Breathing'],
BLO_factors = expiration_BLO_factors(data_registry)['Breathing'],
viable_to_RNA = virus_distributions(data_registry)['SARS_CoV_2_DELTA'
].build_model(SAMPLE_SIZE).viable_to_RNA_ratio,
HI = 0.,
@ -644,17 +645,17 @@ def simple_expo_sr_model_distr(data_registry) -> SimpleExposureModel:
sr_models = (
SimpleShortRangeModel(
interaction_interval = interaction_intervals[0],
distance = short_range_distances.generate_samples(SAMPLE_SIZE),
distance = short_range_distances(data_registry).generate_samples(SAMPLE_SIZE),
breathing_rate = activity_distributions(data_registry)['Seated'].build_model(
SAMPLE_SIZE).exhalation_rate,
BLO_factors = expiration_BLO_factors['Breathing'],
BLO_factors = expiration_BLO_factors(data_registry)['Breathing'],
),
SimpleShortRangeModel(
interaction_interval = interaction_intervals[1],
distance = short_range_distances.generate_samples(SAMPLE_SIZE),
distance = short_range_distances(data_registry).generate_samples(SAMPLE_SIZE),
breathing_rate = activity_distributions(data_registry)['Seated'].build_model(
SAMPLE_SIZE).exhalation_rate,
BLO_factors = expiration_BLO_factors['Speaking'],
BLO_factors = expiration_BLO_factors(data_registry)['Speaking'],
)
),
)
@ -687,14 +688,14 @@ def test_shortrange_concentration(time,c_model,simple_c_model,
)
def test_longrange_exposure(c_model):
def test_longrange_exposure(data_registry, c_model):
simple_expo_model = SimpleExposureModel(
infected_presence = presence,
viral_load = models.Virus.types['SARS_CoV_2_DELTA'].viral_load_in_sputum,
breathing_rate = models.Activity.types['Seated'].exhalation_rate,
room_volume = 50.,
lambda_ventilation= 1.,
BLO_factors = expiration_BLO_factors['Breathing'],
BLO_factors = expiration_BLO_factors(data_registry)['Breathing'],
viable_to_RNA = models.Virus.types['SARS_CoV_2_DELTA'].viable_to_RNA_ratio,
HI = 0.,
ID50 = models.Virus.types['SARS_CoV_2_DELTA'].infectious_dose,
@ -702,6 +703,7 @@ def test_longrange_exposure(c_model):
sr_models = (),
)
expo_model = mc.ExposureModel(
data_registry=data_registry,
concentration_model=c_model.build_model(SAMPLE_SIZE),
short_range=(),
exposed=mc.Population(
@ -735,8 +737,8 @@ def test_longrange_concentration_with_distributions(c_model_distr, time, data_re
SAMPLE_SIZE).exhalation_rate,
room_volume = 50.,
lambda_ventilation= 1.,
BLO_factors = expiration_BLO_factors['Breathing'],
viable_to_RNA = virus_distributions['SARS_CoV_2_DELTA'
BLO_factors = expiration_BLO_factors(data_registry)['Breathing'],
viable_to_RNA = virus_distributions(data_registry)['SARS_CoV_2_DELTA'
].build_model(SAMPLE_SIZE).viable_to_RNA_ratio,
HI = 0.,
)
@ -746,7 +748,7 @@ def test_longrange_concentration_with_distributions(c_model_distr, time, data_re
)
def test_longrange_exposure_with_distributions(c_model_distr):
def test_longrange_exposure_with_distributions(data_registry, c_model_distr):
simple_expo_model = SimpleExposureModel(
infected_presence = presence,
viral_load = virus_distributions(data_registry)['SARS_CoV_2_DELTA'
@ -755,7 +757,7 @@ def test_longrange_exposure_with_distributions(c_model_distr):
SAMPLE_SIZE).exhalation_rate,
room_volume = 50.,
lambda_ventilation= 1.,
BLO_factors = expiration_BLO_factors['Breathing'],
BLO_factors = expiration_BLO_factors(data_registry)['Breathing'],
viable_to_RNA = virus_distributions(data_registry)['SARS_CoV_2_DELTA'
].build_model(SAMPLE_SIZE).viable_to_RNA_ratio,
HI = 0.,
@ -766,13 +768,14 @@ def test_longrange_exposure_with_distributions(c_model_distr):
sr_models = (),
)
expo_model = mc.ExposureModel(
data_registry=data_registry,
concentration_model=c_model_distr.build_model(SAMPLE_SIZE),
short_range=(),
exposed=mc.Population(
number=1,
presence=presence,
mask=models.Mask.types['No mask'],
activity=activity_distributions['Seated'],
activity=activity_distributions(data_registry)['Seated'],
host_immunity=0.,
),
geographical_data=models.Cases(),

42
caimira/tests/test_known_quantities.py
View file

@ -84,11 +84,11 @@ def build_model(data_registry, interval_duration):
return model
def test_concentrations_startup():
def test_concentrations_startup(data_registry):
# The concentrations should be the same until the beginning of the
# first time that the ventilation is disabled.
m1 = build_model(interval_duration=120)
m2 = build_model(interval_duration=65)
m1 = build_model(data_registry, interval_duration=120)
m2 = build_model(data_registry, interval_duration=65)
assert m1.concentration(1.) == m2.concentration(1.)
@ -319,11 +319,11 @@ def build_hourly_dependent_model_multipleventilation(data_registry, month, inter
"time",
[0.5, 1.2, 2., 3.5, 5., 6.5, 7.5, 7.9, 8.],
)
def test_concentrations_hourly_dep_temp_vs_constant(month, temperatures, time):
def test_concentrations_hourly_dep_temp_vs_constant(data_registry, month, temperatures, time):
# The concentrations should be the same up to 8 AM (time when the
# temperature changes DURING the window opening).
m1 = build_hourly_dependent_model(month)
m2 = build_constant_temp_model(temperatures[7] + 273.15)
m1 = build_hourly_dependent_model(data_registry, month)
m2 = build_constant_temp_model(data_registry, temperatures[7] + 273.15)
npt.assert_allclose(m1.concentration(time), m2.concentration(time), rtol=1e-5)
@pytest.mark.parametrize(
@ -334,10 +334,11 @@ def test_concentrations_hourly_dep_temp_vs_constant(month, temperatures, time):
"time",
[0.5, 1.2, 2., 3.5, 5., 6.5, 7.5, 7.9, 8.],
)
def test_concentrations_hourly_dep_temp_startup(month, temperatures, time):
def test_concentrations_hourly_dep_temp_startup(data_registry, month, temperatures, time):
# The concentrations should be the zero up to the first presence time
# of an infected person.
m = build_hourly_dependent_model(
data_registry,
month,
((0., 0.5), (1., 1.5), (4., 4.5), (7.5, 8), ),
((8., 12.), ),
@ -345,8 +346,8 @@ def test_concentrations_hourly_dep_temp_startup(month, temperatures, time):
assert m.concentration(time) == 0.
def test_concentrations_hourly_dep_multipleventilation():
m = build_hourly_dependent_model_multipleventilation('Jan')
def test_concentrations_hourly_dep_multipleventilation(data_registry):
m = build_hourly_dependent_model_multipleventilation(data_registry, 'Jan')
m.concentration(12.)
@ -358,11 +359,11 @@ def test_concentrations_hourly_dep_multipleventilation():
"time",
[0.5, 1.2, 2., 3.5, 5., 6.5, 7.5, 7.9, 8., 8.5, 9., 12.],
)
def test_concentrations_hourly_dep_adding_artificial_transitions(month_temp_item, time):
def test_concentrations_hourly_dep_adding_artificial_transitions(data_registry, month_temp_item, time):
month, temperatures = month_temp_item
# Adding a second opening inside the first one should not change anything
m1 = build_hourly_dependent_model(month, intervals_open=((7.5, 8.5), ))
m2 = build_hourly_dependent_model(month, intervals_open=((7.5, 8.5), (8., 8.1), ))
m1 = build_hourly_dependent_model(data_registry, month, intervals_open=((7.5, 8.5), ))
m2 = build_hourly_dependent_model(data_registry, month, intervals_open=((7.5, 8.5), (8., 8.1), ))
npt.assert_allclose(m1.concentration(time), m2.concentration(time), rtol=1e-5)
@ -373,17 +374,18 @@ def test_concentrations_hourly_dep_adding_artificial_transitions(month_temp_item
+ [float(t) for t in np.arange(0, 24.5, 0.5)]
),
)
def test_concentrations_refine_times(time):
def test_concentrations_refine_times(data_registry, time):
month = 'Jan'
m1 = build_hourly_dependent_model(month, intervals_open=((0., 24.),))
m2 = build_hourly_dependent_model(month, intervals_open=((0., 24.),),
m1 = build_hourly_dependent_model(data_registry, month, intervals_open=((0., 24.),))
m2 = build_hourly_dependent_model(data_registry, month, intervals_open=((0., 24.),),
artificial_refinement=True)
npt.assert_allclose(m1.concentration(time), m2.concentration(time), rtol=1e-8)
def build_exposure_model(concentration_model, short_range_model):
def build_exposure_model(data_registry, concentration_model, short_range_model):
infected = concentration_model.infected
return models.ExposureModel(
data_registry=data_registry,
concentration_model=concentration_model,
short_range=short_range_model,
exposed=models.Population(
@ -406,9 +408,11 @@ def build_exposure_model(concentration_model, short_range_model):
['Jun', 1385.917562],
],
)
def test_exposure_hourly_dep(month,expected_deposited_exposure, baseline_sr_model):
def test_exposure_hourly_dep(data_registry, month, expected_deposited_exposure, baseline_sr_model):
m = build_exposure_model(
data_registry,
build_hourly_dependent_model(
data_registry,
month,
intervals_open=((0., 24.), ),
intervals_presence_infected=((8., 12.), (13., 17.))
@ -427,9 +431,11 @@ def test_exposure_hourly_dep(month,expected_deposited_exposure, baseline_sr_mode
['Jun', 1439.267381],
],
)
def test_exposure_hourly_dep_refined(month,expected_deposited_exposure, baseline_sr_model):
def test_exposure_hourly_dep_refined(data_registry, month, expected_deposited_exposure, baseline_sr_model):
m = build_exposure_model(
data_registry,
build_hourly_dependent_model(
data_registry,
month,
intervals_open=((0., 24.),),
intervals_presence_infected=((8., 12.), (13., 17.)),

9
caimira/tests/test_monte_carlo_full_models.py
View file

@ -69,6 +69,7 @@ def shared_office_mc(data_registry):
evaporation_factor=0.3,
)
return mc.ExposureModel(
data_registry=data_registry,
concentration_model=concentration_mc,
short_range=(),
exposed=mc.Population(
@ -115,6 +116,7 @@ def classroom_mc(data_registry):
evaporation_factor=0.3,
)
return mc.ExposureModel(
data_registry=data_registry,
concentration_model=concentration_mc,
short_range=(),
exposed=mc.Population(
@ -152,6 +154,7 @@ def ski_cabin_mc(data_registry):
evaporation_factor=0.3,
)
return mc.ExposureModel(
data_registry=data_registry,
concentration_model=concentration_mc,
short_range=(),
exposed=mc.Population(
@ -195,6 +198,7 @@ def skagit_chorale_mc(data_registry):
evaporation_factor=0.3,
)
return mc.ExposureModel(
data_registry=data_registry,
concentration_model=concentration_mc,
short_range=(),
exposed=mc.Population(
@ -238,6 +242,7 @@ def bus_ride_mc(data_registry):
evaporation_factor=0.3,
)
return mc.ExposureModel(
data_registry=data_registry,
concentration_model=concentration_mc,
short_range=(),
exposed=mc.Population(
@ -270,12 +275,13 @@ def gym_mc(data_registry):
presence=mc.SpecificInterval(((0., 1.),)),
mask=models.Mask.types["No mask"],
activity=activity_distributions(data_registry)['Heavy exercise'],
expiration=expiration_distributions['Breathing'],
expiration=expiration_distributions(data_registry)['Breathing'],
host_immunity=0.,
),
evaporation_factor=0.3,
)
return mc.ExposureModel(
data_registry=data_registry,
concentration_model=concentration_mc,
short_range=(),
exposed=mc.Population(
@ -314,6 +320,7 @@ def waiting_room_mc(data_registry):
evaporation_factor=0.3,
)
return mc.ExposureModel(
data_registry=data_registry,
concentration_model=concentration_mc,
short_range=(),
exposed=mc.Population(