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

from __future__ import annotations

from datetime import datetime
from typing import Any

import pandas as pd
import xarray as xr
from xarray import DataArray

OFFSET = pd.offsets.Nano(1)




def read(
    self,
    name: str,
    varnames: str,
    dates: list[tuple[datetime, datetime]],
    files: list[str],
    tracer: object | None = None,
    **kwargs: Any,
) -> DataArray:
    """Read LMDZ DYNAMICO NetCDF flux files into a pyCIF variable.

    For each requested date/file pair, opens the file (only when the file
    path changes) and selects the exact requested time slice; the native
    ``cell`` dimension is renamed to ``lon`` and dummy ``lev``/``lat``
    dimensions are added to match pyCIF's convention.

    Args:
        self: The flux tracer plugin instance.
        name (str): name of the component
        varnames (str): original name of the variable to read; `name` is
            used if `varnames` is empty
        dates (list[tuple[datetime, datetime]]): list of ``(start, end)``
            date intervals to extract
        files (list[str]): list of files matching `dates`
        tracer: Unused directly, kept for interface consistency with other
            flux plugins.

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

    Raises:
        ValueError: If a requested time slice yields zero or multiple
            matches in a file.
    """
    varnames = varnames if varnames else name

    da_list: list[DataArray] = []
    ref_path = ""

    for (date_i, date_f), file_path in zip(dates, files):
        if file_path != ref_path:

            with xr.open_dataset(file_path) as ds:
                da = ds[varnames].sel(time=slice(date_i, date_f - OFFSET))

                # Excepting a single time value in the slice
                if da.sizes["time"] == 0:
                    raise ValueError(
                        f"no value in file '{file_path}' between "
                        f"datetimes {date_i} and {date_f}"
                    )

                if da.sizes["time"] >= 2:
                    raise ValueError(
                        f"multiple values in file '{file_path}' between "
                        f"datetimes {date_i} and {date_f}"
                    )

                da = da.assign_coords(time=[date_i])
                da_list.append(da)

    da = xr.concat(da_list, dim="time")
    da = da.drop_vars(["lat", "lon"], errors="ignore")
    da = da.rename({"cell": "lon"})
    da = da.expand_dims(["lev", "lat"])
    da = da.transpose("time", "lev", "lat", "lon")

    return da