Source code for pycif.plugins.datastreams.fluxes.dalecbethy.read

import numpy as np
import xarray as xr

from .....utils.check.errclass import CifError
from .....utils.hdf5 import _hdf5_lock




def read(
        self,
        name,
        varnames,
        dates,
        files,
        interpol_flx=False,
        tracer=None,
        model=None,
        ddi=None,
        **kwargs
):
    """Get D&B meteorological forcing values from its dynamic forcing file.

    For each requested hourly period, finds the matching ``time`` record in
    the file and extracts the row of ``var2extract`` for D&B's active
    sample points (``tracer.domain.active``, set by
    :func:`pycif.plugins.domains.dalecbethy.read_domain.read_grid`), so that
    the returned data lines up 1-to-1 with ``tracer.domain.zlon``/``zlat``.

    D&B's own ``time`` records mark the *end* of each hourly period (see
    ``pycif.plugins.datastreams.fluxes.dalecbethy.fetch.fetch``, which
    builds ``period == [hh - 1h, hh]`` from each ``hh`` record), so periods
    are matched on ``period[1]``, not ``period[0]``.

    Args:
        self: the fluxes Plugin.
        name (str): name of the component.
        varnames (list[str] or str): variable name(s) to read; ``name`` is
            used if ``varnames`` is empty.
        dates (list): list of the date intervals to extract.
        files (list): list of files matching ``dates``.
        interpol_flx (bool): unused, kept for interface compatibility.
        tracer: the tracer Plugin, giving access to ``domain.active``.
        model: unused, kept for interface compatibility.
        ddi: unused, kept for interface compatibility.
        **kwargs: unused, kept for interface compatibility.

    Returns:
        xr.DataArray: the forcing data with dimensions
        ``(time, lev, lat, lon)`` (``lev`` and ``lat`` of size 1).

    Raises:
        CifError: if ``tracer.domain`` has no ``active`` attribute (i.e. its
            grid was not read through ``pycif.plugins.domains.dalecbethy``),
            or if a requested hour cannot be found in a file.
    """
    var2extract = varnames if varnames != "" else name

    if not hasattr(tracer.domain, "active"):
        raise CifError(
            f"D&B fluxes tracer {tracer.name} requires a 'dalecbethy' "
            f"domain (with an 'active' sample-point selection) to align "
            f"its 'ng'-indexed forcing rows with the tracer's own grid."
        )
    active = tracer.domain.active

    trcr_flx = []
    for period, ff in zip(dates, files):
        with _hdf5_lock:
            with xr.open_dataset(ff) as ds:
                (hour_idx,) = np.where(
                    ds["time"].values == np.datetime64(period[1]))

                if hour_idx.size == 0:
                    raise CifError(
                        f"Could not find time record {period[1]} in D&B "
                        f"dynamic forcing file '{ff}' for tracer "
                        f"{tracer.name}."
                    )

                data = (
                    ds[var2extract]
                    .isel(time=hour_idx[0])
                    .values[active]
                    .astype(float)
                )

        trcr_flx.append(data)

    xmod = xr.DataArray(
        np.array(trcr_flx)[:, np.newaxis, np.newaxis, :],
        coords={"time": np.array(dates)[:, 0]},
        dims=("time", "lev", "lat", "lon"),
    )
    return xmod