Function get_default_netcdf_write_engine

xarray/backends/writers.py:165–217

Return the default netCDF library to use for writing a netCDF file.

(
    path_or_file: str | IOBase | None,
    format: T_NetcdfTypes | None,
)

def get_default_netcdf_write_engine(
    path_or_file: str | IOBase | None,
    format: T_NetcdfTypes | None,
) -> Literal["netcdf4", "h5netcdf", "scipy"]:
    """Return the default netCDF library to use for writing a netCDF file."""

    module_names = {
        "netcdf4": "netCDF4",
        "scipy": "scipy",
        "h5netcdf": "h5netcdf",
    }
    candidates = list(OPTIONS["netcdf_engine_order"])

    if format is not None:
        format = format.upper()  # type: ignore[assignment]
        if format not in {
            "NETCDF4",
            "NETCDF4_CLASSIC",
            "NETCDF3_64BIT",
            "NETCDF3_CLASSIC",
        }:
            raise ValueError(f"unexpected {format=}")
        # TODO: allow format='NETCDF4_CLASSIC' to default to using h5netcdf,
        # when the oldest supported version of h5netcdf supports it:
        # https://github.com/h5netcdf/h5netcdf/pull/283
        if format != "NETCDF4":
            candidates.remove("h5netcdf")
        if format not in {"NETCDF3_64BIT", "NETCDF3_CLASSIC"}:
            candidates.remove("scipy")

    nczarr_mode = isinstance(path_or_file, str) and path_or_file.endswith(
        "#mode=nczarr"
    )
    if nczarr_mode:
        candidates[:] = ["netcdf4"]

    if isinstance(path_or_file, IOBase):
        candidates.remove("netcdf4")

    for engine in candidates:
        module_name = module_names[engine]
        if importlib.util.find_spec(module_name) is not None:
            return engine

    if nczarr_mode:
        format_str = " in NCZarr format"
    else:
        format_str = f" with {format=}" if format is not None else ""
    libraries = ", ".join(module_names[c] for c in candidates)
    raise ValueError(
        f"cannot write NetCDF files{format_str} because none of the suitable "
        f"backend libraries ({libraries}) are installed"
    )


def _sanitize_unlimited_dims(dataset, unlimited_dims):