diff --git a/cara/apps/calculator/model_generator.py b/cara/apps/calculator/model_generator.py index ad65795e..0d7506ae 100644 --- a/cara/apps/calculator/model_generator.py +++ b/cara/apps/calculator/model_generator.py @@ -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: diff --git a/cara/apps/expert.py b/cara/apps/expert.py index 31903b79..a0f273e1 100644 --- a/cara/apps/expert.py +++ b/cara/apps/expert.py @@ -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): diff --git a/cara/models.py b/cara/models.py index c2500d68..50eaaec2 100644 --- a/cara/models.py +++ b/cara/models.py @@ -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 ) diff --git a/cara/tests/models/test_mask.py b/cara/tests/models/test_mask.py new file mode 100644 index 00000000..f2069cbd --- /dev/null +++ b/cara/tests/models/test_mask.py @@ -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) +