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

from __future__ import annotations

from os import PathLike
from typing import Any

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




def write(
    self,
    name: str,
    path: str | PathLike,
    data: xr.DataArray,
    metadata: dict[str, Any] | None = None,
    **kwargs,
):
    """Append a mixing-ratio initial condition (icosahedral grid) to a NetCDF file.

    Squeezes the singleton ``time`` and ``lat`` dimensions, renames ``lon``
    to ``cell`` (the DYNAMICO cell index), builds coordinates via
    ``self.domain.get_domain_coords()``, and appends the resulting dataset
    to ``path``.

    Args:
        self (object): this plugin; ``self.domain`` supplies the
            coordinates.
        name (str): variable name to write.
        path (str): path to the target file to write.
        data (xarray.DataArray): data to write, with singleton ``time``/
            ``lat`` dimensions.
        metadata: Unused.
        **kwargs: Unused.

    Raises:
        CifTypeError: 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 CifTypeError(
            f"Plugin {plg_name} write method 'data' argument should of type "
            f"xarray.DataArray. Passed 'data' argument type: {type(data)}"
        )

    data = data.squeeze(["time", "lat"])
    data = data.rename({"lon": "cell"})

    ds = xr.Dataset(
        {name: (data.dims, data.values)},
        coords=self.domain.get_domain_coords(),
    )

    with _hdf5_lock:
        ds.to_netcdf(path, mode="a")