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

from logging import debug

import numpy as np
import pandas as pd
import xarray as xr




def read(
    self,
    name,
    varnames,
    dates,
    files,
    interpol_flx=False,
    tracer=None,
    model=None,
    ddi=None,
    **kwargs,
):
    """Get fluxes from raw files and load them into a pyCIF
    variables.


    Args:
        name (str): name of the component
        varnames (list[str]): original names of variables to read; use `name`
            if `varnames` is empty
        dates (list): list of the date intervals to extract
        files (list): list of the files matching dates

    Return:
        xr.DataArray: the actual data with dimension:
            time, levels, latitudes, longitudes

    """
    var2extract = varnames if varnames != "" else name

    # Loop over dates/files and import data
    da_list = []
    out_dates = []

    for (di, df), ff in zip(dates, files):
        debug(f"Reaaaaaaaaaaading the file '{ff}' for the date interval ({di}, {df})")
        print(f"Read {di} -> {ff}")

        # Read the file to fetch dates
        with xr.open_dataset(ff) as ds:
            times = pd.to_datetime(ds["time"].values)

            # Replace by correct year if is_climato
            if tracer.is_climato:
                year_ref = di.year
                year_data = times.year[0]
                times += pd.DateOffset(years=year_ref - year_data)

            da = ds[var2extract].isel(time=(times == di))
            da = da.fillna(0.0)
            da = da.expand_dims("lev", axis=1)
            da = da.rename(longitude="lon", latitude="lat")

        da_list.append(da)
        out_dates.append(di)

        print(f"Read {di} -> {ff}")

    # if only one level for emissions, create the axis
    xmod = xr.concat(da_list, dim="time")
    xmod["time"] = (["time"], out_dates)

    return xmod