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

import numpy as np
import xarray as xr


from logging import info
from .....utils.check.errclass import CifError




def read(
    self,
    name,
    varnames,
    dates,
    files,
    interpol_flx=False,
    tracer=None,
    **kwargs
):
    """Read comma-delimited dummy_txt flux files into a pyCIF array.

    For each requested date/file pair, loads the 2D flux array with
    :func:`numpy.loadtxt`. If the file is missing, the flux field is
    generated on the fly via ``self.make(tracer)`` instead of raising. The
    loaded/generated array shape is checked against the domain's
    ``(nlat, nlon)``.

    Args:
        self: the fluxes Plugin.
        name: the name of the component.
        varnames: unused, kept for interface compatibility.
        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, passed through to ``self.make`` when a
            file needs to be generated.
        **kwargs: unused, kept for interface compatibility.

    Returns:
        xr.DataArray: the flux data with dimensions
        ``(time, lev, lat, lon)``.

    Raises:
        CifError: if a loaded or generated flux array's shape does not match
            the domain's ``(nlat, nlon)``.
    """

    # Reading fluxes for periods within the simulation window
    trcr_flx = []
    trcr_dates = []
    for dd, file_flx in zip(dates, files):
        try:
            flx = np.loadtxt(file_flx, delimiter=",")

        except IOError:
            info(
                f"Fluxes are not available in {file_flx}. \nCreating them from text"
            )
            flx = self.make(tracer)[0, 0]

        # Checking the fluxes shape
        nlat = self.domain.nlat
        nlon = self.domain.nlon

        if flx.shape != (nlat, nlon):
            raise CifError(
                f"Fluxes of shape {flx.shape} are not consistent with the domain definition {nlat}/{nlon}"
            )

        # Appending to list
        trcr_flx.append(flx)
        trcr_dates.append(dd[0])

    # Putting fluxes to an xarray
    xmod = xr.DataArray(
        np.array(trcr_flx)[:, np.newaxis, ...],
        coords={"time": trcr_dates},
        dims=("time", "lev", "lat", "lon"),
    )

    return xmod