PySDM.environments.parcel

Zero-dimensional adiabatic parcel framework

  1"""
  2Zero-dimensional adiabatic parcel framework
  3"""
  4
  5from typing import List, Optional
  6
  7import numpy as np
  8
  9from PySDM.environments.impl.moist import Moist
 10from PySDM.impl.mesh import Mesh
 11from PySDM.initialisation.equilibrate_wet_radii import (
 12    default_rtol,
 13    equilibrate_wet_radii,
 14)
 15from PySDM.environments.impl import register_environment
 16
 17
 18@register_environment()
 19class Parcel(Moist):  # pylint: disable=too-many-instance-attributes
 20    def __init__(
 21        self,
 22        *,
 23        dt,
 24        mass_of_dry_air: float,
 25        p0: float,
 26        initial_water_vapour_mixing_ratio: float,
 27        T0: float,
 28        w: [float, callable],
 29        z0: float = 0,
 30        mixed_phase=False,
 31        variables: Optional[List[str]] = None,
 32    ):
 33        variables = (variables or []) + ["rhod", "z"]
 34        super().__init__(dt, Mesh.mesh_0d(), variables, mixed_phase=mixed_phase)
 35
 36        self.p0 = p0
 37        self.initial_water_vapour_mixing_ratio = initial_water_vapour_mixing_ratio
 38        self.T0 = T0
 39        self.z0 = z0
 40        self.mass_of_dry_air = mass_of_dry_air
 41        self.w = w if callable(w) else lambda _: w
 42        self.delta_liquid_water_mixing_ratio = np.nan
 43
 44    @property
 45    def dv(self):
 46        rhod_mean = (self.get_predicted("rhod")[0] + self["rhod"][0]) / 2
 47        return self.particulator.formulae.trivia.volume_of_density_mass(
 48            rhod_mean, self.mass_of_dry_air
 49        )
 50
 51    def register(self, builder):
 52        formulae = builder.particulator.formulae
 53        pd0 = formulae.trivia.p_d(self.p0, self.initial_water_vapour_mixing_ratio)
 54        rhod0 = formulae.state_variable_triplet.rhod_of_pd_T(pd0, self.T0)
 55        self.mesh.dv = formulae.trivia.volume_of_density_mass(
 56            rhod0, self.mass_of_dry_air
 57        )
 58
 59        Moist.register(self, builder)
 60
 61        self["water_vapour_mixing_ratio"][:] = self.initial_water_vapour_mixing_ratio
 62        self["thd"][:] = formulae.trivia.th_std(pd0, self.T0)
 63        self["rhod"][:] = rhod0
 64        self["z"][:] = self.z0
 65
 66        self._tmp["water_vapour_mixing_ratio"][
 67            :
 68        ] = self.initial_water_vapour_mixing_ratio
 69        self.sync_parcel_vars()
 70        Moist.sync(self)
 71        self.notify()
 72
 73    def init_attributes(
 74        self,
 75        *,
 76        n_in_dv: [float, np.ndarray],
 77        kappa: float,
 78        r_dry: [float, np.ndarray],
 79        rtol=default_rtol,
 80        include_dry_volume_in_attribute: bool = True,
 81    ):
 82        if not isinstance(n_in_dv, np.ndarray):
 83            r_dry = np.array([r_dry])
 84            n_in_dv = np.array([n_in_dv])
 85
 86        attributes = {}
 87        dry_volume = self.particulator.formulae.trivia.volume(radius=r_dry)
 88        attributes["kappa times dry volume"] = dry_volume * kappa
 89        attributes["multiplicity"] = n_in_dv
 90        r_wet = equilibrate_wet_radii(
 91            r_dry=r_dry,
 92            environment=self,
 93            kappa_times_dry_volume=attributes["kappa times dry volume"],
 94            rtol=rtol,
 95        )
 96        attributes["volume"] = self.particulator.formulae.trivia.volume(radius=r_wet)
 97        if include_dry_volume_in_attribute:
 98            attributes["dry volume"] = dry_volume
 99        return attributes
100
101    def advance_parcel_vars(self):
102        """compute new values of displacement, dry-air density and volume,
103        and write them to self._tmp and self.mesh.dv"""
104        dt = self.particulator.dt
105        formulae = self.particulator.formulae
106        T = self["T"][0]
107        p = self["p"][0]
108
109        dz_dt = self.w((self.particulator.n_steps + 1 / 2) * dt)  # "mid-point"
110        water_vapour_mixing_ratio = (
111            self["water_vapour_mixing_ratio"][0]
112            - self.delta_liquid_water_mixing_ratio / 2
113        )
114
115        # derivative evaluated at p_old, T_old, mixrat_mid, w_mid
116        drho_dz = formulae.hydrostatics.drho_dz(
117            p=p,
118            T=T,
119            water_vapour_mixing_ratio=water_vapour_mixing_ratio,
120            lv=formulae.latent_heat_vapourisation.lv(T),
121            d_liquid_water_mixing_ratio__dz=(
122                self.delta_liquid_water_mixing_ratio / dz_dt / dt
123            ),
124        )
125        drhod_dz = drho_dz  # TODO #407
126
127        self.particulator.backend.explicit_euler(self._tmp["z"], dt, dz_dt)
128        self.particulator.backend.explicit_euler(
129            self._tmp["rhod"], dt, dz_dt * drhod_dz
130        )
131
132        self.mesh.dv = formulae.trivia.volume_of_density_mass(
133            (self._tmp["rhod"][0] + self["rhod"][0]) / 2, self.mass_of_dry_air
134        )
135
136    def get_thd(self):
137        return self["thd"]
138
139    def get_water_vapour_mixing_ratio(self):
140        return self["water_vapour_mixing_ratio"]
141
142    def sync_parcel_vars(self):
143        self.delta_liquid_water_mixing_ratio = (
144            self._tmp["water_vapour_mixing_ratio"][0]
145            - self["water_vapour_mixing_ratio"][0]
146        )
147        for var in self.variables:
148            self._tmp[var][:] = self[var][:]
149
150    def sync(self):
151        self.sync_parcel_vars()
152        self.advance_parcel_vars()
153        super().sync()
@register_environment()
class Parcel(PySDM.environments.impl.moist.Moist):
 19@register_environment()
 20class Parcel(Moist):  # pylint: disable=too-many-instance-attributes
 21    def __init__(
 22        self,
 23        *,
 24        dt,
 25        mass_of_dry_air: float,
 26        p0: float,
 27        initial_water_vapour_mixing_ratio: float,
 28        T0: float,
 29        w: [float, callable],
 30        z0: float = 0,
 31        mixed_phase=False,
 32        variables: Optional[List[str]] = None,
 33    ):
 34        variables = (variables or []) + ["rhod", "z"]
 35        super().__init__(dt, Mesh.mesh_0d(), variables, mixed_phase=mixed_phase)
 36
 37        self.p0 = p0
 38        self.initial_water_vapour_mixing_ratio = initial_water_vapour_mixing_ratio
 39        self.T0 = T0
 40        self.z0 = z0
 41        self.mass_of_dry_air = mass_of_dry_air
 42        self.w = w if callable(w) else lambda _: w
 43        self.delta_liquid_water_mixing_ratio = np.nan
 44
 45    @property
 46    def dv(self):
 47        rhod_mean = (self.get_predicted("rhod")[0] + self["rhod"][0]) / 2
 48        return self.particulator.formulae.trivia.volume_of_density_mass(
 49            rhod_mean, self.mass_of_dry_air
 50        )
 51
 52    def register(self, builder):
 53        formulae = builder.particulator.formulae
 54        pd0 = formulae.trivia.p_d(self.p0, self.initial_water_vapour_mixing_ratio)
 55        rhod0 = formulae.state_variable_triplet.rhod_of_pd_T(pd0, self.T0)
 56        self.mesh.dv = formulae.trivia.volume_of_density_mass(
 57            rhod0, self.mass_of_dry_air
 58        )
 59
 60        Moist.register(self, builder)
 61
 62        self["water_vapour_mixing_ratio"][:] = self.initial_water_vapour_mixing_ratio
 63        self["thd"][:] = formulae.trivia.th_std(pd0, self.T0)
 64        self["rhod"][:] = rhod0
 65        self["z"][:] = self.z0
 66
 67        self._tmp["water_vapour_mixing_ratio"][
 68            :
 69        ] = self.initial_water_vapour_mixing_ratio
 70        self.sync_parcel_vars()
 71        Moist.sync(self)
 72        self.notify()
 73
 74    def init_attributes(
 75        self,
 76        *,
 77        n_in_dv: [float, np.ndarray],
 78        kappa: float,
 79        r_dry: [float, np.ndarray],
 80        rtol=default_rtol,
 81        include_dry_volume_in_attribute: bool = True,
 82    ):
 83        if not isinstance(n_in_dv, np.ndarray):
 84            r_dry = np.array([r_dry])
 85            n_in_dv = np.array([n_in_dv])
 86
 87        attributes = {}
 88        dry_volume = self.particulator.formulae.trivia.volume(radius=r_dry)
 89        attributes["kappa times dry volume"] = dry_volume * kappa
 90        attributes["multiplicity"] = n_in_dv
 91        r_wet = equilibrate_wet_radii(
 92            r_dry=r_dry,
 93            environment=self,
 94            kappa_times_dry_volume=attributes["kappa times dry volume"],
 95            rtol=rtol,
 96        )
 97        attributes["volume"] = self.particulator.formulae.trivia.volume(radius=r_wet)
 98        if include_dry_volume_in_attribute:
 99            attributes["dry volume"] = dry_volume
100        return attributes
101
102    def advance_parcel_vars(self):
103        """compute new values of displacement, dry-air density and volume,
104        and write them to self._tmp and self.mesh.dv"""
105        dt = self.particulator.dt
106        formulae = self.particulator.formulae
107        T = self["T"][0]
108        p = self["p"][0]
109
110        dz_dt = self.w((self.particulator.n_steps + 1 / 2) * dt)  # "mid-point"
111        water_vapour_mixing_ratio = (
112            self["water_vapour_mixing_ratio"][0]
113            - self.delta_liquid_water_mixing_ratio / 2
114        )
115
116        # derivative evaluated at p_old, T_old, mixrat_mid, w_mid
117        drho_dz = formulae.hydrostatics.drho_dz(
118            p=p,
119            T=T,
120            water_vapour_mixing_ratio=water_vapour_mixing_ratio,
121            lv=formulae.latent_heat_vapourisation.lv(T),
122            d_liquid_water_mixing_ratio__dz=(
123                self.delta_liquid_water_mixing_ratio / dz_dt / dt
124            ),
125        )
126        drhod_dz = drho_dz  # TODO #407
127
128        self.particulator.backend.explicit_euler(self._tmp["z"], dt, dz_dt)
129        self.particulator.backend.explicit_euler(
130            self._tmp["rhod"], dt, dz_dt * drhod_dz
131        )
132
133        self.mesh.dv = formulae.trivia.volume_of_density_mass(
134            (self._tmp["rhod"][0] + self["rhod"][0]) / 2, self.mass_of_dry_air
135        )
136
137    def get_thd(self):
138        return self["thd"]
139
140    def get_water_vapour_mixing_ratio(self):
141        return self["water_vapour_mixing_ratio"]
142
143    def sync_parcel_vars(self):
144        self.delta_liquid_water_mixing_ratio = (
145            self._tmp["water_vapour_mixing_ratio"][0]
146            - self["water_vapour_mixing_ratio"][0]
147        )
148        for var in self.variables:
149            self._tmp[var][:] = self[var][:]
150
151    def sync(self):
152        self.sync_parcel_vars()
153        self.advance_parcel_vars()
154        super().sync()
Parcel( *, dt, mass_of_dry_air: float, p0: float, initial_water_vapour_mixing_ratio: float, T0: float, w: [<class 'float'>, <built-in function callable>], z0: float = 0, mixed_phase=False, variables: Optional[List[str]] = None)
21    def __init__(
22        self,
23        *,
24        dt,
25        mass_of_dry_air: float,
26        p0: float,
27        initial_water_vapour_mixing_ratio: float,
28        T0: float,
29        w: [float, callable],
30        z0: float = 0,
31        mixed_phase=False,
32        variables: Optional[List[str]] = None,
33    ):
34        variables = (variables or []) + ["rhod", "z"]
35        super().__init__(dt, Mesh.mesh_0d(), variables, mixed_phase=mixed_phase)
36
37        self.p0 = p0
38        self.initial_water_vapour_mixing_ratio = initial_water_vapour_mixing_ratio
39        self.T0 = T0
40        self.z0 = z0
41        self.mass_of_dry_air = mass_of_dry_air
42        self.w = w if callable(w) else lambda _: w
43        self.delta_liquid_water_mixing_ratio = np.nan
p0
initial_water_vapour_mixing_ratio
T0
z0
mass_of_dry_air
w
delta_liquid_water_mixing_ratio
dv
45    @property
46    def dv(self):
47        rhod_mean = (self.get_predicted("rhod")[0] + self["rhod"][0]) / 2
48        return self.particulator.formulae.trivia.volume_of_density_mass(
49            rhod_mean, self.mass_of_dry_air
50        )
def register(self, builder):
52    def register(self, builder):
53        formulae = builder.particulator.formulae
54        pd0 = formulae.trivia.p_d(self.p0, self.initial_water_vapour_mixing_ratio)
55        rhod0 = formulae.state_variable_triplet.rhod_of_pd_T(pd0, self.T0)
56        self.mesh.dv = formulae.trivia.volume_of_density_mass(
57            rhod0, self.mass_of_dry_air
58        )
59
60        Moist.register(self, builder)
61
62        self["water_vapour_mixing_ratio"][:] = self.initial_water_vapour_mixing_ratio
63        self["thd"][:] = formulae.trivia.th_std(pd0, self.T0)
64        self["rhod"][:] = rhod0
65        self["z"][:] = self.z0
66
67        self._tmp["water_vapour_mixing_ratio"][
68            :
69        ] = self.initial_water_vapour_mixing_ratio
70        self.sync_parcel_vars()
71        Moist.sync(self)
72        self.notify()
def init_attributes( self, *, n_in_dv: [<class 'float'>, <class 'numpy.ndarray'>], kappa: float, r_dry: [<class 'float'>, <class 'numpy.ndarray'>], rtol=1e-05, include_dry_volume_in_attribute: bool = True):
 74    def init_attributes(
 75        self,
 76        *,
 77        n_in_dv: [float, np.ndarray],
 78        kappa: float,
 79        r_dry: [float, np.ndarray],
 80        rtol=default_rtol,
 81        include_dry_volume_in_attribute: bool = True,
 82    ):
 83        if not isinstance(n_in_dv, np.ndarray):
 84            r_dry = np.array([r_dry])
 85            n_in_dv = np.array([n_in_dv])
 86
 87        attributes = {}
 88        dry_volume = self.particulator.formulae.trivia.volume(radius=r_dry)
 89        attributes["kappa times dry volume"] = dry_volume * kappa
 90        attributes["multiplicity"] = n_in_dv
 91        r_wet = equilibrate_wet_radii(
 92            r_dry=r_dry,
 93            environment=self,
 94            kappa_times_dry_volume=attributes["kappa times dry volume"],
 95            rtol=rtol,
 96        )
 97        attributes["volume"] = self.particulator.formulae.trivia.volume(radius=r_wet)
 98        if include_dry_volume_in_attribute:
 99            attributes["dry volume"] = dry_volume
100        return attributes
def advance_parcel_vars(self):
102    def advance_parcel_vars(self):
103        """compute new values of displacement, dry-air density and volume,
104        and write them to self._tmp and self.mesh.dv"""
105        dt = self.particulator.dt
106        formulae = self.particulator.formulae
107        T = self["T"][0]
108        p = self["p"][0]
109
110        dz_dt = self.w((self.particulator.n_steps + 1 / 2) * dt)  # "mid-point"
111        water_vapour_mixing_ratio = (
112            self["water_vapour_mixing_ratio"][0]
113            - self.delta_liquid_water_mixing_ratio / 2
114        )
115
116        # derivative evaluated at p_old, T_old, mixrat_mid, w_mid
117        drho_dz = formulae.hydrostatics.drho_dz(
118            p=p,
119            T=T,
120            water_vapour_mixing_ratio=water_vapour_mixing_ratio,
121            lv=formulae.latent_heat_vapourisation.lv(T),
122            d_liquid_water_mixing_ratio__dz=(
123                self.delta_liquid_water_mixing_ratio / dz_dt / dt
124            ),
125        )
126        drhod_dz = drho_dz  # TODO #407
127
128        self.particulator.backend.explicit_euler(self._tmp["z"], dt, dz_dt)
129        self.particulator.backend.explicit_euler(
130            self._tmp["rhod"], dt, dz_dt * drhod_dz
131        )
132
133        self.mesh.dv = formulae.trivia.volume_of_density_mass(
134            (self._tmp["rhod"][0] + self["rhod"][0]) / 2, self.mass_of_dry_air
135        )

compute new values of displacement, dry-air density and volume, and write them to self._tmp and self.mesh.dv

def get_thd(self):
137    def get_thd(self):
138        return self["thd"]
def get_water_vapour_mixing_ratio(self):
140    def get_water_vapour_mixing_ratio(self):
141        return self["water_vapour_mixing_ratio"]
def sync_parcel_vars(self):
143    def sync_parcel_vars(self):
144        self.delta_liquid_water_mixing_ratio = (
145            self._tmp["water_vapour_mixing_ratio"][0]
146            - self["water_vapour_mixing_ratio"][0]
147        )
148        for var in self.variables:
149            self._tmp[var][:] = self[var][:]
def sync(self):
151    def sync(self):
152        self.sync_parcel_vars()
153        self.advance_parcel_vars()
154        super().sync()
def instantiate(self, *, builder):
 8def _instantiate(self, *, builder):
 9    copy = deepcopy(self)
10    copy.register(builder=builder)
11    return copy