Adapters

Classes that structurally satisfy the pipekit_cycle ForwardModel / ObservationOperator protocols. plumax does not import pipekit_cycle at runtime.

adapters

Data-assimilation adapters for the pipekit_cycle protocols.

pipekit_cycle decomposes data assimilation into three runtime-checkable protocols — ForwardModel (predict), ObservationOperator (compare), AnalysisStep (update). plumax ships adapters that satisfy them structurally: the classes below expose exactly the right methods / properties, so isinstance(obj, ForwardModel) succeeds, without plumax importing pipekit_cycle at runtime. This keeps the dependency arrow pointing the right way — assimilation drivers depend on plumax, not the reverse.

• :class:EulerianForwardModel — wraps an :class:~plumax.les_fvm.EulerianDispersionRHS as a ForwardModel, advancing a 3-D concentration field one explicit-Euler step per call.
• :class:RadtranObservationOperator — wraps the normalised Beer–Lambert forward model as an ObservationOperator mapping a scalar plume enhancement (ΔVMR) to a normalised radiance spectrum, with the analytic Jacobian as its tangent-linear.

EulerianForwardModel

EulerianForwardModel(rhs: EulerianDispersionRHS, dt: float, t0: float = 0.0)

Expose an Eulerian dispersion RHS as a pipekit_cycle.ForwardModel.

The adapter advances a tracer field with a single explicit-Euler step, c ← c + dt · RHS(t0, c). It is the minimal forward operator a cycling driver needs; richer time integration (diffrax) is available through :func:~plumax.les_fvm.simulate_eulerian_dispersion.

Parameters:

Name	Type	Description	Default
rhs	EulerianDispersionRHS	The tendency module RHS(t, concentration) -> dC/dt.	required
dt	float	Default integration step [s].	required
t0	float	Time at which the (autonomous-in-practice) RHS is evaluated [s].	0.0

Source code in src/plumax/adapters.py

def __init__(self, rhs: EulerianDispersionRHS, dt: float, t0: float = 0.0) -> None:
    self.rhs = rhs
    self._dt = dt
    self.t0 = t0

step

step(state: Any, dt: float) -> Any

Advance state by dt with one explicit-Euler step.

Source code in src/plumax/adapters.py

def step(self, state: Any, dt: float) -> Any:
    """Advance ``state`` by ``dt`` with one explicit-Euler step."""
    tendency = self.rhs(jnp.asarray(self.t0, dtype=jnp.asarray(state).dtype), state)
    return state + dt * tendency

RadtranObservationOperator

RadtranObservationOperator(ds: Dataset, nu_obs: ndarray, *, T_K: float, p_atm: float, vmr_background: float, path_length_cm: float, amf: float, var: str = 'absorption_cross_section')

Expose the normalised Beer–Lambert forward as an ObservationOperator.

H(ΔVMR) answers "what normalised radiance spectrum would a plume enhancement of ΔVMR produce?". The innovation in a retrieval is then observed_spectrum - H(ΔVMR_forecast). :meth:linearize returns the analytic Jacobian d(L_norm)/d(ΔVMR) the forward model already computes.

Parameters:

Name	Type	Description	Default
ds	Dataset	Absorption-cross-section LUT dataset.	required
nu_obs	ndarray	Observation wavenumbers [cm⁻¹].	required
T_K	float	Layer temperature [K].	required
p_atm	float	Layer pressure [atm].	required
vmr_background	float	Background gas volume mixing ratio.	required
path_length_cm	float	Optical path length [cm].	required
amf	float	Air-mass factor.	required
var	str	LUT cross-section variable name.	'absorption_cross_section'

Source code in src/plumax/adapters.py

def __init__(
    self,
    ds: xr.Dataset,
    nu_obs: np.ndarray,
    *,
    T_K: float,
    p_atm: float,
    vmr_background: float,
    path_length_cm: float,
    amf: float,
    var: str = "absorption_cross_section",
) -> None:
    self.ds = ds
    self.nu_obs = np.asarray(nu_obs, dtype=float)
    self.T_K = T_K
    self.p_atm = p_atm
    self.vmr_background = vmr_background
    self.path_length_cm = path_length_cm
    self.amf = amf
    self.var = var

linearize

linearize(state: Any) -> np.ndarray

Return the tangent-linear d(L_norm)/d(ΔVMR) at state.

Source code in src/plumax/adapters.py

def linearize(self, state: Any) -> np.ndarray:
    """Return the tangent-linear ``d(L_norm)/d(ΔVMR)`` at ``state``."""
    return self._forward(state).jacobian