Skip to content

HITRAN look-up tables

HITRAN line-by-line absorption cross-section LUTs and a Beer–Lambert forward model. hitran-api is imported lazily. RTM stack.

hapi_lut

HAPI-driven absorption cross-section look-up tables.

Submodules
  • config — GasConfig / LUTGridConfig / ATMOSPHERIC_GASES.
  • generator — single-gas fetch / compute / wrap / save pipeline.
  • multi — multi-gas orchestration (separate-per-gas + combined).
  • beers — LUT interpolation + Beer-Lambert transmittance + the differential-ratio form used for plume-enhancement retrievals.

HAPI (hitran-api on PyPI) is imported lazily inside the generator + multi pipelines — importing this sub-package does not require HAPI, only building / fetching a LUT does.

GasConfig dataclass

GasConfig(name: str, molecule_id: int, isotopologue_id: int, nu_min: float, nu_max: float, description: str = '')

Configuration for a single HITRAN gas species.

Attributes:

Name Type Description
name str

Short name used as HAPI SourceTable id (e.g. 'CH4').

molecule_id int

HITRAN molecule id (see hitran.org).

isotopologue_id int

HITRAN isotopologue id (1 = most abundant).

nu_min float

Lower wavenumber bound [cm⁻¹].

nu_max float

Upper wavenumber bound [cm⁻¹].

description str

Free-text description for dataset metadata.

LUTGridConfig dataclass

LUTGridConfig(T_grid: ndarray = (lambda: np.array([200, 220, 240, 260, 280, 300, 320], dtype=float))(), P_grid: ndarray = (lambda: np.array([0.1, 0.2, 0.4, 0.6, 0.8, 1.0], dtype=float))(), nu_step: float = 0.01, diluent_composition: dict[str, float] | None = None)

Grid configuration shared across gases for a single LUT build.

Attributes:

Name Type Description
T_grid ndarray

Temperature knots [K]. Should bracket the atmospheric range of interest (upper-troposphere ~200 K to surface ~320 K).

P_grid ndarray

Pressure knots [atm]. Should bracket the altitude range of interest (upper-tropo ~0.1 atm to surface ~1.0 atm).

nu_step float

Spectral step [cm⁻¹] on the output wavenumber grid. HITRAN lines have HWHMs down to ~0.005 cm⁻¹ at low pressure — step ≤ 0.01 cm⁻¹ for hyperspectral work, 0.05 cm⁻¹ for tutorials.

diluent_composition dict[str, float] | None

Optional override for the air diluent fractions. If None, each gas gets {'air': 1-VMR, 'self': VMR}.

get_diluent_for_gas

get_diluent_for_gas(gas_name: str, vmr_nominal: float | None = None) -> dict[str, float]

Return the HAPI Diluent dict for a given gas.

If an explicit diluent_composition override was supplied at construction time it is returned verbatim; otherwise a standard {'air': 1 - VMR, 'self': VMR} split is built from DEFAULT_VMR_NOMINAL[gas_name] (falling back to 1e-6).

Source code in src/plumax/hapi_lut/config.py
def get_diluent_for_gas(
    self,
    gas_name: str,
    vmr_nominal: float | None = None,
) -> dict[str, float]:
    """Return the HAPI ``Diluent`` dict for a given gas.

    If an explicit ``diluent_composition`` override was supplied at
    construction time it is returned verbatim; otherwise a standard
    ``{'air': 1 - VMR, 'self': VMR}`` split is built from
    ``DEFAULT_VMR_NOMINAL[gas_name]`` (falling back to 1e-6).
    """
    if self.diluent_composition is not None:
        return dict(self.diluent_composition)
    if vmr_nominal is None:
        vmr_nominal = DEFAULT_VMR_NOMINAL.get(gas_name, 1e-6)
    return {"air": 1.0 - vmr_nominal, "self": vmr_nominal}

absorption_coefficient

absorption_coefficient(sigma: DataArray | ndarray, *, vmr: float, p_atm: float, T_K: float) -> xr.DataArray | np.ndarray

α(ν) = σ(ν) · N_gas [1/cm].

Source code in src/plumax/hapi_lut/beers.py
def absorption_coefficient(
    sigma: xr.DataArray | np.ndarray,
    *,
    vmr: float,
    p_atm: float,
    T_K: float,
) -> xr.DataArray | np.ndarray:
    """α(ν) = σ(ν) · N_gas [1/cm]."""
    n_total = number_density(p_atm, T_K)
    return sigma * (n_total * vmr)

air_mass_factor

air_mass_factor(sza_deg: float, vza_deg: float) -> float

Plane-parallel two-way AMF for sun → surface → sensor.

Breaks down past ~75° zenith; fine for the tutorial regime.

Source code in src/plumax/hapi_lut/beers.py
def air_mass_factor(sza_deg: float, vza_deg: float) -> float:
    """Plane-parallel two-way AMF for sun → surface → sensor.

    Breaks down past ~75° zenith; fine for the tutorial regime.
    """
    sza = np.deg2rad(sza_deg)
    vza = np.deg2rad(vza_deg)
    return 1.0 / np.cos(sza) + 1.0 / np.cos(vza)

beers_law_from_lut

beers_law_from_lut(ds: Dataset, *, vmr: float, T_K: float, p_atm: float, l_vert_cm: float, sza_deg: float, vza_deg: float, var: str = 'absorption_cross_section') -> xr.DataArray

End-to-end: σ-from-LUT → α → τ(ν) for a single homogeneous layer.

Returns an xarray.DataArray carrying the wavenumber + wavelength coords so callers can plot τ(λ) without re-introducing units.

Source code in src/plumax/hapi_lut/beers.py
def beers_law_from_lut(
    ds: xr.Dataset,
    *,
    vmr: float,
    T_K: float,
    p_atm: float,
    l_vert_cm: float,
    sza_deg: float,
    vza_deg: float,
    var: str = "absorption_cross_section",
) -> xr.DataArray:
    """End-to-end: σ-from-LUT → α → τ(ν) for a single homogeneous layer.

    Returns an ``xarray.DataArray`` carrying the ``wavenumber`` + ``wavelength``
    coords so callers can plot τ(λ) without re-introducing units.
    """
    sigma = interpolate_cross_section(ds, T_K=T_K, p_atm=p_atm, var=var)
    alpha = absorption_coefficient(sigma, vmr=vmr, p_atm=p_atm, T_K=T_K)
    tau = transmittance(alpha, l_vert_cm=l_vert_cm, sza_deg=sza_deg, vza_deg=vza_deg)
    return tau.assign_attrs(
        description="Single-layer Beer-Lambert transmittance from LUT",
        vmr=vmr,
        T_K=T_K,
        p_atm=p_atm,
        l_vert_cm=l_vert_cm,
        sza_deg=sza_deg,
        vza_deg=vza_deg,
    )

interpolate_cross_section

interpolate_cross_section(ds: Dataset, T_K: float, p_atm: float, *, var: str = 'absorption_cross_section') -> xr.DataArray

Return σ(ν) at a given (T, p) via bilinear LUT interpolation.

The LUT dataset must carry temperature [K] and pressure [atm] coords as in build_lut_dataset. No extrapolation — callers should clip to the LUT range before interpolating.

Source code in src/plumax/hapi_lut/beers.py
def interpolate_cross_section(
    ds: xr.Dataset,
    T_K: float,
    p_atm: float,
    *,
    var: str = "absorption_cross_section",
) -> xr.DataArray:
    """Return σ(ν) at a given (T, p) via bilinear LUT interpolation.

    The LUT dataset must carry ``temperature`` [K] and ``pressure`` [atm]
    coords as in ``build_lut_dataset``. No extrapolation — callers should
    clip to the LUT range before interpolating.
    """
    if var not in ds:
        raise KeyError(f"variable {var!r} not in dataset (have {list(ds.data_vars)})")
    return ds[var].interp(temperature=T_K, pressure=p_atm)

number_density

number_density(p_atm: float, T_K: float) -> float

Return total number density [molecules / cm³] from the ideal-gas law.

Source code in src/plumax/hapi_lut/beers.py
def number_density(p_atm: float, T_K: float) -> float:
    """Return total number density [molecules / cm³] from the ideal-gas law."""
    p_pa = p_atm * ATM_TO_PA
    n_per_m3 = p_pa / (BOLTZMANN_J_PER_K * T_K)
    return n_per_m3 / CM3_PER_M3

plume_ratio_spectrum

plume_ratio_spectrum(ds: Dataset, *, vmr_background: float, vmr_total: float, T_K: float, p_atm: float, l_vert_cm: float, sza_deg: float, vza_deg: float, var: str = 'absorption_cross_section') -> xr.DataArray

Differential-Beer-Lambert ratio τ_total(ν) / τ_bg(ν).

Callers typically pass vmr_total = vmr_background * (1 + enhancement) where enhancement is the fitting parameter in the plume retrieval.

Source code in src/plumax/hapi_lut/beers.py
def plume_ratio_spectrum(
    ds: xr.Dataset,
    *,
    vmr_background: float,
    vmr_total: float,
    T_K: float,
    p_atm: float,
    l_vert_cm: float,
    sza_deg: float,
    vza_deg: float,
    var: str = "absorption_cross_section",
) -> xr.DataArray:
    """Differential-Beer-Lambert ratio τ_total(ν) / τ_bg(ν).

    Callers typically pass ``vmr_total = vmr_background * (1 + enhancement)``
    where ``enhancement`` is the fitting parameter in the plume retrieval.
    """
    tau_bg = beers_law_from_lut(
        ds,
        vmr=vmr_background,
        T_K=T_K,
        p_atm=p_atm,
        l_vert_cm=l_vert_cm,
        sza_deg=sza_deg,
        vza_deg=vza_deg,
        var=var,
    )
    tau_total = beers_law_from_lut(
        ds,
        vmr=vmr_total,
        T_K=T_K,
        p_atm=p_atm,
        l_vert_cm=l_vert_cm,
        sza_deg=sza_deg,
        vza_deg=vza_deg,
        var=var,
    )
    ratio = tau_total / tau_bg
    return ratio.assign_attrs(
        description="Plume/background transmittance ratio (differential Beer-Lambert)",
        vmr_background=vmr_background,
        vmr_total=vmr_total,
    )

transmittance

transmittance(alpha: DataArray | ndarray, *, l_vert_cm: float, sza_deg: float, vza_deg: float) -> xr.DataArray | np.ndarray

τ(ν) = exp(-α · L_vert · AMF).

Source code in src/plumax/hapi_lut/beers.py
def transmittance(
    alpha: xr.DataArray | np.ndarray,
    *,
    l_vert_cm: float,
    sza_deg: float,
    vza_deg: float,
) -> xr.DataArray | np.ndarray:
    """τ(ν) = exp(-α · L_vert · AMF)."""
    amf = air_mass_factor(sza_deg, vza_deg)
    return np.exp(-alpha * l_vert_cm * amf)

build_lut_dataset

build_lut_dataset(gas_config: GasConfig, grid_config: LUTGridConfig, nu_grid: ndarray, sigma_lut: ndarray, wavelength_nm: ndarray) -> xr.Dataset

Wrap the raw LUT arrays in a CF-conformant xarray.Dataset.

Source code in src/plumax/hapi_lut/generator.py
def build_lut_dataset(
    gas_config: GasConfig,
    grid_config: LUTGridConfig,
    nu_grid: np.ndarray,
    sigma_lut: np.ndarray,
    wavelength_nm: np.ndarray,
) -> xr.Dataset:
    """Wrap the raw LUT arrays in a CF-conformant ``xarray.Dataset``."""
    now = datetime.now(UTC).isoformat()
    ds = xr.Dataset(
        data_vars={
            "absorption_cross_section": (
                ["wavenumber", "temperature", "pressure"],
                sigma_lut,
                {
                    "units": "cm^2 / molecule",
                    "long_name": f"{gas_config.name} absorption cross-section",
                    "description": (
                        "Voigt-profile line-by-line absorption cross-section "
                        "from HITRAN, computed with HAPI."
                    ),
                    "standard_name": (
                        f"absorption_cross_section_of_{gas_config.name.lower()}_in_air"
                    ),
                },
            ),
        },
        coords={
            "wavenumber": (
                ["wavenumber"],
                nu_grid,
                {
                    "units": "cm^-1",
                    "long_name": "Wavenumber",
                    "standard_name": "wavenumber",
                },
            ),
            "wavelength": (
                ["wavenumber"],
                wavelength_nm,
                {
                    "units": "nm",
                    "long_name": "Wavelength",
                    "standard_name": "radiation_wavelength",
                },
            ),
            "temperature": (
                ["temperature"],
                np.asarray(grid_config.T_grid, dtype=float),
                {
                    "units": "K",
                    "long_name": "Temperature",
                    "standard_name": "air_temperature",
                },
            ),
            "pressure": (
                ["pressure"],
                np.asarray(grid_config.P_grid, dtype=float),
                {
                    "units": "atm",
                    "long_name": "Pressure",
                    "standard_name": "air_pressure",
                },
            ),
        },
        attrs={
            "title": f"{gas_config.name} Absorption Cross-Section Look-Up Table",
            "institution": "Generated using HITRAN API (HAPI)",
            "source": "HITRAN molecular spectroscopic database",
            "molecule": gas_config.name,
            "molecule_id": gas_config.molecule_id,
            "isotopologue_id": gas_config.isotopologue_id,
            "spectral_range_cm_1": f"{nu_grid.min():.1f}-{nu_grid.max():.1f}",
            "spectral_range_nm": f"{wavelength_nm[-1]:.1f}-{wavelength_nm[0]:.1f}",
            "temperature_range_K": f"{float(grid_config.T_grid.min()):.0f}-{float(grid_config.T_grid.max()):.0f}",
            "pressure_range_atm": f"{float(grid_config.P_grid.min()):.2f}-{float(grid_config.P_grid.max()):.2f}",
            "line_shape": "Voigt profile",
            "description": gas_config.description,
            "conventions": "CF-1.8",
            "creation_date": now,
        },
    )
    return ds

compute_absorption_lut

compute_absorption_lut(gas_config: GasConfig, grid_config: LUTGridConfig, *, cache_dir: Path | str | None = None, progress: bool = True) -> tuple[np.ndarray, np.ndarray, np.ndarray]

Evaluate σ(ν, T, P) on the full LUT grid.

Assumes fetch_hitran_data has already populated the cache. Fails fast — a HAPI failure at any (T, P) knot, or any non-finite cross-section in the result, raises rather than leaving silent NaN holes in the LUT.

Parameters:

Name Type Description Default
gas_config GasConfig

Target gas (molecule + isotopologue + ν range).

required
grid_config LUTGridConfig

Common T/P/ν-step grid.

required
cache_dir Path | str | None

Where the fetched HITRAN .data/.header pair lives.

None
progress bool

Emit per-(T, P) log lines at INFO.

True

Raises:

Type Description
RuntimeError

if absorptionCoefficient_Voigt raises at any grid knot, or if the returned cross-section contains any non-finite values.

Returns:

Type Description
ndarray

(nu_grid, sigma_lut, wavelength_nm):

ndarray
  • nu_grid: wavenumbers [cm⁻¹], shape (n_nu,)
ndarray
  • sigma_lut: cross-sections [cm²/molecule], shape (n_nu, n_T, n_P).
tuple[ndarray, ndarray, ndarray]
  • wavelength_nm: wavelengths [nm], shape (n_nu,).
Source code in src/plumax/hapi_lut/generator.py
def compute_absorption_lut(
    gas_config: GasConfig,
    grid_config: LUTGridConfig,
    *,
    cache_dir: Path | str | None = None,
    progress: bool = True,
) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
    """Evaluate σ(ν, T, P) on the full LUT grid.

    Assumes ``fetch_hitran_data`` has already populated the cache. Fails
    fast — a HAPI failure at any ``(T, P)`` knot, or any non-finite
    cross-section in the result, raises rather than leaving silent NaN
    holes in the LUT.

    Args:
        gas_config:   Target gas (molecule + isotopologue + ν range).
        grid_config:  Common T/P/ν-step grid.
        cache_dir:    Where the fetched HITRAN ``.data/.header`` pair lives.
        progress:     Emit per-(T, P) log lines at INFO.

    Raises:
        RuntimeError: if ``absorptionCoefficient_Voigt`` raises at any
            grid knot, or if the returned cross-section contains any
            non-finite values.

    Returns:
        ``(nu_grid, sigma_lut, wavelength_nm)``:
        - ``nu_grid``: wavenumbers [cm⁻¹], shape ``(n_nu,)``
        - ``sigma_lut``: cross-sections [cm²/molecule], shape ``(n_nu, n_T, n_P)``.
        - ``wavelength_nm``: wavelengths [nm], shape ``(n_nu,)``.
    """
    from hapi import absorptionCoefficient_Voigt

    if cache_dir is not None:
        _init_hapi_cache(Path(cache_dir))

    nu_grid = np.arange(gas_config.nu_min, gas_config.nu_max, grid_config.nu_step)
    wavelength_nm = 1e7 / nu_grid

    n_T = len(grid_config.T_grid)
    n_P = len(grid_config.P_grid)
    sigma_lut = np.zeros((nu_grid.size, n_T, n_P), dtype=np.float64)

    vmr_nominal = DEFAULT_VMR_NOMINAL.get(gas_config.name, 1e-6)
    diluent = grid_config.get_diluent_for_gas(gas_config.name, vmr_nominal)

    total = n_T * n_P
    for i_T, T in enumerate(grid_config.T_grid):
        for i_P, P in enumerate(grid_config.P_grid):
            if progress:
                logger.info(
                    "  [%d/%d] %s T=%.0fK P=%.2fatm",
                    i_T * n_P + i_P + 1,
                    total,
                    gas_config.name,
                    T,
                    P,
                )
            try:
                _, coef = absorptionCoefficient_Voigt(
                    SourceTables=gas_config.name,
                    WavenumberGrid=nu_grid,
                    Environment={"T": float(T), "p": float(P)},
                    Diluent=diluent,
                    HITRAN_units=True,  # cm²/molecule
                )
            except Exception as exc:
                raise RuntimeError(
                    f"absorptionCoefficient_Voigt failed for {gas_config.name} "
                    f"at T={T:.1f} K, p={P:.3f} atm."
                ) from exc
            sigma_lut[:, i_T, i_P] = coef

    if not np.all(np.isfinite(sigma_lut)):
        n_bad = int(np.sum(~np.isfinite(sigma_lut)))
        raise RuntimeError(
            f"compute_absorption_lut produced {n_bad} non-finite entries for "
            f"{gas_config.name}; refusing to return a partial LUT."
        )

    return nu_grid, sigma_lut, wavelength_nm

default_cache_dir

default_cache_dir() -> Path

Return the HITRAN line-data cache directory.

Resolution order
  1. $HAPI_CACHE_DIR if set.
  2. ~/.cache/plumax/hitran — a per-user cache, so that invoking the generator from anywhere on disk does not leave unignored artefacts in the repository tree. The projects/plumax/data/hitran_cache path used by the notebooks is always passed explicitly.
Source code in src/plumax/hapi_lut/generator.py
def default_cache_dir() -> Path:
    """Return the HITRAN line-data cache directory.

    Resolution order:
      1. ``$HAPI_CACHE_DIR`` if set.
      2. ``~/.cache/plumax/hitran`` — a per-user cache, so that
         invoking the generator from anywhere on disk does not leave
         unignored artefacts in the repository tree. The
         ``projects/plumax/data/hitran_cache`` path used by the
         notebooks is always passed explicitly.
    """
    env = os.environ.get(HAPI_CACHE_ENV)
    if env:
        return Path(env).expanduser().resolve()
    return Path.home() / ".cache" / "plumax" / "hitran"

fetch_hitran_data

fetch_hitran_data(gas_config: GasConfig, cache_dir: Path | str | None = None) -> Path

Fetch HITRAN line parameters for gas_config into cache_dir.

HAPI writes a <GasConfig.name>.data + .header pair under cache_dir, overwriting any existing pair (HAPI's fetch does not no-op on a cache hit). A successful fetch also records the fetched ν range in a <name>.range.json sidecar.

When fetch() fails (no network, API error) the previously-cached pair is only reused if its recorded range covers the requested [nu_min, nu_max] — otherwise a narrowed cache (e.g. left by :func:plumax.hapi_lut.multi.create_combined_lut, which fetches over the gas intersection range) would silently serve a truncated line list, and absorptionCoefficient_Voigt would return σ ≈ 0 outside it.

Raises:

Type Description
RuntimeError

if fetch() raises and either no cached pair is present or its recorded range does not cover the request. The underlying HAPI exception is chained so the network / API error is visible.

Returns:

Type Description
Path

The resolved cache directory.

Source code in src/plumax/hapi_lut/generator.py
def fetch_hitran_data(
    gas_config: GasConfig,
    cache_dir: Path | str | None = None,
) -> Path:
    """Fetch HITRAN line parameters for ``gas_config`` into ``cache_dir``.

    HAPI writes a ``<GasConfig.name>.data`` + ``.header`` pair under
    ``cache_dir``, **overwriting** any existing pair (HAPI's ``fetch`` does not
    no-op on a cache hit). A successful fetch also records the fetched ν range
    in a ``<name>.range.json`` sidecar.

    When ``fetch()`` fails (no network, API error) the previously-cached pair is
    only reused if its recorded range **covers** the requested
    ``[nu_min, nu_max]`` — otherwise a narrowed cache (e.g. left by
    :func:`plumax.hapi_lut.multi.create_combined_lut`, which fetches over the
    gas intersection range) would silently serve a truncated line list, and
    ``absorptionCoefficient_Voigt`` would return σ ≈ 0 outside it.

    Raises:
        RuntimeError: if ``fetch()`` raises and either no cached pair is present
            or its recorded range does not cover the request. The underlying
            HAPI exception is chained so the network / API error is visible.

    Returns:
        The resolved cache directory.
    """
    from hapi import fetch

    cache = Path(cache_dir) if cache_dir is not None else default_cache_dir()
    _init_hapi_cache(cache)

    data_path = cache / f"{gas_config.name}.data"
    header_path = cache / f"{gas_config.name}.header"
    already_cached = data_path.exists() and header_path.exists()

    logger.info(
        "Fetching HITRAN %s (M%d I%d, %.1f-%.1f cm^-1)%s",
        gas_config.name,
        gas_config.molecule_id,
        gas_config.isotopologue_id,
        gas_config.nu_min,
        gas_config.nu_max,
        " [cache present]" if already_cached else "",
    )

    fetched = True
    try:
        fetch(
            gas_config.name,
            gas_config.molecule_id,
            gas_config.isotopologue_id,
            gas_config.nu_min,
            gas_config.nu_max,
        )
    except Exception as exc:
        # HAPI raises plain Exceptions on network failures and API errors. A
        # cached pair is only trustworthy if its recorded range covers the
        # request — otherwise proceeding would feed a truncated line list to
        # the Voigt sum and silently produce σ ≈ 0 outside the cached window.
        fetched = False
        cached_range = _read_cached_range(cache, gas_config.name)
        if (
            not already_cached
            or cached_range is None
            or not _range_covers(cached_range, gas_config)
        ):
            detail = (
                "no cached data was found"
                if not already_cached
                else (
                    "the cached line list has no recorded range"
                    if cached_range is None
                    else f"the cached line list covers only "
                    f"{cached_range[0]:.1f}-{cached_range[1]:.1f} cm^-1"
                )
            )
            raise RuntimeError(
                f"HITRAN fetch failed for {gas_config.name} "
                f"(M{gas_config.molecule_id} I{gas_config.isotopologue_id}, "
                f"{gas_config.nu_min:.1f}-{gas_config.nu_max:.1f} cm^-1) and "
                f"{detail} at {cache}. Re-fetch online or point `cache_dir` at "
                "a cache whose range spans the request."
            ) from exc
        logger.warning(
            "fetch() raised %s — proceeding with cached %s (range %.1f-%.1f "
            "cm^-1 covers the request).",
            type(exc).__name__,
            data_path,
            cached_range[0],
            cached_range[1],
        )

    if not (data_path.exists() and header_path.exists()):
        raise RuntimeError(
            f"HITRAN fetch for {gas_config.name} returned without error but "
            f"no {gas_config.name}.data/.header pair was produced under {cache}."
        )
    if fetched:
        _write_cached_range(
            cache, gas_config.name, gas_config.nu_min, gas_config.nu_max
        )
    return cache

generate_single_gas_lut

generate_single_gas_lut(gas_config: GasConfig, grid_config: LUTGridConfig | None = None, *, cache_dir: Path | str | None = None, output_dir: Path | str | None = None, save: bool = False) -> xr.Dataset

End-to-end single-gas pipeline: fetch → compute → wrap → (optionally) save.

Source code in src/plumax/hapi_lut/generator.py
def generate_single_gas_lut(
    gas_config: GasConfig,
    grid_config: LUTGridConfig | None = None,
    *,
    cache_dir: Path | str | None = None,
    output_dir: Path | str | None = None,
    save: bool = False,
) -> xr.Dataset:
    """End-to-end single-gas pipeline: fetch → compute → wrap → (optionally) save."""
    grid_config = grid_config if grid_config is not None else LUTGridConfig()
    cache = Path(cache_dir) if cache_dir is not None else default_cache_dir()

    fetch_hitran_data(gas_config, cache_dir=cache)
    nu_grid, sigma_lut, wavelength_nm = compute_absorption_lut(
        gas_config, grid_config, cache_dir=cache
    )
    ds = build_lut_dataset(gas_config, grid_config, nu_grid, sigma_lut, wavelength_nm)
    if save:
        if output_dir is None:
            raise ValueError("save=True requires output_dir.")
        save_lut(ds, output_dir, gas_config.name)
    return ds

save_lut

save_lut(ds: Dataset, output_dir: Path | str, gas_name: str) -> Path

Persist ds to <output_dir>/<gas_name>_absorption_lut.nc (zlib).

Source code in src/plumax/hapi_lut/generator.py
def save_lut(ds: xr.Dataset, output_dir: Path | str, gas_name: str) -> Path:
    """Persist ``ds`` to ``<output_dir>/<gas_name>_absorption_lut.nc`` (zlib)."""
    out = Path(output_dir)
    out.mkdir(parents=True, exist_ok=True)
    path = out / f"{gas_name.lower()}_absorption_lut.nc"
    ds.to_netcdf(
        path,
        encoding={
            "absorption_cross_section": {
                "zlib": True,
                "complevel": 4,
                "dtype": "float32",
            }
        },
    )
    logger.info(
        "Saved %s LUT to %s (%.1f MB)", gas_name, path, path.stat().st_size / 1e6
    )
    return path

create_combined_lut

create_combined_lut(gases: list[str], grid_config: LUTGridConfig | None = None, *, output_dir: Path | str, cache_dir: Path | str | None = None, filename: str = 'combined_atmospheric_lut.nc') -> Path

Build a single NetCDF with one cross-section variable per gas.

The shared wavenumber axis is the intersection of the per-gas ranges: [max(nu_min), min(nu_max)]. All variables therefore live on one (wavenumber, temperature, pressure) coordinate system — convenient for matched-filter retrievals where CH₄ and CO₂ share the same SWIR window.

output_dir has no default for the same reason as :func:create_multi_gas_luts — forcing an explicit path keeps generated artefacts out of the current working directory.

Source code in src/plumax/hapi_lut/multi.py
def create_combined_lut(
    gases: list[str],
    grid_config: LUTGridConfig | None = None,
    *,
    output_dir: Path | str,
    cache_dir: Path | str | None = None,
    filename: str = "combined_atmospheric_lut.nc",
) -> Path:
    """Build a single NetCDF with one cross-section variable per gas.

    The shared wavenumber axis is the intersection of the per-gas ranges:
    ``[max(nu_min), min(nu_max)]``. All variables therefore live on one
    ``(wavenumber, temperature, pressure)`` coordinate system — convenient
    for matched-filter retrievals where CH₄ and CO₂ share the same SWIR window.

    ``output_dir`` has no default for the same reason as
    :func:`create_multi_gas_luts` — forcing an explicit path keeps
    generated artefacts out of the current working directory.
    """
    grid_config = grid_config if grid_config is not None else LUTGridConfig()
    cache = Path(cache_dir) if cache_dir is not None else default_cache_dir()
    out = Path(output_dir)
    out.mkdir(parents=True, exist_ok=True)

    missing = [g for g in gases if g not in ATMOSPHERIC_GASES]
    if missing:
        raise KeyError(f"Unknown gases: {missing}")

    nu_min = max(ATMOSPHERIC_GASES[g].nu_min for g in gases)
    nu_max = min(ATMOSPHERIC_GASES[g].nu_max for g in gases)
    if nu_min >= nu_max:
        raise ValueError(
            f"Gases {gases} have no common spectral range ({nu_min:.1f} >= {nu_max:.1f})."
        )
    nu_grid = np.arange(nu_min, nu_max, grid_config.nu_step)
    wavelength_nm = 1e7 / nu_grid

    logger.info(
        "Building combined LUT for %s on %.1f-%.1f cm^-1 (%d pts).",
        gases,
        nu_min,
        nu_max,
        nu_grid.size,
    )

    data_vars: dict[str, tuple] = {}
    for gas_name in gases:
        base = ATMOSPHERIC_GASES[gas_name]
        sub = GasConfig(
            name=base.name,
            molecule_id=base.molecule_id,
            isotopologue_id=base.isotopologue_id,
            nu_min=nu_min,
            nu_max=nu_max,
            description=base.description,
        )
        fetch_hitran_data(sub, cache_dir=cache)
        _, sigma_lut, _ = compute_absorption_lut(sub, grid_config, cache_dir=cache)
        data_vars[f"{gas_name}_cross_section"] = (
            ["wavenumber", "temperature", "pressure"],
            sigma_lut,
            {
                "units": "cm^2 / molecule",
                "long_name": f"{gas_name} absorption cross-section",
                "molecule_id": base.molecule_id,
                "isotopologue_id": base.isotopologue_id,
            },
        )

    ds = xr.Dataset(
        data_vars=data_vars,
        coords={
            "wavenumber": (["wavenumber"], nu_grid, {"units": "cm^-1"}),
            "wavelength": (["wavenumber"], wavelength_nm, {"units": "nm"}),
            "temperature": (
                ["temperature"],
                np.asarray(grid_config.T_grid, dtype=float),
                {"units": "K"},
            ),
            "pressure": (
                ["pressure"],
                np.asarray(grid_config.P_grid, dtype=float),
                {"units": "atm"},
            ),
        },
        attrs={
            "title": "Combined Atmospheric Absorption LUT",
            "gases": ", ".join(gases),
            "line_shape": "Voigt profile",
            "conventions": "CF-1.8",
        },
    )

    path = out / filename
    ds.to_netcdf(
        path,
        encoding={
            var: {"zlib": True, "complevel": 4, "dtype": "float32"} for var in data_vars
        },
    )
    logger.info("Saved combined LUT to %s (%.1f MB)", path, path.stat().st_size / 1e6)
    return path

create_multi_gas_luts

create_multi_gas_luts(gases: list[str], grid_config: LUTGridConfig | None = None, *, output_dir: Path | str, cache_dir: Path | str | None = None, force_recompute: bool = False) -> dict[str, Path]

Build one separate <gas>_absorption_lut.nc per gas.

Gases are looked up in ATMOSPHERIC_GASES. Unknown names are skipped with a warning rather than raising — lets users pass a superset list.

Parameters:

Name Type Description Default
gases list[str]

List of gas names (e.g. ['CH4', 'CO2', 'H2O']).

required
grid_config LUTGridConfig | None

Common grid; defaults to LUTGridConfig().

None
output_dir Path | str

Required. Where to write NetCDF LUTs. No default is provided to avoid dropping unignored artefacts into the current working directory; callers should point this at the project-local git-ignored projects/plumax/data/hapi_lut/ path (or an out-of-tree scratch directory).

required
cache_dir Path | str | None

HITRAN cache dir; defaults to default_cache_dir().

None
force_recompute bool

If False, skip gases whose NetCDF already exists.

False

Returns:

Type Description
dict[str, Path]

Mapping {gas_name: Path} for successfully-built LUTs.

Source code in src/plumax/hapi_lut/multi.py
def create_multi_gas_luts(
    gases: list[str],
    grid_config: LUTGridConfig | None = None,
    *,
    output_dir: Path | str,
    cache_dir: Path | str | None = None,
    force_recompute: bool = False,
) -> dict[str, Path]:
    """Build one separate ``<gas>_absorption_lut.nc`` per gas.

    Gases are looked up in ``ATMOSPHERIC_GASES``. Unknown names are skipped
    with a warning rather than raising — lets users pass a superset list.

    Args:
        gases:           List of gas names (e.g. ``['CH4', 'CO2', 'H2O']``).
        grid_config:     Common grid; defaults to ``LUTGridConfig()``.
        output_dir:      Required. Where to write NetCDF LUTs. No default
                         is provided to avoid dropping unignored artefacts
                         into the current working directory; callers should
                         point this at the project-local git-ignored
                         ``projects/plumax/data/hapi_lut/`` path
                         (or an out-of-tree scratch directory).
        cache_dir:       HITRAN cache dir; defaults to ``default_cache_dir()``.
        force_recompute: If False, skip gases whose NetCDF already exists.

    Returns:
        Mapping ``{gas_name: Path}`` for successfully-built LUTs.
    """
    grid_config = grid_config if grid_config is not None else LUTGridConfig()
    cache = Path(cache_dir) if cache_dir is not None else default_cache_dir()
    out = Path(output_dir)
    out.mkdir(parents=True, exist_ok=True)

    lut_files: dict[str, Path] = {}
    for gas_name in gases:
        if gas_name not in ATMOSPHERIC_GASES:
            logger.warning("Unknown gas %r — skipping.", gas_name)
            continue
        gas_config = ATMOSPHERIC_GASES[gas_name]

        path = out / f"{gas_name.lower()}_absorption_lut.nc"
        if path.exists() and not force_recompute:
            logger.info("LUT already exists for %s (%s); skipping.", gas_name, path)
            lut_files[gas_name] = path
            continue

        fetch_hitran_data(gas_config, cache_dir=cache)
        nu_grid, sigma_lut, wavelength_nm = compute_absorption_lut(
            gas_config, grid_config, cache_dir=cache
        )
        ds = build_lut_dataset(
            gas_config, grid_config, nu_grid, sigma_lut, wavelength_nm
        )
        lut_files[gas_name] = save_lut(ds, out, gas_name)

    logger.info("Built %d of %d requested LUTs.", len(lut_files), len(gases))
    return lut_files