Source code for scmdata.netcdf

NetCDF4 file operations

Reading and writing :class:`ScmRun <>` to disk as binary
from __future__ import annotations

    import netCDF4 as nc

    has_netcdf = True
except ImportError:  # pragma: no cover
    nc = None
    has_netcdf = False

from datetime import datetime
from logging import getLogger
from typing import TYPE_CHECKING, Any, Iterable

import pandas as pd
import xarray as xr

from . import __version__
from ._typing import FilePath

logger = getLogger(__name__)

    from import BaseScmRun

def _var_to_nc(var):
    return var.replace("|", "__").replace(" ", "_")

def _rename_variables(xr_ds):
    name_mapping = {}
    for data_var in xr_ds.data_vars:
        serialised_name = _var_to_nc(data_var)
        name_mapping[data_var] = serialised_name
        xr_ds[data_var].attrs["long_name"] = data_var

    xr_ds = xr_ds.rename_vars(name_mapping)

    return xr_ds

def _get_xr_dataset_to_write(run, dimensions, extras):
    xr_ds = run.to_xarray(dimensions, extras)
    xr_ds = _rename_variables(xr_ds)

    return xr_ds

def _write_nc(
    fname: FilePath,
    run: BaseScmRun,
    dimensions: list[str],
    extras: list[str],
    **kwargs: Any,
) -> None:
    Low level function to write the dimensions, variables and metadata to disk
    xr_ds = _get_xr_dataset_to_write(run, dimensions, extras)

    xr_ds.attrs["created_at"] = datetime.utcnow().isoformat()
    xr_ds.attrs["_scmdata_version"] = __version__

    if run.metadata:

    write_kwargs = _update_kwargs_to_match_serialised_variable_names(xr_ds, kwargs)
    xr_ds.to_netcdf(fname, **write_kwargs)

def _read_nc(cls: BaseScmRun, fname: FilePath) -> BaseScmRun:
    loaded = xr.load_dataset(fname, use_cftime=True)
    dataframe = loaded.to_dataframe()

    dataframe = _reshape_to_scmrun_dataframe(dataframe, loaded)
    run = _convert_to_cls_and_add_metadata(dataframe, loaded, cls)

    return run  # type: ignore

def _reshape_to_scmrun_dataframe(dataframe, loaded):
    index_cols = list(set(dataframe.columns) - set(loaded.data_vars))
    dataframe = dataframe.set_index(index_cols, append=True)
    if "_id" in dataframe.index.names:
        dataframe = dataframe.reset_index("_id", drop=True)

    variable_name_map = {k: v.attrs["long_name"] for k, v in loaded.data_vars.items()} = "variable"
    dataframe.columns =

    # Can revert to below once is resolved
    # dataframe = dataframe.stack("variable").unstack("time").reset_index()
    dataframe = dataframe.stack("variable").unstack("time")
    dataframe.columns = pd.Index(dataframe.columns.values)
    dataframe = dataframe.reset_index()

    unit_map = {
        data_var: loaded[data_var].attrs["units"] for data_var in loaded.data_vars
    dataframe["unit"] = dataframe["variable"].map(_var_to_nc).map(unit_map).values

    return dataframe

def _convert_to_cls_and_add_metadata(dataframe, loaded, cls):
    for k in list(loaded.attrs.keys()):
        if k.startswith("scmdata_metadata_"):
            dataframe[k.replace("scmdata_metadata_", "")] = loaded.attrs.pop(k)

    run = cls(dataframe)

    return run

def _update_kwargs_to_match_serialised_variable_names(xr_ds, in_kwargs):
    variable_name_map = {v.attrs["long_name"]: k for k, v in xr_ds.data_vars.items()}

    def _update_kwargs(dict_in):
        dict_out = {}
        for key, value in dict_in.items():
            if isinstance(value, dict):
                new_val = _update_kwargs(value)
            elif value in variable_name_map:
                new_val = variable_name_map[value]
                new_val = value

            if key in variable_name_map:
                dict_out[variable_name_map[key]] = new_val
                dict_out[key] = new_val

        return dict_out

    return _update_kwargs(in_kwargs)

[docs]def run_to_nc( run: BaseScmRun, fname: FilePath, dimensions: Iterable[str] = ("region",), extras: Iterable[str] = (), **kwargs: Any, ) -> None: """ Write timeseries to disk as a netCDF4 file Each unique variable will be written as a variable within the netCDF file. Choosing the dimensions and extras such that there are as few empty (or nan) values as possible will lead to the best compression on disk. Parameters ---------- fname : str | pathlib.Path Path to write the file into dimensions : iterable of str Dimensions to include in the netCDF file. The time dimension is always included (if not provided it will be the last dimension). An additional dimension (specifically a co-ordinate in xarray terms), "_id", will be included if ``extras`` is provided and any of the metadata in ``extras`` is not uniquely defined by ``dimensions``. "_id" maps the timeseries in each variable to their relevant metadata. extras : iterable of str Metadata columns to write as variables in the netCDF file (specifically as "non-dimension co-ordinates" in xarray terms, see `xarray terminology <>`_ for more details). Where possible, these non-dimension co-ordinates will use dimension co-ordinates as their own co-ordinates. However, if the metadata in ``extras`` is not defined by a single dimension in ``dimensions``, then the ``extras`` co-ordinates will have dimensions of "_id". This "_id" co-ordinate maps the values in the ``extras`` co-ordinates to each timeseries in the serialised dataset. Where "_id" is required, an extra "_id" dimension will also be added to ``dimensions``. kwargs Passed through to :meth:`xarray.Dataset.to_netcdf` See Also -------- :meth:`` """ if not has_netcdf: raise ImportError("netcdf4 is not installed. Run 'pip install netcdf4'") _dimensions = list(dimensions) if "time" in _dimensions: _dimensions.remove("time") if "variable" in _dimensions: _dimensions.remove("variable") _write_nc(fname, run, _dimensions, list(extras), **kwargs)
[docs]def nc_to_run(cls: BaseScmRun, fname: FilePath) -> BaseScmRun: """ Read a netCDF4 file from disk Parameters ---------- fname: str Filename to read See Also -------- :meth:`` """ if not has_netcdf: raise ImportError("netcdf4 is not installed. Run 'pip install netcdf4'") try: return _read_nc(cls, fname) except Exception: logger.exception(f"Failed reading netcdf file: {fname}") raise
[docs]def inject_nc_methods(cls: type[BaseScmRun]) -> None: """ Add the to/from nc methods to a class Parameters ---------- cls Class to add methods to """ name = "to_nc" func = run_to_nc func.__name__ = name func.__doc__ = func.__doc__ setattr(cls, name, func) name = "from_nc" func = classmethod(nc_to_run) func.__name__ = name func.__doc__ = func.__doc__ setattr(cls, name, func)