Source code for pycif.plugins.datastreams.fluxes.unstructured_NetCDF.write

import os
from logging import debug, info

import xarray as xr




def write(self, name, flx_file, flx, mode="a", metadata=None, **kwargs):
    if not isinstance(metadata, dict):
        raise TypeError("'metadata' must be a dict")

    domain = metadata.get('domain', None)
    if domain is None:
        raise KeyError("'metadata' has no 'domain' key")

    lat = domain.zlat[0, :]
    lon = domain.zlon[0, :]

    lat_vertices = domain.zlatc.T
    lon_vertices = domain.zlonc.T

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

    ds = xr.Dataset(
        {name: (flx.dims, flx.data)},
        # fmt: off
        coords={
            "time": (["time"], flx.time.data, {
                "standard_name": "time",
                "long_name":     "time",
                "axis":          "T",
            }),
            "lat": (["cell"], lat, {
                "standard_name": "latitude",
                "long_name":     "latitude",
                "units":         "degrees_north",
                "bounds":        "lat_bnds"
            }),
            "lon": (["cell"], lon, {
                "standard_name": "longitude",
                "long_name":     "longitude",
                "units":         "degrees_east",
                "bounds":        "lon_bnds"
            }),
            "lat_bnds": (["cell", "vertex"], lat_vertices, {
                "standard_name": "latitude_bounds",
                "long_name":     "latitude bounds",
                "units":         "degrees_north"
            }),
            "lon_bnds": (["cell", "vertex"], lon_vertices, {
                "standard_name": "longitude_bounds",
                "long_name":     "longitude bounds",
                "units":         "degrees_east"
            })
        }
        # fmt: on
    )

    if not os.path.exists(flx_file):
        info(f"writing gridded NetCDF flux '{name}' fluxes to '{flx_file}'")
        ds.to_netcdf(flx_file, mode='w')

    else:
        info(f"appending gridded NetCDF flux '{name}' fluxes to '{flx_file}'")
        ds.to_netcdf(flx_file, mode='a')