import os
import numpy as np
import xarray as xr
from .....utils.check.errclass import CifTypeError
from .....utils.hdf5 import _hdf5_lock
[docs]
def write(self, name, path, data, lmdz_version='std', metadata=None, **kwargs):
"""Write a restart/initial-condition variable to an AEMISSION-style CHIMERE-compatible NetCDF file.
Builds ``lat``/``lon`` coordinates (with bounds) from
``metadata['domain']`` if given, adds a single-valued ``temps`` time
coordinate (for compatibility with LMDZ std/OpenMPI), renames
dimensions to LMDZ's ``temps``/``sigs``/``rlatu``/``rlonv`` convention,
and creates or appends to the target NetCDF file.
Args:
self (object): this plugin.
name (str): variable name to write.
path (str): path to the target file to write.
data (xarray.DataArray): data to write.
lmdz_version (str): LMDZ plugin version, ``'std'`` or ``'acc'``
(currently unused in the body).
metadata: Optional dict with a ``"domain"`` entry, used to build
the ``lat``/``lon`` coordinates; if absent, no horizontal
coordinates are written.
**kwargs: Unused.
Raises:
CifTypeError: If ``data`` is not an ``xarray.DataArray``.
"""
plg = self.plugin
plg_name = f"{plg.name} / {plg.version} / {plg.type}"
if not isinstance(data, xr.DataArray):
raise CifTypeError(
f"Plugin {plg_name} write method 'data' argument should of type "
f"xarray.DataArray. Passed 'data' argument type: {type(data)}"
)
# Reading domain if it exists
if metadata is not None and 'domain' in metadata:
domain = metadata['domain']
lat = domain.zlat[:, 0]
lat_bnds = domain.zlatc[:, 0]
lat_bnds = np.concatenate([lat_bnds[:-1, np.newaxis],
lat_bnds[1:, np.newaxis]], axis=1)
lon = domain.zlon[0, :-1]
lon_bnds = domain.zlonc[0, :-1]
lon_bnds = np.concatenate([lon_bnds[:-1, np.newaxis],
lon_bnds[1:, np.newaxis]], axis=1)
coords = {
'lat': (['lat'], lat, {
'standard_name': "latitude",
'long_name': "latitude",
'units': "degrees_north",
'axis': "Y",
'bounds': "lat_bnds"
}),
'lon': (['lon'], lon, {
'standard_name': "longitude",
'long_name': "longitude",
'units': "degrees_east",
'axis': "X",
'bounds': "lon_bnds"
}),
'lat_bnds': (['lat', 'bnds'], lat_bnds, {
'standard_name': "latitude_bounds",
'long_name': "latitude bounds",
'units': "degrees_north"
}),
'lon_bnds': (['lon', 'bnds'], lon_bnds, {
'standard_name': "longitude_bounds",
'long_name': "longitude bounds",
'units': "degrees_east"
})
}
else:
coords = {}
# For compatibility with LMDZ std (OpenMPI version)
coords['temps'] = (['temps'], [0], {
'standard_name': "time",
'long_name': "time",
'axis': "T"
})
data = data.rename({'time': "temps", 'lev': "sigs",
'lat': "rlatu", 'lon': "rlonv"})
ds = xr.Dataset({name: (data.dims, data.values)}, coords=coords)
with _hdf5_lock:
if not os.path.exists(path):
ds.to_netcdf(path, mode='w')
else:
ds.to_netcdf(path, mode='a')
