Source code for pycif.plugins.datastreams.fields.lmdz_chemfield_ico.write

from __future__ import annotations

from os import PathLike

import xarray as xr
from .....utils.check.errclass import CifTypeError
from .....utils.hdf5 import _hdf5_lock


# pylint: disable=unused-argument


def write(
    self,
    name: str,
    path: str | PathLike,
    data: xr.DataArray,
    **kwargs,
) -> None:
    """Append an LMDz chemical field (icosahedral grid) to a NetCDF file.

    Squeezes the ``lat`` dimension (unused on the icosahedral grid), renames
    ``lon`` to ``cell`` (the DYNAMICO cell index), squeezes ``lev`` for
    single-level (surface-only) fields, builds coordinates via
    ``self.domain.get_domain_coords()``, and appends the resulting dataset
    to ``path``.

    Args:
        self: This plugin; ``self.domain`` supplies the coordinates.
        name: Name of the variable to write.
        path: Path to the NetCDF file to append to.
        data: Data to write, as an ``xarray.DataArray`` with ``lat``,
            ``lon`` and ``lev`` dimensions.
        **kwargs: Unused.

    Raises:
        TypeError: If ``data`` is not an ``xarray.DataArray``.
    """
    if not isinstance(data, xr.DataArray):
        plg = self.plugin
        plg_name = f"{plg.name} / {plg.version} / {plg.type}"
        raise TypeError(
            f"Plugin {plg_name} write method 'data' argument should of type "
            f"xarray.DataArray. Passed 'data' argument type: {type(data)}"
        )

    data = data.squeeze("lat")
    data = data.rename({"lon": "cell"})

    # For deposition velocity fields
    if data.sizes["lev"] == 1:
        data = data.squeeze("lev")

    coords = self.domain.get_domain_coords()

    if "time" in data.coords:
        # fmt: off
        coords["time"] = (["time"], data.time.values, {
            "standard_name": "time",
            "long_name":     "time axis",
            "axis":          "T",
        })
        # fmt: on

    # Formating dataset
    ds = xr.Dataset({name: (data.dims, data.values)}, coords=coords)
    with _hdf5_lock:
        ds.to_netcdf(path, mode="a", encoding={"time": {'dtype': "int32"}})