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

import numpy as np
import xarray as xr


from logging import info




def read(
    self,
    name,
    varnames,
    dates,
    files,
    interpol_flx=False,
    tracer=None,
    **kwargs
):
    """Get fluxes from pre-computed fluxes and load them into a pycif
    variables

    Args:
        self: the model Plugin
        name: the name of the component
        tracdir, tracfile: flux directory and file format
        dates: list of dates to extract
        interpol_flx (bool): if True, interpolates fluxes at time t from
        values of surrounding available files

    """

    # 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(
                "Fluxes are not available in {}. \n"
                "Creating them from text".format(file_flx)
            )
            flx = self.make(tracer)[0, 0]

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

        if flx.shape != (nlat, nlon):
            raise Exception(
                "Fluxes of shape {} are not consistent "
                "with the domain definition {}/{}".format(
                    flx.shape, 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