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

Merge branch 'feature/refined_mask_model' into 'feature/MonteCarlo'

Browse source 
New Mask model

See merge request cara/cara!190
This commit is contained in:
Nicolas Mounet 2021-05-31 21:10:42 +00:00
commit 1e6e291501
7 changed files with 72 additions and 146 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
cara/apps/calculator/model_generator.py
View file

@ -269,10 +269,10 @@ class FormData:
else:
return ventilation
def mask(self) -> models._MaskBase:
def mask(self) -> models.Mask:
# Initializes the mask type if mask wearing is "continuous", otherwise instantiates the mask attribute as
# the "No mask"-mask
mask = models._MaskBase.types[self.mask_type if self.mask_wearing_option == "mask_on" else 'No mask']
mask = models.Mask.types[self.mask_type if self.mask_wearing_option == "mask_on" else 'No mask']
return mask
def infected_population(self) -> models.InfectedPopulation:

14
cara/apps/expert.py
View file

@ -390,10 +390,10 @@ class ModelWidgets(View):
def _build_mask(self, node):
mask = node.dcs_instance()
for name, mask_ in models._MaskBase.types.items():
for name, mask_ in models.Mask.types.items():
if mask == mask_:
break
mask_choice = widgets.Select(options=list(models._MaskBase.types.keys()), value=name)
mask_choice = widgets.Select(options=list(models.Mask.types.keys()), value=name)
def on_mask_change(change):
node.dcs_select(change['new'])
@ -496,7 +496,7 @@ baseline_model = models.ExposureModel(
number=1,
virus=models.Virus.types['SARS_CoV_2'],
presence=models.SpecificInterval(((8, 12), (13, 17))),
mask=models._MaskBase.types['No mask'],
mask=models.Mask.types['No mask'],
activity=models.Activity.types['Seated'],
expiration=models.Expiration.types['Talking'],
),
@ -505,7 +505,7 @@ baseline_model = models.ExposureModel(
number=10,
presence=models.SpecificInterval(((8, 12), (13, 17))),
activity=models.Activity.types['Seated'],
mask=models._MaskBase.types['No mask'],
mask=models.Mask.types['No mask'],
),
)
@ -515,10 +515,10 @@ class CARAStateBuilder(state.StateBuilder):
# For example, build_type__VentilationBase is called when dealing with ConcentrationModel
# types as it has a ventilation: _VentilationBase field.
def build_type__MaskBase(self, _: dataclasses.Field):
def build_type_Mask(self, _: dataclasses.Field):
return state.DataclassStatePredefined(
models._MaskBase,
choices=models._MaskBase.types,
models.Mask,
choices=models.Mask.types,
)
def build_type_Virus(self, _: dataclasses.Field):

112
cara/models.py
View file

@ -472,92 +472,48 @@ Virus.types = {
@dataclass(frozen=True)
class _MaskBase:
"""
Represents the filtration of aerosols by a mask, both inward and
outward.
The nature of the various mask models means that it is expected
for subclasses of _MaskBase to exist.
"""
class Mask:
#: Filtration efficiency of masks when inhaling.
η_inhale: _VectorisedFloat
#: Global factor applied to filtration efficiency of masks when exhaling.
factor_exhale: _VectorisedFloat = 1.
#: Pre-populated examples of Masks.
types: typing.ClassVar[typing.Dict[str, "_MaskBase"]]
types: typing.ClassVar[typing.Dict[str, "Mask"]]
def exhale_efficiency(self, diameter: float) -> _VectorisedFloat:
# Overall exhale efficiency, including the effect of the leaks.
raise NotImplementedError("Subclass must implement")
def inhale_efficiency(self) -> _VectorisedFloat:
# Overall inhale efficiency, including the effect of the leaks.
raise NotImplementedError("Subclass must implement")
@dataclass(frozen=True)
class Mask(_MaskBase):
#: Filtration efficiency.
η_exhale: _VectorisedFloat
#: Leakage through side of masks.
η_leaks: _VectorisedFloat
#: Filtration efficiency of masks when inhaling.
η_inhale: _VectorisedFloat
def exhale_efficiency(self, diameter: float) -> _VectorisedFloat:
# Overall efficiency with the effect of the leaks for aerosol emission
# Gammaitoni et al (1997). Diameter is in cm.
if diameter < 3e-4:
"""
Overall exhale efficiency, including the effect of the leaks.
See CERN-OPEN-2021-004 (doi: 10.17181/CERN.1GDQ.5Y75), and Ref.
therein (Asadi 2020).
Obtained from measurements of filtration efficiency and of
the leakage through the sides.
Diameter is in microns.
"""
if diameter < 0.5:
eta_out = 0.
else:
eta_out = self.η_exhale * (1 - self.η_leaks)
return eta_out
def inhale_efficiency(self) -> _VectorisedFloat:
# Overall inhale efficiency, including the effect of the leaks.
return self.η_inhale
@dataclass(frozen=True)
class MeasuredMask(_MaskBase):
#: Filtration efficiency of masks when inhaling.
η_inhale: _VectorisedFloat
def exhale_efficiency(self, diameter: float) -> _VectorisedFloat:
# See CERN-OPEN-2021-004 (doi: 10.17181/CERN.1GDQ.5Y75), and Ref.
# therein (Asadi 2020).
# Obtained from measurements of filtration efficiency and of
# the leakage through the sides.
# Diameter is in cm.
if diameter < 0.5e-4:
eta_out = 0.
elif diameter < 0.94614e-4:
eta_out = 0.5893 * diameter * 1e4 + 0.1546
elif diameter < 3e-4:
eta_out = 0.0509 * diameter * 1e4 + 0.664
elif diameter < 0.94614:
eta_out = 0.5893 * diameter + 0.1546
elif diameter < 3.:
eta_out = 0.0509 * diameter + 0.664
else:
eta_out = 0.8167
return eta_out
return eta_out*self.factor_exhale
def inhale_efficiency(self) -> _VectorisedFloat:
# Overall inhale efficiency, including the effect of the leaks.
"""
Overall inhale efficiency, including the effect of the leaks.
"""
return self.η_inhale
_MaskBase.types = {
'No mask': Mask(0, 0, 0),
Mask.types = {
'No mask': Mask(0, 0),
'Type I': Mask(
η_exhale=0.95,
η_leaks=0.15, # (Huang 2007)
η_inhale=0.3, # (Browen 2010)
),
'FFP2': Mask(
η_exhale=0.95, # (same outward effect as type 1 - Asadi 2020)
η_leaks=0.15, # (same outward effect as type 1 - Asadi 2020)
η_inhale=0.865, # (94% penetration efficiency + 8% max inward leakage -> EN 149)
),
'Type I measured': MeasuredMask(
η_inhale=0.5, # (CERN-OPEN-2021-004)
),
'FFP2 measured': MeasuredMask(
'FFP2': Mask(
η_inhale=0.865, # (94% penetration efficiency + 8% max inward leakage -> EN 149)
),
}
@ -569,10 +525,10 @@ class _ExpirationBase:
Represents the expiration of aerosols by a person.
Subclasses of _ExpirationBase represent different models.
"""
#: Pre-populated examples of Masks.
#: Pre-populated examples of Expirations.
types: typing.ClassVar[typing.Dict[str, "_ExpirationBase"]]
def aerosols(self, mask: _MaskBase):
def aerosols(self, mask: Mask):
# total volume of aerosols expired (cm^3).
raise NotImplementedError("Subclass must implement")
@ -589,13 +545,13 @@ class Expiration(_ExpirationBase):
ejection_factor: typing.Tuple[float, ...]
particle_sizes: typing.Tuple[float, ...] = (0.8e-4, 1.8e-4, 3.5e-4, 5.5e-4) # In cm.
def aerosols(self, mask: _MaskBase):
def aerosols(self, mask: Mask):
def volume(diameter):
return (4 * np.pi * (diameter/2)**3) / 3
total = 0
for diameter, factor in zip(self.particle_sizes, self.ejection_factor):
contribution = (volume(diameter) * factor *
(1 - mask.exhale_efficiency(diameter)))
(1 - mask.exhale_efficiency(diameter*1e4)))
total += contribution
return total
@ -617,7 +573,7 @@ class MultipleExpiration(_ExpirationBase):
raise ValueError("expirations and weigths should contain the"
"same number of elements")
def aerosols(self, mask: _MaskBase):
def aerosols(self, mask: Mask):
return np.array([
weight * expiration.aerosols(mask) / sum(self.weights)
for weight,expiration in zip(self.weights,self.expirations)
@ -665,7 +621,7 @@ class Population:
presence: Interval
#: The kind of mask being worn by the people.
mask: _MaskBase
mask: Mask
#: The physical activity being carried out by the people.
activity: Activity

12
cara/tests/models/test_concentration_model.py
View file

@ -1,6 +1,7 @@
import re
import numpy as np
import numpy.testing as npt
import pytest
from cara import models
@ -13,8 +14,7 @@ from cara import models
{'air_change': np.array([100, 120])},
{'viral_load_in_sputum': np.array([5e8, 1e9])},
{'quantum_infectious_dose': np.array([50, 20])},
{'η_exhale': np.array([0.92, 0.95])},
{'η_leaks': np.array([0.15, 0.20])},
{'factor_exhale': np.array([0.92, 0.95])},
]
)
def test_concentration_model_vectorisation(override_params):
@ -24,8 +24,7 @@ def test_concentration_model_vectorisation(override_params):
'air_change': 100,
'viral_load_in_sputum': 1e9,
'quantum_infectious_dose': 50,
'η_exhale': 0.95,
'η_leaks': 0.15,
'factor_exhale': 0.95,
}
defaults.update(override_params)
@ -37,8 +36,7 @@ def test_concentration_model_vectorisation(override_params):
number=1,
presence=always,
mask=models.Mask(
η_exhale=defaults['η_exhale'],
η_leaks=defaults['η_leaks'],
factor_exhale=defaults['factor_exhale'],
η_inhale=0.3,
),
activity=models.Activity(
@ -128,4 +126,4 @@ def test_integrated_concentration(simple_conc_model):
c2 = simple_conc_model.integrated_concentration(0, 1)
c3 = simple_conc_model.integrated_concentration(1, 2)
assert c1 != 0
assert c1 == c2 + c3
npt.assert_almost_equal(c1, c2 + c3, decimal=15)

22
cara/tests/models/test_exposure_model.py
View file

@ -43,7 +43,7 @@ populations = [
),
# A population with some array component for η_inhale.
models.Population(
10, halftime, models.Mask(0.95, 0.15, np.array([0.3, 0.35])),
10, halftime, models.Mask(np.array([0.3, 0.35])),
models.Activity.types['Standing'],
),
# A population with some array component for inhalation_rate.
@ -60,10 +60,10 @@ populations = [
np.array([14.4, 14.4]), np.array([99.6803184113, 99.5181053773])],
[populations[2], KnownConcentrations(lambda t: 1.2),
np.array([14.4, 14.4]), np.array([99.4146994564, 99.6803184113])],
np.array([14.4, 14.4]), np.array([97.4574432074, 98.3493482895])],
[populations[0], KnownConcentrations(lambda t: np.array([1.2, 2.4])),
np.array([14.4, 28.8]), np.array([99.6803184113, 99.9989780368])],
np.array([14.4, 28.8]), np.array([98.3493482895, 99.9727534893])],
[populations[1], KnownConcentrations(lambda t: np.array([1.2, 2.4])),
np.array([14.4, 28.8]), np.array([99.6803184113, 99.9976777757])],
@ -123,22 +123,22 @@ def conc_model():
models.InfectedPopulation(
number=1,
presence=interesting_times,
mask=models.Mask.types['Type I'],
mask=models.Mask.types['No mask'],
activity=models.Activity.types['Seated'],
virus=models.Virus.types['SARS_CoV_2'],
expiration=models.Expiration.types['Breathing'],
expiration=models.Expiration.types['Superspreading event'],
)
)
# expected quanta were computed with a trapezoidal integration, using
# a mesh of 10'000 pts per exposed presence interval.
@pytest.mark.parametrize("exposed_time_interval, expected_quanta", [
[(0, 1), 0.0055680845],
[(1, 1.01), 6.4960491e-05],
[(1.01, 1.02), 6.3187723e-05],
[(12, 12.01), 1.9307359e-06],
[(12, 24), 0.079347465],
[(0, 24), 0.086122050],
[(0, 1), 5.4869151],
[(1, 1.01), 0.064013521],
[(1.01, 1.02), 0.062266596],
[(12, 12.01), 0.0019025904],
[(12, 24), 78.190763],
[(0, 24), 84.866592],
]
)
def test_exposure_model_integral_accuracy(exposed_time_interval,

45
cara/tests/models/test_mask.py
View file

@ -1,6 +1,3 @@
import dataclasses
import numpy as np
import numpy.testing as npt
import pytest
@ -14,45 +11,23 @@ from cara import models
[np.array([0.3, 0.5]), np.array([0.3, 0.5])],
],
)
def test_masks_inhale(η_inhale, expected_inhale_efficiency):
mask = models.Mask(η_inhale=η_inhale,η_exhale=0.95,η_leaks=0.15)
measuredmask = models.MeasuredMask(η_inhale=η_inhale)
def test_mask_inhale(η_inhale, expected_inhale_efficiency):
mask = models.Mask(η_inhale=η_inhale)
npt.assert_equal(mask.inhale_efficiency(),
expected_inhale_efficiency)
npt.assert_equal(measuredmask.inhale_efficiency(),
expected_inhale_efficiency)
@pytest.mark.parametrize(
"η_exhale, η_leaks, expected_exhale_efficiency_small, expected_exhale_efficiency_large",
"diameter, factor_exhale, expected_exhale_efficiency",
[
[0.95, 0.15, 0., 0.8075],
[np.array([0.95, 1.]), 0.15, np.zeros(2), np.array([0.8075, 0.85])],
[0.95, np.array([0.15, 0.]), np.zeros(2), np.array([0.8075, 0.95])],
[np.array([0.95, 1.]), np.array([0.15, 0.]), np.zeros(2), np.array([0.8075, 1.])],
[0.3, 1., 0.],
[0.7, 0.3, 0.56711*0.3],
[1., 1., 0.7149],
[4., 0.5, 0.8167*0.5],
[5., 0., 0.],
],
)
def test_mask_exhale(η_exhale, η_leaks, expected_exhale_efficiency_small,
expected_exhale_efficiency_large):
mask = models.Mask(η_inhale=0.3,η_exhale=η_exhale,η_leaks=η_leaks)
# we test one small and one large diameter (resp. 1 and 4 microns)
npt.assert_equal(mask.exhale_efficiency(1.e-4),
expected_exhale_efficiency_small)
npt.assert_equal(mask.exhale_efficiency(4.e-4),
expected_exhale_efficiency_large)
@pytest.mark.parametrize(
"diameter, expected_exhale_efficiency",
[
[0.3e-4, 0.],
[0.7e-4, 0.56711],
[1.e-4, 0.7149],
[4.e-4, 0.8167],
],
)
def test_measuredmask_exhale(diameter, expected_exhale_efficiency):
mask = models.MeasuredMask(η_inhale=0.3)
def test_mask_exhale(diameter, factor_exhale, expected_exhale_efficiency):
mask = models.Mask(η_inhale=0.3, factor_exhale=factor_exhale)
npt.assert_almost_equal(mask.exhale_efficiency(diameter),
expected_exhale_efficiency)

9
cara/tests/test_infected_population.py
View file

@ -8,8 +8,7 @@ import cara.models
"override_params", [
{'viral_load_in_sputum': np.array([5e8, 1e9])},
{'quantum_infectious_dose': np.array([50, 20])},
{'η_exhale': np.array([0.92, 0.95])},
{'η_leaks': np.array([0.15, 0.20])},
{'factor_exhale': np.array([0.92, 0.95])},
{'exhalation_rate': np.array([0.75, 0.81])},
]
)
@ -17,8 +16,7 @@ def test_infected_population_vectorisation(override_params):
defaults = {
'viral_load_in_sputum': 1e9,
'quantum_infectious_dose': 50,
'η_exhale': 0.95,
'η_leaks': 0.15,
'factor_exhale': 0.95,
'exhalation_rate': 0.75,
}
defaults.update(override_params)
@ -28,8 +26,7 @@ def test_infected_population_vectorisation(override_params):
number=1,
presence=office_hours,
mask=cara.models.Mask(
η_exhale=defaults['η_exhale'],
η_leaks=defaults['η_leaks'],
factor_exhale=defaults['factor_exhale'],
η_inhale=0.3,
),
activity=cara.models.Activity(