dubfi.fluxes.util

Utility functions for flux inversion problem.

Added in version 0.1.0: (initial release)

Functions

get_b_prior(→ numpy.ndarray)

Construct a priori B matrix from configuration.

get_s_prior(→ numpy.ndarray)

Construct a priori scaling factors from config.

get_prior_cycling(→ tuple[numpy.ndarray, ...)

Construct a priori B matrix in cycling step from results of previous cycle.

mix_b(→ numpy.ndarray)

Compute error covariance matrix for alpha * x1 + (1 - alpha) * x2 from error covariance matricess of x1 and x2.

get_localization(→ numpy.ndarray)

Construct localization matrix based on coordinates.

get_localization_sparse(→ scipy.sparse.csc_array)

Construct localization matrix based on coordinates, see get_localization().

chunk_corr_std(→ tuple[numpy.ndarray, numpy.ndarray, ...)

Compute correlation coefficients and standard deviation of chunks of data.

fill_chunks(→ numpy.ndarray)

Fill array chunks: result[split[i-1]:split[i]] = a[i].

Module Contents

dubfi.fluxes.util.get_b_prior(flux_cat: numpy.ndarray, config: dict) → numpy.ndarray

Construct a priori B matrix from configuration.

Parameters:

  • flux_cat (np.ndarray) – names of flux categories (as used in configuration), result will be aligned to this list.

  • config (dict) – configuration

Returns:

b_prior – prior uncertainty matrix (B)

Return type:

np.ndarray

dubfi.fluxes.util.get_s_prior(flux_cat: numpy.ndarray, config: dict) → numpy.ndarray

Construct a priori scaling factors from config.

dubfi.fluxes.util.get_prior_cycling(ds: xarray.Dataset, flux_cat: numpy.ndarray, config: dict) → tuple[numpy.ndarray, numpy.ndarray, numpy.ndarray, numpy.ndarray]

Construct a priori B matrix in cycling step from results of previous cycle.

Parameters:

  • ds (xr.Dataset) – output of previous cycle, coordinates must agree with provided coordinates.

  • flux_cat (array) – names of flux categories, result will be aligned to this list.

  • config (dict) – configuration

Returns:

  • s_prior (np.ndarray) – prior scaling factors for simpler Kalman filter, shape (flux_cat,)

  • b_prior (np.ndarray) – error covariance matrix of s_prior, shape (flux_cat, flux_cat)

  • s_prior_norm_pref (np.ndarray) – prior scaling factors for extended inversion, shape (norm_prefactor, flux_cat)

  • b_prior_norm_pref (np.ndarray) – error covariance matrix of s_prior_norm_pref, shape (norm_prefactor, flux_cat, flux_cat)

  • Remarks

  • ——-

  • This function mixes the initial prior and the posterior from a

  • previous inversion step. The posterior uncertainty is optionally

  • inflated before mixing with the initial prior. It assumes that

  • the uncertainties of the initial prior and the posterior are

  • maximally correlated, see mix_b().

  • .. versionadded:: 0.1.1

dubfi.fluxes.util.mix_b(alpha: float, b1: numpy.ndarray, b2: numpy.ndarray, sqrt_b1: numpy.ndarray | None = None) → numpy.ndarray

Compute error covariance matrix for alpha * x1 + (1 - alpha) * x2 from error covariance matricess of x1 and x2.

Consider two vectors x1, x2 with error covariance matrices B1, B2. Denote the covariance matrix of x1, x2 by B12, i.e., using cumulante notation: B1[i,j] = << x1[i] x1[j] >>, B2[i,j] = << x2[i] x2[j] >>, B12[i,j] = << x1[i] x2[j] >>. Then the error covariance matrix of x1 + x2 is Bsum = B1 + B2 + B12 + B12.T.

We assume that B12 = sqrt(B1) sqrt(B2) such that the error covariance matrix of the combined vector (x1, x2) is singular with minimal rank (equal to rank(B1)=rank(B2)). This describes the strongest possible uncertainty correlation between x1 and x2.

The output of this function is therefore (denote a=alpha):

a**2 * B1 + (1-a)**2 * B2 + a*(1-a) * ( sqrt(B1) sqrt(B2) + sqrt(B2) sqrt(B1) )

Note

b1 and b2 must be real-symmetric and positive-semidefinite.

Added in version 0.1.1.

dubfi.fluxes.util.get_localization(lon: numpy.ndarray, lat: numpy.ndarray, height: numpy.ndarray, rtime: numpy.ndarray, hscale: float, vscale: float) → numpy.ndarray

Construct localization matrix based on coordinates.

All input arrays must be one-dimensional and aligned.

Parameters:

  • lon (np.ndarray) – longitude (degrees east)

  • lat (np.ndarray) – latitude (degrees east)

  • height (np.ndarray) – vertical coordinate in meters

  • rtime (np.ndarray) – time relative to localization time scale with arbitrary offset

  • hscale (np.ndarray) – horizontal localization scale in meters

  • vscale (np.ndarray) – vertical localization scale in meters

Returns:

localization_weights – localization matrix specifying weights between each combination of coordiniates

Return type:

np.ndarray

dubfi.fluxes.util.get_localization_sparse(lon: numpy.ndarray, lat: numpy.ndarray, height: numpy.ndarray, rtime: numpy.ndarray, hscale: float, vscale: float, threshold: float = 1e-05, chunk: int = 500) → scipy.sparse.csc_array

Construct localization matrix based on coordinates, see get_localization().

dubfi.fluxes.util.chunk_corr_std(a: numpy.ndarray, b: numpy.ndarray, split: numpy.ndarray) → tuple[numpy.ndarray, numpy.ndarray, numpy.ndarray]

Compute correlation coefficients and standard deviation of chunks of data.

result[0] = corr(a[0:split[i]], b[0:split[i]]), result[i] = corr(a[split[i-1]:split[i]], b[split[i-1]:split[i]]) for i > 0 split[-1] == a.size == b.size

dubfi.fluxes.util.fill_chunks(a: numpy.ndarray, split: numpy.ndarray) → numpy.ndarray

Fill array chunks: result[split[i-1]:split[i]] = a[i].