lair.inversion.flux#

Flux inversion module.

This module provides classes and utilities for performing atmospheric flux inversion, a technique used to estimate surface fluxes (such as greenhouse gas emissions or uptake) from observed atmospheric concentrations. Flux inversion is a specific application of the general inverse problem framework, where the goal is to infer unknown fluxes (posterior) given observed concentrations, a prior flux inventory, and a model of atmospheric transport.

Key terminology differences from the base inverse problem: - “Prior” refers to the initial estimate of surface fluxes (e.g., from inventories or models). - “Posterior” refers to the updated estimate of fluxes after assimilating observations. - “Jacobian” (or “Forward Operator”) maps fluxes to concentrations via atmospheric transport. - “Concentrations” are the observed values at receptor locations/times. - “Background” is the baseline (constant) concentration not attributed to local fluxes.

The Jacobian matrix is constructed using atmospheric transport models, currently STILT (Stochastic Time-Inverted Lagrangian Transport), which simulates the influence of surface fluxes on observed concentrations by generating footprints for each observation. These footprints quantify the sensitivity of each observation to fluxes at different locations and times, forming the basis of the Jacobian. This module supports building the Jacobian from STILT simulations, specifying time bins, grid resolutions, and parallel computation. It also provides the FluxInversion class, which extends the base InverseProblem to handle flux-specific terminology and plotting interfaces for visualizing results.

Classes

`FluxInversion`(concentrations, inventory, ...)	FluxInversion: Atmospheric Flux Inversion Problem.
`StiltJacobianBuilder`(simulations)	Build the Jacobian matrix from STILT simulations.

class lair.inversion.flux.StiltJacobianBuilder(simulations: List[Simulation | Path])[source]#

Build the Jacobian matrix from STILT simulations.

Attributes

simulations	(list[stilt.Simulation \| Path]) List of STILT simulations or paths to simulation directories.
failed_sims	(list[str]) List of simulation IDs that failed to process.

Methods

build_from_coords(coords, flux_times, resolution=None, subset_hours=None, num_processes=1)

Build the Jacobian matrix H from specified coordinates (x, y) and flux time bins.

__init__(simulations: List[Simulation | Path])[source]#

Initialize the Jacobian builder with a list of STILT simulations or paths to simulation directories.

Parameters:

simulationslist[stilt.Simulation | Path]: List of STILT simulations or paths to simulation directories.

build_from_coords(coords: list[tuple[float, float]] | dict[str, list[tuple[float, float]]], flux_times: IntervalIndex, resolution: str | None = None, subset_hours: int | list[int] | None = None, num_processes: int | Literal['max'] = 1) → ForwardOperator | dict[str, ForwardOperator][source]#

Build the Jacobian matrix H from specified coordinates (x, y) and flux time bins.

Parameters:

coordslist[tuple[float, float]] | dict[str, list[tuple[float, float]]]: Coordinates of the output grid points. Multiple sets of coordinates can be provided as a dictionary of lists of coordinate tuples. Otherwise, a single list of coordinate tuples can be provided. Coordinates should be in the same CRS as the STILT footprints and specified as (x, y) tuples.
flux_timespd.IntervalIndex: Time bins for the fluxes
resolutionstr | None, optional: Resolution of the footprints to use, by default None (use highest resolution available)
subset_hoursint | list[int] | None, optional: Subset the simulations to specific hours of the day, by default None
num_processesint | Literal[‘max’], optional: Number of processes to use for parallel computation, by default 1

Returns:

Jacobian | dict[str, Jacobian]: If coords is a dict, returns a dict of Jacobians for each set of coordinates. Otherwise, returns a single Jacobian.

class lair.inversion.flux.FluxInversion(concentrations: Series, inventory: Series, jacobian: ForwardOperator | DataFrame, prior_error: SymmetricMatrix, modeldata_mismatch: SymmetricMatrix, background: Series | float | None = None, estimator: type[Estimator] | str = 'bayesian', **kwargs)[source]#

FluxInversion: Atmospheric Flux Inversion Problem.

Subclass of InverseProblem for estimating spatial and temporal surface fluxes (e.g., greenhouse gas emissions or uptake) from observed atmospheric concentrations. Combines observations, prior flux estimates, and a forward model (Jacobian) within a statistical estimation framework.

Attributes

concentrations	(pd.Series) Observed concentrations used in the inversion.
inventory	(pd.Series) Prior flux inventory.
jacobian	(pd.DataFrame) Forward operator mapping fluxes to concentrations.
prior_error	(CovarianceMatrix) Covariance matrix representing uncertainty in the prior flux inventory.
modeldata_mismatch	(CovarianceMatrix) Covariance matrix representing uncertainty in observed concentrations and model-data mismatch.
background	(pd.Series, float, or None) Background concentration.
estimator	(type[Estimator] or str, optional) Estimation method or class to use for the inversion (e.g., ‘bayesian’). Default is ‘bayesian’.
posterior_fluxes	(pd.Series) Estimated fluxes after inversion (posterior).
posterior_concentrations	(pd.Series) Modelled concentrations using posterior fluxes.
prior_concentrations	(pd.Series) Modelled concentrations using prior fluxes.
plot	(_Plotter) Diagnostic and plotting interface.

__init__(concentrations: Series, inventory: Series, jacobian: ForwardOperator | DataFrame, prior_error: SymmetricMatrix, modeldata_mismatch: SymmetricMatrix, background: Series | float | None = None, estimator: type[Estimator] | str = 'bayesian', **kwargs) → None[source]#

Initialize a flux inversion problem.

Parameters:

concentrationspd.Series: Observed concentrations with a multi-index of (obs_location, obs_time).
inventorypd.Series: Prior flux inventory with a multi-index of (time, lat, lon).
jacobianJacobian | pd.DataFrame: Jacobian matrix mapping fluxes to concentrations.
prior_errorCovarianceMatrix: Prior error covariance matrix.
modeldata_mismatchCovarianceMatrix: Model-data mismatch covariance matrix.
backgroundpd.Series | float | None, optional: Background concentration to add to modelled concentrations, by default None.
estimatortype[Estimator] | str, optional: Estimator class or name to use for the inversion, by default ‘bayesian’.
kwargsdict, optional: Additional keyword arguments to pass to the InverseProblem constructor.