Function _check_compression_codec_available

xarray/tests/test_backends.py:182–240

Check if a compression codec is available in the netCDF4 library. Parameters ---------- codec : str or None The compression codec name (e.g., 'zstd', 'blosc_lz', etc.) Returns ------- bool True if the codec is available, False otherwise.

(codec: str | None)

def _check_compression_codec_available(codec: str | None) -> bool:
    """Check if a compression codec is available in the netCDF4 library.

    Parameters
    ----------
    codec : str or None
        The compression codec name (e.g., 'zstd', 'blosc_lz', etc.)

    Returns
    -------
    bool
        True if the codec is available, False otherwise.
    """
    if codec is None or codec in ("zlib", "szip"):
        # These are standard and should be available
        return True

    if not has_netCDF4:
        return False

    try:
        import os

        import netCDF4

        # Try to create a file with the compression to test availability
        with tempfile.NamedTemporaryFile(suffix=".nc", delete=False) as tmp:
            tmp_path = tmp.name

        try:
            nc = netCDF4.Dataset(tmp_path, "w", format="NETCDF4")
            nc.createDimension("x", 10)

            # Attempt to create a variable with the compression
            if codec and codec.startswith("blosc"):
                nc.createVariable(  # type: ignore[call-overload, unused-ignore]
                    varname="test",
                    datatype="f4",
                    dimensions=("x",),
                    compression=codec,
                    blosc_shuffle=1,
                )
            else:
                nc.createVariable(  # type: ignore[call-overload, unused-ignore]
                    varname="test", datatype="f4", dimensions=("x",), compression=codec
                )

            nc.close()
            os.unlink(tmp_path)
            return True
        except (RuntimeError, netCDF4.NetCDF4MissingFeatureException):
            # Codec not available
            if os.path.exists(tmp_path):
                with contextlib.suppress(OSError):
                    os.unlink(tmp_path)
            return False
    except Exception:
        # Any other error, assume codec is not available