Function guess_chunkmanager

xarray/namedarray/parallelcompat.py:94–140

Get namespace of chunk-handling methods, guessing from what's available. If the name of a specific ChunkManager is given (e.g. "dask"), then use that. Else use whatever is installed, defaulting to dask if there are multiple options.

(
    manager: str | ChunkManagerEntrypoint[Any] | None,
)

def guess_chunkmanager(
    manager: str | ChunkManagerEntrypoint[Any] | None,
) -> ChunkManagerEntrypoint[Any]:
    """
    Get namespace of chunk-handling methods, guessing from what's available.

    If the name of a specific ChunkManager is given (e.g. "dask"), then use that.
    Else use whatever is installed, defaulting to dask if there are multiple options.
    """

    available_chunkmanagers = list_chunkmanagers()

    if manager is None:
        if len(available_chunkmanagers) == 1:
            # use the only option available
            manager = next(iter(available_chunkmanagers.keys()))
        else:
            # use the one in options (default dask)
            manager = OPTIONS["chunk_manager"]

    if isinstance(manager, str):
        if manager not in available_chunkmanagers and manager in KNOWN_CHUNKMANAGERS:
            raise ImportError(
                f"chunk manager {manager!r} is not available."
                f" Please make sure {KNOWN_CHUNKMANAGERS[manager]!r} is installed"
                " and importable."
            )
        elif len(available_chunkmanagers) == 0:
            raise ImportError(
                "no chunk managers available. Try installing `dask` or another package"
                " that provides a chunk manager."
            )
        elif manager not in available_chunkmanagers:
            raise ValueError(
                f"unrecognized chunk manager {manager!r} - must be one of the installed"
                f" chunk managers: {list(available_chunkmanagers)}"
            )

        return available_chunkmanagers[manager]
    elif isinstance(manager, ChunkManagerEntrypoint):
        # already a valid ChunkManager so just pass through
        return manager
    else:
        raise TypeError(
            "manager must be a string or instance of ChunkManagerEntrypoint,"
            f" but received type {type(manager)}"
        )


def get_chunked_array_type(*args: Any) -> ChunkManagerEntrypoint[Any]: