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

Merge branch 'feature/MonteCarlo' of https://gitlab.cern.ch/cara/cara into feature/MonteCarlo

Browse source
This commit is contained in:
Nicolas Mounet 2021-05-31 06:47:11 +02:00
commit 1f39bbb292
4 changed files with 132 additions and 21 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.Mask:
def mask(self) -> models._MaskBase:
# Initializes the mask type if mask wearing is "continuous", otherwise instantiates the mask attribute as
# the "No mask"-mask
mask = models.Mask.types[self.mask_type if self.mask_wearing_option == "mask_on" else 'No mask']
mask = models._MaskBase.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.Mask.types.items():
for name, mask_ in models._MaskBase.types.items():
if mask == mask_:
break
mask_choice = widgets.Select(options=list(models.Mask.types.keys()), value=name)
mask_choice = widgets.Select(options=list(models._MaskBase.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.Mask.types['No mask'],
mask=models._MaskBase.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.Mask.types['No mask'],
mask=models._MaskBase.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_Mask(self, _: dataclasses.Field):
def build_type__MaskBase(self, _: dataclasses.Field):
return state.DataclassStatePredefined(
models.Mask,
choices=models.Mask.types,
models._MaskBase,
choices=models._MaskBase.types,
)
def build_type_Virus(self, _: dataclasses.Field):

77
cara/models.py
View file

@ -472,8 +472,28 @@ Virus.types = {
@dataclass(frozen=True)
class Mask:
#: Filtration efficiency. (In %/100)
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.
"""
#: Pre-populated examples of Masks.
types: typing.ClassVar[typing.Dict[str, "_MaskBase"]]
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.
@ -482,9 +502,6 @@ class Mask:
#: Filtration efficiency of masks when inhaling.
η_inhale: _VectorisedFloat
#: Pre-populated examples of Masks.
types: typing.ClassVar[typing.Dict[str, "Mask"]]
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.
@ -494,8 +511,38 @@ class Mask:
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
Mask.types = {
@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
else:
eta_out = 0.8167
return eta_out
def inhale_efficiency(self) -> _VectorisedFloat:
# Overall inhale efficiency, including the effect of the leaks.
return self.η_inhale
_MaskBase.types = {
'No mask': Mask(0, 0, 0),
'Type I': Mask(
η_exhale=0.95,
@ -507,6 +554,12 @@ Mask.types = {
η_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(
η_inhale=0.865, # (94% penetration efficiency + 8% max inward leakage -> EN 149)
),
}
@ -519,8 +572,8 @@ class _ExpirationBase:
#: Pre-populated examples of Masks.
types: typing.ClassVar[typing.Dict[str, "_ExpirationBase"]]
def aerosols(self, mask: Mask):
# total volume of aerosols expired per volume of air (mL/cm^3).
def aerosols(self, mask: _MaskBase):
# total volume of aerosols expired (cm^3).
raise NotImplementedError("Subclass must implement")
@ -536,7 +589,7 @@ 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: Mask):
def aerosols(self, mask: _MaskBase):
def volume(diameter):
return (4 * np.pi * (diameter/2)**3) / 3
total = 0
@ -564,7 +617,7 @@ class MultipleExpiration(_ExpirationBase):
raise ValueError("expirations and weigths should contain the"
"same number of elements")
def aerosols(self, mask: Mask):
def aerosols(self, mask: _MaskBase):
return np.array([
weight * expiration.aerosols(mask) / sum(self.weights)
for weight,expiration in zip(self.weights,self.expirations)
@ -612,7 +665,7 @@ class Population:
presence: Interval
#: The kind of mask being worn by the people.
mask: Mask
mask: _MaskBase
#: The physical activity being carried out by the people.
activity: Activity
@ -835,7 +888,7 @@ class ExposureModel:
inf_aero = (
self.exposed.activity.inhalation_rate *
(1 - self.exposed.mask.η_inhale) *
(1 - self.exposed.mask.inhale_efficiency()) *
exposure
)

58
cara/tests/models/test_mask.py Normal file
View file

@ -0,0 +1,58 @@
import dataclasses
import numpy as np
import numpy.testing as npt
import pytest
from cara import models
@pytest.mark.parametrize(
"η_inhale, expected_inhale_efficiency",
[
[0.5, 0.5],
[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)
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",
[
[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.])],
],
)
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)
npt.assert_almost_equal(mask.exhale_efficiency(diameter),
expected_exhale_efficiency)