Operators

pipekit.Operator wrappers over the forward models, with config round-trip via ConfigMixin.

operators

pipekit Operator wrappers over the plumax forward models.

Each forward model in plumax is a pure function with a long, model-specific signature. These wrappers split that signature the pipekit way: the static grid / domain / scheme configuration is captured at construction time, and the dynamic source / wind parameters flow through _apply as a single mapping carrier. The result is an :class:pipekit.Operator that composes with |, supports the config round-trip via :class:pipekit.ConfigMixin, and can be dropped into a :class:pipekit.Sequential pipeline.

Example

from plumax.operators import GaussianPlume plume = GaussianPlume( ... stability_class="D", ... domain_x=(-100.0, 1000.0, 56), ... domain_y=(-300.0, 300.0, 31), ... domain_z=(0.0, 100.0, 11), ... ) ds = plume({ ... "emission_rate": 1.0, ... "source_location": (0.0, 0.0, 10.0), ... "wind_speed": 5.0, ... "wind_direction": 270.0, ... }) ds["concentration"].dims ('x', 'y', 'z')

GaussianPlume

GaussianPlume(stability_class: str, domain_x: tuple[float, float, int], domain_y: tuple[float, float, int], domain_z: tuple[float, float, int], background_conc: float = 0.0)

Bases: Operator

Steady-state Gaussian plume forward model as a pipekit operator.

Construction captures the static grid and stability configuration. Calling the operator with a mapping of dynamic parameters (emission_rate, source_location, wind_speed, wind_direction) returns the simulated :class:xarray.Dataset.

Parameters:

Name	Type	Description	Default
stability_class	str	Pasquill stability class 'A'–'F'.	required
domain_x	tuple[float, float, int]	(start, stop, n_points) for the x axis [m].	required
domain_y	tuple[float, float, int]	(start, stop, n_points) for the y axis [m].	required
domain_z	tuple[float, float, int]	(start, stop, n_points) for the z axis [m].	required
background_conc	float	Additive background concentration [kg/m³].	0.0

Source code in src/plumax/operators.py

def __init__(
    self,
    stability_class: str,
    domain_x: tuple[float, float, int],
    domain_y: tuple[float, float, int],
    domain_z: tuple[float, float, int],
    background_conc: float = 0.0,
) -> None:
    self.stability_class = stability_class
    self.domain_x = domain_x
    self.domain_y = domain_y
    self.domain_z = domain_z
    self.background_conc = background_conc

GaussianPuff

GaussianPuff(stability_class: str, domain_x: tuple[float, float, int], domain_y: tuple[float, float, int], domain_z: tuple[float, float, int], time_array: Any, release_frequency: float = 1.0, scheme: str = 'pg', background_conc: float = 0.0, turbulence: Any = None, turbulence_seed: Any = None)

Bases: Operator

Time-resolved Gaussian puff forward model as a pipekit operator.

Static grid, time axis, and scheme configuration are captured at construction; the dynamic parameters (emission_rate, source_location, wind_speed, wind_direction) flow through _apply. wind_speed / wind_direction are per-timestep arrays.

Note: because time_array (and an optional turbulence model) are not JSON primitives, this operator's :meth:get_config is a debug repr rather than a faithful YAML round-trip; hence forbid_in_yaml = True.

Parameters:

Name	Type	Description	Default
stability_class	str	Pasquill stability class 'A'–'F'.	required
domain_x	tuple[float, float, int]	(start, stop, n_points) along x.	required
domain_y	tuple[float, float, int]	(start, stop, n_points) along y.	required
domain_z	tuple[float, float, int]	(start, stop, n_points) along z.	required
time_array	Any	Output times [s].	required
release_frequency	float	Puff release frequency [Hz].	1.0
scheme	str	Dispersion scheme, 'pg' (default) or 'briggs'.	'pg'
background_conc	float	Additive background concentration [kg/m³].	0.0
turbulence	Any	Optional :class:~plumax.gauss_puff.OUTurbulence model.	None
turbulence_seed	Any	Seed / generator for the turbulence draws.	None

Source code in src/plumax/operators.py

def __init__(
    self,
    stability_class: str,
    domain_x: tuple[float, float, int],
    domain_y: tuple[float, float, int],
    domain_z: tuple[float, float, int],
    time_array: Any,
    release_frequency: float = 1.0,
    scheme: str = "pg",
    background_conc: float = 0.0,
    turbulence: Any = None,
    turbulence_seed: Any = None,
) -> None:
    self.stability_class = stability_class
    self.domain_x = domain_x
    self.domain_y = domain_y
    self.domain_z = domain_z
    self.time_array = time_array
    self.release_frequency = release_frequency
    self.scheme = scheme
    self.background_conc = background_conc
    self.turbulence = turbulence
    self.turbulence_seed = turbulence_seed

EulerianDispersion

EulerianDispersion(**config: Any)

Bases: Operator

Eulerian finite-volume dispersion runner as a pipekit operator.

:func:~plumax.les_fvm.simulate_eulerian_dispersion is keyword-only with a large static configuration (domain, time window, flow, eddy diffusivity, boundary conditions). That configuration is captured at construction and merged with the per-call params mapping (typically emission_rate and source_location) inside _apply.

Because the stored configuration may hold non-serialisable objects (WindSchedule, boundary-condition atoms), :meth:get_config reports the configured keys only and forbid_in_yaml = True.

Parameters:

Name	Type	Description	Default
**config	Any	Keyword arguments forwarded verbatim to :func:~plumax.les_fvm.simulate_eulerian_dispersion.	{}

Source code in src/plumax/operators.py

def __init__(self, **config: Any) -> None:
    self.config = dict(config)

LagrangianDispersion

LagrangianDispersion(turbulence: Any, domain_x: tuple[float, float, int], domain_y: tuple[float, float, int], domain_z: tuple[float, float, int], n_particles: int = 5000, t_end: float = 600.0, dt: float = 1.0, pbl_height: float | None = None, background_conc: float = 0.0, seed: int = 0)

Bases: Operator

Markov-1 Lagrangian dispersion forward model as a pipekit operator.

Static grid, turbulence and integration configuration are captured at construction; the dynamic parameters (emission_rate, source_location and either wind or wind_speed / wind_direction) flow through _apply. The turbulence model is a non-primitive object, so forbid_in_yaml = True and :meth:get_config is a debug repr.

Parameters:

Name	Type	Description	Default
turbulence	Any	A turbulence model (e.g. :class:~plumax.lagrangian.HomogeneousTurbulence).	required
domain_x	tuple[float, float, int]	(start, stop, n_cells) along x.	required
domain_y	tuple[float, float, int]	(start, stop, n_cells) along y.	required
domain_z	tuple[float, float, int]	(start, stop, n_cells) along z.	required
n_particles	int	Ensemble size.	5000
t_end	float	Integration horizon [s].	600.0
dt	float	Time step [s].	1.0
pbl_height	float | None	Optional reflecting PBL lid [m].	None
background_conc	float	Additive background concentration [kg/m³].	0.0
seed	int	PRNG seed.	0

Source code in src/plumax/operators.py

def __init__(
    self,
    turbulence: Any,
    domain_x: tuple[float, float, int],
    domain_y: tuple[float, float, int],
    domain_z: tuple[float, float, int],
    n_particles: int = 5000,
    t_end: float = 600.0,
    dt: float = 1.0,
    pbl_height: float | None = None,
    background_conc: float = 0.0,
    seed: int = 0,
) -> None:
    self.turbulence = turbulence
    self.domain_x = domain_x
    self.domain_y = domain_y
    self.domain_z = domain_z
    self.n_particles = n_particles
    self.t_end = t_end
    self.dt = dt
    self.pbl_height = pbl_height
    self.background_conc = background_conc
    self.seed = seed