Merge branch 'master' into model/humidity_function
This commit is contained in:
commit
645bdd0197
5 changed files with 364 additions and 125 deletions
|
|
@ -33,7 +33,7 @@ from .user import AuthenticatedUser, AnonymousUser
|
|||
# calculator version. If the calculator needs to make breaking changes (e.g. change
|
||||
# form attributes) then it can also increase its MAJOR version without needing to
|
||||
# increase the overall CARA version (found at ``cara.__version__``).
|
||||
__version__ = "4.1.1"
|
||||
__version__ = "4.1.2"
|
||||
|
||||
|
||||
class BaseRequestHandler(RequestHandler):
|
||||
|
|
|
|||
115
cara/models.py
115
cara/models.py
|
|
@ -1171,28 +1171,70 @@ class ShortRangeModel:
|
|||
# calculations for the same time (e.g. at state change times).
|
||||
return self._normed_concentration(concentration_model, time)
|
||||
|
||||
def normed_exposure_between_bounds(self, concentration_model: ConcentrationModel, time1: float, time2: float):
|
||||
@method_cache
|
||||
def extract_between_bounds(self, time1: float, time2: float) -> typing.Tuple[float,float]:
|
||||
"""
|
||||
Get the integrated short-range concentration of viruses in the air between the times start and stop,
|
||||
normalized by the virus viral load.
|
||||
Extract the bounds of the interval resulting from the
|
||||
intersection of [time1, time2] and the presence interval.
|
||||
If [time1, time2] has nothing common to the presence interval,
|
||||
we return (0, 0).
|
||||
Raise an error if time1 and time2 are not in ascending order.
|
||||
"""
|
||||
start_bound, stop_bound = self.presence.boundaries()[0]
|
||||
|
||||
jet_origin = self.expiration.jet_origin_concentration()
|
||||
dilution = self.dilution_factor()
|
||||
if time1>time2:
|
||||
raise ValueError("time1 must be less or equal to time2")
|
||||
|
||||
total_normed_concentration_diluted = (
|
||||
concentration_model.integrated_concentration(start_bound,
|
||||
stop_bound)/dilution/
|
||||
concentration_model.virus.viral_load_in_sputum
|
||||
start, stop = self.presence.boundaries()[0]
|
||||
if (stop < time1) or (start > time2):
|
||||
return (0, 0)
|
||||
elif start <= time1 and time2<= stop:
|
||||
return time1, time2
|
||||
elif start <= time1 and stop < time2:
|
||||
return time1, stop
|
||||
elif time1 < start and time2 <= stop:
|
||||
return start, time2
|
||||
elif time1 <= start and stop < time2:
|
||||
return start, stop
|
||||
|
||||
def _normed_jet_exposure_between_bounds(self,
|
||||
concentration_model: ConcentrationModel,
|
||||
time1: float, time2: float):
|
||||
"""
|
||||
Get the part of the integrated short-range concentration of
|
||||
viruses in the air, between the times start and stop, coming
|
||||
from the jet concentration, normalized by the viral load, and
|
||||
without dilution.
|
||||
"""
|
||||
start, stop = self.extract_between_bounds(time1, time2)
|
||||
jet_origin = self.expiration.jet_origin_concentration()
|
||||
return jet_origin * (stop - start)
|
||||
|
||||
def _normed_interpolated_longrange_exposure_between_bounds(
|
||||
self, concentration_model: ConcentrationModel,
|
||||
time1: float, time2: float):
|
||||
"""
|
||||
Get the part of the integrated short-range concentration due
|
||||
to the background concentration, normalized by the viral load
|
||||
and the breathing rate, and without dilution.
|
||||
One needs to interpolate the integrated long-range concentration
|
||||
for the particle diameters defined here.
|
||||
TODO: make sure any potential extrapolation has a
|
||||
negligible effect.
|
||||
"""
|
||||
start, stop = self.extract_between_bounds(time1, time2)
|
||||
if stop<=start:
|
||||
return 0.
|
||||
|
||||
normed_int_concentration = (
|
||||
concentration_model.integrated_concentration(start, stop)
|
||||
/concentration_model.virus.viral_load_in_sputum
|
||||
/concentration_model.infected.activity.exhalation_rate
|
||||
)
|
||||
total_normed_concentration_interpolated = np.interp(
|
||||
normed_int_concentration_interpolated = np.interp(
|
||||
self.expiration.particle.diameter,
|
||||
concentration_model.infected.particle.diameter,
|
||||
total_normed_concentration_diluted
|
||||
normed_int_concentration
|
||||
)
|
||||
return (jet_origin/dilution * (stop_bound - start_bound)
|
||||
) - total_normed_concentration_interpolated
|
||||
return normed_int_concentration_interpolated
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
|
|
@ -1270,7 +1312,7 @@ class ExposureModel:
|
|||
# we compute first the mean of all diameter-dependent quantities
|
||||
# to perform properly the Monte-Carlo integration over
|
||||
# particle diameters (doing things in another order would
|
||||
# lead to wrong results).
|
||||
# lead to wrong results for the probability of infection).
|
||||
dep_exposure_integrated = np.array(self._long_range_normed_exposure_between_bounds(time1, time2) *
|
||||
aerosols *
|
||||
fdep).mean()
|
||||
|
|
@ -1300,46 +1342,45 @@ class ExposureModel:
|
|||
"""
|
||||
deposited_exposure = 0.
|
||||
for interaction in self.short_range:
|
||||
start, stop = interaction.presence.boundaries()[0]
|
||||
if stop < time1:
|
||||
continue
|
||||
elif start > time2:
|
||||
break
|
||||
elif start <= time1 and time2<= stop:
|
||||
start_bound, stop_bound = time1, time2
|
||||
elif start <= time1 and stop < time2:
|
||||
start_bound, stop_bound = time1, stop
|
||||
elif time1 < start and time2 <= stop:
|
||||
start_bound, stop_bound = start, time2
|
||||
elif time1 <= start and stop < time2:
|
||||
start_bound, stop_bound = start, stop
|
||||
short_range_exposure = interaction.normed_exposure_between_bounds(self.concentration_model, start_bound, stop_bound)
|
||||
start, stop = interaction.extract_between_bounds(time1, time2)
|
||||
short_range_jet_exposure = interaction._normed_jet_exposure_between_bounds(
|
||||
self.concentration_model, start, stop)
|
||||
short_range_lr_exposure = interaction._normed_interpolated_longrange_exposure_between_bounds(
|
||||
self.concentration_model, start, stop)
|
||||
dilution = interaction.dilution_factor()
|
||||
|
||||
fdep = interaction.expiration.particle.fraction_deposited(evaporation_factor=1.0)
|
||||
diameter = interaction.expiration.particle.diameter
|
||||
|
||||
# Aerosols not considered given the formula for the initial concentration at mouth/nose.
|
||||
# Aerosols not considered given the formula for the initial
|
||||
# concentration at mouth/nose.
|
||||
if diameter is not None and not np.isscalar(diameter):
|
||||
# we compute first the mean of all diameter-dependent quantities
|
||||
# to perform properly the Monte-Carlo integration over
|
||||
# particle diameters (doing things in another order would
|
||||
# lead to wrong results).
|
||||
deposited_exposure += np.array(short_range_exposure *
|
||||
fdep).mean()
|
||||
# lead to wrong results for the probability of infection).
|
||||
this_deposited_exposure = (np.array(short_range_jet_exposure
|
||||
* fdep).mean()
|
||||
- np.array(short_range_lr_exposure * fdep).mean()
|
||||
* self.concentration_model.infected.activity.exhalation_rate)
|
||||
else:
|
||||
# in the case of a single diameter or no diameter defined,
|
||||
# one should not take any mean at this stage.
|
||||
deposited_exposure += short_range_exposure*fdep
|
||||
this_deposited_exposure = (short_range_jet_exposure * fdep
|
||||
- short_range_lr_exposure * fdep
|
||||
* self.concentration_model.infected.activity.exhalation_rate)
|
||||
|
||||
# multiply by the (diameter-independent) inhalation rate
|
||||
deposited_exposure *= interaction.activity.inhalation_rate
|
||||
deposited_exposure += (this_deposited_exposure *
|
||||
interaction.activity.inhalation_rate
|
||||
/dilution)
|
||||
|
||||
# then we multiply by diameter-independent quantities: viral load
|
||||
# and fraction of infected virions
|
||||
f_inf = self.concentration_model.infected.fraction_of_infectious_virus()
|
||||
deposited_exposure *= (f_inf
|
||||
* self.concentration_model.virus.viral_load_in_sputum
|
||||
)
|
||||
* (1 - self.exposed.mask.inhale_efficiency()))
|
||||
# long-range concentration
|
||||
deposited_exposure += self.long_range_deposited_exposure_between_bounds(time1, time2)
|
||||
|
||||
|
|
|
|||
|
|
@ -6,10 +6,13 @@ import pytest
|
|||
from cara import models
|
||||
import cara.monte_carlo as mc_models
|
||||
from cara.apps.calculator.model_generator import build_expiration
|
||||
from cara.monte_carlo.data import short_range_expiration_distributions, short_range_distances, activity_distributions
|
||||
from cara.monte_carlo.data import short_range_expiration_distributions,\
|
||||
expiration_distributions, short_range_distances, activity_distributions
|
||||
|
||||
# TODO: seed better the random number generators
|
||||
np.random.seed(2000)
|
||||
SAMPLE_SIZE = 250_000
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def concentration_model() -> mc_models.ConcentrationModel:
|
||||
|
|
@ -41,11 +44,12 @@ def short_range_model():
|
|||
|
||||
|
||||
def test_short_range_model_ndarray(concentration_model, short_range_model):
|
||||
concentration_model = concentration_model.build_model(250_000)
|
||||
model = short_range_model.build_model(250_000)
|
||||
concentration_model = concentration_model.build_model(SAMPLE_SIZE)
|
||||
model = short_range_model.build_model(SAMPLE_SIZE)
|
||||
assert isinstance(model._normed_concentration(concentration_model, 10.75), np.ndarray)
|
||||
assert isinstance(model.short_range_concentration(concentration_model, 10.75), np.ndarray)
|
||||
assert isinstance(model.normed_exposure_between_bounds(concentration_model, 10.75, 10.85), np.ndarray)
|
||||
assert isinstance(model._normed_jet_exposure_between_bounds(concentration_model, 10.75, 10.85), np.ndarray)
|
||||
assert isinstance(model._normed_interpolated_longrange_exposure_between_bounds(concentration_model, 10.75, 10.85), np.ndarray)
|
||||
assert isinstance(model.short_range_concentration(concentration_model, 14.0), float)
|
||||
|
||||
|
||||
|
|
@ -59,16 +63,42 @@ def test_short_range_model_ndarray(concentration_model, short_range_model):
|
|||
]
|
||||
)
|
||||
def test_dilution_factor(activity, expected_dilution):
|
||||
model = models.ShortRangeModel(expiration="Breathing",
|
||||
model = mc_models.ShortRangeModel(expiration=short_range_expiration_distributions['Breathing'],
|
||||
activity=models.Activity.types[activity],
|
||||
presence=models.SpecificInterval(present_times=((10.5, 11.0),)),
|
||||
distance=0.854)
|
||||
distance=0.854).build_model(SAMPLE_SIZE)
|
||||
assert isinstance(model.dilution_factor(), np.ndarray)
|
||||
np.testing.assert_almost_equal(
|
||||
model.dilution_factor(), expected_dilution, decimal=10
|
||||
)
|
||||
|
||||
|
||||
def test_extract_between_bounds_raise_on_wrong_order(short_range_model):
|
||||
model = short_range_model.build_model(1)
|
||||
with pytest.raises(ValueError, match='time1 must be less or equal to time2'):
|
||||
model.extract_between_bounds(11.,10.)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"time1, time2, expected_start, expected_stop", [
|
||||
[10., 12., 10.5, 11.],
|
||||
[10., 10.7, 10.5, 10.7],
|
||||
[10., 10.45, 0., 0.],
|
||||
[11.01, 11.5, 0., 0.],
|
||||
[10.8, 10.9, 10.8, 10.9],
|
||||
[10.8, 11.5, 10.8, 11.],
|
||||
[10.5, 11., 10.5, 11.],
|
||||
]
|
||||
)
|
||||
def test_extract_between_bounds(short_range_model, time1, time2,
|
||||
expected_start, expected_stop):
|
||||
model = short_range_model.build_model(1)
|
||||
np.testing.assert_equal(
|
||||
model.extract_between_bounds(time1, time2),
|
||||
(expected_start, expected_stop),
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"time, expected_short_range_concentration", [
|
||||
[8.5, 0.],
|
||||
|
|
@ -78,28 +108,50 @@ def test_dilution_factor(activity, expected_dilution):
|
|||
[12.0, 0.],
|
||||
]
|
||||
)
|
||||
def test_short_range_concentration(time, expected_short_range_concentration, concentration_model, short_range_model):
|
||||
concentration_model = concentration_model.build_model(250_000)
|
||||
model = short_range_model.build_model(250_000)
|
||||
def test_short_range_concentration(time, expected_short_range_concentration,
|
||||
concentration_model, short_range_model):
|
||||
concentration_model = concentration_model.build_model(SAMPLE_SIZE)
|
||||
model = short_range_model.build_model(SAMPLE_SIZE)
|
||||
np.testing.assert_allclose(
|
||||
np.array(model.short_range_concentration(concentration_model, time)).mean(),
|
||||
expected_short_range_concentration, rtol=0.01
|
||||
expected_short_range_concentration, rtol=0.02
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"start, stop, expected_exposure", [
|
||||
[8.5, 12.5, 7.875963317294013e-09],
|
||||
[10.5, 11.0, 7.875963317294013e-09],
|
||||
[10.4, 11.1, 7.875963317294013e-09],
|
||||
[10.5, 11.1, 7.875963317294013e-09],
|
||||
[10.6, 11.1, 7.66539809488759e-09],
|
||||
[10.4, 10.9, 7.66539809488759e-09],
|
||||
|
||||
]
|
||||
)
|
||||
def test_normed_exposure_between_bounds(start, stop, expected_exposure, concentration_model, short_range_model):
|
||||
concentration_model = concentration_model.build_model(250_000)
|
||||
model = short_range_model.build_model(250_000)
|
||||
np.testing.assert_almost_equal(
|
||||
model.normed_exposure_between_bounds(concentration_model, start, stop).mean(), expected_exposure
|
||||
|
||||
def test_short_range_exposure_with_ndarray_mask():
|
||||
c_model = mc_models.ConcentrationModel(
|
||||
room=models.Room(volume=50, humidity=0.3),
|
||||
ventilation=models.AirChange(active=models.PeriodicInterval(period=120, duration=120),
|
||||
air_exch=10_000_000,),
|
||||
infected=mc_models.InfectedPopulation(
|
||||
number=1,
|
||||
presence=models.SpecificInterval(present_times=((8.5, 12.5), (13.5, 17.5))),
|
||||
virus=models.Virus.types['SARS_CoV_2_DELTA'],
|
||||
mask=models.Mask.types['No mask'],
|
||||
activity=models.Activity.types['Seated'],
|
||||
expiration=expiration_distributions['Breathing'],
|
||||
host_immunity=0.,
|
||||
),
|
||||
evaporation_factor=0.3,
|
||||
)
|
||||
sr_model = mc_models.ShortRangeModel(expiration=short_range_expiration_distributions['Shouting'],
|
||||
activity=models.Activity.types['Heavy exercise'],
|
||||
presence=models.SpecificInterval(present_times=((10.5, 11.0),)),
|
||||
distance=0.854)
|
||||
e_model = mc_models.ExposureModel(
|
||||
concentration_model = c_model,
|
||||
short_range = (sr_model,),
|
||||
exposed = mc_models.Population(
|
||||
number=1,
|
||||
presence=models.SpecificInterval(present_times=((8.5, 12.5), (13.5, 17.5))),
|
||||
mask=models.Mask(η_inhale=np.array([0., 0.3, 0.5])),
|
||||
activity=models.Activity.types['Light activity'],
|
||||
host_immunity=0.,
|
||||
),
|
||||
).build_model(SAMPLE_SIZE)
|
||||
assert isinstance(e_model.deposited_exposure(), np.ndarray)
|
||||
assert len(e_model.deposited_exposure()) == 3
|
||||
np.testing.assert_allclose(e_model.deposited_exposure(),
|
||||
e_model.deposited_exposure()[0]*np.array([1., 0.7, 0.5]),
|
||||
rtol=1e-8)
|
||||
|
||||
|
|
|
|||
|
|
@ -19,7 +19,7 @@ from cara.monte_carlo.data import (expiration_distributions,
|
|||
# TODO: seed better the random number generators
|
||||
np.random.seed(2000)
|
||||
SAMPLE_SIZE = 1_000_000
|
||||
TOLERANCE = 0.02
|
||||
TOLERANCE = 0.04
|
||||
|
||||
sqrt2pi = np.sqrt(2.*np.pi)
|
||||
sqrt2 = np.sqrt(2.)
|
||||
|
|
@ -248,7 +248,6 @@ class SimpleShortRangeModel:
|
|||
|
||||
return dilution
|
||||
|
||||
@method_cache
|
||||
def jet_concentration(self,conc_model: SimpleConcentrationModel) -> _VectorisedFloat:
|
||||
"""
|
||||
virion concentration at the origin of the jet (close to
|
||||
|
|
@ -271,13 +270,13 @@ class SimpleShortRangeModel:
|
|||
def concentration(self, conc_model: SimpleConcentrationModel, time: float) -> _VectorisedFloat:
|
||||
"""
|
||||
compute the short-range part of the concentration, and add it
|
||||
to the background concentration
|
||||
to the long-range concentration
|
||||
"""
|
||||
if self.interaction_interval.triggered(time):
|
||||
background_concentration = conc_model.concentration(time)
|
||||
lr_concentration = conc_model.concentration(time)
|
||||
S = self.dilution_factor()
|
||||
return (self.jet_concentration(conc_model)
|
||||
- background_concentration) / S
|
||||
- lr_concentration) / S
|
||||
else:
|
||||
return 0.
|
||||
|
||||
|
|
@ -356,8 +355,17 @@ class SimpleExposureModel(SimpleConcentrationModel):
|
|||
epsabs=0.,limit=500)[0]
|
||||
* self.viral_load * self.breathing_rate)
|
||||
|
||||
def total_concentration(self, t: float):
|
||||
"""
|
||||
total concentration at time t
|
||||
"""
|
||||
res = self.concentration(t)
|
||||
for sr_mod in self.sr_models:
|
||||
res += sr_mod.concentration(self,t)
|
||||
return res
|
||||
|
||||
@method_cache
|
||||
def integrated_background_concentration(self,t1: float,t2: float,
|
||||
def integrated_longrange_concentration(self,t1: float,t2: float,
|
||||
evaporation: float) -> _VectorisedFloat:
|
||||
"""
|
||||
background (long-range) concentration integrated from t1 to t2
|
||||
|
|
@ -421,7 +429,7 @@ class SimpleExposureModel(SimpleConcentrationModel):
|
|||
epsabs=0.,limit=500)[0]
|
||||
* self.viral_load * 1e-6 * (t2-t1) )
|
||||
result += sr_model.breathing_rate * (
|
||||
res-self.integrated_background_concentration(t1,t2,evaporation)
|
||||
res-self.integrated_longrange_concentration(t1,t2,evaporation)
|
||||
)/sr_model.dilution_factor()
|
||||
|
||||
return result
|
||||
|
|
@ -433,7 +441,7 @@ class SimpleExposureModel(SimpleConcentrationModel):
|
|||
"""
|
||||
result = 0.
|
||||
for t1,t2 in self.infected_presence.boundaries():
|
||||
result += (self.integrated_background_concentration(t1,t2,self.evaporation)
|
||||
result += (self.integrated_longrange_concentration(t1,t2,self.evaporation)
|
||||
* self.breathing_rate)
|
||||
|
||||
result += self.integrated_shortrange_concentration()
|
||||
|
|
@ -472,18 +480,37 @@ def c_model() -> mc.ConcentrationModel:
|
|||
).build_model(SAMPLE_SIZE)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def c_model_distr() -> mc.ConcentrationModel:
|
||||
return mc.ConcentrationModel(
|
||||
room=models.Room(volume=50, humidity=0.3),
|
||||
ventilation=models.AirChange(active=models.PeriodicInterval(
|
||||
period=120, duration=120), air_exch=1.),
|
||||
infected=mc.InfectedPopulation(
|
||||
number=1,
|
||||
presence=presence,
|
||||
virus=virus_distributions['SARS_CoV_2_DELTA'],
|
||||
mask=models.Mask.types['No mask'],
|
||||
activity=activity_distributions['Seated'],
|
||||
expiration=expiration_distributions['Breathing'],
|
||||
host_immunity=0.,
|
||||
),
|
||||
evaporation_factor=0.3,
|
||||
).build_model(SAMPLE_SIZE)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sr_models() -> typing.Tuple[mc.ShortRangeModel, ...]:
|
||||
return (
|
||||
mc.ShortRangeModel(
|
||||
expiration = short_range_expiration_distributions['Breathing'],
|
||||
expiration = short_range_expiration_distributions['Speaking'],
|
||||
activity = models.Activity.types['Seated'],
|
||||
presence = interaction_intervals[0],
|
||||
distance = 0.854,
|
||||
).build_model(SAMPLE_SIZE),
|
||||
mc.ShortRangeModel(
|
||||
expiration = short_range_expiration_distributions['Speaking'],
|
||||
activity = models.Activity.types['Seated'],
|
||||
expiration = short_range_expiration_distributions['Breathing'],
|
||||
activity = models.Activity.types['Heavy exercise'],
|
||||
presence = interaction_intervals[1],
|
||||
distance = 0.854,
|
||||
).build_model(SAMPLE_SIZE),
|
||||
|
|
@ -509,21 +536,118 @@ def simple_sr_models() -> typing.Tuple[SimpleShortRangeModel, ...]:
|
|||
interaction_interval = interaction_intervals[0],
|
||||
distance = 0.854,
|
||||
breathing_rate = models.Activity.types['Seated'].exhalation_rate,
|
||||
BLO_factors = expiration_BLO_factors['Breathing'],
|
||||
BLO_factors = expiration_BLO_factors['Speaking'],
|
||||
),
|
||||
SimpleShortRangeModel(
|
||||
interaction_interval = interaction_intervals[1],
|
||||
distance = 0.854,
|
||||
breathing_rate = models.Activity.types['Seated'].exhalation_rate,
|
||||
BLO_factors = expiration_BLO_factors['Speaking'],
|
||||
)
|
||||
breathing_rate = models.Activity.types['Heavy exercise'].exhalation_rate,
|
||||
BLO_factors = expiration_BLO_factors['Breathing'],
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def expo_sr_model(c_model,sr_models) -> mc.ExposureModel:
|
||||
return mc.ExposureModel(
|
||||
concentration_model=c_model,
|
||||
short_range=sr_models,
|
||||
exposed=mc.Population(
|
||||
number=1,
|
||||
presence=presence,
|
||||
mask=models.Mask.types['No mask'],
|
||||
activity=models.Activity.types['Seated'],
|
||||
host_immunity=0.,
|
||||
),
|
||||
).build_model(SAMPLE_SIZE)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def simple_expo_sr_model(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'],
|
||||
finf = models.Virus.types['SARS_CoV_2_DELTA'].viable_to_RNA_ratio,
|
||||
HI = 0.,
|
||||
ID50 = models.Virus.types['SARS_CoV_2_DELTA'].infectious_dose,
|
||||
transmissibility = models.Virus.types['SARS_CoV_2_DELTA'].transmissibility_factor,
|
||||
sr_models = simple_sr_models,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def expo_sr_model_distr(c_model_distr) -> mc.ExposureModel:
|
||||
return mc.ExposureModel(
|
||||
concentration_model=c_model_distr,
|
||||
short_range=(
|
||||
mc.ShortRangeModel(
|
||||
expiration = short_range_expiration_distributions['Breathing'],
|
||||
activity = activity_distributions['Seated'],
|
||||
presence = interaction_intervals[0],
|
||||
distance = short_range_distances,
|
||||
).build_model(SAMPLE_SIZE),
|
||||
mc.ShortRangeModel(
|
||||
expiration = short_range_expiration_distributions['Speaking'],
|
||||
activity = activity_distributions['Seated'],
|
||||
presence = interaction_intervals[1],
|
||||
distance = short_range_distances,
|
||||
).build_model(SAMPLE_SIZE),
|
||||
),
|
||||
exposed=mc.Population(
|
||||
number=1,
|
||||
presence=presence,
|
||||
mask=models.Mask.types['No mask'],
|
||||
activity=models.Activity.types['Seated'],
|
||||
host_immunity=0.,
|
||||
),
|
||||
).build_model(SAMPLE_SIZE)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def simple_expo_sr_model_distr(c_model_distr) -> SimpleExposureModel:
|
||||
return SimpleExposureModel(
|
||||
infected_presence = presence,
|
||||
viral_load = virus_distributions['SARS_CoV_2_DELTA'
|
||||
].build_model(SAMPLE_SIZE).viral_load_in_sputum,
|
||||
breathing_rate = activity_distributions['Seated'].build_model(
|
||||
SAMPLE_SIZE).exhalation_rate,
|
||||
room_volume = 50.,
|
||||
lambda_ventilation= 1.,
|
||||
BLO_factors = expiration_BLO_factors['Breathing'],
|
||||
finf = virus_distributions['SARS_CoV_2_DELTA'
|
||||
].build_model(SAMPLE_SIZE).viable_to_RNA_ratio,
|
||||
HI = 0.,
|
||||
ID50 = virus_distributions['SARS_CoV_2_DELTA'
|
||||
].build_model(SAMPLE_SIZE).infectious_dose,
|
||||
transmissibility = virus_distributions['SARS_CoV_2_DELTA'
|
||||
].transmissibility_factor,
|
||||
sr_models = (
|
||||
SimpleShortRangeModel(
|
||||
interaction_interval = interaction_intervals[0],
|
||||
distance = short_range_distances.generate_samples(SAMPLE_SIZE),
|
||||
breathing_rate = activity_distributions['Seated'].build_model(
|
||||
SAMPLE_SIZE).exhalation_rate,
|
||||
BLO_factors = expiration_BLO_factors['Breathing'],
|
||||
),
|
||||
SimpleShortRangeModel(
|
||||
interaction_interval = interaction_intervals[1],
|
||||
distance = short_range_distances.generate_samples(SAMPLE_SIZE),
|
||||
breathing_rate = activity_distributions['Seated'].build_model(
|
||||
SAMPLE_SIZE).exhalation_rate,
|
||||
BLO_factors = expiration_BLO_factors['Speaking'],
|
||||
)
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"time", np.linspace(8.5,17.5,12),
|
||||
)
|
||||
def test_background_concentration(time,c_model,simple_c_model):
|
||||
def test_longrange_concentration(time,c_model,simple_c_model):
|
||||
npt.assert_allclose(
|
||||
c_model.concentration(time).mean(),
|
||||
simple_c_model.concentration(time), rtol=TOLERANCE
|
||||
|
|
@ -546,7 +670,7 @@ def test_shortrange_concentration(time,c_model,simple_c_model,
|
|||
)
|
||||
|
||||
|
||||
def test_background_exposure(c_model):
|
||||
def test_longrange_exposure(c_model):
|
||||
simple_expo_model = SimpleExposureModel(
|
||||
infected_presence = presence,
|
||||
viral_load = models.Virus.types['SARS_CoV_2_DELTA'].viral_load_in_sputum,
|
||||
|
|
@ -581,7 +705,27 @@ def test_background_exposure(c_model):
|
|||
)
|
||||
|
||||
|
||||
def test_background_exposure_with_distributions():
|
||||
@pytest.mark.parametrize(
|
||||
"time", [11., 12.5, 17.]
|
||||
)
|
||||
def test_longrange_concentration_with_distributions(c_model_distr,time):
|
||||
simple_expo_model = SimpleConcentrationModel(
|
||||
infected_presence = presence,
|
||||
viral_load = virus_distributions['SARS_CoV_2_DELTA'
|
||||
].build_model(SAMPLE_SIZE).viral_load_in_sputum,
|
||||
breathing_rate = activity_distributions['Seated'].build_model(
|
||||
SAMPLE_SIZE).exhalation_rate,
|
||||
room_volume = 50.,
|
||||
lambda_ventilation= 1.,
|
||||
BLO_factors = expiration_BLO_factors['Breathing'],
|
||||
)
|
||||
npt.assert_allclose(
|
||||
c_model_distr.concentration(time).mean(),
|
||||
simple_expo_model.concentration(time).mean(), rtol=TOLERANCE
|
||||
)
|
||||
|
||||
|
||||
def test_longrange_exposure_with_distributions(c_model_distr):
|
||||
simple_expo_model = SimpleExposureModel(
|
||||
infected_presence = presence,
|
||||
viral_load = virus_distributions['SARS_CoV_2_DELTA'
|
||||
|
|
@ -601,21 +745,7 @@ def test_background_exposure_with_distributions():
|
|||
sr_models = (),
|
||||
)
|
||||
expo_model = mc.ExposureModel(
|
||||
concentration_model=mc.ConcentrationModel(
|
||||
room=models.Room(volume=50, humidity=0.3),
|
||||
ventilation=models.AirChange(active=models.PeriodicInterval(
|
||||
period=120, duration=120), air_exch=1.),
|
||||
infected=mc.InfectedPopulation(
|
||||
number=1,
|
||||
presence=presence,
|
||||
virus=virus_distributions['SARS_CoV_2_DELTA'],
|
||||
mask=models.Mask.types['No mask'],
|
||||
activity=activity_distributions['Seated'],
|
||||
expiration=expiration_distributions['Breathing'],
|
||||
host_immunity=0.,
|
||||
),
|
||||
evaporation_factor=0.3,
|
||||
),
|
||||
concentration_model=c_model_distr,
|
||||
short_range=(),
|
||||
exposed=mc.Population(
|
||||
number=1,
|
||||
|
|
@ -635,31 +765,21 @@ def test_background_exposure_with_distributions():
|
|||
)
|
||||
|
||||
|
||||
def test_exposure_with_shortrange(c_model,sr_models,simple_sr_models):
|
||||
simple_expo_sr_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'],
|
||||
finf = models.Virus.types['SARS_CoV_2_DELTA'].viable_to_RNA_ratio,
|
||||
HI = 0.,
|
||||
ID50 = models.Virus.types['SARS_CoV_2_DELTA'].infectious_dose,
|
||||
transmissibility = models.Virus.types['SARS_CoV_2_DELTA'].transmissibility_factor,
|
||||
sr_models = simple_sr_models,
|
||||
)
|
||||
expo_sr_model = mc.ExposureModel(
|
||||
concentration_model=c_model,
|
||||
short_range=sr_models,
|
||||
exposed=mc.Population(
|
||||
number=1,
|
||||
presence=presence,
|
||||
mask=models.Mask.types['No mask'],
|
||||
activity=models.Activity.types['Seated'],
|
||||
host_immunity=0.,
|
||||
),
|
||||
).build_model(SAMPLE_SIZE)
|
||||
# tests on the concentration with short-range should be skipped until
|
||||
# one finds a way to avoid the large variability of the concentration
|
||||
# with short-range 'Speaking' or 'Shouting' interactions
|
||||
@pytest.mark.skip
|
||||
@pytest.mark.parametrize(
|
||||
"time", [10.75, 14.75, 16.]
|
||||
)
|
||||
def test_concentration_with_shortrange(expo_sr_model,simple_expo_sr_model,time):
|
||||
npt.assert_allclose(
|
||||
expo_sr_model.concentration(time).mean(),
|
||||
simple_expo_sr_model.total_concentration(time).mean(), rtol=TOLERANCE
|
||||
)
|
||||
|
||||
|
||||
def test_exposure_with_shortrange(expo_sr_model,simple_expo_sr_model):
|
||||
npt.assert_allclose(
|
||||
expo_sr_model.deposited_exposure().mean(),
|
||||
simple_expo_sr_model.dose().mean(), rtol=TOLERANCE
|
||||
|
|
@ -669,3 +789,29 @@ def test_exposure_with_shortrange(c_model,sr_models,simple_sr_models):
|
|||
simple_expo_sr_model.probability_infection().mean(), rtol=TOLERANCE
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.skip
|
||||
@pytest.mark.parametrize(
|
||||
"time", [10.75, 14.75, 16.]
|
||||
)
|
||||
def test_concentration_with_shortrange_and_distributions(
|
||||
expo_sr_model_distr,simple_expo_sr_model_distr,time):
|
||||
npt.assert_allclose(
|
||||
expo_sr_model_distr.concentration(time).mean(),
|
||||
simple_expo_sr_model_distr.total_concentration(time).mean(),
|
||||
rtol=TOLERANCE
|
||||
)
|
||||
|
||||
|
||||
def test_exposure_with_shortrange_and_distributions(expo_sr_model_distr,
|
||||
simple_expo_sr_model_distr):
|
||||
npt.assert_allclose(
|
||||
expo_sr_model_distr.deposited_exposure().mean(),
|
||||
simple_expo_sr_model_distr.dose().mean(), rtol=0.05
|
||||
)
|
||||
npt.assert_allclose(
|
||||
expo_sr_model_distr.infection_probability().mean(),
|
||||
simple_expo_sr_model_distr.probability_infection().mean(),
|
||||
rtol=0.03
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -9,8 +9,8 @@ from cara.apps.calculator.model_generator import build_expiration
|
|||
|
||||
# TODO: seed better the random number generators
|
||||
np.random.seed(2000)
|
||||
SAMPLE_SIZE = 600_000
|
||||
TOLERANCE = 0.06
|
||||
SAMPLE_SIZE = 500_000
|
||||
TOLERANCE = 0.05
|
||||
|
||||
# Load the weather data (temperature in kelvin) for Toronto.
|
||||
toronto_coordinates = (43.667, 79.400)
|
||||
|
|
@ -340,7 +340,7 @@ def test_report_models(mc_model, expected_pi, expected_new_cases,
|
|||
["No mask", "Jul", 11.81, 14.137, 809],
|
||||
["Type I", "Jul", 2.33, 1.305, 149],
|
||||
["FFP2", "Jul", 0.73, 0.351, 149],
|
||||
["Type I", "Feb", 0.62, 0.291, 162],
|
||||
["Type I", "Feb", 0.62, 0.291, 149],
|
||||
],
|
||||
)
|
||||
def test_small_shared_office_Geneva(mask_type, month, expected_pi,
|
||||
|
|
|
|||
Loading…
Reference in a new issue