Function _get_chunk

xarray/namedarray/utils.py:203–264

Return map from each dim to chunk sizes, accounting for backend's preferred chunks.

(  # type: ignore[no-untyped-def]
    data: DuckArray[Any],
    chunks,
    chunkmanager: ChunkManagerEntrypoint[Any],
    *,
    preferred_chunks,
    dims=None,
)

def _get_chunk(  # type: ignore[no-untyped-def]
    data: DuckArray[Any],
    chunks,
    chunkmanager: ChunkManagerEntrypoint[Any],
    *,
    preferred_chunks,
    dims=None,
) -> Mapping[Any, T_ChunkDim]:
    """
    Return map from each dim to chunk sizes, accounting for backend's preferred chunks.
    """
    from xarray.core.common import _contains_cftime_datetimes
    from xarray.core.utils import emit_user_level_warning
    from xarray.structure.chunks import _get_breaks_cached

    dims = chunks.keys() if dims is None else dims
    shape = data.shape

    # Determine the explicit requested chunks.
    preferred_chunk_shape = tuple(
        itertools.starmap(preferred_chunks.get, zip(dims, shape, strict=True))
    )
    if isinstance(chunks, Number) or (chunks == "auto"):
        chunks = dict.fromkeys(dims, chunks)
    chunk_shape = tuple(
        chunks.get(dim, None) or preferred_chunk_sizes
        for dim, preferred_chunk_sizes in zip(dims, preferred_chunk_shape, strict=True)
    )

    limit: int | None
    if _contains_cftime_datetimes(data):
        limit, dtype = fake_target_chunksize(data, chunkmanager.get_auto_chunk_size())
    else:
        limit = None
        dtype = data.dtype

    chunk_shape = chunkmanager.normalize_chunks(
        chunk_shape,
        shape=shape,
        dtype=dtype,
        limit=limit,
        previous_chunks=preferred_chunk_shape,
    )

    # Warn where requested chunks break preferred chunks, provided that the variable
    # contains data.
    if data.size:  # type: ignore[unused-ignore,attr-defined]  # DuckArray protocol doesn't include 'size' - should it?
        for dim, size, chunk_sizes in zip(dims, shape, chunk_shape, strict=True):
            if preferred_chunk_sizes := preferred_chunks.get(dim):
                disagreement = _get_breaks_cached(
                    size=size,
                    chunk_sizes=chunk_sizes,
                    preferred_chunk_sizes=preferred_chunk_sizes,
                )
                if disagreement:
                    emit_user_level_warning(
                        "The specified chunks separate the stored chunks along "
                        f'dimension "{dim}" starting at index {disagreement}. This could '