zarr.codecs

zarr.codecs

SubchunkWriteOrder module-attribute

SubchunkWriteOrder = Literal[
    "morton",
    "unordered",
    "lexicographic",
    "colexicographic",
]

__all__ module-attribute

__all__ = [
    "BloscCname",
    "BloscCodec",
    "BloscShuffle",
    "BytesCodec",
    "CastValue",
    "Crc32cCodec",
    "Endian",
    "GzipCodec",
    "ScaleOffset",
    "ShardingCodec",
    "ShardingCodecIndexLocation",
    "SubchunkWriteOrder",
    "TransposeCodec",
    "VLenBytesCodec",
    "VLenUTF8Codec",
    "ZstdCodec",
]

BloscCname

Deprecated. Pass a literal string (one of "lz4", "lz4hc", "blosclz", "snappy", "zlib", "zstd") directly to BloscCodec instead.

Source code in zarr/codecs/blosc.py

class BloscCname(metaclass=_DeprecatedStrEnumMeta):
    """
    Deprecated. Pass a literal string (one of `"lz4"`, `"lz4hc"`,
    `"blosclz"`, `"snappy"`, `"zlib"`, `"zstd"`) directly to
    `BloscCodec` instead.
    """

    _members: ClassVar[dict[str, str]] = {
        "lz4": "lz4",
        "lz4hc": "lz4hc",
        "blosclz": "blosclz",
        "snappy": "snappy",
        "zstd": "zstd",
        "zlib": "zlib",
    }

BloscCodec dataclass

Bases: BytesBytesCodec

Blosc compression codec for zarr.

Blosc is a high-performance compressor optimized for binary data. It uses a combination of blocking, shuffling, and fast compression algorithms to achieve excellent compression ratios and speed.

Attributes:

• is_fixed_size (bool) –

  Always False for Blosc codec, as compression produces variable-sized output.

• typesize (int) –

  The data type size in bytes used for shuffle filtering.

• cname (BloscCnameLiteral) –

  The compression algorithm being used; one of "lz4", "lz4hc", "blosclz", "snappy", "zlib", or "zstd".

• clevel (int) –

  The compression level (0-9).

• shuffle (BloscShuffleLiteral) –

  The shuffle filter mode; one of "noshuffle", "shuffle", or "bitshuffle".

• blocksize (int) –

  The size of compressed blocks in bytes (0 for automatic).

Parameters:

• typesize (int, default: None ) –

  The data type size in bytes. This affects how the shuffle filter processes the data. If None, defaults to 1 and the attribute is marked as tunable. Default: 1.

• cname (BloscCnameLiteral, default: 'zstd' ) –

  The compression algorithm to use; one of "lz4", "lz4hc", "blosclz", "snappy", "zlib", or "zstd". Default is "zstd". Passing a BloscCname enum is deprecated.

• clevel (int, default: 5 ) –

  The compression level, from 0 (no compression) to 9 (maximum compression). Higher values provide better compression at the cost of speed. Default: 5.

• shuffle (BloscShuffleLiteral or None, default: None ) –

  The shuffle filter to apply before compression; one of "noshuffle", "shuffle", or "bitshuffle":

  • 'noshuffle': No shuffling
  • 'shuffle': Byte shuffling (better for typesize > 1)
  • 'bitshuffle': Bit shuffling (better for typesize == 1)

  If None, defaults to 'bitshuffle' and the attribute is marked as tunable. Default: 'bitshuffle'.

• blocksize (int, default: 0 ) –

  The requested size of compressed blocks in bytes. A value of 0 means automatic block size selection. Default: 0.

Notes

Tunable attributes: If typesize or shuffle are set to None during initialization, they are marked as tunable attributes. This means they can be adjusted later based on the data type of the array being compressed.

Thread Safety: This codec sets numcodecs.blosc.use_threads = False at module import time to avoid threading issues in asyncio contexts.

Examples:

Create a Blosc codec with default settings:

>>> codec = BloscCodec()
>>> codec.typesize
1
>>> codec.shuffle
'bitshuffle'

Create a codec with specific compression settings:

>>> codec = BloscCodec(cname='zstd', clevel=9, shuffle='shuffle')
>>> codec.cname
'zstd'

Source code in zarr/codecs/blosc.py

@dataclass(frozen=True)
class BloscCodec(BytesBytesCodec):
    """
    Blosc compression codec for zarr.

    Blosc is a high-performance compressor optimized for binary data. It uses a
    combination of blocking, shuffling, and fast compression algorithms to achieve
    excellent compression ratios and speed.

    Attributes
    ----------
    is_fixed_size : bool
        Always False for Blosc codec, as compression produces variable-sized output.
    typesize : int
        The data type size in bytes used for shuffle filtering.
    cname : BloscCnameLiteral
        The compression algorithm being used; one of "lz4", "lz4hc",
        "blosclz", "snappy", "zlib", or "zstd".
    clevel : int
        The compression level (0-9).
    shuffle : BloscShuffleLiteral
        The shuffle filter mode; one of "noshuffle", "shuffle", or
        "bitshuffle".
    blocksize : int
        The size of compressed blocks in bytes (0 for automatic).

    Parameters
    ----------
    typesize : int, optional
        The data type size in bytes. This affects how the shuffle filter processes
        the data. If None, defaults to 1 and the attribute is marked as tunable.
        Default: 1.
    cname : BloscCnameLiteral, optional
        The compression algorithm to use; one of "lz4", "lz4hc", "blosclz",
        "snappy", "zlib", or "zstd". Default is "zstd". Passing a `BloscCname`
        enum is deprecated.
    clevel : int, optional
        The compression level, from 0 (no compression) to 9 (maximum compression).
        Higher values provide better compression at the cost of speed. Default: 5.
    shuffle : BloscShuffleLiteral or None, optional
        The shuffle filter to apply before compression; one of "noshuffle",
        "shuffle", or "bitshuffle":

        - 'noshuffle': No shuffling
        - 'shuffle': Byte shuffling (better for typesize > 1)
        - 'bitshuffle': Bit shuffling (better for typesize == 1)

        If None, defaults to 'bitshuffle' and the attribute is marked
        as tunable. Default: 'bitshuffle'.
    blocksize : int, optional
        The requested size of compressed blocks in bytes. A value of 0 means
        automatic block size selection. Default: 0.

    Notes
    -----
    **Tunable attributes**: If `typesize` or `shuffle` are set to None during
    initialization, they are marked as tunable attributes. This means they can be
    adjusted later based on the data type of the array being compressed.

    **Thread Safety**: This codec sets `numcodecs.blosc.use_threads = False` at
    module import time to avoid threading issues in asyncio contexts.

    Examples
    --------
    Create a Blosc codec with default settings:

    >>> codec = BloscCodec()
    >>> codec.typesize
    1
    >>> codec.shuffle
    'bitshuffle'

    Create a codec with specific compression settings:

    >>> codec = BloscCodec(cname='zstd', clevel=9, shuffle='shuffle')
    >>> codec.cname
    'zstd'
    """

    # This attribute tracks parameters were set to None at init time, and thus tunable
    _tunable_attrs: set[Literal["typesize", "shuffle"]] = field(init=False)
    is_fixed_size = False

    typesize: int
    cname: BloscCnameLiteral
    clevel: int
    shuffle: BloscShuffleLiteral
    blocksize: int

    def __init__(
        self,
        *,
        typesize: int | None = None,
        cname: BloscCname | BloscCnameLiteral = "zstd",
        clevel: int = 5,
        shuffle: BloscShuffle | BloscShuffleLiteral | None = None,
        blocksize: int = 0,
    ) -> None:
        object.__setattr__(self, "_tunable_attrs", set())

        if typesize is None:
            typesize = 1
            self._tunable_attrs.update({"typesize"})

        if shuffle is None:
            shuffle = "bitshuffle"
            self._tunable_attrs.update({"shuffle"})

        cname = _coerce_enum_input(cname, "cname", "BloscCodec")  # type: ignore[assignment]
        shuffle = _coerce_enum_input(shuffle, "shuffle", "BloscCodec")  # type: ignore[assignment]

        typesize_parsed = parse_typesize(typesize)
        cname_parsed = _parse_cname(cname)
        clevel_parsed = parse_clevel(clevel)
        shuffle_parsed = _parse_shuffle(shuffle)
        blocksize_parsed = parse_blocksize(blocksize)

        object.__setattr__(self, "typesize", typesize_parsed)
        object.__setattr__(self, "cname", cname_parsed)
        object.__setattr__(self, "clevel", clevel_parsed)
        object.__setattr__(self, "shuffle", shuffle_parsed)
        object.__setattr__(self, "blocksize", blocksize_parsed)

    @classmethod
    def from_dict(cls, data: dict[str, JSON]) -> Self:
        _, configuration_parsed = parse_named_configuration(data, "blosc")
        return cls(**configuration_parsed)  # type: ignore[arg-type]

    def to_dict(self) -> dict[str, JSON]:
        result: BloscJSON_V3 = {
            "name": "blosc",
            "configuration": {
                "typesize": self.typesize,
                "cname": self.cname,
                "clevel": self.clevel,
                "shuffle": self.shuffle,
                "blocksize": self.blocksize,
            },
        }
        return result  # type: ignore[return-value]

    def evolve_from_array_spec(self, array_spec: ArraySpec) -> Self:
        """
        Create a new codec with typesize and shuffle parameters adjusted
        according to the size of each element in the data type
        associated with array_spec. Parameters are only updated if they were set to
        None when self.__init__ was called.
        """
        item_size = 1
        if isinstance(array_spec.dtype, HasItemSize):
            item_size = array_spec.dtype.item_size
        new_codec = self
        if "typesize" in self._tunable_attrs:
            new_codec = replace(new_codec, typesize=item_size)
        if "shuffle" in self._tunable_attrs:
            new_codec = replace(
                new_codec,
                shuffle=("bitshuffle" if item_size == 1 else "shuffle"),
            )

        return new_codec

    @cached_property
    def _blosc_codec(self) -> Blosc:
        map_shuffle_str_to_int: dict[BloscShuffleLiteral, int] = {
            "noshuffle": 0,
            "shuffle": 1,
            "bitshuffle": 2,
        }
        config_dict: BloscConfigV2 = {
            "cname": self.cname,
            "clevel": self.clevel,
            "shuffle": map_shuffle_str_to_int[self.shuffle],
            "blocksize": self.blocksize,
        }
        # See https://github.com/zarr-developers/numcodecs/pull/713
        if Version(numcodecs.__version__) >= Version("0.16.0"):
            config_dict["typesize"] = self.typesize
        return Blosc.from_config(config_dict)

    def _decode_sync(
        self,
        chunk_bytes: Buffer,
        chunk_spec: ArraySpec,
    ) -> Buffer:
        return as_numpy_array_wrapper(self._blosc_codec.decode, chunk_bytes, chunk_spec.prototype)

    async def _decode_single(
        self,
        chunk_bytes: Buffer,
        chunk_spec: ArraySpec,
    ) -> Buffer:
        return await asyncio.to_thread(self._decode_sync, chunk_bytes, chunk_spec)

    def _encode_sync(
        self,
        chunk_bytes: Buffer,
        chunk_spec: ArraySpec,
    ) -> Buffer | None:
        # Since blosc only support host memory, we convert the input and output of the encoding
        # between numpy array and buffer
        return chunk_spec.prototype.buffer.from_bytes(
            self._blosc_codec.encode(chunk_bytes.as_numpy_array())
        )

    async def _encode_single(
        self,
        chunk_bytes: Buffer,
        chunk_spec: ArraySpec,
    ) -> Buffer | None:
        return await asyncio.to_thread(self._encode_sync, chunk_bytes, chunk_spec)

    def compute_encoded_size(self, _input_byte_length: int, _chunk_spec: ArraySpec) -> int:
        raise NotImplementedError

blocksize instance-attribute

blocksize: int

clevel instance-attribute

clevel: int

cname instance-attribute

cname: BloscCnameLiteral

is_fixed_size class-attribute instance-attribute

is_fixed_size = False

shuffle instance-attribute

shuffle: BloscShuffleLiteral

typesize instance-attribute

typesize: int

__init__

__init__(
    *,
    typesize: int | None = None,
    cname: BloscCname | BloscCnameLiteral = "zstd",
    clevel: int = 5,
    shuffle: BloscShuffle
    | BloscShuffleLiteral
    | None = None,
    blocksize: int = 0,
) -> None

Source code in zarr/codecs/blosc.py

def __init__(
    self,
    *,
    typesize: int | None = None,
    cname: BloscCname | BloscCnameLiteral = "zstd",
    clevel: int = 5,
    shuffle: BloscShuffle | BloscShuffleLiteral | None = None,
    blocksize: int = 0,
) -> None:
    object.__setattr__(self, "_tunable_attrs", set())

    if typesize is None:
        typesize = 1
        self._tunable_attrs.update({"typesize"})

    if shuffle is None:
        shuffle = "bitshuffle"
        self._tunable_attrs.update({"shuffle"})

    cname = _coerce_enum_input(cname, "cname", "BloscCodec")  # type: ignore[assignment]
    shuffle = _coerce_enum_input(shuffle, "shuffle", "BloscCodec")  # type: ignore[assignment]

    typesize_parsed = parse_typesize(typesize)
    cname_parsed = _parse_cname(cname)
    clevel_parsed = parse_clevel(clevel)
    shuffle_parsed = _parse_shuffle(shuffle)
    blocksize_parsed = parse_blocksize(blocksize)

    object.__setattr__(self, "typesize", typesize_parsed)
    object.__setattr__(self, "cname", cname_parsed)
    object.__setattr__(self, "clevel", clevel_parsed)
    object.__setattr__(self, "shuffle", shuffle_parsed)
    object.__setattr__(self, "blocksize", blocksize_parsed)

compute_encoded_size

compute_encoded_size(
    _input_byte_length: int, _chunk_spec: ArraySpec
) -> int

Given an input byte length, this method returns the output byte length. Raises a NotImplementedError for codecs with variable-sized outputs (e.g. compressors).

Parameters:

• input_byte_length (int) –
• chunk_spec (ArraySpec) –

Returns:

• int –

Source code in zarr/codecs/blosc.py

def compute_encoded_size(self, _input_byte_length: int, _chunk_spec: ArraySpec) -> int:
    raise NotImplementedError

decode async

decode(
    chunks_and_specs: Iterable[tuple[CO | None, ArraySpec]],
) -> Iterable[CI | None]

Decodes a batch of chunks. Chunks can be None in which case they are ignored by the codec.

Parameters:

• chunks_and_specs (Iterable[tuple[CodecOutput | None, ArraySpec]]) –

  Ordered set of encoded chunks with their accompanying chunk spec.

Returns:

• Iterable[CI | None] –

Source code in zarr/abc/codec.py

async def decode(
    self,
    chunks_and_specs: Iterable[tuple[CO | None, ArraySpec]],
) -> Iterable[CI | None]:
    """Decodes a batch of chunks.
    Chunks can be None in which case they are ignored by the codec.

    Parameters
    ----------
    chunks_and_specs : Iterable[tuple[CodecOutput | None, ArraySpec]]
        Ordered set of encoded chunks with their accompanying chunk spec.

    Returns
    -------
    Iterable[CI | None]
    """
    return await _batching_helper(self._decode_single, chunks_and_specs)

encode async

encode(
    chunks_and_specs: Iterable[tuple[CI | None, ArraySpec]],
) -> Iterable[CO | None]

Encodes a batch of chunks. Chunks can be None in which case they are ignored by the codec.

Parameters:

• chunks_and_specs (Iterable[tuple[CI | None, ArraySpec]]) –

  Ordered set of to-be-encoded chunks with their accompanying chunk spec.

Returns:

• Iterable[CodecOutput | None] –

Source code in zarr/abc/codec.py

async def encode(
    self,
    chunks_and_specs: Iterable[tuple[CI | None, ArraySpec]],
) -> Iterable[CO | None]:
    """Encodes a batch of chunks.
    Chunks can be None in which case they are ignored by the codec.

    Parameters
    ----------
    chunks_and_specs : Iterable[tuple[CI | None, ArraySpec]]
        Ordered set of to-be-encoded chunks with their accompanying chunk spec.

    Returns
    -------
    Iterable[CodecOutput | None]
    """
    return await _batching_helper(self._encode_single, chunks_and_specs)

evolve_from_array_spec

evolve_from_array_spec(array_spec: ArraySpec) -> Self

Create a new codec with typesize and shuffle parameters adjusted according to the size of each element in the data type associated with array_spec. Parameters are only updated if they were set to None when self.init was called.

Source code in zarr/codecs/blosc.py

def evolve_from_array_spec(self, array_spec: ArraySpec) -> Self:
    """
    Create a new codec with typesize and shuffle parameters adjusted
    according to the size of each element in the data type
    associated with array_spec. Parameters are only updated if they were set to
    None when self.__init__ was called.
    """
    item_size = 1
    if isinstance(array_spec.dtype, HasItemSize):
        item_size = array_spec.dtype.item_size
    new_codec = self
    if "typesize" in self._tunable_attrs:
        new_codec = replace(new_codec, typesize=item_size)
    if "shuffle" in self._tunable_attrs:
        new_codec = replace(
            new_codec,
            shuffle=("bitshuffle" if item_size == 1 else "shuffle"),
        )

    return new_codec

from_dict classmethod

from_dict(data: dict[str, JSON]) -> Self

Create an instance of the model from a dictionary

Source code in zarr/codecs/blosc.py

@classmethod
def from_dict(cls, data: dict[str, JSON]) -> Self:
    _, configuration_parsed = parse_named_configuration(data, "blosc")
    return cls(**configuration_parsed)  # type: ignore[arg-type]

resolve_metadata

resolve_metadata(chunk_spec: ArraySpec) -> ArraySpec

Computed the spec of the chunk after it has been encoded by the codec. This is important for codecs that change the shape, data type or fill value of a chunk. The spec will then be used for subsequent codecs in the pipeline.

Parameters:

• chunk_spec (ArraySpec) –

Returns:

• ArraySpec –

Source code in zarr/abc/codec.py

def resolve_metadata(self, chunk_spec: ArraySpec) -> ArraySpec:
    """Computed the spec of the chunk after it has been encoded by the codec.
    This is important for codecs that change the shape, data type or fill value of a chunk.
    The spec will then be used for subsequent codecs in the pipeline.

    Parameters
    ----------
    chunk_spec : ArraySpec

    Returns
    -------
    ArraySpec
    """
    return chunk_spec

to_dict

to_dict() -> dict[str, JSON]

Recursively serialize this model to a dictionary. This method inspects the fields of self and calls x.to_dict() for any fields that are instances of Metadata. Sequences of Metadata are similarly recursed into, and the output of that recursion is collected in a list.

Source code in zarr/codecs/blosc.py

def to_dict(self) -> dict[str, JSON]:
    result: BloscJSON_V3 = {
        "name": "blosc",
        "configuration": {
            "typesize": self.typesize,
            "cname": self.cname,
            "clevel": self.clevel,
            "shuffle": self.shuffle,
            "blocksize": self.blocksize,
        },
    }
    return result  # type: ignore[return-value]

validate

validate(
    *,
    shape: tuple[int, ...],
    dtype: ZDType[TBaseDType, TBaseScalar],
    chunk_grid: ChunkGridMetadata,
) -> None

Validates that the codec configuration is compatible with the array metadata. Raises errors when the codec configuration is not compatible.

Parameters:

• shape (tuple[int, ...]) –

  The array shape

• dtype (dtype[Any]) –

  The array data type

• chunk_grid (ChunkGridMetadata) –

  The array chunk grid metadata

Source code in zarr/abc/codec.py

def validate(
    self,
    *,
    shape: tuple[int, ...],
    dtype: ZDType[TBaseDType, TBaseScalar],
    chunk_grid: ChunkGridMetadata,
) -> None:
    """Validates that the codec configuration is compatible with the array metadata.
    Raises errors when the codec configuration is not compatible.

    Parameters
    ----------
    shape : tuple[int, ...]
        The array shape
    dtype : np.dtype[Any]
        The array data type
    chunk_grid : ChunkGridMetadata
        The array chunk grid metadata
    """

BloscShuffle

Deprecated. Pass a literal string ("noshuffle", "shuffle", or "bitshuffle") directly to BloscCodec instead.

Source code in zarr/codecs/blosc.py

class BloscShuffle(metaclass=_DeprecatedStrEnumMeta):
    """
    Deprecated. Pass a literal string (`"noshuffle"`, `"shuffle"`, or
    `"bitshuffle"`) directly to `BloscCodec` instead.
    """

    _members: ClassVar[dict[str, str]] = {
        "noshuffle": "noshuffle",
        "shuffle": "shuffle",
        "bitshuffle": "bitshuffle",
    }

    @staticmethod
    def from_int(num: int) -> BloscShuffleLiteral:
        mapping: dict[int, BloscShuffleLiteral] = {
            0: "noshuffle",
            1: "shuffle",
            2: "bitshuffle",
        }
        if num not in mapping:
            raise ValueError(f"Value must be between 0 and 2. Got {num}.")
        return mapping[num]

from_int staticmethod

from_int(num: int) -> BloscShuffleLiteral

Source code in zarr/codecs/blosc.py

@staticmethod
def from_int(num: int) -> BloscShuffleLiteral:
    mapping: dict[int, BloscShuffleLiteral] = {
        0: "noshuffle",
        1: "shuffle",
        2: "bitshuffle",
    }
    if num not in mapping:
        raise ValueError(f"Value must be between 0 and 2. Got {num}.")
    return mapping[num]

BytesCodec dataclass

Bases: ArrayBytesCodec

bytes codec

Source code in zarr/codecs/bytes.py

@dataclass(frozen=True)
class BytesCodec(ArrayBytesCodec):
    """bytes codec"""

    is_fixed_size = True

    endian: EndianLiteral | None

    def __init__(self, *, endian: Endian | EndianLiteral | None = sys.byteorder) -> None:
        if endian is None:
            endian_parsed: EndianLiteral | None = None
        else:
            coerced = _coerce_enum_input(endian, "endian", "BytesCodec")
            endian_parsed = _parse_endian(coerced)

        object.__setattr__(self, "endian", endian_parsed)

    @classmethod
    def from_dict(cls, data: dict[str, JSON]) -> Self:
        _, configuration_parsed = parse_named_configuration(
            data, "bytes", require_configuration=False
        )
        configuration_parsed = configuration_parsed or {}
        configuration_parsed.setdefault("endian", None)
        return cls(**configuration_parsed)  # type: ignore[arg-type]

    def to_dict(self) -> dict[str, JSON]:
        if self.endian is None:
            return {"name": "bytes"}
        else:
            return {"name": "bytes", "configuration": {"endian": self.endian}}

    def evolve_from_array_spec(self, array_spec: ArraySpec) -> Self:
        if isinstance(array_spec.dtype, Struct):
            if array_spec.dtype.has_multi_byte_fields():
                if self.endian is None:
                    warnings.warn(
                        "Missing 'endian' for structured dtype with multi-byte fields. "
                        "Assuming little-endian for legacy compatibility.",
                        UserWarning,
                        stacklevel=2,
                    )
                    return replace(self, endian="little")
            else:
                if self.endian is not None:
                    return replace(self, endian=None)
        elif not isinstance(array_spec.dtype, HasEndianness):
            if self.endian is not None:
                return replace(self, endian=None)
        elif self.endian is None:
            raise ValueError(
                "The `endian` configuration needs to be specified for multi-byte data types."
            )
        return self

    def _decode_sync(
        self,
        chunk_bytes: Buffer,
        chunk_spec: ArraySpec,
    ) -> NDBuffer:
        endian_str = self.endian
        dtype = chunk_spec.dtype.to_native_dtype()
        # The byte order of the stored data is set by this codec's `endian`
        # configuration; the byte order of the decoded array is set by the array's
        # data type. The two are independent: the raw bytes are viewed with a dtype
        # in the stored byte order, then converted to the declared dtype if needed.
        if isinstance(chunk_spec.dtype, HasEndianness):
            view_dtype = replace(chunk_spec.dtype, endianness=endian_str).to_native_dtype()  # type: ignore[call-arg]
        elif isinstance(chunk_spec.dtype, Struct) and endian_str is not None:
            # Per the struct data type spec, all multi-byte fields are stored in the
            # byte order configured on this codec.
            view_dtype = dtype.newbyteorder(endian_str)
        else:
            view_dtype = dtype
        as_array_like = chunk_bytes.as_array_like()
        chunk_array = chunk_spec.prototype.nd_buffer.from_ndarray_like(
            as_array_like.view(dtype=view_dtype)  # type: ignore[attr-defined]
        )
        if view_dtype != dtype:
            # This byte-swapping conversion copies the chunk. The dtype inequality
            # guard keeps the common case, where the stored and declared byte orders
            # already match, on the zero-copy view path above.
            chunk_array = chunk_array.astype(dtype)

        # ensure correct chunk shape
        if chunk_array.shape != chunk_spec.shape:
            chunk_array = chunk_array.reshape(
                chunk_spec.shape,
            )
        return chunk_array

    async def _decode_single(
        self,
        chunk_bytes: Buffer,
        chunk_spec: ArraySpec,
    ) -> NDBuffer:
        return self._decode_sync(chunk_bytes, chunk_spec)

    def _encode_sync(
        self,
        chunk_array: NDBuffer,
        chunk_spec: ArraySpec,
    ) -> Buffer | None:
        assert isinstance(chunk_array, NDBuffer)
        if chunk_array.dtype.itemsize > 1 and self.endian is not None:
            # Compare full dtypes rather than the top-level byteorder: numpy reports
            # byteorder '|' for structured dtypes even when their fields are
            # byte-order-sensitive, so newbyteorder is the only reliable way to
            # detect (and normalize) a byte-order mismatch.
            new_dtype = chunk_array.dtype.newbyteorder(self.endian)
            if new_dtype != chunk_array.dtype:
                chunk_array = chunk_array.astype(new_dtype)

        nd_array = chunk_array.as_ndarray_like()
        # Flatten the nd-array (only copy if needed) and reinterpret as bytes
        nd_array = nd_array.ravel().view(dtype="B")
        return chunk_spec.prototype.buffer.from_array_like(nd_array)

    async def _encode_single(
        self,
        chunk_array: NDBuffer,
        chunk_spec: ArraySpec,
    ) -> Buffer | None:
        return self._encode_sync(chunk_array, chunk_spec)

    def compute_encoded_size(self, input_byte_length: int, _chunk_spec: ArraySpec) -> int:
        return input_byte_length

endian instance-attribute

endian: EndianLiteral | None

is_fixed_size class-attribute instance-attribute

is_fixed_size = True

__init__

__init__(
    *, endian: Endian | EndianLiteral | None = byteorder
) -> None

Source code in zarr/codecs/bytes.py

def __init__(self, *, endian: Endian | EndianLiteral | None = sys.byteorder) -> None:
    if endian is None:
        endian_parsed: EndianLiteral | None = None
    else:
        coerced = _coerce_enum_input(endian, "endian", "BytesCodec")
        endian_parsed = _parse_endian(coerced)

    object.__setattr__(self, "endian", endian_parsed)

compute_encoded_size

compute_encoded_size(
    input_byte_length: int, _chunk_spec: ArraySpec
) -> int

Given an input byte length, this method returns the output byte length. Raises a NotImplementedError for codecs with variable-sized outputs (e.g. compressors).

Parameters:

• input_byte_length (int) –
• chunk_spec (ArraySpec) –

Returns:

• int –

Source code in zarr/codecs/bytes.py

def compute_encoded_size(self, input_byte_length: int, _chunk_spec: ArraySpec) -> int:
    return input_byte_length

decode async

decode(
    chunks_and_specs: Iterable[tuple[CO | None, ArraySpec]],
) -> Iterable[CI | None]

Decodes a batch of chunks. Chunks can be None in which case they are ignored by the codec.

Parameters:

• chunks_and_specs (Iterable[tuple[CodecOutput | None, ArraySpec]]) –

  Ordered set of encoded chunks with their accompanying chunk spec.

Returns:

• Iterable[CI | None] –

Source code in zarr/abc/codec.py

async def decode(
    self,
    chunks_and_specs: Iterable[tuple[CO | None, ArraySpec]],
) -> Iterable[CI | None]:
    """Decodes a batch of chunks.
    Chunks can be None in which case they are ignored by the codec.

    Parameters
    ----------
    chunks_and_specs : Iterable[tuple[CodecOutput | None, ArraySpec]]
        Ordered set of encoded chunks with their accompanying chunk spec.

    Returns
    -------
    Iterable[CI | None]
    """
    return await _batching_helper(self._decode_single, chunks_and_specs)

encode async

encode(
    chunks_and_specs: Iterable[tuple[CI | None, ArraySpec]],
) -> Iterable[CO | None]

Encodes a batch of chunks. Chunks can be None in which case they are ignored by the codec.

Parameters:

• chunks_and_specs (Iterable[tuple[CI | None, ArraySpec]]) –

  Ordered set of to-be-encoded chunks with their accompanying chunk spec.

Returns:

• Iterable[CodecOutput | None] –

Source code in zarr/abc/codec.py

async def encode(
    self,
    chunks_and_specs: Iterable[tuple[CI | None, ArraySpec]],
) -> Iterable[CO | None]:
    """Encodes a batch of chunks.
    Chunks can be None in which case they are ignored by the codec.

    Parameters
    ----------
    chunks_and_specs : Iterable[tuple[CI | None, ArraySpec]]
        Ordered set of to-be-encoded chunks with their accompanying chunk spec.

    Returns
    -------
    Iterable[CodecOutput | None]
    """
    return await _batching_helper(self._encode_single, chunks_and_specs)

evolve_from_array_spec

evolve_from_array_spec(array_spec: ArraySpec) -> Self

Fills in codec configuration parameters that can be automatically inferred from the array metadata.

Parameters:

• array_spec (ArraySpec) –

Returns:

• Self –

Source code in zarr/codecs/bytes.py

def evolve_from_array_spec(self, array_spec: ArraySpec) -> Self:
    if isinstance(array_spec.dtype, Struct):
        if array_spec.dtype.has_multi_byte_fields():
            if self.endian is None:
                warnings.warn(
                    "Missing 'endian' for structured dtype with multi-byte fields. "
                    "Assuming little-endian for legacy compatibility.",
                    UserWarning,
                    stacklevel=2,
                )
                return replace(self, endian="little")
        else:
            if self.endian is not None:
                return replace(self, endian=None)
    elif not isinstance(array_spec.dtype, HasEndianness):
        if self.endian is not None:
            return replace(self, endian=None)
    elif self.endian is None:
        raise ValueError(
            "The `endian` configuration needs to be specified for multi-byte data types."
        )
    return self

from_dict classmethod

from_dict(data: dict[str, JSON]) -> Self

Create an instance of the model from a dictionary

Source code in zarr/codecs/bytes.py

@classmethod
def from_dict(cls, data: dict[str, JSON]) -> Self:
    _, configuration_parsed = parse_named_configuration(
        data, "bytes", require_configuration=False
    )
    configuration_parsed = configuration_parsed or {}
    configuration_parsed.setdefault("endian", None)
    return cls(**configuration_parsed)  # type: ignore[arg-type]

resolve_metadata

resolve_metadata(chunk_spec: ArraySpec) -> ArraySpec

Computed the spec of the chunk after it has been encoded by the codec. This is important for codecs that change the shape, data type or fill value of a chunk. The spec will then be used for subsequent codecs in the pipeline.

Parameters:

• chunk_spec (ArraySpec) –

Returns:

• ArraySpec –

Source code in zarr/abc/codec.py

def resolve_metadata(self, chunk_spec: ArraySpec) -> ArraySpec:
    """Computed the spec of the chunk after it has been encoded by the codec.
    This is important for codecs that change the shape, data type or fill value of a chunk.
    The spec will then be used for subsequent codecs in the pipeline.

    Parameters
    ----------
    chunk_spec : ArraySpec

    Returns
    -------
    ArraySpec
    """
    return chunk_spec

to_dict

to_dict() -> dict[str, JSON]

Recursively serialize this model to a dictionary. This method inspects the fields of self and calls x.to_dict() for any fields that are instances of Metadata. Sequences of Metadata are similarly recursed into, and the output of that recursion is collected in a list.

Source code in zarr/codecs/bytes.py

def to_dict(self) -> dict[str, JSON]:
    if self.endian is None:
        return {"name": "bytes"}
    else:
        return {"name": "bytes", "configuration": {"endian": self.endian}}

validate

validate(
    *,
    shape: tuple[int, ...],
    dtype: ZDType[TBaseDType, TBaseScalar],
    chunk_grid: ChunkGridMetadata,
) -> None

Validates that the codec configuration is compatible with the array metadata. Raises errors when the codec configuration is not compatible.

Parameters:

• shape (tuple[int, ...]) –

  The array shape

• dtype (dtype[Any]) –

  The array data type

• chunk_grid (ChunkGridMetadata) –

  The array chunk grid metadata

Source code in zarr/abc/codec.py

def validate(
    self,
    *,
    shape: tuple[int, ...],
    dtype: ZDType[TBaseDType, TBaseScalar],
    chunk_grid: ChunkGridMetadata,
) -> None:
    """Validates that the codec configuration is compatible with the array metadata.
    Raises errors when the codec configuration is not compatible.

    Parameters
    ----------
    shape : tuple[int, ...]
        The array shape
    dtype : np.dtype[Any]
        The array data type
    chunk_grid : ChunkGridMetadata
        The array chunk grid metadata
    """

CastValue dataclass

Bases: ArrayArrayCodec

Cast-value array-to-array codec.

Value-converts array elements to a new data type during encoding, and back to the original data type during decoding.

Requires the cast-value-rs package for the actual casting logic.

Parameters:

• data_type (str or ZDType) –

  Target zarr v3 data type. Strings are looked up by spec name (e.g. "uint8", "float32"); a ZDType instance is used as-is.

• rounding (RoundingMode, default: 'nearest-even' ) –

  How to round when exact representation is impossible. Default is "nearest-even".

• out_of_range (OutOfRangeMode or None, default: None ) –

  What to do when a value is outside the target's range. None means error; "clamp" clips to range; "wrap" uses modular arithmetic (only valid for integer types). Default is None.

• scalar_map (ScalarMap, ScalarMapJSON, or None, default: None ) –

  Explicit mapping from input scalars to output scalars. Default is None.

Attributes:

• dtype (ZDType) –

  Resolved target data type (a ZDType instance, regardless of whether the constructor received a string or a ZDType).

• rounding (RoundingMode) –

  The rounding mode, as supplied to the constructor.

• out_of_range (OutOfRangeMode or None) –

  The out-of-range behaviour, as supplied to the constructor.

• scalar_map (ScalarMap or None) –

  Parsed scalar map (always normalized to ScalarMap form).

References

• The cast_value codec spec: github.com/zarr-developers/zarr-extensions/tree/main/codecs/cast_value

Source code in zarr/codecs/cast_value.py

@dataclass(frozen=True)
class CastValue(ArrayArrayCodec):
    """Cast-value array-to-array codec.

    Value-converts array elements to a new data type during encoding,
    and back to the original data type during decoding.

    Requires the `cast-value-rs` package for the actual casting logic.

    Parameters
    ----------
    data_type : str or ZDType
        Target zarr v3 data type. Strings are looked up by spec name
        (e.g. "uint8", "float32"); a `ZDType` instance is used as-is.
    rounding : RoundingMode
        How to round when exact representation is impossible. Default is
        "nearest-even".
    out_of_range : OutOfRangeMode or None
        What to do when a value is outside the target's range. `None` means
        error; "clamp" clips to range; "wrap" uses modular arithmetic
        (only valid for integer types). Default is `None`.
    scalar_map : ScalarMap, ScalarMapJSON, or None
        Explicit mapping from input scalars to output scalars. Default is
        `None`.

    Attributes
    ----------
    dtype : ZDType
        Resolved target data type (a `ZDType` instance, regardless of
        whether the constructor received a string or a `ZDType`).
    rounding : RoundingMode
        The rounding mode, as supplied to the constructor.
    out_of_range : OutOfRangeMode or None
        The out-of-range behaviour, as supplied to the constructor.
    scalar_map : ScalarMap or None
        Parsed scalar map (always normalized to `ScalarMap` form).

    References
    ----------

    - The `cast_value` codec spec: https://github.com/zarr-developers/zarr-extensions/tree/main/codecs/cast_value
    """

    is_fixed_size = True

    dtype: ZDType[TBaseDType, TBaseScalar]
    rounding: RoundingMode
    out_of_range: OutOfRangeMode | None
    scalar_map: ScalarMap | None

    def __init__(
        self,
        *,
        data_type: str | ZDType[TBaseDType, TBaseScalar],
        rounding: RoundingMode = "nearest-even",
        out_of_range: OutOfRangeMode | None = None,
        scalar_map: ScalarMapJSON | ScalarMap | None = None,
    ) -> None:
        if isinstance(data_type, str):
            zdtype = get_data_type_from_json(data_type, zarr_format=3)
        else:
            zdtype = data_type
        if zdtype.to_json(zarr_format=3) not in PERMITTED_DATA_TYPE_NAMES:
            raise ValueError(
                f"Invalid target data type {data_type!r}. "
                f"cast_value codec only supports integer and floating-point data types. "
                f"Got {zdtype}."
            )
        object.__setattr__(self, "dtype", zdtype)
        object.__setattr__(self, "rounding", rounding)
        object.__setattr__(self, "out_of_range", out_of_range)
        if scalar_map is not None:
            parsed = parse_scalar_map(scalar_map)
        else:
            parsed = None
        object.__setattr__(self, "scalar_map", parsed)

    @classmethod
    def from_dict(cls, data: dict[str, JSON]) -> Self:
        _, configuration_parsed = parse_named_configuration(
            data, "cast_value", require_configuration=True
        )
        return cls(**configuration_parsed)  # type: ignore[arg-type]

    def to_dict(self) -> dict[str, JSON]:
        config: dict[str, JSON] = {"data_type": cast("JSON", self.dtype.to_json(zarr_format=3))}
        if self.rounding != "nearest-even":
            config["rounding"] = self.rounding
        if self.out_of_range is not None:
            config["out_of_range"] = self.out_of_range
        if self.scalar_map is not None:
            json_map: dict[str, list[tuple[object, object]]] = {}
            for direction in ("encode", "decode"):
                if direction in self.scalar_map:
                    json_map[direction] = [(k, v) for k, v in self.scalar_map[direction].items()]
            config["scalar_map"] = cast("JSON", json_map)
        return {"name": "cast_value", "configuration": config}

    def validate(
        self,
        *,
        shape: tuple[int, ...],
        dtype: ZDType[TBaseDType, TBaseScalar],
        chunk_grid: ChunkGridMetadata,
    ) -> None:
        # `dtype` is the source (the array's dtype); `self.dtype` is the
        # cast target. The spec requires both to be permitted, and rules
        # like `out_of_range="wrap"` apply to the target.
        source_name = dtype.to_json(zarr_format=3)
        target_name = self.dtype.to_json(zarr_format=3)
        for role, name in (("source", source_name), ("target", target_name)):
            if name not in PERMITTED_DATA_TYPE_NAMES:
                raise ValueError(
                    f"The cast_value codec only supports integer and floating-point data types. "
                    f"Got {role} dtype {name}."
                )
        if self.out_of_range == "wrap" and target_name not in CAST_VALUE_INT_DTYPES:
            raise ValueError(
                f"out_of_range='wrap' is only valid for integer target types. "
                f"Got target dtype {target_name}."
            )

        if self.scalar_map is not None:
            self._validate_scalar_map(dtype, self.dtype)

    def _validate_scalar_map(
        self,
        source_zdtype: ZDType[TBaseDType, TBaseScalar],
        target_zdtype: ZDType[TBaseDType, TBaseScalar],
    ) -> None:
        """Validate that scalar map entries are compatible with source/target dtypes."""
        assert self.scalar_map is not None
        # For encode: keys are source values, values are target values.
        # For decode: keys are target values, values are source values.
        direction_dtypes: dict[
            str, tuple[ZDType[TBaseDType, TBaseScalar], ZDType[TBaseDType, TBaseScalar]]
        ] = {
            "encode": (source_zdtype, target_zdtype),
            "decode": (target_zdtype, source_zdtype),
        }
        for direction, (key_zdtype, val_zdtype) in direction_dtypes.items():
            if direction not in self.scalar_map:
                continue
            sub_map = self.scalar_map[direction]  # type: ignore[literal-required]
            for k, v in sub_map.items():
                _check_representable(k, key_zdtype, f"scalar_map {direction} key")
                _check_representable(v, val_zdtype, f"scalar_map {direction} value")

    def _do_cast(
        self,
        arr: np.ndarray,
        *,
        target_dtype: np.dtype,
        scalar_map: Mapping[str | float | int, str | float | int] | None,
    ) -> np.ndarray:
        if not _HAS_RUST_BACKEND:
            raise ImportError(
                "The cast_value codec requires the 'cast-value-rs' package. "
                "Install it with: pip install cast-value-rs"
            )
        scalar_map_entries: dict[float | int, float | int] | None = None
        if scalar_map is not None:
            src_dtype = arr.dtype
            to_src = int if np.issubdtype(src_dtype, np.integer) else float
            to_tgt = int if np.issubdtype(target_dtype, np.integer) else float
            scalar_map_entries = {to_src(k): to_tgt(v) for k, v in scalar_map.items()}
        return cast_array_rs(  # type: ignore[no-any-return]
            arr,
            target_dtype=target_dtype,
            rounding_mode=self.rounding,
            out_of_range_mode=self.out_of_range,
            scalar_map_entries=scalar_map_entries,
        )

    def _get_scalar_map(
        self, direction: str
    ) -> Mapping[str | float | int, str | float | int] | None:
        """Extract the encode or decode mapping from scalar_map, or None."""
        if self.scalar_map is None:
            return None
        return self.scalar_map.get(direction)  # type: ignore[return-value]

    def resolve_metadata(self, chunk_spec: ArraySpec) -> ArraySpec:
        """
        Update the fill value of the output spec by applying casting procedure.
        """
        target_zdtype = self.dtype
        target_native = target_zdtype.to_native_dtype()
        source_native = chunk_spec.dtype.to_native_dtype()

        fill = chunk_spec.fill_value
        fill_arr = np.array([fill], dtype=source_native)

        new_fill_arr = self._do_cast(
            fill_arr, target_dtype=target_native, scalar_map=self._get_scalar_map("encode")
        )
        new_fill = target_native.type(new_fill_arr[0])

        return replace(chunk_spec, dtype=target_zdtype, fill_value=new_fill)

    def _encode_sync(
        self,
        chunk_array: NDBuffer,
        _chunk_spec: ArraySpec,
    ) -> NDBuffer | None:
        arr = chunk_array.as_ndarray_like()
        target_native = self.dtype.to_native_dtype()

        result = self._do_cast(
            np.asarray(arr), target_dtype=target_native, scalar_map=self._get_scalar_map("encode")
        )
        return chunk_array.__class__.from_ndarray_like(result)

    async def _encode_single(
        self,
        chunk_data: NDBuffer,
        chunk_spec: ArraySpec,
    ) -> NDBuffer | None:
        return self._encode_sync(chunk_data, chunk_spec)

    def _decode_sync(
        self,
        chunk_array: NDBuffer,
        chunk_spec: ArraySpec,
    ) -> NDBuffer:
        arr = chunk_array.as_ndarray_like()
        target_native = chunk_spec.dtype.to_native_dtype()

        result = self._do_cast(
            np.asarray(arr), target_dtype=target_native, scalar_map=self._get_scalar_map("decode")
        )
        return chunk_array.__class__.from_ndarray_like(result)

    async def _decode_single(
        self,
        chunk_data: NDBuffer,
        chunk_spec: ArraySpec,
    ) -> NDBuffer:
        return self._decode_sync(chunk_data, chunk_spec)

    def compute_encoded_size(self, input_byte_length: int, chunk_spec: ArraySpec) -> int:
        dtype_name = chunk_spec.dtype.to_json(zarr_format=3)
        if dtype_name not in PERMITTED_DATA_TYPE_NAMES:
            raise ValueError(
                "cast_value codec only supports fixed-size integer and floating-point data types. "
                f"Got source dtype: {chunk_spec.dtype}."
            )
        source_itemsize = chunk_spec.dtype.to_native_dtype().itemsize
        target_itemsize = self.dtype.to_native_dtype().itemsize
        if source_itemsize == 0 or target_itemsize == 0:
            raise ValueError(
                "cast_value codec requires fixed-size data types. "
                f"Got source itemsize={source_itemsize}, target itemsize={target_itemsize}."
            )
        num_elements = input_byte_length // source_itemsize
        return num_elements * target_itemsize

dtype instance-attribute

dtype: ZDType[TBaseDType, TBaseScalar]

is_fixed_size class-attribute instance-attribute

is_fixed_size = True

out_of_range instance-attribute

out_of_range: OutOfRangeMode | None

rounding instance-attribute

rounding: RoundingMode

scalar_map instance-attribute

scalar_map: ScalarMap | None

__init__

__init__(
    *,
    data_type: str | ZDType[TBaseDType, TBaseScalar],
    rounding: RoundingMode = "nearest-even",
    out_of_range: OutOfRangeMode | None = None,
    scalar_map: ScalarMapJSON | ScalarMap | None = None,
) -> None

Source code in zarr/codecs/cast_value.py

def __init__(
    self,
    *,
    data_type: str | ZDType[TBaseDType, TBaseScalar],
    rounding: RoundingMode = "nearest-even",
    out_of_range: OutOfRangeMode | None = None,
    scalar_map: ScalarMapJSON | ScalarMap | None = None,
) -> None:
    if isinstance(data_type, str):
        zdtype = get_data_type_from_json(data_type, zarr_format=3)
    else:
        zdtype = data_type
    if zdtype.to_json(zarr_format=3) not in PERMITTED_DATA_TYPE_NAMES:
        raise ValueError(
            f"Invalid target data type {data_type!r}. "
            f"cast_value codec only supports integer and floating-point data types. "
            f"Got {zdtype}."
        )
    object.__setattr__(self, "dtype", zdtype)
    object.__setattr__(self, "rounding", rounding)
    object.__setattr__(self, "out_of_range", out_of_range)
    if scalar_map is not None:
        parsed = parse_scalar_map(scalar_map)
    else:
        parsed = None
    object.__setattr__(self, "scalar_map", parsed)

compute_encoded_size

compute_encoded_size(
    input_byte_length: int, chunk_spec: ArraySpec
) -> int

Given an input byte length, this method returns the output byte length. Raises a NotImplementedError for codecs with variable-sized outputs (e.g. compressors).

Parameters:

• input_byte_length (int) –
• chunk_spec (ArraySpec) –

Returns:

• int –

Source code in zarr/codecs/cast_value.py

def compute_encoded_size(self, input_byte_length: int, chunk_spec: ArraySpec) -> int:
    dtype_name = chunk_spec.dtype.to_json(zarr_format=3)
    if dtype_name not in PERMITTED_DATA_TYPE_NAMES:
        raise ValueError(
            "cast_value codec only supports fixed-size integer and floating-point data types. "
            f"Got source dtype: {chunk_spec.dtype}."
        )
    source_itemsize = chunk_spec.dtype.to_native_dtype().itemsize
    target_itemsize = self.dtype.to_native_dtype().itemsize
    if source_itemsize == 0 or target_itemsize == 0:
        raise ValueError(
            "cast_value codec requires fixed-size data types. "
            f"Got source itemsize={source_itemsize}, target itemsize={target_itemsize}."
        )
    num_elements = input_byte_length // source_itemsize
    return num_elements * target_itemsize

decode async

decode(
    chunks_and_specs: Iterable[tuple[CO | None, ArraySpec]],
) -> Iterable[CI | None]

Decodes a batch of chunks. Chunks can be None in which case they are ignored by the codec.

Parameters:

• chunks_and_specs (Iterable[tuple[CodecOutput | None, ArraySpec]]) –

  Ordered set of encoded chunks with their accompanying chunk spec.

Returns:

• Iterable[CI | None] –

Source code in zarr/abc/codec.py

async def decode(
    self,
    chunks_and_specs: Iterable[tuple[CO | None, ArraySpec]],
) -> Iterable[CI | None]:
    """Decodes a batch of chunks.
    Chunks can be None in which case they are ignored by the codec.

    Parameters
    ----------
    chunks_and_specs : Iterable[tuple[CodecOutput | None, ArraySpec]]
        Ordered set of encoded chunks with their accompanying chunk spec.

    Returns
    -------
    Iterable[CI | None]
    """
    return await _batching_helper(self._decode_single, chunks_and_specs)

encode async

encode(
    chunks_and_specs: Iterable[tuple[CI | None, ArraySpec]],
) -> Iterable[CO | None]

Encodes a batch of chunks. Chunks can be None in which case they are ignored by the codec.

Parameters:

• chunks_and_specs (Iterable[tuple[CI | None, ArraySpec]]) –

  Ordered set of to-be-encoded chunks with their accompanying chunk spec.

Returns:

• Iterable[CodecOutput | None] –

Source code in zarr/abc/codec.py

async def encode(
    self,
    chunks_and_specs: Iterable[tuple[CI | None, ArraySpec]],
) -> Iterable[CO | None]:
    """Encodes a batch of chunks.
    Chunks can be None in which case they are ignored by the codec.

    Parameters
    ----------
    chunks_and_specs : Iterable[tuple[CI | None, ArraySpec]]
        Ordered set of to-be-encoded chunks with their accompanying chunk spec.

    Returns
    -------
    Iterable[CodecOutput | None]
    """
    return await _batching_helper(self._encode_single, chunks_and_specs)

evolve_from_array_spec

evolve_from_array_spec(array_spec: ArraySpec) -> Self

Fills in codec configuration parameters that can be automatically inferred from the array metadata.

Parameters:

• array_spec (ArraySpec) –

Returns:

• Self –

Source code in zarr/abc/codec.py

def evolve_from_array_spec(self, array_spec: ArraySpec) -> Self:
    """Fills in codec configuration parameters that can be automatically
    inferred from the array metadata.

    Parameters
    ----------
    array_spec : ArraySpec

    Returns
    -------
    Self
    """
    return self

from_dict classmethod

from_dict(data: dict[str, JSON]) -> Self

Create an instance of the model from a dictionary

Source code in zarr/codecs/cast_value.py

@classmethod
def from_dict(cls, data: dict[str, JSON]) -> Self:
    _, configuration_parsed = parse_named_configuration(
        data, "cast_value", require_configuration=True
    )
    return cls(**configuration_parsed)  # type: ignore[arg-type]

resolve_metadata

resolve_metadata(chunk_spec: ArraySpec) -> ArraySpec

Update the fill value of the output spec by applying casting procedure.

Source code in zarr/codecs/cast_value.py

def resolve_metadata(self, chunk_spec: ArraySpec) -> ArraySpec:
    """
    Update the fill value of the output spec by applying casting procedure.
    """
    target_zdtype = self.dtype
    target_native = target_zdtype.to_native_dtype()
    source_native = chunk_spec.dtype.to_native_dtype()

    fill = chunk_spec.fill_value
    fill_arr = np.array([fill], dtype=source_native)

    new_fill_arr = self._do_cast(
        fill_arr, target_dtype=target_native, scalar_map=self._get_scalar_map("encode")
    )
    new_fill = target_native.type(new_fill_arr[0])

    return replace(chunk_spec, dtype=target_zdtype, fill_value=new_fill)

to_dict

to_dict() -> dict[str, JSON]

Recursively serialize this model to a dictionary. This method inspects the fields of self and calls x.to_dict() for any fields that are instances of Metadata. Sequences of Metadata are similarly recursed into, and the output of that recursion is collected in a list.

Source code in zarr/codecs/cast_value.py

def to_dict(self) -> dict[str, JSON]:
    config: dict[str, JSON] = {"data_type": cast("JSON", self.dtype.to_json(zarr_format=3))}
    if self.rounding != "nearest-even":
        config["rounding"] = self.rounding
    if self.out_of_range is not None:
        config["out_of_range"] = self.out_of_range
    if self.scalar_map is not None:
        json_map: dict[str, list[tuple[object, object]]] = {}
        for direction in ("encode", "decode"):
            if direction in self.scalar_map:
                json_map[direction] = [(k, v) for k, v in self.scalar_map[direction].items()]
        config["scalar_map"] = cast("JSON", json_map)
    return {"name": "cast_value", "configuration": config}

validate

validate(
    *,
    shape: tuple[int, ...],
    dtype: ZDType[TBaseDType, TBaseScalar],
    chunk_grid: ChunkGridMetadata,
) -> None

Validates that the codec configuration is compatible with the array metadata. Raises errors when the codec configuration is not compatible.

Parameters:

• shape (tuple[int, ...]) –

  The array shape

• dtype (dtype[Any]) –

  The array data type

• chunk_grid (ChunkGridMetadata) –

  The array chunk grid metadata

Source code in zarr/codecs/cast_value.py

def validate(
    self,
    *,
    shape: tuple[int, ...],
    dtype: ZDType[TBaseDType, TBaseScalar],
    chunk_grid: ChunkGridMetadata,
) -> None:
    # `dtype` is the source (the array's dtype); `self.dtype` is the
    # cast target. The spec requires both to be permitted, and rules
    # like `out_of_range="wrap"` apply to the target.
    source_name = dtype.to_json(zarr_format=3)
    target_name = self.dtype.to_json(zarr_format=3)
    for role, name in (("source", source_name), ("target", target_name)):
        if name not in PERMITTED_DATA_TYPE_NAMES:
            raise ValueError(
                f"The cast_value codec only supports integer and floating-point data types. "
                f"Got {role} dtype {name}."
            )
    if self.out_of_range == "wrap" and target_name not in CAST_VALUE_INT_DTYPES:
        raise ValueError(
            f"out_of_range='wrap' is only valid for integer target types. "
            f"Got target dtype {target_name}."
        )

    if self.scalar_map is not None:
        self._validate_scalar_map(dtype, self.dtype)

Crc32cCodec dataclass

Bases: BytesBytesCodec

crc32c codec

Source code in zarr/codecs/crc32c_.py

@dataclass(frozen=True)
class Crc32cCodec(BytesBytesCodec):
    """crc32c codec"""

    is_fixed_size = True

    @classmethod
    def from_dict(cls, data: dict[str, JSON]) -> Self:
        parse_named_configuration(data, "crc32c", require_configuration=False)
        return cls()

    def to_dict(self) -> dict[str, JSON]:
        return {"name": "crc32c"}

    def _decode_sync(
        self,
        chunk_bytes: Buffer,
        chunk_spec: ArraySpec,
    ) -> Buffer:
        data = chunk_bytes.as_numpy_array()
        crc32_bytes = data[-4:]
        inner_bytes = data[:-4]

        # Need to do a manual cast until https://github.com/numpy/numpy/issues/26783 is resolved
        computed_checksum = np.uint32(google_crc32c.value(cast(ABCBuffer, inner_bytes))).tobytes()
        stored_checksum = bytes(crc32_bytes)
        if computed_checksum != stored_checksum:
            raise ValueError(
                f"Stored and computed checksum do not match. Stored: {stored_checksum!r}. Computed: {computed_checksum!r}."
            )
        return chunk_spec.prototype.buffer.from_array_like(inner_bytes)

    async def _decode_single(
        self,
        chunk_bytes: Buffer,
        chunk_spec: ArraySpec,
    ) -> Buffer:
        return self._decode_sync(chunk_bytes, chunk_spec)

    def _encode_sync(
        self,
        chunk_bytes: Buffer,
        chunk_spec: ArraySpec,
    ) -> Buffer | None:
        data = chunk_bytes.as_numpy_array()
        # Calculate the checksum and "cast" it to a numpy array
        checksum = np.array([google_crc32c.value(cast(ABCBuffer, data))], dtype=np.uint32)
        # Append the checksum (as bytes) to the data
        return chunk_spec.prototype.buffer.from_array_like(np.append(data, checksum.view("B")))

    async def _encode_single(
        self,
        chunk_bytes: Buffer,
        chunk_spec: ArraySpec,
    ) -> Buffer | None:
        return self._encode_sync(chunk_bytes, chunk_spec)

    def compute_encoded_size(self, input_byte_length: int, _chunk_spec: ArraySpec) -> int:
        return input_byte_length + 4

is_fixed_size class-attribute instance-attribute

is_fixed_size = True

__init__

__init__() -> None

compute_encoded_size

compute_encoded_size(
    input_byte_length: int, _chunk_spec: ArraySpec
) -> int

Given an input byte length, this method returns the output byte length. Raises a NotImplementedError for codecs with variable-sized outputs (e.g. compressors).

Parameters:

• input_byte_length (int) –
• chunk_spec (ArraySpec) –

Returns:

• int –

Source code in zarr/codecs/crc32c_.py

def compute_encoded_size(self, input_byte_length: int, _chunk_spec: ArraySpec) -> int:
    return input_byte_length + 4

decode async

decode(
    chunks_and_specs: Iterable[tuple[CO | None, ArraySpec]],
) -> Iterable[CI | None]

Decodes a batch of chunks. Chunks can be None in which case they are ignored by the codec.

Parameters:

• chunks_and_specs (Iterable[tuple[CodecOutput | None, ArraySpec]]) –

  Ordered set of encoded chunks with their accompanying chunk spec.

Returns:

• Iterable[CI | None] –

Source code in zarr/abc/codec.py

async def decode(
    self,
    chunks_and_specs: Iterable[tuple[CO | None, ArraySpec]],
) -> Iterable[CI | None]:
    """Decodes a batch of chunks.
    Chunks can be None in which case they are ignored by the codec.

    Parameters
    ----------
    chunks_and_specs : Iterable[tuple[CodecOutput | None, ArraySpec]]
        Ordered set of encoded chunks with their accompanying chunk spec.

    Returns
    -------
    Iterable[CI | None]
    """
    return await _batching_helper(self._decode_single, chunks_and_specs)

encode async

encode(
    chunks_and_specs: Iterable[tuple[CI | None, ArraySpec]],
) -> Iterable[CO | None]

Encodes a batch of chunks. Chunks can be None in which case they are ignored by the codec.

Parameters:

• chunks_and_specs (Iterable[tuple[CI | None, ArraySpec]]) –

  Ordered set of to-be-encoded chunks with their accompanying chunk spec.

Returns:

• Iterable[CodecOutput | None] –

Source code in zarr/abc/codec.py

async def encode(
    self,
    chunks_and_specs: Iterable[tuple[CI | None, ArraySpec]],
) -> Iterable[CO | None]:
    """Encodes a batch of chunks.
    Chunks can be None in which case they are ignored by the codec.

    Parameters
    ----------
    chunks_and_specs : Iterable[tuple[CI | None, ArraySpec]]
        Ordered set of to-be-encoded chunks with their accompanying chunk spec.

    Returns
    -------
    Iterable[CodecOutput | None]
    """
    return await _batching_helper(self._encode_single, chunks_and_specs)

evolve_from_array_spec

evolve_from_array_spec(array_spec: ArraySpec) -> Self

Fills in codec configuration parameters that can be automatically inferred from the array metadata.

Parameters:

• array_spec (ArraySpec) –

Returns:

• Self –

Source code in zarr/abc/codec.py

def evolve_from_array_spec(self, array_spec: ArraySpec) -> Self:
    """Fills in codec configuration parameters that can be automatically
    inferred from the array metadata.

    Parameters
    ----------
    array_spec : ArraySpec

    Returns
    -------
    Self
    """
    return self

from_dict classmethod

from_dict(data: dict[str, JSON]) -> Self

Create an instance of the model from a dictionary

Source code in zarr/codecs/crc32c_.py

@classmethod
def from_dict(cls, data: dict[str, JSON]) -> Self:
    parse_named_configuration(data, "crc32c", require_configuration=False)
    return cls()

resolve_metadata

resolve_metadata(chunk_spec: ArraySpec) -> ArraySpec

Computed the spec of the chunk after it has been encoded by the codec. This is important for codecs that change the shape, data type or fill value of a chunk. The spec will then be used for subsequent codecs in the pipeline.

Parameters:

• chunk_spec (ArraySpec) –

Returns:

• ArraySpec –

Source code in zarr/abc/codec.py

def resolve_metadata(self, chunk_spec: ArraySpec) -> ArraySpec:
    """Computed the spec of the chunk after it has been encoded by the codec.
    This is important for codecs that change the shape, data type or fill value of a chunk.
    The spec will then be used for subsequent codecs in the pipeline.

    Parameters
    ----------
    chunk_spec : ArraySpec

    Returns
    -------
    ArraySpec
    """
    return chunk_spec

to_dict

to_dict() -> dict[str, JSON]

Recursively serialize this model to a dictionary. This method inspects the fields of self and calls x.to_dict() for any fields that are instances of Metadata. Sequences of Metadata are similarly recursed into, and the output of that recursion is collected in a list.

Source code in zarr/codecs/crc32c_.py

def to_dict(self) -> dict[str, JSON]:
    return {"name": "crc32c"}

validate

validate(
    *,
    shape: tuple[int, ...],
    dtype: ZDType[TBaseDType, TBaseScalar],
    chunk_grid: ChunkGridMetadata,
) -> None

Validates that the codec configuration is compatible with the array metadata. Raises errors when the codec configuration is not compatible.

Parameters:

• shape (tuple[int, ...]) –

  The array shape

• dtype (dtype[Any]) –

  The array data type

• chunk_grid (ChunkGridMetadata) –

  The array chunk grid metadata

Source code in zarr/abc/codec.py

def validate(
    self,
    *,
    shape: tuple[int, ...],
    dtype: ZDType[TBaseDType, TBaseScalar],
    chunk_grid: ChunkGridMetadata,
) -> None:
    """Validates that the codec configuration is compatible with the array metadata.
    Raises errors when the codec configuration is not compatible.

    Parameters
    ----------
    shape : tuple[int, ...]
        The array shape
    dtype : np.dtype[Any]
        The array data type
    chunk_grid : ChunkGridMetadata
        The array chunk grid metadata
    """

Endian

Deprecated. Pass a literal string ("little" or "big") directly to BytesCodec instead.

Source code in zarr/codecs/bytes.py

class Endian(metaclass=_DeprecatedStrEnumMeta):
    """
    Deprecated. Pass a literal string (`"little"` or `"big"`) directly to
    `BytesCodec` instead.
    """

    _members: ClassVar[dict[str, str]] = {"little": "little", "big": "big"}

GzipCodec dataclass

Bases: BytesBytesCodec

gzip codec

Source code in zarr/codecs/gzip.py

@dataclass(frozen=True)
class GzipCodec(BytesBytesCodec):
    """gzip codec"""

    is_fixed_size = False

    level: int = 5

    def __init__(self, *, level: int = 5) -> None:
        level_parsed = parse_gzip_level(level)

        object.__setattr__(self, "level", level_parsed)

    @classmethod
    def from_dict(cls, data: dict[str, JSON]) -> Self:
        _, configuration_parsed = parse_named_configuration(data, "gzip")
        return cls(**configuration_parsed)  # type: ignore[arg-type]

    def to_dict(self) -> dict[str, JSON]:
        return {"name": "gzip", "configuration": {"level": self.level}}

    @cached_property
    def _gzip_codec(self) -> GZip:
        return GZip(self.level)

    def _decode_sync(
        self,
        chunk_bytes: Buffer,
        chunk_spec: ArraySpec,
    ) -> Buffer:
        return as_numpy_array_wrapper(self._gzip_codec.decode, chunk_bytes, chunk_spec.prototype)

    async def _decode_single(
        self,
        chunk_bytes: Buffer,
        chunk_spec: ArraySpec,
    ) -> Buffer:
        return await asyncio.to_thread(self._decode_sync, chunk_bytes, chunk_spec)

    def _encode_sync(
        self,
        chunk_bytes: Buffer,
        chunk_spec: ArraySpec,
    ) -> Buffer | None:
        return as_numpy_array_wrapper(self._gzip_codec.encode, chunk_bytes, chunk_spec.prototype)

    async def _encode_single(
        self,
        chunk_bytes: Buffer,
        chunk_spec: ArraySpec,
    ) -> Buffer | None:
        return await asyncio.to_thread(self._encode_sync, chunk_bytes, chunk_spec)

    def compute_encoded_size(
        self,
        _input_byte_length: int,
        _chunk_spec: ArraySpec,
    ) -> int:
        raise NotImplementedError

is_fixed_size class-attribute instance-attribute

is_fixed_size = False

level class-attribute instance-attribute

level: int = 5

__init__

__init__(*, level: int = 5) -> None

Source code in zarr/codecs/gzip.py

def __init__(self, *, level: int = 5) -> None:
    level_parsed = parse_gzip_level(level)

    object.__setattr__(self, "level", level_parsed)

compute_encoded_size

compute_encoded_size(
    _input_byte_length: int, _chunk_spec: ArraySpec
) -> int

Given an input byte length, this method returns the output byte length. Raises a NotImplementedError for codecs with variable-sized outputs (e.g. compressors).

Parameters:

• input_byte_length (int) –
• chunk_spec (ArraySpec) –

Returns:

• int –

Source code in zarr/codecs/gzip.py

def compute_encoded_size(
    self,
    _input_byte_length: int,
    _chunk_spec: ArraySpec,
) -> int:
    raise NotImplementedError

decode async

decode(
    chunks_and_specs: Iterable[tuple[CO | None, ArraySpec]],
) -> Iterable[CI | None]

Decodes a batch of chunks. Chunks can be None in which case they are ignored by the codec.

Parameters:

• chunks_and_specs (Iterable[tuple[CodecOutput | None, ArraySpec]]) –

  Ordered set of encoded chunks with their accompanying chunk spec.

Returns:

• Iterable[CI | None] –

Source code in zarr/abc/codec.py

async def decode(
    self,
    chunks_and_specs: Iterable[tuple[CO | None, ArraySpec]],
) -> Iterable[CI | None]:
    """Decodes a batch of chunks.
    Chunks can be None in which case they are ignored by the codec.

    Parameters
    ----------
    chunks_and_specs : Iterable[tuple[CodecOutput | None, ArraySpec]]
        Ordered set of encoded chunks with their accompanying chunk spec.

    Returns
    -------
    Iterable[CI | None]
    """
    return await _batching_helper(self._decode_single, chunks_and_specs)

encode async

encode(
    chunks_and_specs: Iterable[tuple[CI | None, ArraySpec]],
) -> Iterable[CO | None]

Encodes a batch of chunks. Chunks can be None in which case they are ignored by the codec.

Parameters:

• chunks_and_specs (Iterable[tuple[CI | None, ArraySpec]]) –

  Ordered set of to-be-encoded chunks with their accompanying chunk spec.

Returns:

• Iterable[CodecOutput | None] –

Source code in zarr/abc/codec.py

async def encode(
    self,
    chunks_and_specs: Iterable[tuple[CI | None, ArraySpec]],
) -> Iterable[CO | None]:
    """Encodes a batch of chunks.
    Chunks can be None in which case they are ignored by the codec.

    Parameters
    ----------
    chunks_and_specs : Iterable[tuple[CI | None, ArraySpec]]
        Ordered set of to-be-encoded chunks with their accompanying chunk spec.

    Returns
    -------
    Iterable[CodecOutput | None]
    """
    return await _batching_helper(self._encode_single, chunks_and_specs)

evolve_from_array_spec

evolve_from_array_spec(array_spec: ArraySpec) -> Self

Fills in codec configuration parameters that can be automatically inferred from the array metadata.

Parameters:

• array_spec (ArraySpec) –

Returns:

• Self –

Source code in zarr/abc/codec.py

def evolve_from_array_spec(self, array_spec: ArraySpec) -> Self:
    """Fills in codec configuration parameters that can be automatically
    inferred from the array metadata.

    Parameters
    ----------
    array_spec : ArraySpec

    Returns
    -------
    Self
    """
    return self

from_dict classmethod

from_dict(data: dict[str, JSON]) -> Self

Create an instance of the model from a dictionary

Source code in zarr/codecs/gzip.py

@classmethod
def from_dict(cls, data: dict[str, JSON]) -> Self:
    _, configuration_parsed = parse_named_configuration(data, "gzip")
    return cls(**configuration_parsed)  # type: ignore[arg-type]

resolve_metadata

resolve_metadata(chunk_spec: ArraySpec) -> ArraySpec

Computed the spec of the chunk after it has been encoded by the codec. This is important for codecs that change the shape, data type or fill value of a chunk. The spec will then be used for subsequent codecs in the pipeline.

Parameters:

• chunk_spec (ArraySpec) –

Returns:

• ArraySpec –

Source code in zarr/abc/codec.py

def resolve_metadata(self, chunk_spec: ArraySpec) -> ArraySpec:
    """Computed the spec of the chunk after it has been encoded by the codec.
    This is important for codecs that change the shape, data type or fill value of a chunk.
    The spec will then be used for subsequent codecs in the pipeline.

    Parameters
    ----------
    chunk_spec : ArraySpec

    Returns
    -------
    ArraySpec
    """
    return chunk_spec

to_dict

to_dict() -> dict[str, JSON]

Recursively serialize this model to a dictionary. This method inspects the fields of self and calls x.to_dict() for any fields that are instances of Metadata. Sequences of Metadata are similarly recursed into, and the output of that recursion is collected in a list.

Source code in zarr/codecs/gzip.py

def to_dict(self) -> dict[str, JSON]:
    return {"name": "gzip", "configuration": {"level": self.level}}

validate

validate(
    *,
    shape: tuple[int, ...],
    dtype: ZDType[TBaseDType, TBaseScalar],
    chunk_grid: ChunkGridMetadata,
) -> None

Validates that the codec configuration is compatible with the array metadata. Raises errors when the codec configuration is not compatible.

Parameters:

• shape (tuple[int, ...]) –

  The array shape

• dtype (dtype[Any]) –

  The array data type

• chunk_grid (ChunkGridMetadata) –

  The array chunk grid metadata

Source code in zarr/abc/codec.py

def validate(
    self,
    *,
    shape: tuple[int, ...],
    dtype: ZDType[TBaseDType, TBaseScalar],
    chunk_grid: ChunkGridMetadata,
) -> None:
    """Validates that the codec configuration is compatible with the array metadata.
    Raises errors when the codec configuration is not compatible.

    Parameters
    ----------
    shape : tuple[int, ...]
        The array shape
    dtype : np.dtype[Any]
        The array data type
    chunk_grid : ChunkGridMetadata
        The array chunk grid metadata
    """

ScaleOffset dataclass

Bases: ArrayArrayCodec

Scale-offset array-to-array codec.

Encodes values with out = (in - offset) * scale and decodes with out = (in / scale) + offset, using the input array's data type semantics. Intermediate or final values that are not representable in that dtype are reported as errors (integer overflow, unsigned underflow, non-exact integer division).

Parameters:

• offset (int, float, or str, default: 0 ) –

  Value subtracted during encoding. Strings preserve the exact JSON representation when round-tripping metadata. Default is 0.

• scale (int, float, or str, default: 1 ) –

  Value multiplied during encoding (after offset subtraction). Strings preserve the exact JSON representation when round-tripping metadata. Default is 1.

Attributes:

• offset (int, float, or str) –

  The offset value, as supplied to the constructor.

• scale (int, float, or str) –

  The scale value, as supplied to the constructor.

References

• The scale_offset codec spec: github.com/zarr-developers/zarr-extensions/tree/main/codecs/scale_offset

Source code in zarr/codecs/scale_offset.py

@dataclass(frozen=True)
class ScaleOffset(ArrayArrayCodec):
    """Scale-offset array-to-array codec.

    Encodes values with `out = (in - offset) * scale` and decodes with
    `out = (in / scale) + offset`, using the input array's data type semantics.
    Intermediate or final values that are not representable in that dtype are reported
    as errors (integer overflow, unsigned underflow, non-exact integer division).

    Parameters
    ----------
    offset : int, float, or str
        Value subtracted during encoding. Strings preserve the exact JSON
        representation when round-tripping metadata. Default is 0.
    scale : int, float, or str
        Value multiplied during encoding (after offset subtraction). Strings
        preserve the exact JSON representation when round-tripping metadata.
        Default is 1.

    Attributes
    ----------
    offset : int, float, or str
        The offset value, as supplied to the constructor.
    scale : int, float, or str
        The scale value, as supplied to the constructor.

    References
    ----------

    - The `scale_offset` codec spec: https://github.com/zarr-developers/zarr-extensions/tree/main/codecs/scale_offset
    """

    is_fixed_size = True

    offset: int | float | str
    scale: int | float | str

    def __init__(self, *, offset: object = 0, scale: object = 1) -> None:
        if not isinstance(offset, int | float | str):
            raise TypeError(f"offset must be a number or string, got {type(offset).__name__}")
        if not isinstance(scale, int | float | str):
            raise TypeError(f"scale must be a number or string, got {type(scale).__name__}")
        object.__setattr__(self, "offset", offset)
        object.__setattr__(self, "scale", scale)

    @classmethod
    def from_dict(cls, data: dict[str, JSON]) -> Self:
        _, configuration_parsed = parse_named_configuration(
            data, "scale_offset", require_configuration=False
        )
        configuration_parsed = configuration_parsed or {}
        return cls(**configuration_parsed)

    def to_dict(self) -> dict[str, JSON]:
        if self.offset == 0 and self.scale == 1:
            return {"name": "scale_offset"}
        config: dict[str, JSON] = {}
        if self.offset != 0:
            config["offset"] = self.offset
        if self.scale != 1:
            config["scale"] = self.scale
        return {"name": "scale_offset", "configuration": config}

    def validate(
        self,
        *,
        shape: tuple[int, ...],
        dtype: ZDType[TBaseDType, TBaseScalar],
        chunk_grid: ChunkGridMetadata,
    ) -> None:
        native = dtype.to_native_dtype()
        if not np.issubdtype(native, np.integer) and not np.issubdtype(native, np.floating):
            raise ValueError(
                f"scale_offset codec only supports integer and floating-point data types. "
                f"Got {dtype}."
            )
        if self.scale == 0:
            raise ValueError("scale_offset scale must be non-zero.")
        for name, value in [("offset", self.offset), ("scale", self.scale)]:
            try:
                dtype.from_json_scalar(value, zarr_format=3)
            except (TypeError, ValueError, OverflowError) as e:
                raise ValueError(
                    f"scale_offset {name} value {value!r} is not representable in dtype {native}."
                ) from e

    def resolve_metadata(self, chunk_spec: ArraySpec) -> ArraySpec:
        zdtype = chunk_spec.dtype
        fill = np.asarray(zdtype.cast_scalar(chunk_spec.fill_value))
        offset = cast("np.generic", zdtype.from_json_scalar(self.offset, zarr_format=3))
        scale = cast("np.generic", zdtype.from_json_scalar(self.scale, zarr_format=3))
        new_fill = _encode(fill, offset, scale)
        return replace(chunk_spec, fill_value=new_fill.reshape(()).item())

    def _decode_sync(
        self,
        chunk_array: NDBuffer,
        chunk_spec: ArraySpec,
    ) -> NDBuffer:
        arr = cast("np.ndarray[tuple[Any, ...], np.dtype[Any]]", chunk_array.as_ndarray_like())
        zdtype = chunk_spec.dtype
        offset = cast("np.generic", zdtype.from_json_scalar(self.offset, zarr_format=3))
        scale = cast("np.generic", zdtype.from_json_scalar(self.scale, zarr_format=3))
        result = _decode(arr, offset, scale, scale_repr=self.scale)
        return chunk_spec.prototype.nd_buffer.from_ndarray_like(result)

    async def _decode_single(
        self,
        chunk_array: NDBuffer,
        chunk_spec: ArraySpec,
    ) -> NDBuffer:
        return self._decode_sync(chunk_array, chunk_spec)

    def _encode_sync(
        self,
        chunk_array: NDBuffer,
        chunk_spec: ArraySpec,
    ) -> NDBuffer | None:
        arr = cast("np.ndarray[tuple[Any, ...], np.dtype[Any]]", chunk_array.as_ndarray_like())
        zdtype = chunk_spec.dtype
        offset = cast("np.generic", zdtype.from_json_scalar(self.offset, zarr_format=3))
        scale = cast("np.generic", zdtype.from_json_scalar(self.scale, zarr_format=3))
        result = _encode(arr, offset, scale)
        return chunk_spec.prototype.nd_buffer.from_ndarray_like(result)

    async def _encode_single(
        self,
        chunk_array: NDBuffer,
        _chunk_spec: ArraySpec,
    ) -> NDBuffer | None:
        return self._encode_sync(chunk_array, _chunk_spec)

    def compute_encoded_size(self, input_byte_length: int, _chunk_spec: ArraySpec) -> int:
        return input_byte_length

is_fixed_size class-attribute instance-attribute

is_fixed_size = True

offset instance-attribute

offset: int | float | str

scale instance-attribute

scale: int | float | str

__init__

__init__(*, offset: object = 0, scale: object = 1) -> None

Source code in zarr/codecs/scale_offset.py

def __init__(self, *, offset: object = 0, scale: object = 1) -> None:
    if not isinstance(offset, int | float | str):
        raise TypeError(f"offset must be a number or string, got {type(offset).__name__}")
    if not isinstance(scale, int | float | str):
        raise TypeError(f"scale must be a number or string, got {type(scale).__name__}")
    object.__setattr__(self, "offset", offset)
    object.__setattr__(self, "scale", scale)

compute_encoded_size

compute_encoded_size(
    input_byte_length: int, _chunk_spec: ArraySpec
) -> int

Given an input byte length, this method returns the output byte length. Raises a NotImplementedError for codecs with variable-sized outputs (e.g. compressors).

Parameters:

• input_byte_length (int) –
• chunk_spec (ArraySpec) –

Returns:

• int –

Source code in zarr/codecs/scale_offset.py

def compute_encoded_size(self, input_byte_length: int, _chunk_spec: ArraySpec) -> int:
    return input_byte_length

decode async

decode(
    chunks_and_specs: Iterable[tuple[CO | None, ArraySpec]],
) -> Iterable[CI | None]

Decodes a batch of chunks. Chunks can be None in which case they are ignored by the codec.

Parameters:

• chunks_and_specs (Iterable[tuple[CodecOutput | None, ArraySpec]]) –

  Ordered set of encoded chunks with their accompanying chunk spec.

Returns:

• Iterable[CI | None] –

Source code in zarr/abc/codec.py

async def decode(
    self,
    chunks_and_specs: Iterable[tuple[CO | None, ArraySpec]],
) -> Iterable[CI | None]:
    """Decodes a batch of chunks.
    Chunks can be None in which case they are ignored by the codec.

    Parameters
    ----------
    chunks_and_specs : Iterable[tuple[CodecOutput | None, ArraySpec]]
        Ordered set of encoded chunks with their accompanying chunk spec.

    Returns
    -------
    Iterable[CI | None]
    """
    return await _batching_helper(self._decode_single, chunks_and_specs)

encode async

encode(
    chunks_and_specs: Iterable[tuple[CI | None, ArraySpec]],
) -> Iterable[CO | None]

Encodes a batch of chunks. Chunks can be None in which case they are ignored by the codec.

Parameters:

• chunks_and_specs (Iterable[tuple[CI | None, ArraySpec]]) –

  Ordered set of to-be-encoded chunks with their accompanying chunk spec.

Returns:

• Iterable[CodecOutput | None] –

Source code in zarr/abc/codec.py

async def encode(
    self,
    chunks_and_specs: Iterable[tuple[CI | None, ArraySpec]],
) -> Iterable[CO | None]:
    """Encodes a batch of chunks.
    Chunks can be None in which case they are ignored by the codec.

    Parameters
    ----------
    chunks_and_specs : Iterable[tuple[CI | None, ArraySpec]]
        Ordered set of to-be-encoded chunks with their accompanying chunk spec.

    Returns
    -------
    Iterable[CodecOutput | None]
    """
    return await _batching_helper(self._encode_single, chunks_and_specs)

evolve_from_array_spec

evolve_from_array_spec(array_spec: ArraySpec) -> Self

Fills in codec configuration parameters that can be automatically inferred from the array metadata.

Parameters:

• array_spec (ArraySpec) –

Returns:

• Self –

Source code in zarr/abc/codec.py

def evolve_from_array_spec(self, array_spec: ArraySpec) -> Self:
    """Fills in codec configuration parameters that can be automatically
    inferred from the array metadata.

    Parameters
    ----------
    array_spec : ArraySpec

    Returns
    -------
    Self
    """
    return self

from_dict classmethod

from_dict(data: dict[str, JSON]) -> Self

Create an instance of the model from a dictionary

Source code in zarr/codecs/scale_offset.py

@classmethod
def from_dict(cls, data: dict[str, JSON]) -> Self:
    _, configuration_parsed = parse_named_configuration(
        data, "scale_offset", require_configuration=False
    )
    configuration_parsed = configuration_parsed or {}
    return cls(**configuration_parsed)

resolve_metadata

resolve_metadata(chunk_spec: ArraySpec) -> ArraySpec

Computed the spec of the chunk after it has been encoded by the codec. This is important for codecs that change the shape, data type or fill value of a chunk. The spec will then be used for subsequent codecs in the pipeline.

Parameters:

• chunk_spec (ArraySpec) –

Returns:

• ArraySpec –

Source code in zarr/codecs/scale_offset.py

def resolve_metadata(self, chunk_spec: ArraySpec) -> ArraySpec:
    zdtype = chunk_spec.dtype
    fill = np.asarray(zdtype.cast_scalar(chunk_spec.fill_value))
    offset = cast("np.generic", zdtype.from_json_scalar(self.offset, zarr_format=3))
    scale = cast("np.generic", zdtype.from_json_scalar(self.scale, zarr_format=3))
    new_fill = _encode(fill, offset, scale)
    return replace(chunk_spec, fill_value=new_fill.reshape(()).item())

to_dict

to_dict() -> dict[str, JSON]

Recursively serialize this model to a dictionary. This method inspects the fields of self and calls x.to_dict() for any fields that are instances of Metadata. Sequences of Metadata are similarly recursed into, and the output of that recursion is collected in a list.

Source code in zarr/codecs/scale_offset.py

def to_dict(self) -> dict[str, JSON]:
    if self.offset == 0 and self.scale == 1:
        return {"name": "scale_offset"}
    config: dict[str, JSON] = {}
    if self.offset != 0:
        config["offset"] = self.offset
    if self.scale != 1:
        config["scale"] = self.scale
    return {"name": "scale_offset", "configuration": config}

validate

validate(
    *,
    shape: tuple[int, ...],
    dtype: ZDType[TBaseDType, TBaseScalar],
    chunk_grid: ChunkGridMetadata,
) -> None

Validates that the codec configuration is compatible with the array metadata. Raises errors when the codec configuration is not compatible.

Parameters:

• shape (tuple[int, ...]) –

  The array shape

• dtype (dtype[Any]) –

  The array data type

• chunk_grid (ChunkGridMetadata) –

  The array chunk grid metadata

Source code in zarr/codecs/scale_offset.py

def validate(
    self,
    *,
    shape: tuple[int, ...],
    dtype: ZDType[TBaseDType, TBaseScalar],
    chunk_grid: ChunkGridMetadata,
) -> None:
    native = dtype.to_native_dtype()
    if not np.issubdtype(native, np.integer) and not np.issubdtype(native, np.floating):
        raise ValueError(
            f"scale_offset codec only supports integer and floating-point data types. "
            f"Got {dtype}."
        )
    if self.scale == 0:
        raise ValueError("scale_offset scale must be non-zero.")
    for name, value in [("offset", self.offset), ("scale", self.scale)]:
        try:
            dtype.from_json_scalar(value, zarr_format=3)
        except (TypeError, ValueError, OverflowError) as e:
            raise ValueError(
                f"scale_offset {name} value {value!r} is not representable in dtype {native}."
            ) from e

ShardingCodec dataclass

Bases: ArrayBytesCodec, ArrayBytesCodecPartialDecodeMixin, ArrayBytesCodecPartialEncodeMixin

Sharding codec.

subchunk_write_order controls the physical order of subchunks within a shard. It is a write-time setting only: it is not stored in array metadata, so reopening a sharded array does not recover it (the setting reverts to the morton default per codec instance).

Source code in zarr/codecs/sharding.py

@dataclass(frozen=True)
class ShardingCodec(
    ArrayBytesCodec, ArrayBytesCodecPartialDecodeMixin, ArrayBytesCodecPartialEncodeMixin
):
    """Sharding codec.

    `subchunk_write_order` controls the physical order of subchunks within a shard. It is a
    write-time setting only: it is not stored in array metadata, so reopening a sharded array
    does not recover it (the setting reverts to the `morton` default per codec instance).
    """

    is_fixed_size = False

    chunk_shape: tuple[int, ...]
    codecs: tuple[Codec, ...]
    index_codecs: tuple[Codec, ...]
    index_location: IndexLocation = "end"
    subchunk_write_order: SubchunkWriteOrder = "morton"

    def __init__(
        self,
        *,
        chunk_shape: ShapeLike,
        codecs: Iterable[Codec | dict[str, JSON]] = (BytesCodec(),),
        index_codecs: Iterable[Codec | dict[str, JSON]] = (BytesCodec(), Crc32cCodec()),
        index_location: ShardingCodecIndexLocation | IndexLocation = "end",
        subchunk_write_order: SubchunkWriteOrder = "morton",
    ) -> None:
        chunk_shape_parsed = parse_shapelike(chunk_shape)
        codecs_parsed = parse_codecs(codecs)
        index_codecs_parsed = parse_codecs(index_codecs)
        index_location_coerced = _coerce_enum_input(
            index_location, "index_location", "ShardingCodec"
        )
        index_location_parsed = _parse_index_location(index_location_coerced)
        if subchunk_write_order not in SUBCHUNK_WRITE_ORDER:
            raise ValueError(
                f"Unrecognized subchunk write order: {subchunk_write_order}. Only {SUBCHUNK_WRITE_ORDER} are allowed."
            )

        object.__setattr__(self, "chunk_shape", chunk_shape_parsed)
        object.__setattr__(self, "codecs", codecs_parsed)
        object.__setattr__(self, "index_codecs", index_codecs_parsed)
        object.__setattr__(self, "index_location", index_location_parsed)
        object.__setattr__(self, "subchunk_write_order", subchunk_write_order)

        # Use instance-local lru_cache to avoid memory leaks

        # numpy void scalars are not hashable, which means an array spec with a fill value that is
        # a numpy void scalar will break the lru_cache. This is commented for now but should be
        # fixed. See https://github.com/zarr-developers/zarr-python/issues/3054
        # object.__setattr__(self, "_get_chunk_spec", lru_cache()(self._get_chunk_spec))
        object.__setattr__(self, "_get_index_chunk_spec", lru_cache()(self._get_index_chunk_spec))
        object.__setattr__(self, "_get_chunks_per_shard", lru_cache()(self._get_chunks_per_shard))
        object.__setattr__(self, "_shard_index_size", lru_cache()(self._shard_index_size))
        object.__setattr__(
            self, "_get_inner_chunk_transform", lru_cache()(self._get_inner_chunk_transform)
        )
        object.__setattr__(
            self, "_get_index_chunk_transform", lru_cache()(self._get_index_chunk_transform)
        )

    # todo: typedict return type
    def __getstate__(self) -> dict[str, Any]:
        # `subchunk_write_order` is not part of codec metadata (`to_dict`), so carry it
        # explicitly to survive a pickle round-trip (otherwise it reverts to `morton`).
        return {"subchunk_write_order": self.subchunk_write_order, **self.to_dict()}

    def __setstate__(self, state: dict[str, Any]) -> None:
        config = state["configuration"]
        object.__setattr__(self, "chunk_shape", parse_shapelike(config["chunk_shape"]))
        object.__setattr__(self, "codecs", parse_codecs(config["codecs"]))
        object.__setattr__(self, "index_codecs", parse_codecs(config["index_codecs"]))
        object.__setattr__(self, "index_location", _parse_index_location(config["index_location"]))
        object.__setattr__(self, "subchunk_write_order", state["subchunk_write_order"])

        # Use instance-local lru_cache to avoid memory leaks
        # object.__setattr__(self, "_get_chunk_spec", lru_cache()(self._get_chunk_spec))
        object.__setattr__(self, "_get_index_chunk_spec", lru_cache()(self._get_index_chunk_spec))
        object.__setattr__(self, "_get_chunks_per_shard", lru_cache()(self._get_chunks_per_shard))
        object.__setattr__(self, "_shard_index_size", lru_cache()(self._shard_index_size))
        object.__setattr__(
            self, "_get_inner_chunk_transform", lru_cache()(self._get_inner_chunk_transform)
        )
        object.__setattr__(
            self, "_get_index_chunk_transform", lru_cache()(self._get_index_chunk_transform)
        )

    @classmethod
    def from_dict(cls, data: dict[str, JSON]) -> Self:
        _, configuration_parsed = parse_named_configuration(data, "sharding_indexed")
        return cls(**configuration_parsed)  # type: ignore[arg-type]

    @property
    def codec_pipeline(self) -> CodecPipeline:
        return get_pipeline_class().from_codecs(self.codecs)

    def _get_inner_pipeline(self, shard_spec: ArraySpec) -> CodecPipeline:
        """The nested pipeline for inner-chunk IO, evolved against the inner
        chunk spec.

        Evolving matters for two reasons: it threads the spec through the inner
        codec chain (spec-changing codecs see the spec they will actually
        operate on), and — for a synchronous pipeline — it builds the sync
        transform, so inner-chunk IO over the (sync-capable) sharding byte
        getters takes the pipeline's sync fast path instead of scheduling one
        coroutine per inner chunk. The bare `codec_pipeline` property returns an
        unevolved pipeline, which a synchronous pipeline can only run through
        its async fallback.

        Memoized per (pipeline class, batch size, shard_spec): evolving builds
        a ChunkTransform, which is wasteful to redo on every shard operation.
        The pipeline class and batch size participate in the key so the
        `codec_pipeline.path` and `codec_pipeline.batch_size` configs are still
        honored after the first use (`from_codecs` captures batch_size at
        construction). A benign construction race between threads is possible
        (last writer wins) — same as the other caches here.
        """
        cache: dict[tuple[type[CodecPipeline], int, ArraySpec], CodecPipeline] | None = getattr(
            self, "_inner_pipeline_cache", None
        )
        if cache is None:
            cache = {}
            object.__setattr__(self, "_inner_pipeline_cache", cache)
        key = (get_pipeline_class(), zarr_config.get("codec_pipeline.batch_size"), shard_spec)
        pipeline = cache.get(key)
        if pipeline is None:
            chunk_spec = self._get_chunk_spec(shard_spec)
            pipeline = self.codec_pipeline.evolve_from_array_spec(chunk_spec)
            cache[key] = pipeline
        return pipeline

    def to_dict(self) -> dict[str, JSON]:
        return {
            "name": "sharding_indexed",
            "configuration": {
                "chunk_shape": self.chunk_shape,
                "codecs": tuple(s.to_dict() for s in self.codecs),
                "index_codecs": tuple(s.to_dict() for s in self.index_codecs),
                "index_location": self.index_location,
            },
        }

    def evolve_from_array_spec(self, array_spec: ArraySpec) -> Self:
        """Thread the spec through the inner chain.

        Each codec is evolved against the spec produced by the previous one.
        Evolving every codec against the same unthreaded spec is the bug shape that
        strips `BytesCodec.endian` behind a dtype-changing codec — and this
        method runs on the real array-creation path, baking the damaged chain
        into the evolved instance before the transform builders ever run.

        Parameters
        ----------
        array_spec
            The base spec to be evolved as we thread it.

        Returns
        -------
            This codec with the evolved code chain.
        """
        from zarr.core.chunk_utils import evolve_codecs

        shard_spec = self._get_chunk_spec(array_spec)
        evolved_codecs = evolve_codecs(self.codecs, shard_spec)
        if evolved_codecs != self.codecs:
            return replace(self, codecs=evolved_codecs)
        return self

    def validate(
        self,
        *,
        shape: tuple[int, ...],
        dtype: ZDType[TBaseDType, TBaseScalar],
        chunk_grid: ChunkGridMetadata,
    ) -> None:
        if len(self.chunk_shape) != len(shape):
            raise ValueError(
                "The shard's `chunk_shape` and array's `shape` need to have the same number of dimensions."
            )
        if isinstance(chunk_grid, RegularChunkGridMetadata):
            edges_per_dim: tuple[tuple[int, ...], ...] = tuple((s,) for s in chunk_grid.chunk_shape)
        elif isinstance(chunk_grid, RectilinearChunkGridMetadata):
            edges_per_dim = tuple(
                (s,) if isinstance(s, int) else s for s in chunk_grid.chunk_shapes
            )
        else:
            raise TypeError(
                f"Sharding is only compatible with regular and rectilinear chunk grids, "
                f"got {type(chunk_grid)}"
            )
        for i, (edges, inner) in enumerate(zip(edges_per_dim, self.chunk_shape, strict=False)):
            for edge in set(edges):
                if edge % inner != 0:
                    raise ValueError(
                        f"Chunk edge length {edge} in dimension {i} is not "
                        f"divisible by the shard's inner chunk size {inner}."
                    )

    def _get_inner_chunk_transform(self, shard_spec: ArraySpec) -> Any:
        """The synchronous transform for the inner codec chain.

        Memoized by the instance-local `lru_cache` wrapping installed in
        `__init__`/`__setstate__` (the single cache mechanism for these
        builders — do not add another layer inside the body).

        Codecs are evolved with the spec THREADED forward (`evolve_codecs`):
        each inner codec is evolved against the spec produced by the previous
        one, not the original chunk spec. Evolving every codec against the same
        unthreaded spec is the bug shape that stripped `BytesCodec.endian` at
        the pipeline level (see `evolve_codecs`) — the inner chain must use the
        same single source of truth.
        """

        chunk_spec = self._get_chunk_spec(shard_spec)
        return ChunkTransform(codecs=evolve_codecs(self.codecs, chunk_spec))

    def _get_index_chunk_transform(self, chunks_per_shard: tuple[int, ...]) -> Any:
        """The synchronous transform for the index codec chain.

        Memoized via instance-local `lru_cache`.
        """

        index_spec = self._get_index_chunk_spec(chunks_per_shard)
        return ChunkTransform(codecs=evolve_codecs(self.index_codecs, index_spec))

    def _decode_shard_index_sync(
        self, index_bytes: Buffer, chunks_per_shard: tuple[int, ...]
    ) -> _ShardIndex:
        """Decode shard index synchronously using ChunkTransform."""
        index_transform = self._get_index_chunk_transform(chunks_per_shard)
        index_spec = self._get_index_chunk_spec(chunks_per_shard)
        index_array = index_transform.decode_chunk(index_bytes, index_spec)
        return _ShardIndex(chunks_per_shard, index_array.as_numpy_array())

    def _encode_shard_index_sync(self, index: _ShardIndex) -> Buffer:
        """Encode shard index synchronously using ChunkTransform."""
        index_transform = self._get_index_chunk_transform(index.chunks_per_shard)
        index_spec = self._get_index_chunk_spec(index.chunks_per_shard)
        index_nd = get_ndbuffer_class().from_numpy_array(index.offsets_and_lengths)
        result: Buffer | None = index_transform.encode_chunk(index_nd, index_spec)
        assert result is not None
        return result

    def _shard_reader_from_bytes_sync(
        self, buf: Buffer, chunks_per_shard: tuple[int, ...]
    ) -> _ShardReader:
        """Sync version of _ShardReader.from_bytes."""
        shard_index_size = self._shard_index_size(chunks_per_shard)
        if self.index_location == "start":
            shard_index_bytes = buf[:shard_index_size]
        else:
            shard_index_bytes = buf[-shard_index_size:]
        index = self._decode_shard_index_sync(shard_index_bytes, chunks_per_shard)
        reader = _ShardReader()
        reader.buf = buf
        reader.index = index
        return reader

    def _decode_sync(
        self,
        shard_bytes: Buffer,
        shard_spec: ArraySpec,
    ) -> NDBuffer:
        """Decode a full shard synchronously.

        Sync counterpart to `_decode_single`. Same semantics (decode every
        inner chunk and assemble the full shard array) but routes through
        `ChunkTransform` instead of the async codec pipeline, so it can
        run on the sync codec-pipeline fast path without an event loop.

        For a partial read where the caller only needs a slice of the shard,
        use `_decode_partial_sync` instead — it fetches only the byte
        ranges that overlap the selection.

        This method does not parallelize decompression, but should.
        See TODO: make issue for handling subchunk parallelism
        """
        shard_shape = shard_spec.shape
        chunk_shape = self.chunk_shape
        chunks_per_shard = self._get_chunks_per_shard(shard_spec)
        chunk_spec = self._get_chunk_spec(shard_spec)
        inner_transform = self._get_inner_chunk_transform(shard_spec)

        indexer = BasicIndexer(
            tuple(slice(0, s) for s in shard_shape),
            shape=shard_shape,
            chunk_grid=ChunkGrid.from_sizes(shard_shape, chunk_shape),
        )

        out = chunk_spec.prototype.nd_buffer.empty(
            shape=shard_shape,
            dtype=shard_spec.dtype.to_native_dtype(),
            order=shard_spec.order,
        )

        shard_dict = self._shard_reader_from_bytes_sync(shard_bytes, chunks_per_shard)

        if shard_dict.index.is_all_empty():
            out.fill(shard_spec.fill_value)
            return out

        for chunk_coords, chunk_selection, out_selection, _ in indexer:
            # the GetResult status is discarded: missing INNER chunks of a
            # present shard always fill (read_missing_chunks is a store-key
            # level promise, applied to top-level statuses at the array layer)
            decode_and_scatter_chunk(
                shard_dict.get(chunk_coords),
                out,
                chunk_spec=chunk_spec,
                chunk_selection=chunk_selection,
                out_selection=out_selection,
                drop_axes=(),
                decode=inner_transform.decode_chunk,
            )

        return out

    def _encode_sync(
        self,
        shard_array: NDBuffer,
        shard_spec: ArraySpec,
    ) -> Buffer | None:
        """Encode a full shard synchronously.

        Sync counterpart to `_encode_single`. This is reached when a
        `ShardingCodec` is an *inner* codec of another sharding codec (nested
        sharding): the outer codec encodes each inner chunk through its
        `ChunkTransform`, which calls this method on the inner `ShardingCodec`.

        Each inner chunk is encoded through the inner `ChunkTransform` and
        collected into an intermediate `dict`. The dict's key order is
        immaterial — the physical on-disk layout is decided downstream by the
        `subchunk_write_order` loop in `_encode_shard_dict_sync` (this method
        does NOT impose a layout). Empty inner chunks become `None` entries when
        `write_empty_chunks` is False, signalling `_encode_shard_dict_sync` to
        elide them from the data section and mark them empty in the shard index.

        Returns `None` if every inner chunk was elided (an all-empty shard) —
        callers treat that as "delete the shard key".

        This method does not parallelize compression, but should.
        See TODO: make issue for handling subchunk parallelism

        For a partial write that only touches some inner chunks, use
        `_encode_partial_sync` instead.
        """
        shard_shape = shard_spec.shape
        chunks_per_shard = self._get_chunks_per_shard(shard_spec)
        chunk_spec = self._get_chunk_spec(shard_spec)
        inner_transform = self._get_inner_chunk_transform(shard_spec)

        indexer = BasicIndexer(
            tuple(slice(0, s) for s in shard_shape),
            shape=shard_shape,
            chunk_grid=ChunkGrid.from_sizes(shard_shape, self.chunk_shape),
        )

        # Key order here is immaterial; _encode_shard_dict_sync lays the present
        # chunks out in subchunk_write_order.
        shard_builder: dict[tuple[int, ...], Buffer | None] = dict.fromkeys(
            lexicographic_order_coords(chunks_per_shard)
        )

        for chunk_coords, _chunk_selection, out_selection, _ in indexer:
            # None = chunk normalized to missing (see encode_or_elide_chunk)
            shard_builder[chunk_coords] = encode_or_elide_chunk(
                shard_array[out_selection], chunk_spec, inner_transform.encode_chunk
            )

        return self._encode_shard_dict_sync(
            shard_builder,
            chunks_per_shard=chunks_per_shard,
            buffer_prototype=default_buffer_prototype(),
        )

    def _encode_partial_sync(
        self,
        byte_setter: Any,
        value: NDBuffer,
        selection: SelectorTuple,
        shard_spec: ArraySpec,
    ) -> None:
        """Sync equivalent of `_encode_partial_single`.

        Receives the source data for the written region (not a pre-merged
        shard array) and the selection within the shard, matching the
        calling convention of the async partial-encode path used by
        `BatchedCodecPipeline`.

        This method does not parallelize compression, but should.
        See TODO: make issue for handling subchunk parallelism

        Loads the existing shard, merges the written region into the affected
        inner chunks, and rewrites the whole shard.
        """
        shard_shape = shard_spec.shape
        chunks_per_shard = self._get_chunks_per_shard(shard_spec)
        chunk_spec = self._get_chunk_spec(shard_spec)
        inner_transform = self._get_inner_chunk_transform(shard_spec)

        indexer = list(
            get_indexer(
                selection,
                shape=shard_shape,
                chunk_grid=ChunkGrid.from_sizes(shard_shape, self.chunk_shape),
            )
        )

        is_complete = self._is_complete_shard_write(indexer, chunks_per_shard)

        is_scalar = len(value.shape) == 0

        # Load existing inner-chunk bytes into a dict (same structure as
        # the async path's shard_dict).
        if is_complete:
            shard_dict: dict[tuple[int, ...], Buffer | None] = dict.fromkeys(
                lexicographic_order_coords(chunks_per_shard)
            )
        else:
            existing_bytes = byte_setter.get_sync(prototype=shard_spec.prototype)
            if existing_bytes is not None:
                shard_reader_fb = self._shard_reader_from_bytes_sync(
                    existing_bytes, chunks_per_shard
                )
                # Build the dict with one vectorized index lookup over all chunks,
                # matching the async _encode_partial_single path. A per-coordinate
                # __getitem__ loop here is O(n_chunks) Python overhead that dominates
                # partial writes into shards with many inner chunks. The coordinate
                # array and keys are cached on the reader, so neither is rebuilt here.
                shard_dict = shard_reader_fb.to_dict_vectorized()
            else:
                shard_dict = dict.fromkeys(lexicographic_order_coords(chunks_per_shard))

        # Merge, encode, and store each affected inner chunk into shard_dict via
        # the canonical merge_and_encode_chunk (None = normalized to missing).
        #
        # Scalar fast path: when the written value is a scalar broadcast, every
        # *complete* inner chunk is byte-for-byte identical — same fill, same
        # empty-check, same encoded bytes. Compute that outcome once and reuse it
        # for all complete chunks instead of re-merging, re-checking, and
        # re-encoding tens of thousands of identical chunks. Incomplete (edge)
        # chunks still merge against their own existing data individually.
        # `_sentinel` distinguishes "not computed yet" from a memoized `None`
        # (an empty chunk).
        _sentinel = object()
        scalar_complete_result: Buffer | None | object = _sentinel

        for chunk_coords, chunk_sel, out_sel, is_complete_chunk in indexer:
            if is_scalar and is_complete_chunk:
                if scalar_complete_result is _sentinel:
                    scalar_complete_result = merge_and_encode_chunk(
                        None,
                        value,
                        chunk_spec=chunk_spec,
                        chunk_selection=chunk_sel,
                        out_selection=out_sel,
                        is_complete=is_complete_chunk,
                        drop_axes=(),
                        decode=inner_transform.decode_chunk,
                        encode=inner_transform.encode_chunk,
                    )
                shard_dict[chunk_coords] = scalar_complete_result  # type: ignore[assignment]
                continue

            # A complete chunk fully overwrites: skip decoding what it replaces.
            existing_raw = None if is_complete_chunk else shard_dict.get(chunk_coords)
            shard_dict[chunk_coords] = merge_and_encode_chunk(
                existing_raw,
                value,
                chunk_spec=chunk_spec,
                chunk_selection=chunk_sel,
                out_selection=out_sel,
                is_complete=is_complete_chunk,
                drop_axes=(),
                decode=inner_transform.decode_chunk,
                encode=inner_transform.encode_chunk,
            )

        blob = self._encode_shard_dict_sync(
            shard_dict,
            chunks_per_shard=chunks_per_shard,
            buffer_prototype=default_buffer_prototype(),
        )
        if blob is None:
            byte_setter.delete_sync()
        else:
            byte_setter.set_sync(blob)

    def _build_shard_layout(
        self,
        shard_dict: ShardMapping,
        chunks_per_shard: tuple[int, ...],
    ) -> tuple[_ShardIndex, list[Buffer]] | None:
        """Lay out the present inner chunks of a shard. Pure compute, no IO.

        Packs the encoded inner chunks (in the codec's `subchunk_write_order`)
        into a contiguous data section and builds a shard index pointing each
        present chunk at its ABSOLUTE byte offset within the final blob: when
        the index is stored at the start, offsets are pre-shifted by the index
        size (known without encoding — the index codecs are fixed-size, which
        every index read path already relies on), so the index can be encoded
        exactly once by the caller.

        Returns `(index, data_buffers)`, or `None` for an all-empty shard (no
        chunks present). Shared by the sync and async `_encode_shard_dict*` so
        the layout/offset logic cannot drift between them.
        """
        index = _ShardIndex.create_empty(chunks_per_shard)
        buffers: list[Buffer] = []
        chunk_start = (
            self._shard_index_size(chunks_per_shard) if self.index_location == "start" else 0
        )

        for chunk_coords in self._subchunk_order_iter(chunks_per_shard, self.subchunk_write_order):
            value = shard_dict.get(chunk_coords)
            if value is None or len(value) == 0:
                continue
            chunk_length = len(value)
            buffers.append(value)
            index.set_chunk_slice(chunk_coords, slice(chunk_start, chunk_start + chunk_length))
            chunk_start += chunk_length

        if len(buffers) == 0:
            return None
        return index, buffers

    def _assemble_shard(
        self,
        index_bytes: Buffer,
        buffers: list[Buffer],
        buffer_prototype: BufferPrototype,
        *,
        chunks_per_shard: tuple[int, ...],
    ) -> Buffer:
        """Concatenate the encoded index and data buffers into the shard blob.

        The layout from `_build_shard_layout` already assumes the index size, so
        the encoded index length must match `_shard_index_size` exactly — guard
        that assumption rather than silently corrupt offsets. The guard lives
        here, once, for both the sync and async encode paths.
        """
        if len(index_bytes) != self._shard_index_size(chunks_per_shard):
            raise RuntimeError(
                "encoded shard index size does not match _shard_index_size; "
                "variable-size index codecs are not supported"
            )
        if self.index_location == "start":
            buffers.insert(0, index_bytes)
        else:
            buffers.append(index_bytes)
        template = buffer_prototype.buffer.create_zero_length()
        return template.combine(buffers)

    def _encode_shard_dict_sync(
        self,
        shard_dict: ShardMapping,
        chunks_per_shard: tuple[int, ...],
        buffer_prototype: BufferPrototype,
    ) -> Buffer | None:
        """Sync version of _encode_shard_dict.

        Layout via the shared `_build_shard_layout` (offsets already absolute),
        then a single index encode and concatenation.

        Returns `None` for an all-empty shard (no chunks present).
        """
        layout = self._build_shard_layout(shard_dict, chunks_per_shard)
        if layout is None:
            return None
        index, buffers = layout
        index_bytes = self._encode_shard_index_sync(index)
        return self._assemble_shard(
            index_bytes, buffers, buffer_prototype, chunks_per_shard=chunks_per_shard
        )

    async def _decode_single(
        self,
        shard_bytes: Buffer,
        shard_spec: ArraySpec,
    ) -> NDBuffer:
        shard_shape = shard_spec.shape
        chunk_shape = self.chunk_shape
        chunks_per_shard = self._get_chunks_per_shard(shard_spec)
        chunk_spec = self._get_chunk_spec(shard_spec)

        indexer = BasicIndexer(
            tuple(slice(0, s) for s in shard_shape),
            shape=shard_shape,
            chunk_grid=ChunkGrid.from_sizes(shard_shape, chunk_shape),
        )

        # setup output array
        out = chunk_spec.prototype.nd_buffer.empty(
            shape=shard_shape,
            dtype=shard_spec.dtype.to_native_dtype(),
            order=shard_spec.order,
        )
        shard_dict = await _ShardReader.from_bytes(shard_bytes, self, chunks_per_shard)

        if shard_dict.index.is_all_empty():
            out.fill(shard_spec.fill_value)
            return out

        # decoding chunks and writing them into the output buffer
        await self._get_inner_pipeline(shard_spec).read(
            [
                (
                    _ShardingByteGetter(shard_dict, chunk_coords),
                    chunk_spec,
                    chunk_selection,
                    out_selection,
                    is_complete_shard,
                )
                for chunk_coords, chunk_selection, out_selection, is_complete_shard in indexer
            ],
            out,
        )

        return out

    async def _decode_partial_single(
        self,
        byte_getter: ByteGetter,
        selection: SelectorTuple,
        shard_spec: ArraySpec,
    ) -> NDBuffer | None:
        shard_shape = shard_spec.shape
        chunk_shape = self.chunk_shape
        chunks_per_shard = self._get_chunks_per_shard(shard_spec)
        chunk_spec = self._get_chunk_spec(shard_spec)

        indexer = get_indexer(
            selection,
            shape=shard_shape,
            chunk_grid=ChunkGrid.from_sizes(shard_shape, chunk_shape),
        )

        # setup output array
        out = shard_spec.prototype.nd_buffer.empty(
            shape=indexer.shape,
            dtype=shard_spec.dtype.to_native_dtype(),
            order=shard_spec.order,
        )

        indexed_chunks = list(indexer)
        all_chunk_coords = {chunk_coords for chunk_coords, *_ in indexed_chunks}

        # reading bytes of all requested chunks
        shard_dict_maybe: ShardMapping | None
        if self._is_total_shard(all_chunk_coords, chunks_per_shard):
            # read entire shard
            shard_dict_maybe = await self._load_full_shard_maybe(
                byte_getter=byte_getter,
                prototype=chunk_spec.prototype,
                chunks_per_shard=chunks_per_shard,
            )
        else:
            # read some chunks within the shard
            shard_dict_maybe = await self._load_partial_shard_maybe(
                byte_getter,
                chunk_spec.prototype,
                chunks_per_shard,
                all_chunk_coords,
                max_gap_bytes=shard_spec.config.sharding_coalesce_max_gap_bytes,
                max_coalesced_bytes=shard_spec.config.sharding_coalesce_max_bytes,
            )

        if shard_dict_maybe is None:
            return None
        shard_dict = shard_dict_maybe

        # decoding chunks and writing them into the output buffer
        await self._get_inner_pipeline(shard_spec).read(
            [
                (
                    _ShardingByteGetter(shard_dict, chunk_coords),
                    chunk_spec,
                    chunk_selection,
                    out_selection,
                    is_complete_shard,
                )
                for chunk_coords, chunk_selection, out_selection, is_complete_shard in indexer
            ],
            out,
        )

        if hasattr(indexer, "sel_shape"):
            return out.reshape(indexer.sel_shape)
        else:
            return out

    def _subchunk_order_iter(
        self, chunks_per_shard: tuple[int, ...], subchunk_write_order: SubchunkWriteOrder
    ) -> Iterable[tuple[int, ...]]:
        subchunk_iter: Iterable[tuple[int, ...]]
        match subchunk_write_order:
            case "morton":
                subchunk_iter = morton_order_coords(chunks_per_shard)
            case "lexicographic":
                subchunk_iter = lexicographic_order_coords(chunks_per_shard)
            case "colexicographic":
                subchunk_iter = colexicographic_order_coords(chunks_per_shard)
            case "unordered":
                # "unordered" promises no particular layout; today it happens to be
                # lexicographic, but callers must not rely on that.
                subchunk_iter = np.ndindex(chunks_per_shard)
            case _:
                raise ValueError(f"Unrecognized subchunk write order: {subchunk_write_order!r}.")
        return subchunk_iter

    def _decode_full_shard_bulk_if_uncompressed(
        self,
        shard_bytes: Buffer,
        shard_spec: ArraySpec,
        indexer: Any,
    ) -> NDBuffer | None:
        """Vectorized whole-shard decode for dense, fixed-size, uncompressed shards.

        Returns the assembled shard array, or None if the fast path does not
        apply (so the caller falls back to the per-chunk loop). Conditions:
        - inner codec chain is fixed-size (no compression / variable-length);
        - the inner codec chain is exactly a single BytesCodec — decode is a
          dtype/endian view with no reordering. A trailing crc32c is NOT accepted
          (the bulk path can't verify per-chunk checksums, so crc shards keep the
          per-chunk path's corruption detection);
        - the stored index is dense (every chunk present, equal fixed length,
          contiguous) so the data section is a regular grid of chunk payloads.

        Chunk positions are read from the stored index, so this is correct for
        any `subchunk_write_order` (morton / lexicographic / colexicographic /
        unordered). The on-disk byte order is taken from the BytesCodec's
        `endian`, so big- and little-endian shards both decode correctly.
        """
        # --- gate on a trivial, fixed-size inner codec chain ---
        if not self._inner_codecs_fixed_size:
            return None
        # The inner chain must be exactly a single BytesCodec (a dtype/endian
        # view, no reordering). A trailing Crc32cCodec is excluded on purpose:
        # the bulk path would have to strip-and-discard the per-chunk checksum
        # bytes, silently dropping the corruption detection the per-chunk path
        # enforces (Crc32cCodec._decode_sync raises on mismatch). crc-protected
        # shards therefore fall through to the per-chunk path.
        if len(self.codecs) != 1 or not isinstance(self.codecs[0], BytesCodec):
            return None
        ab_codec = self.codecs[0]

        chunks_per_shard = self._get_chunks_per_shard(shard_spec)
        chunk_spec = self._get_chunk_spec(shard_spec)
        n_chunks = product(chunks_per_shard)
        if n_chunks == 0:
            return None

        # Only valid for a plain contiguous full-shard read, where each chunk
        # lands at its natural grid position. The `sel_shape` check is
        # load-bearing: a gather indexer (CoordinateIndexer, from vindex / an
        # oindex with an integer-array selection) reorders points and exposes
        # `sel_shape`, but its `.shape` is the FLATTENED point count, which can
        # equal the shard shape by coincidence (trivially in 1-D). Gating on
        # shape alone lets such a selection through, and the bulk path then
        # returns the shard in natural order, silently dropping the reordering.
        # A contiguous full read (BasicIndexer, or a non-gathering
        # OrthogonalIndexer from `arr[:]`) has no `sel_shape` and is served here.
        # Anything that gathers must fall through to the per-chunk path so
        # chunk_selection / out_selection are honored.
        if getattr(indexer, "sel_shape", None) is not None:
            return None
        if tuple(indexer.shape) != tuple(shard_spec.shape):
            return None
        chunk_byte_length = self._inner_chunk_byte_length(chunk_spec)

        shard_index_size = self._shard_index_size(chunks_per_shard)
        if len(shard_bytes) != n_chunks * chunk_byte_length + shard_index_size:
            return None  # not a dense fixed-size shard

        # --- decode the index; require dense layout ---
        if self.index_location == "start":
            index_bytes = shard_bytes[:shard_index_size]
        else:
            index_bytes = shard_bytes[-shard_index_size:]
        index = self._decode_shard_index_sync(index_bytes, chunks_per_shard)
        if not index.is_dense(chunk_byte_length):
            return None

        # --- bulk reconstruct ---
        # The index gives each chunk's absolute byte offset within the blob; with
        # a dense, crc-free, fixed-size layout the payload length is exactly the
        # encoded item-bytes of one chunk.
        native_dtype = shard_spec.dtype.to_native_dtype()
        raw = shard_bytes.as_numpy_array().view(np.uint8)
        payload = chunk_byte_length
        cs = self.chunk_shape

        # On-disk byte order is carried by the BytesCodec's `endian`, NOT by the
        # data type (zarr v3). Build the read-view dtype from the codec's endian
        # exactly as BytesCodec._decode_sync does, so a big-endian shard read on a
        # little-endian host (or vice versa) is interpreted correctly. Assigning
        # into the native-dtype `out` then performs any needed byteswap. `endian`
        # is now a plain Literal['little', 'big'] | None string (no longer an enum).
        endian_str = ab_codec.endian
        if isinstance(chunk_spec.dtype, HasEndianness):
            stored_dtype = replace(chunk_spec.dtype, endianness=endian_str).to_native_dtype()  # type: ignore[call-arg]
        else:
            stored_dtype = chunk_spec.dtype.to_native_dtype()

        offsets = index.offsets_and_lengths[..., 0].reshape(-1)  # localized coords, C-order
        coords_c = list(np.ndindex(chunks_per_shard))
        out = shard_spec.prototype.nd_buffer.empty(
            shape=indexer.shape, dtype=native_dtype, order=shard_spec.order
        )
        for flat, coord in enumerate(coords_c):
            start = int(offsets[flat])
            chunk = raw[start : start + payload].view(stored_dtype).reshape(cs)
            sel = tuple(slice(c * s, c * s + s) for c, s in zip(coord, cs, strict=True))
            out[sel] = chunk
        return out

    def _decode_partial_sync(
        self,
        byte_getter: Any,
        selection: SelectorTuple,
        shard_spec: ArraySpec,
    ) -> NDBuffer | None:
        """Sync equivalent of `_decode_partial_single`.

        Reads only the inner-chunk byte ranges that overlap `selection`
        (plus the shard index) and decodes them through the inner codec
        chain.  The store must support `get_sync` with byte ranges.

        This method does not parallelize decompression, but should.
        See TODO: make issue for handling subchunk parallelism

        Two sub-paths:
        - If `selection` covers the entire shard, just fetch the whole
          blob — that's strictly cheaper than two round trips (index, then
          data) plus the per-chunk overhead of partial fetches.
        - Otherwise fetch the index alone, look up only the byte slices of
          the inner chunks the selection touches, fetch those, and decode.
        """
        shard_shape = shard_spec.shape
        chunk_shape = self.chunk_shape
        chunks_per_shard = self._get_chunks_per_shard(shard_spec)
        chunk_spec = self._get_chunk_spec(shard_spec)
        inner_transform = self._get_inner_chunk_transform(shard_spec)

        indexer = get_indexer(
            selection,
            shape=shard_shape,
            chunk_grid=ChunkGrid.from_sizes(shard_shape, chunk_shape),
        )

        out = shard_spec.prototype.nd_buffer.empty(
            shape=indexer.shape,
            dtype=shard_spec.dtype.to_native_dtype(),
            order=shard_spec.order,
        )

        indexed_chunks = list(indexer)
        all_chunk_coords = {chunk_coords for chunk_coords, *_ in indexed_chunks}

        # Read just the inner chunks we need.
        if self._is_total_shard(all_chunk_coords, chunks_per_shard):
            shard_bytes = byte_getter.get_sync(prototype=chunk_spec.prototype)
            if shard_bytes is None:
                return None
            bulk = self._decode_full_shard_bulk_if_uncompressed(shard_bytes, shard_spec, indexer)
            if bulk is not None:
                # The bulk path only fires for a contiguous full-shard read (it
                # returns None for any gather indexer that exposes `sel_shape`),
                # so the result is already shard-shaped — no reshape needed.
                return bulk
            shard_reader = self._shard_reader_from_bytes_sync(shard_bytes, chunks_per_shard)
            shard_dict: ShardMapping = shard_reader
        else:
            # Partial read: fetch only the touched inner chunks, coalescing
            # adjacent byte ranges (mirrors the async _load_partial_shard_maybe
            # / #3004). Returns None if the shard is absent.
            partial = self._load_partial_shard_maybe_sync(
                byte_getter,
                chunk_spec.prototype,
                chunks_per_shard,
                all_chunk_coords,
                max_gap_bytes=shard_spec.config.sharding_coalesce_max_gap_bytes,
                max_coalesced_bytes=shard_spec.config.sharding_coalesce_max_bytes,
            )
            if partial is None:
                return None
            shard_dict = partial

        # Decode each needed inner chunk and scatter into out (statuses
        # discarded: missing inner chunks fill, see _decode_sync).
        for chunk_coords, chunk_selection, out_selection, _ in indexed_chunks:
            decode_and_scatter_chunk(
                shard_dict.get(chunk_coords),
                out,
                chunk_spec=chunk_spec,
                chunk_selection=chunk_selection,
                out_selection=out_selection,
                drop_axes=(),
                decode=inner_transform.decode_chunk,
            )

        if hasattr(indexer, "sel_shape"):
            return out.reshape(indexer.sel_shape)
        return out

    async def _encode_single(
        self,
        shard_array: NDBuffer,
        shard_spec: ArraySpec,
    ) -> Buffer | None:
        shard_shape = shard_spec.shape
        chunk_shape = self.chunk_shape
        chunks_per_shard = self._get_chunks_per_shard(shard_spec)
        chunk_spec = self._get_chunk_spec(shard_spec)

        indexer = list(
            BasicIndexer(
                tuple(slice(0, s) for s in shard_shape),
                shape=shard_shape,
                chunk_grid=ChunkGrid.from_sizes(shard_shape, chunk_shape),
            )
        )
        shard_builder = dict.fromkeys(lexicographic_order_coords(chunks_per_shard))

        await self._get_inner_pipeline(shard_spec).write(
            [
                (
                    _ShardingByteSetter(shard_builder, chunk_coords),
                    chunk_spec,
                    chunk_selection,
                    out_selection,
                    is_complete_shard,
                )
                for chunk_coords, chunk_selection, out_selection, is_complete_shard in indexer
            ],
            shard_array,
        )

        return await self._encode_shard_dict(
            shard_builder,
            chunks_per_shard=chunks_per_shard,
            buffer_prototype=default_buffer_prototype(),
        )

    async def _encode_partial_single(
        self,
        byte_setter: ByteSetter,
        shard_array: NDBuffer,
        selection: SelectorTuple,
        shard_spec: ArraySpec,
    ) -> None:
        shard_shape = shard_spec.shape
        chunk_shape = self.chunk_shape
        chunks_per_shard = self._get_chunks_per_shard(shard_spec)
        chunk_spec = self._get_chunk_spec(shard_spec)

        indexer = list(
            get_indexer(
                selection,
                shape=shard_shape,
                chunk_grid=ChunkGrid.from_sizes(shard_shape, chunk_shape),
            )
        )

        if self._is_complete_shard_write(indexer, chunks_per_shard):
            shard_dict = dict.fromkeys(lexicographic_order_coords(chunks_per_shard))
        else:
            shard_reader = await self._load_full_shard_maybe(
                byte_getter=byte_setter,
                prototype=chunk_spec.prototype,
                chunks_per_shard=chunks_per_shard,
            )
            shard_reader = shard_reader or _ShardReader.create_empty(chunks_per_shard)
            # Use vectorized lookup for better performance. The lexicographic
            # coordinate array and keys are cached, so neither is rebuilt on
            # every write.
            shard_dict = shard_reader.to_dict_vectorized()

        await self._get_inner_pipeline(shard_spec).write(
            [
                (
                    _ShardingByteSetter(shard_dict, chunk_coords),
                    chunk_spec,
                    chunk_selection,
                    out_selection,
                    is_complete_shard,
                )
                for chunk_coords, chunk_selection, out_selection, is_complete_shard in indexer
            ],
            shard_array,
        )
        buf = await self._encode_shard_dict(
            shard_dict,
            chunks_per_shard=chunks_per_shard,
            buffer_prototype=default_buffer_prototype(),
        )

        if buf is None:
            await byte_setter.delete()
        else:
            await byte_setter.set(buf)

    async def _encode_shard_dict(
        self,
        map: ShardMapping,
        chunks_per_shard: tuple[int, ...],
        buffer_prototype: BufferPrototype,
    ) -> Buffer | None:
        """Layout via the shared `_build_shard_layout` (offsets already
        absolute), then a single index encode and concatenation. Async twin of
        `_encode_shard_dict_sync`."""
        layout = self._build_shard_layout(map, chunks_per_shard)
        if layout is None:
            return None
        index, buffers = layout
        index_bytes = await self._encode_shard_index(index)
        return self._assemble_shard(
            index_bytes, buffers, buffer_prototype, chunks_per_shard=chunks_per_shard
        )

    def _is_total_shard(
        self, all_chunk_coords: set[tuple[int, ...]], chunks_per_shard: tuple[int, ...]
    ) -> bool:
        # `all_chunk_coords` comes from an indexer over this shard's chunk grid, so
        # it is always a subset of that grid (`validate` requires the shard shape to
        # be divisible by the inner chunk shape, so the indexer cannot produce an
        # out-of-grid coordinate). A subset whose size equals the grid's is the
        # whole grid, so the count check alone proves totality — no need to build
        # and membership-test the full coordinate set on this hot path.
        return len(all_chunk_coords) == product(chunks_per_shard)

    def _is_complete_shard_write(
        self,
        indexed_chunks: Sequence[ChunkProjection],
        chunks_per_shard: tuple[int, ...],
    ) -> bool:
        all_chunk_coords = {chunk_coords for chunk_coords, *_ in indexed_chunks}
        return self._is_total_shard(all_chunk_coords, chunks_per_shard) and all(
            is_complete_chunk for *_, is_complete_chunk in indexed_chunks
        )

    def _index_codecs_sync_capable(self) -> bool:
        return all(isinstance(c, SupportsSyncCodec) for c in self.index_codecs)

    async def _decode_shard_index(
        self, index_bytes: Buffer, chunks_per_shard: tuple[int, ...]
    ) -> _ShardIndex:
        # Pure compute (the bytes are already in hand): delegate to the sync
        # implementation instead of spinning up a pipeline + per-call
        # AsyncChunkTransform for a tiny fixed-size decode. The default
        # (bytes + crc32c) index chain is sync-capable; an async-only
        # third-party index codec falls back to the full async pipeline, which
        # the synchronous read paths cannot use but this async path still can.
        if self._index_codecs_sync_capable():
            return self._decode_shard_index_sync(index_bytes, chunks_per_shard)
        index_array = next(
            iter(
                await get_pipeline_class()
                .from_codecs(self.index_codecs)
                .decode([(index_bytes, self._get_index_chunk_spec(chunks_per_shard))])
            )
        )
        assert index_array is not None  # the bytes are already in hand
        return _ShardIndex(chunks_per_shard, index_array.as_numpy_array())

    async def _encode_shard_index(self, index: _ShardIndex) -> Buffer:
        # Pure compute: delegate to the sync implementation, with the same
        # async-pipeline fallback as _decode_shard_index.
        if self._index_codecs_sync_capable():
            return self._encode_shard_index_sync(index)
        index_bytes = next(
            iter(
                await get_pipeline_class()
                .from_codecs(self.index_codecs)
                .encode(
                    [
                        (
                            get_ndbuffer_class().from_numpy_array(index.offsets_and_lengths),
                            self._get_index_chunk_spec(index.chunks_per_shard),
                        )
                    ]
                )
            )
        )
        assert index_bytes is not None
        return index_bytes

    def _shard_index_size(self, chunks_per_shard: tuple[int, ...]) -> int:
        return (
            get_pipeline_class()
            .from_codecs(self.index_codecs)
            .compute_encoded_size(
                16 * product(chunks_per_shard), self._get_index_chunk_spec(chunks_per_shard)
            )
        )

    def _get_index_chunk_spec(self, chunks_per_shard: tuple[int, ...]) -> ArraySpec:
        return ArraySpec(
            shape=chunks_per_shard + (2,),
            dtype=UInt64(endianness="little"),
            fill_value=MAX_UINT_64,
            config=ArrayConfig(
                order="C", write_empty_chunks=False
            ),  # Note: this is hard-coded for simplicity -- it is not surfaced into user code,
            prototype=default_buffer_prototype(),
        )

    def _get_chunk_spec(self, shard_spec: ArraySpec) -> ArraySpec:
        return ArraySpec(
            shape=self.chunk_shape,
            dtype=shard_spec.dtype,
            fill_value=shard_spec.fill_value,
            config=shard_spec.config,
            prototype=shard_spec.prototype,
        )

    def _get_chunks_per_shard(self, shard_spec: ArraySpec) -> tuple[int, ...]:
        return tuple(
            s // c
            for s, c in zip(
                shard_spec.shape,
                self.chunk_shape,
                strict=False,
            )
        )

    def _shard_index_byte_range(
        self, chunks_per_shard: tuple[int, ...]
    ) -> RangeByteRequest | SuffixByteRequest:
        """Byte range of the shard index within the shard blob.

        Single source of truth for the index-location arithmetic, shared by the
        sync and async index loaders so they cannot drift.
        """
        shard_index_size = self._shard_index_size(chunks_per_shard)
        if self.index_location == "start":
            return RangeByteRequest(0, shard_index_size)
        return SuffixByteRequest(shard_index_size)

    @staticmethod
    def _pair_chunks_with_byte_ranges(
        shard_index: _ShardIndex, all_chunk_coords: set[tuple[int, ...]]
    ) -> list[tuple[tuple[int, ...], RangeByteRequest]]:
        """Pair each requested chunk coord with its byte range in the shard.

        Coords whose chunk is absent from the index are omitted. Shared by the
        sync and async partial-shard loaders.
        """
        chunk_coord_byte_ranges: list[tuple[tuple[int, ...], RangeByteRequest]] = []
        for chunk_coord in all_chunk_coords:
            chunk_byte_slice = shard_index.get_chunk_slice(chunk_coord)
            if chunk_byte_slice is not None:
                chunk_coord_byte_ranges.append(
                    (chunk_coord, RangeByteRequest(chunk_byte_slice[0], chunk_byte_slice[1]))
                )
        return chunk_coord_byte_ranges

    async def _load_shard_index_maybe(
        self, byte_getter: ByteGetter, chunks_per_shard: tuple[int, ...]
    ) -> _ShardIndex | None:
        index_bytes = await byte_getter.get(
            prototype=numpy_buffer_prototype(),
            byte_range=self._shard_index_byte_range(chunks_per_shard),
        )
        if index_bytes is not None:
            return await self._decode_shard_index(index_bytes, chunks_per_shard)
        return None

    async def _load_full_shard_maybe(
        self, byte_getter: ByteGetter, prototype: BufferPrototype, chunks_per_shard: tuple[int, ...]
    ) -> _ShardReader | None:
        shard_bytes = await byte_getter.get(prototype=prototype)

        return (
            await _ShardReader.from_bytes(shard_bytes, self, chunks_per_shard)
            if shard_bytes
            else None
        )

    @property
    def _inner_codecs_fixed_size(self) -> bool:
        """True when all inner codecs produce fixed-size output (no compression)."""
        return all(c.is_fixed_size for c in self.codecs)

    def _inner_chunk_byte_length(self, chunk_spec: ArraySpec) -> int:
        """Encoded byte length of a single inner chunk. Only valid when _inner_codecs_fixed_size."""
        raw_byte_length = 1
        for s in self.chunk_shape:
            raw_byte_length *= s
        raw_byte_length *= chunk_spec.dtype.item_size  # type: ignore[attr-defined]
        return int(self.codec_pipeline.compute_encoded_size(raw_byte_length, chunk_spec))

    async def _load_partial_shard_maybe(
        self,
        byte_getter: ByteGetter,
        prototype: BufferPrototype,
        chunks_per_shard: tuple[int, ...],
        all_chunk_coords: set[tuple[int, ...]],
        max_gap_bytes: int,
        max_coalesced_bytes: int,
    ) -> ShardMapping | None:
        """
        Read chunks from `byte_getter` for the case where the read is less than a full shard.
        Returns a mapping of chunk coordinates to bytes or None.

        `max_gap_bytes` and `max_coalesced_bytes` are forwarded to
        `Store.get_ranges` to control byte-range coalescing across the requested
        chunks.
        """
        shard_index = await self._load_shard_index_maybe(byte_getter, chunks_per_shard)
        if shard_index is None:
            return None

        chunk_coord_byte_ranges = self._pair_chunks_with_byte_ranges(shard_index, all_chunk_coords)

        if not chunk_coord_byte_ranges:
            return {}

        shard_dict: ShardMutableMapping = {}
        if isinstance(byte_getter, StorePath):
            # External store: use Store.get_ranges for coalescing + concurrency.
            byte_ranges = [byte_range for _, byte_range in chunk_coord_byte_ranges]
            try:
                async for group in byte_getter.store.get_ranges(
                    byte_getter.path,
                    byte_ranges,
                    prototype=prototype,
                    max_gap_bytes=max_gap_bytes,
                    max_coalesced_bytes=max_coalesced_bytes,
                ):
                    for idx, buf in group:
                        if buf is not None:
                            chunk_coord, _ = chunk_coord_byte_ranges[idx]
                            shard_dict[chunk_coord] = buf
            except BaseExceptionGroup as eg:
                # `Store.get_ranges` raises FileNotFoundError (wrapped in a
                # BaseExceptionGroup) if any underlying fetch indicates the key is
                # absent. The shard index loaded above, so this typically means a
                # race where the shard was deleted mid-read; treat it as "shard
                # gone" to match the index-missing branch (return None). Anything
                # else in the group (e.g. IO errors) is re-raised.
                _, rest = eg.split(FileNotFoundError)
                if rest is not None:
                    raise rest from None
                return None
        else:
            # Any other ByteGetter. In practice only `_ShardingByteGetter` for
            # nested sharding, which slices an in-memory buffer (no I/O to coalesce).
            for chunk_coord, byte_range in chunk_coord_byte_ranges:
                buf = await byte_getter.get(prototype, byte_range)
                if buf is not None:
                    shard_dict[chunk_coord] = buf

        return shard_dict

    def _load_shard_index_maybe_sync(
        self, byte_getter: Any, chunks_per_shard: tuple[int, ...]
    ) -> _ShardIndex | None:
        """Sync counterpart of `_load_shard_index_maybe`."""
        index_bytes = byte_getter.get_sync(
            prototype=numpy_buffer_prototype(),
            byte_range=self._shard_index_byte_range(chunks_per_shard),
        )
        if index_bytes is not None:
            return self._decode_shard_index_sync(index_bytes, chunks_per_shard)
        return None

    def _load_partial_shard_maybe_sync(
        self,
        byte_getter: Any,
        prototype: BufferPrototype,
        chunks_per_shard: tuple[int, ...],
        all_chunk_coords: set[tuple[int, ...]],
        *,
        max_gap_bytes: int,
        max_coalesced_bytes: int,
    ) -> ShardMapping | None:
        """Sync counterpart of `_load_partial_shard_maybe` (the #3004 read path).

        Reads the shard index, then fetches only the touched inner chunks via the
        store's coalescing `get_ranges_sync` (merging adjacent ranges into fewer
        reads), matching the async path's IO shape without an event loop.
        `max_gap_bytes` and `max_coalesced_bytes` control the coalescing, forwarded
        from the array's `sharding_coalesce_*` config exactly as the async path.
        """
        shard_index = self._load_shard_index_maybe_sync(byte_getter, chunks_per_shard)
        if shard_index is None:
            return None

        chunk_coord_byte_ranges = self._pair_chunks_with_byte_ranges(shard_index, all_chunk_coords)

        if not chunk_coord_byte_ranges:
            return {}

        shard_dict: ShardMutableMapping = {}
        store = byte_getter.store if hasattr(byte_getter, "store") else None
        if isinstance(store, Store) and isinstance(store, SupportsGetSync):
            # External store: coalesce via get_ranges_sync (mirrors get_ranges).
            byte_ranges = [byte_range for _, byte_range in chunk_coord_byte_ranges]
            try:
                for idx, buf in store.get_ranges_sync(
                    byte_getter.path,
                    byte_ranges,
                    prototype=prototype,
                    max_gap_bytes=max_gap_bytes,
                    max_coalesced_bytes=max_coalesced_bytes,
                ):
                    if buf is not None:
                        chunk_coord, _ = chunk_coord_byte_ranges[idx]
                        shard_dict[chunk_coord] = buf
            except BaseExceptionGroup as eg:
                # Mirror the async path: a FileNotFoundError means the shard was
                # deleted mid-read -> treat as "gone" (None). Re-raise anything else.
                _, rest = eg.split(FileNotFoundError)
                if rest is not None:
                    raise rest from None
                return None
        else:
            # Nested sharding: an in-memory _ShardingByteGetter, no IO to coalesce.
            for chunk_coord, byte_range in chunk_coord_byte_ranges:
                buf = byte_getter.get_sync(prototype=prototype, byte_range=byte_range)
                if buf is not None:
                    shard_dict[chunk_coord] = buf

        return shard_dict

    def compute_encoded_size(self, input_byte_length: int, shard_spec: ArraySpec) -> int:
        chunks_per_shard = self._get_chunks_per_shard(shard_spec)
        return input_byte_length + self._shard_index_size(chunks_per_shard)

chunk_shape instance-attribute

chunk_shape: tuple[int, ...]

codec_pipeline property

codec_pipeline: CodecPipeline

codecs instance-attribute

codecs: tuple[Codec, ...]

index_codecs instance-attribute

index_codecs: tuple[Codec, ...]

index_location class-attribute instance-attribute

index_location: IndexLocation = 'end'

is_fixed_size class-attribute instance-attribute

is_fixed_size = False

subchunk_write_order class-attribute instance-attribute

subchunk_write_order: SubchunkWriteOrder = 'morton'

__getstate__

__getstate__() -> dict[str, Any]

Source code in zarr/codecs/sharding.py

def __getstate__(self) -> dict[str, Any]:
    # `subchunk_write_order` is not part of codec metadata (`to_dict`), so carry it
    # explicitly to survive a pickle round-trip (otherwise it reverts to `morton`).
    return {"subchunk_write_order": self.subchunk_write_order, **self.to_dict()}

__init__

__init__(
    *,
    chunk_shape: ShapeLike,
    codecs: Iterable[Codec | dict[str, JSON]] = (
        BytesCodec(),
    ),
    index_codecs: Iterable[Codec | dict[str, JSON]] = (
        BytesCodec(),
        Crc32cCodec(),
    ),
    index_location: ShardingCodecIndexLocation
    | IndexLocation = "end",
    subchunk_write_order: SubchunkWriteOrder = "morton",
) -> None

Source code in zarr/codecs/sharding.py

def __init__(
    self,
    *,
    chunk_shape: ShapeLike,
    codecs: Iterable[Codec | dict[str, JSON]] = (BytesCodec(),),
    index_codecs: Iterable[Codec | dict[str, JSON]] = (BytesCodec(), Crc32cCodec()),
    index_location: ShardingCodecIndexLocation | IndexLocation = "end",
    subchunk_write_order: SubchunkWriteOrder = "morton",
) -> None:
    chunk_shape_parsed = parse_shapelike(chunk_shape)
    codecs_parsed = parse_codecs(codecs)
    index_codecs_parsed = parse_codecs(index_codecs)
    index_location_coerced = _coerce_enum_input(
        index_location, "index_location", "ShardingCodec"
    )
    index_location_parsed = _parse_index_location(index_location_coerced)
    if subchunk_write_order not in SUBCHUNK_WRITE_ORDER:
        raise ValueError(
            f"Unrecognized subchunk write order: {subchunk_write_order}. Only {SUBCHUNK_WRITE_ORDER} are allowed."
        )

    object.__setattr__(self, "chunk_shape", chunk_shape_parsed)
    object.__setattr__(self, "codecs", codecs_parsed)
    object.__setattr__(self, "index_codecs", index_codecs_parsed)
    object.__setattr__(self, "index_location", index_location_parsed)
    object.__setattr__(self, "subchunk_write_order", subchunk_write_order)

    # Use instance-local lru_cache to avoid memory leaks

    # numpy void scalars are not hashable, which means an array spec with a fill value that is
    # a numpy void scalar will break the lru_cache. This is commented for now but should be
    # fixed. See https://github.com/zarr-developers/zarr-python/issues/3054
    # object.__setattr__(self, "_get_chunk_spec", lru_cache()(self._get_chunk_spec))
    object.__setattr__(self, "_get_index_chunk_spec", lru_cache()(self._get_index_chunk_spec))
    object.__setattr__(self, "_get_chunks_per_shard", lru_cache()(self._get_chunks_per_shard))
    object.__setattr__(self, "_shard_index_size", lru_cache()(self._shard_index_size))
    object.__setattr__(
        self, "_get_inner_chunk_transform", lru_cache()(self._get_inner_chunk_transform)
    )
    object.__setattr__(
        self, "_get_index_chunk_transform", lru_cache()(self._get_index_chunk_transform)
    )

__setstate__

__setstate__(state: dict[str, Any]) -> None

Source code in zarr/codecs/sharding.py

def __setstate__(self, state: dict[str, Any]) -> None:
    config = state["configuration"]
    object.__setattr__(self, "chunk_shape", parse_shapelike(config["chunk_shape"]))
    object.__setattr__(self, "codecs", parse_codecs(config["codecs"]))
    object.__setattr__(self, "index_codecs", parse_codecs(config["index_codecs"]))
    object.__setattr__(self, "index_location", _parse_index_location(config["index_location"]))
    object.__setattr__(self, "subchunk_write_order", state["subchunk_write_order"])

    # Use instance-local lru_cache to avoid memory leaks
    # object.__setattr__(self, "_get_chunk_spec", lru_cache()(self._get_chunk_spec))
    object.__setattr__(self, "_get_index_chunk_spec", lru_cache()(self._get_index_chunk_spec))
    object.__setattr__(self, "_get_chunks_per_shard", lru_cache()(self._get_chunks_per_shard))
    object.__setattr__(self, "_shard_index_size", lru_cache()(self._shard_index_size))
    object.__setattr__(
        self, "_get_inner_chunk_transform", lru_cache()(self._get_inner_chunk_transform)
    )
    object.__setattr__(
        self, "_get_index_chunk_transform", lru_cache()(self._get_index_chunk_transform)
    )

compute_encoded_size

compute_encoded_size(
    input_byte_length: int, shard_spec: ArraySpec
) -> int

Given an input byte length, this method returns the output byte length. Raises a NotImplementedError for codecs with variable-sized outputs (e.g. compressors).

Parameters:

• input_byte_length (int) –
• chunk_spec (ArraySpec) –

Returns:

• int –

Source code in zarr/codecs/sharding.py

def compute_encoded_size(self, input_byte_length: int, shard_spec: ArraySpec) -> int:
    chunks_per_shard = self._get_chunks_per_shard(shard_spec)
    return input_byte_length + self._shard_index_size(chunks_per_shard)

decode async

decode(
    chunks_and_specs: Iterable[tuple[CO | None, ArraySpec]],
) -> Iterable[CI | None]

Decodes a batch of chunks. Chunks can be None in which case they are ignored by the codec.

Parameters:

• chunks_and_specs (Iterable[tuple[CodecOutput | None, ArraySpec]]) –

  Ordered set of encoded chunks with their accompanying chunk spec.

Returns:

• Iterable[CI | None] –

Source code in zarr/abc/codec.py

async def decode(
    self,
    chunks_and_specs: Iterable[tuple[CO | None, ArraySpec]],
) -> Iterable[CI | None]:
    """Decodes a batch of chunks.
    Chunks can be None in which case they are ignored by the codec.

    Parameters
    ----------
    chunks_and_specs : Iterable[tuple[CodecOutput | None, ArraySpec]]
        Ordered set of encoded chunks with their accompanying chunk spec.

    Returns
    -------
    Iterable[CI | None]
    """
    return await _batching_helper(self._decode_single, chunks_and_specs)

decode_partial async

decode_partial(
    batch_info: Iterable[
        tuple[ByteGetter, SelectorTuple, ArraySpec]
    ],
) -> Iterable[NDBuffer | None]

Partially decodes a batch of chunks. This method determines parts of a chunk from the slice selection, fetches these parts from the store (via ByteGetter) and decodes them.

Parameters:

• batch_info (Iterable[tuple[ByteGetter, SelectorTuple, ArraySpec]]) –

  Ordered set of information about slices of encoded chunks. The slice selection determines which parts of the chunk will be fetched. The ByteGetter is used to fetch the necessary bytes. The chunk spec contains information about the construction of an array from the bytes.

Returns:

• Iterable[NDBuffer | None] –

Source code in zarr/abc/codec.py

async def decode_partial(
    self,
    batch_info: Iterable[tuple[ByteGetter, SelectorTuple, ArraySpec]],
) -> Iterable[NDBuffer | None]:
    """Partially decodes a batch of chunks.
    This method determines parts of a chunk from the slice selection,
    fetches these parts from the store (via ByteGetter) and decodes them.

    Parameters
    ----------
    batch_info : Iterable[tuple[ByteGetter, SelectorTuple, ArraySpec]]
        Ordered set of information about slices of encoded chunks.
        The slice selection determines which parts of the chunk will be fetched.
        The ByteGetter is used to fetch the necessary bytes.
        The chunk spec contains information about the construction of an array from the bytes.

    Returns
    -------
    Iterable[NDBuffer | None]
    """
    return await concurrent_map(
        list(batch_info),
        self._decode_partial_single,
        config.get("async.concurrency"),
    )

encode async

encode(
    chunks_and_specs: Iterable[tuple[CI | None, ArraySpec]],
) -> Iterable[CO | None]

Encodes a batch of chunks. Chunks can be None in which case they are ignored by the codec.

Parameters:

• chunks_and_specs (Iterable[tuple[CI | None, ArraySpec]]) –

  Ordered set of to-be-encoded chunks with their accompanying chunk spec.

Returns:

• Iterable[CodecOutput | None] –

Source code in zarr/abc/codec.py

async def encode(
    self,
    chunks_and_specs: Iterable[tuple[CI | None, ArraySpec]],
) -> Iterable[CO | None]:
    """Encodes a batch of chunks.
    Chunks can be None in which case they are ignored by the codec.

    Parameters
    ----------
    chunks_and_specs : Iterable[tuple[CI | None, ArraySpec]]
        Ordered set of to-be-encoded chunks with their accompanying chunk spec.

    Returns
    -------
    Iterable[CodecOutput | None]
    """
    return await _batching_helper(self._encode_single, chunks_and_specs)

encode_partial async

encode_partial(
    batch_info: Iterable[
        tuple[
            ByteSetter, NDBuffer, SelectorTuple, ArraySpec
        ]
    ],
) -> None

Partially encodes a batch of chunks. This method determines parts of a chunk from the slice selection, encodes them and writes these parts to the store (via ByteSetter). If merging with existing chunk data in the store is necessary, this method will read from the store first and perform the merge.

Parameters:

• batch_info (Iterable[tuple[ByteSetter, NDBuffer, SelectorTuple, ArraySpec]]) –

  Ordered set of information about slices of to-be-encoded chunks. The slice selection determines which parts of the chunk will be encoded. The ByteSetter is used to write the necessary bytes and fetch bytes for existing chunk data. The chunk spec contains information about the chunk.

Source code in zarr/abc/codec.py

async def encode_partial(
    self,
    batch_info: Iterable[tuple[ByteSetter, NDBuffer, SelectorTuple, ArraySpec]],
) -> None:
    """Partially encodes a batch of chunks.
    This method determines parts of a chunk from the slice selection, encodes them and
    writes these parts to the store (via ByteSetter).
    If merging with existing chunk data in the store is necessary, this method will
    read from the store first and perform the merge.

    Parameters
    ----------
    batch_info : Iterable[tuple[ByteSetter, NDBuffer, SelectorTuple, ArraySpec]]
        Ordered set of information about slices of to-be-encoded chunks.
        The slice selection determines which parts of the chunk will be encoded.
        The ByteSetter is used to write the necessary bytes and fetch bytes for existing chunk data.
        The chunk spec contains information about the chunk.
    """
    await concurrent_map(
        list(batch_info),
        self._encode_partial_single,
        config.get("async.concurrency"),
    )

evolve_from_array_spec

evolve_from_array_spec(array_spec: ArraySpec) -> Self

Thread the spec through the inner chain.

Each codec is evolved against the spec produced by the previous one. Evolving every codec against the same unthreaded spec is the bug shape that strips BytesCodec.endian behind a dtype-changing codec — and this method runs on the real array-creation path, baking the damaged chain into the evolved instance before the transform builders ever run.

Parameters:

• array_spec (ArraySpec) –

  The base spec to be evolved as we thread it.

Returns:

• This codec with the evolved code chain. –

Source code in zarr/codecs/sharding.py

def evolve_from_array_spec(self, array_spec: ArraySpec) -> Self:
    """Thread the spec through the inner chain.

    Each codec is evolved against the spec produced by the previous one.
    Evolving every codec against the same unthreaded spec is the bug shape that
    strips `BytesCodec.endian` behind a dtype-changing codec — and this
    method runs on the real array-creation path, baking the damaged chain
    into the evolved instance before the transform builders ever run.

    Parameters
    ----------
    array_spec
        The base spec to be evolved as we thread it.

    Returns
    -------
        This codec with the evolved code chain.
    """
    from zarr.core.chunk_utils import evolve_codecs

    shard_spec = self._get_chunk_spec(array_spec)
    evolved_codecs = evolve_codecs(self.codecs, shard_spec)
    if evolved_codecs != self.codecs:
        return replace(self, codecs=evolved_codecs)
    return self

from_dict classmethod

from_dict(data: dict[str, JSON]) -> Self

Create an instance of the model from a dictionary

Source code in zarr/codecs/sharding.py

@classmethod
def from_dict(cls, data: dict[str, JSON]) -> Self:
    _, configuration_parsed = parse_named_configuration(data, "sharding_indexed")
    return cls(**configuration_parsed)  # type: ignore[arg-type]

resolve_metadata

resolve_metadata(chunk_spec: ArraySpec) -> ArraySpec

Computed the spec of the chunk after it has been encoded by the codec. This is important for codecs that change the shape, data type or fill value of a chunk. The spec will then be used for subsequent codecs in the pipeline.

Parameters:

• chunk_spec (ArraySpec) –

Returns:

• ArraySpec –

Source code in zarr/abc/codec.py

def resolve_metadata(self, chunk_spec: ArraySpec) -> ArraySpec:
    """Computed the spec of the chunk after it has been encoded by the codec.
    This is important for codecs that change the shape, data type or fill value of a chunk.
    The spec will then be used for subsequent codecs in the pipeline.

    Parameters
    ----------
    chunk_spec : ArraySpec

    Returns
    -------
    ArraySpec
    """
    return chunk_spec

to_dict

to_dict() -> dict[str, JSON]

Recursively serialize this model to a dictionary. This method inspects the fields of self and calls x.to_dict() for any fields that are instances of Metadata. Sequences of Metadata are similarly recursed into, and the output of that recursion is collected in a list.

Source code in zarr/codecs/sharding.py

def to_dict(self) -> dict[str, JSON]:
    return {
        "name": "sharding_indexed",
        "configuration": {
            "chunk_shape": self.chunk_shape,
            "codecs": tuple(s.to_dict() for s in self.codecs),
            "index_codecs": tuple(s.to_dict() for s in self.index_codecs),
            "index_location": self.index_location,
        },
    }

validate

validate(
    *,
    shape: tuple[int, ...],
    dtype: ZDType[TBaseDType, TBaseScalar],
    chunk_grid: ChunkGridMetadata,
) -> None

Validates that the codec configuration is compatible with the array metadata. Raises errors when the codec configuration is not compatible.

Parameters:

• shape (tuple[int, ...]) –

  The array shape

• dtype (dtype[Any]) –

  The array data type

• chunk_grid (ChunkGridMetadata) –

  The array chunk grid metadata

Source code in zarr/codecs/sharding.py

def validate(
    self,
    *,
    shape: tuple[int, ...],
    dtype: ZDType[TBaseDType, TBaseScalar],
    chunk_grid: ChunkGridMetadata,
) -> None:
    if len(self.chunk_shape) != len(shape):
        raise ValueError(
            "The shard's `chunk_shape` and array's `shape` need to have the same number of dimensions."
        )
    if isinstance(chunk_grid, RegularChunkGridMetadata):
        edges_per_dim: tuple[tuple[int, ...], ...] = tuple((s,) for s in chunk_grid.chunk_shape)
    elif isinstance(chunk_grid, RectilinearChunkGridMetadata):
        edges_per_dim = tuple(
            (s,) if isinstance(s, int) else s for s in chunk_grid.chunk_shapes
        )
    else:
        raise TypeError(
            f"Sharding is only compatible with regular and rectilinear chunk grids, "
            f"got {type(chunk_grid)}"
        )
    for i, (edges, inner) in enumerate(zip(edges_per_dim, self.chunk_shape, strict=False)):
        for edge in set(edges):
            if edge % inner != 0:
                raise ValueError(
                    f"Chunk edge length {edge} in dimension {i} is not "
                    f"divisible by the shard's inner chunk size {inner}."
                )

ShardingCodecIndexLocation

Deprecated. Pass a literal string ("start" or "end") directly to ShardingCodec instead.

Source code in zarr/codecs/sharding.py

class ShardingCodecIndexLocation(metaclass=_DeprecatedStrEnumMeta):
    """
    Deprecated. Pass a literal string (`"start"` or `"end"`) directly to
    `ShardingCodec` instead.
    """

    _members: ClassVar[dict[str, str]] = {"start": "start", "end": "end"}

TransposeCodec dataclass

Bases: ArrayArrayCodec

Transpose codec

Source code in zarr/codecs/transpose.py

@dataclass(frozen=True)
class TransposeCodec(ArrayArrayCodec):
    """Transpose codec"""

    is_fixed_size = True

    order: tuple[int, ...]

    def __init__(self, *, order: Iterable[int]) -> None:
        order_parsed = parse_transpose_order(order)

        object.__setattr__(self, "order", order_parsed)

    @classmethod
    def from_dict(cls, data: dict[str, JSON]) -> Self:
        _, configuration_parsed = parse_named_configuration(data, "transpose")
        return cls(**configuration_parsed)  # type: ignore[arg-type]

    def to_dict(self) -> dict[str, JSON]:
        return {"name": "transpose", "configuration": {"order": tuple(self.order)}}

    def validate(
        self,
        shape: tuple[int, ...],
        dtype: ZDType[TBaseDType, TBaseScalar],
        chunk_grid: ChunkGridMetadata,
    ) -> None:
        if len(self.order) != len(shape):
            raise ValueError(
                f"The `order` tuple must have as many entries as there are dimensions in the array. Got {self.order}."
            )
        if len(self.order) != len(set(self.order)):
            raise ValueError(
                f"There must not be duplicates in the `order` tuple. Got {self.order}."
            )
        if not all(0 <= x < len(shape) for x in self.order):
            raise ValueError(
                f"All entries in the `order` tuple must be between 0 and the number of dimensions in the array. Got {self.order}."
            )

    def evolve_from_array_spec(self, array_spec: ArraySpec) -> Self:
        ndim = array_spec.ndim
        if len(self.order) != ndim:
            raise ValueError(
                f"The `order` tuple must have as many entries as there are dimensions in the array. Got {self.order}."
            )
        if len(self.order) != len(set(self.order)):
            raise ValueError(
                f"There must not be duplicates in the `order` tuple. Got {self.order}."
            )
        if not all(0 <= x < ndim for x in self.order):
            raise ValueError(
                f"All entries in the `order` tuple must be between 0 and the number of dimensions in the array. Got {self.order}."
            )
        order = tuple(self.order)

        if order != self.order:
            return replace(self, order=order)
        return self

    def resolve_metadata(self, chunk_spec: ArraySpec) -> ArraySpec:
        return ArraySpec(
            shape=tuple(chunk_spec.shape[self.order[i]] for i in range(chunk_spec.ndim)),
            dtype=chunk_spec.dtype,
            fill_value=chunk_spec.fill_value,
            config=chunk_spec.config,
            prototype=chunk_spec.prototype,
        )

    def _decode_sync(
        self,
        chunk_array: NDBuffer,
        chunk_spec: ArraySpec,
    ) -> NDBuffer:
        inverse_order = tuple(int(i) for i in np.argsort(self.order))
        return chunk_array.transpose(inverse_order)

    async def _decode_single(
        self,
        chunk_array: NDBuffer,
        chunk_spec: ArraySpec,
    ) -> NDBuffer:
        return self._decode_sync(chunk_array, chunk_spec)

    def _encode_sync(
        self,
        chunk_array: NDBuffer,
        _chunk_spec: ArraySpec,
    ) -> NDBuffer | None:
        return chunk_array.transpose(self.order)

    async def _encode_single(
        self,
        chunk_array: NDBuffer,
        _chunk_spec: ArraySpec,
    ) -> NDBuffer | None:
        return self._encode_sync(chunk_array, _chunk_spec)

    def compute_encoded_size(self, input_byte_length: int, _chunk_spec: ArraySpec) -> int:
        return input_byte_length

is_fixed_size class-attribute instance-attribute

is_fixed_size = True

order instance-attribute

order: tuple[int, ...]

__init__

__init__(*, order: Iterable[int]) -> None

Source code in zarr/codecs/transpose.py

def __init__(self, *, order: Iterable[int]) -> None:
    order_parsed = parse_transpose_order(order)

    object.__setattr__(self, "order", order_parsed)

compute_encoded_size

compute_encoded_size(
    input_byte_length: int, _chunk_spec: ArraySpec
) -> int

Given an input byte length, this method returns the output byte length. Raises a NotImplementedError for codecs with variable-sized outputs (e.g. compressors).

Parameters:

• input_byte_length (int) –
• chunk_spec (ArraySpec) –

Returns:

• int –

Source code in zarr/codecs/transpose.py

def compute_encoded_size(self, input_byte_length: int, _chunk_spec: ArraySpec) -> int:
    return input_byte_length

decode async

decode(
    chunks_and_specs: Iterable[tuple[CO | None, ArraySpec]],
) -> Iterable[CI | None]

Decodes a batch of chunks. Chunks can be None in which case they are ignored by the codec.

Parameters:

• chunks_and_specs (Iterable[tuple[CodecOutput | None, ArraySpec]]) –

  Ordered set of encoded chunks with their accompanying chunk spec.

Returns:

• Iterable[CI | None] –

Source code in zarr/abc/codec.py

async def decode(
    self,
    chunks_and_specs: Iterable[tuple[CO | None, ArraySpec]],
) -> Iterable[CI | None]:
    """Decodes a batch of chunks.
    Chunks can be None in which case they are ignored by the codec.

    Parameters
    ----------
    chunks_and_specs : Iterable[tuple[CodecOutput | None, ArraySpec]]
        Ordered set of encoded chunks with their accompanying chunk spec.

    Returns
    -------
    Iterable[CI | None]
    """
    return await _batching_helper(self._decode_single, chunks_and_specs)

encode async

encode(
    chunks_and_specs: Iterable[tuple[CI | None, ArraySpec]],
) -> Iterable[CO | None]

Encodes a batch of chunks. Chunks can be None in which case they are ignored by the codec.

Parameters:

• chunks_and_specs (Iterable[tuple[CI | None, ArraySpec]]) –

  Ordered set of to-be-encoded chunks with their accompanying chunk spec.

Returns:

• Iterable[CodecOutput | None] –

Source code in zarr/abc/codec.py

async def encode(
    self,
    chunks_and_specs: Iterable[tuple[CI | None, ArraySpec]],
) -> Iterable[CO | None]:
    """Encodes a batch of chunks.
    Chunks can be None in which case they are ignored by the codec.

    Parameters
    ----------
    chunks_and_specs : Iterable[tuple[CI | None, ArraySpec]]
        Ordered set of to-be-encoded chunks with their accompanying chunk spec.

    Returns
    -------
    Iterable[CodecOutput | None]
    """
    return await _batching_helper(self._encode_single, chunks_and_specs)

evolve_from_array_spec

evolve_from_array_spec(array_spec: ArraySpec) -> Self

Fills in codec configuration parameters that can be automatically inferred from the array metadata.

Parameters:

• array_spec (ArraySpec) –

Returns:

• Self –

Source code in zarr/codecs/transpose.py

def evolve_from_array_spec(self, array_spec: ArraySpec) -> Self:
    ndim = array_spec.ndim
    if len(self.order) != ndim:
        raise ValueError(
            f"The `order` tuple must have as many entries as there are dimensions in the array. Got {self.order}."
        )
    if len(self.order) != len(set(self.order)):
        raise ValueError(
            f"There must not be duplicates in the `order` tuple. Got {self.order}."
        )
    if not all(0 <= x < ndim for x in self.order):
        raise ValueError(
            f"All entries in the `order` tuple must be between 0 and the number of dimensions in the array. Got {self.order}."
        )
    order = tuple(self.order)

    if order != self.order:
        return replace(self, order=order)
    return self

from_dict classmethod

from_dict(data: dict[str, JSON]) -> Self

Create an instance of the model from a dictionary

Source code in zarr/codecs/transpose.py

@classmethod
def from_dict(cls, data: dict[str, JSON]) -> Self:
    _, configuration_parsed = parse_named_configuration(data, "transpose")
    return cls(**configuration_parsed)  # type: ignore[arg-type]

resolve_metadata

resolve_metadata(chunk_spec: ArraySpec) -> ArraySpec

Computed the spec of the chunk after it has been encoded by the codec. This is important for codecs that change the shape, data type or fill value of a chunk. The spec will then be used for subsequent codecs in the pipeline.

Parameters:

• chunk_spec (ArraySpec) –

Returns:

• ArraySpec –

Source code in zarr/codecs/transpose.py

def resolve_metadata(self, chunk_spec: ArraySpec) -> ArraySpec:
    return ArraySpec(
        shape=tuple(chunk_spec.shape[self.order[i]] for i in range(chunk_spec.ndim)),
        dtype=chunk_spec.dtype,
        fill_value=chunk_spec.fill_value,
        config=chunk_spec.config,
        prototype=chunk_spec.prototype,
    )

to_dict

to_dict() -> dict[str, JSON]

Recursively serialize this model to a dictionary. This method inspects the fields of self and calls x.to_dict() for any fields that are instances of Metadata. Sequences of Metadata are similarly recursed into, and the output of that recursion is collected in a list.

Source code in zarr/codecs/transpose.py

def to_dict(self) -> dict[str, JSON]:
    return {"name": "transpose", "configuration": {"order": tuple(self.order)}}

validate

validate(
    shape: tuple[int, ...],
    dtype: ZDType[TBaseDType, TBaseScalar],
    chunk_grid: ChunkGridMetadata,
) -> None

Validates that the codec configuration is compatible with the array metadata. Raises errors when the codec configuration is not compatible.

Parameters:

• shape (tuple[int, ...]) –

  The array shape

• dtype (dtype[Any]) –

  The array data type

• chunk_grid (ChunkGridMetadata) –

  The array chunk grid metadata

Source code in zarr/codecs/transpose.py

def validate(
    self,
    shape: tuple[int, ...],
    dtype: ZDType[TBaseDType, TBaseScalar],
    chunk_grid: ChunkGridMetadata,
) -> None:
    if len(self.order) != len(shape):
        raise ValueError(
            f"The `order` tuple must have as many entries as there are dimensions in the array. Got {self.order}."
        )
    if len(self.order) != len(set(self.order)):
        raise ValueError(
            f"There must not be duplicates in the `order` tuple. Got {self.order}."
        )
    if not all(0 <= x < len(shape) for x in self.order):
        raise ValueError(
            f"All entries in the `order` tuple must be between 0 and the number of dimensions in the array. Got {self.order}."
        )

VLenBytesCodec dataclass

Bases: ArrayBytesCodec

Source code in zarr/codecs/vlen_utf8.py

@dataclass(frozen=True)
class VLenBytesCodec(ArrayBytesCodec):
    @classmethod
    def from_dict(cls, data: dict[str, JSON]) -> Self:
        _, configuration_parsed = parse_named_configuration(
            data, "vlen-bytes", require_configuration=False
        )
        configuration_parsed = configuration_parsed or {}
        return cls(**configuration_parsed)

    def to_dict(self) -> dict[str, JSON]:
        return {"name": "vlen-bytes", "configuration": {}}

    def evolve_from_array_spec(self, array_spec: ArraySpec) -> Self:
        return self

    def _decode_sync(
        self,
        chunk_bytes: Buffer,
        chunk_spec: ArraySpec,
    ) -> NDBuffer:
        assert isinstance(chunk_bytes, Buffer)

        raw_bytes = chunk_bytes.as_array_like()
        decoded = _vlen_bytes_codec.decode(raw_bytes)
        assert decoded.dtype == np.object_
        decoded = _reshape_view(decoded, chunk_spec.shape)
        return chunk_spec.prototype.nd_buffer.from_numpy_array(decoded)

    async def _decode_single(
        self,
        chunk_bytes: Buffer,
        chunk_spec: ArraySpec,
    ) -> NDBuffer:
        return self._decode_sync(chunk_bytes, chunk_spec)

    def _encode_sync(
        self,
        chunk_array: NDBuffer,
        chunk_spec: ArraySpec,
    ) -> Buffer | None:
        assert isinstance(chunk_array, NDBuffer)
        # numcodecs vlen codecs flatten with order="A", so an F-contiguous chunk
        # would be encoded in transposed element order (gh-3558)
        return chunk_spec.prototype.buffer.from_bytes(
            _vlen_bytes_codec.encode(np.ascontiguousarray(chunk_array.as_numpy_array()))
        )

    async def _encode_single(
        self,
        chunk_array: NDBuffer,
        chunk_spec: ArraySpec,
    ) -> Buffer | None:
        return self._encode_sync(chunk_array, chunk_spec)

    def compute_encoded_size(self, input_byte_length: int, _chunk_spec: ArraySpec) -> int:
        # what is input_byte_length for an object dtype?
        raise NotImplementedError("compute_encoded_size is not implemented for VLen codecs")

is_fixed_size class-attribute instance-attribute

is_fixed_size: bool = False

__init__

__init__() -> None

compute_encoded_size

compute_encoded_size(
    input_byte_length: int, _chunk_spec: ArraySpec
) -> int

Given an input byte length, this method returns the output byte length. Raises a NotImplementedError for codecs with variable-sized outputs (e.g. compressors).

Parameters:

• input_byte_length (int) –
• chunk_spec (ArraySpec) –

Returns:

• int –

Source code in zarr/codecs/vlen_utf8.py

def compute_encoded_size(self, input_byte_length: int, _chunk_spec: ArraySpec) -> int:
    # what is input_byte_length for an object dtype?
    raise NotImplementedError("compute_encoded_size is not implemented for VLen codecs")

decode async

decode(
    chunks_and_specs: Iterable[tuple[CO | None, ArraySpec]],
) -> Iterable[CI | None]

Decodes a batch of chunks. Chunks can be None in which case they are ignored by the codec.

Parameters:

• chunks_and_specs (Iterable[tuple[CodecOutput | None, ArraySpec]]) –

  Ordered set of encoded chunks with their accompanying chunk spec.

Returns:

• Iterable[CI | None] –

Source code in zarr/abc/codec.py

async def decode(
    self,
    chunks_and_specs: Iterable[tuple[CO | None, ArraySpec]],
) -> Iterable[CI | None]:
    """Decodes a batch of chunks.
    Chunks can be None in which case they are ignored by the codec.

    Parameters
    ----------
    chunks_and_specs : Iterable[tuple[CodecOutput | None, ArraySpec]]
        Ordered set of encoded chunks with their accompanying chunk spec.

    Returns
    -------
    Iterable[CI | None]
    """
    return await _batching_helper(self._decode_single, chunks_and_specs)

encode async

encode(
    chunks_and_specs: Iterable[tuple[CI | None, ArraySpec]],
) -> Iterable[CO | None]

Encodes a batch of chunks. Chunks can be None in which case they are ignored by the codec.

Parameters:

• chunks_and_specs (Iterable[tuple[CI | None, ArraySpec]]) –

  Ordered set of to-be-encoded chunks with their accompanying chunk spec.

Returns:

• Iterable[CodecOutput | None] –

Source code in zarr/abc/codec.py

async def encode(
    self,
    chunks_and_specs: Iterable[tuple[CI | None, ArraySpec]],
) -> Iterable[CO | None]:
    """Encodes a batch of chunks.
    Chunks can be None in which case they are ignored by the codec.

    Parameters
    ----------
    chunks_and_specs : Iterable[tuple[CI | None, ArraySpec]]
        Ordered set of to-be-encoded chunks with their accompanying chunk spec.

    Returns
    -------
    Iterable[CodecOutput | None]
    """
    return await _batching_helper(self._encode_single, chunks_and_specs)

evolve_from_array_spec

evolve_from_array_spec(array_spec: ArraySpec) -> Self

Fills in codec configuration parameters that can be automatically inferred from the array metadata.

Parameters:

• array_spec (ArraySpec) –

Returns:

• Self –

Source code in zarr/codecs/vlen_utf8.py

def evolve_from_array_spec(self, array_spec: ArraySpec) -> Self:
    return self

from_dict classmethod

from_dict(data: dict[str, JSON]) -> Self

Create an instance of the model from a dictionary

Source code in zarr/codecs/vlen_utf8.py

@classmethod
def from_dict(cls, data: dict[str, JSON]) -> Self:
    _, configuration_parsed = parse_named_configuration(
        data, "vlen-bytes", require_configuration=False
    )
    configuration_parsed = configuration_parsed or {}
    return cls(**configuration_parsed)

resolve_metadata

resolve_metadata(chunk_spec: ArraySpec) -> ArraySpec

Computed the spec of the chunk after it has been encoded by the codec. This is important for codecs that change the shape, data type or fill value of a chunk. The spec will then be used for subsequent codecs in the pipeline.

Parameters:

• chunk_spec (ArraySpec) –

Returns:

• ArraySpec –

Source code in zarr/abc/codec.py

def resolve_metadata(self, chunk_spec: ArraySpec) -> ArraySpec:
    """Computed the spec of the chunk after it has been encoded by the codec.
    This is important for codecs that change the shape, data type or fill value of a chunk.
    The spec will then be used for subsequent codecs in the pipeline.

    Parameters
    ----------
    chunk_spec : ArraySpec

    Returns
    -------
    ArraySpec
    """
    return chunk_spec

to_dict

to_dict() -> dict[str, JSON]

Recursively serialize this model to a dictionary. This method inspects the fields of self and calls x.to_dict() for any fields that are instances of Metadata. Sequences of Metadata are similarly recursed into, and the output of that recursion is collected in a list.

Source code in zarr/codecs/vlen_utf8.py

def to_dict(self) -> dict[str, JSON]:
    return {"name": "vlen-bytes", "configuration": {}}

validate

validate(
    *,
    shape: tuple[int, ...],
    dtype: ZDType[TBaseDType, TBaseScalar],
    chunk_grid: ChunkGridMetadata,
) -> None

Validates that the codec configuration is compatible with the array metadata. Raises errors when the codec configuration is not compatible.

Parameters:

• shape (tuple[int, ...]) –

  The array shape

• dtype (dtype[Any]) –

  The array data type

• chunk_grid (ChunkGridMetadata) –

  The array chunk grid metadata

Source code in zarr/abc/codec.py

def validate(
    self,
    *,
    shape: tuple[int, ...],
    dtype: ZDType[TBaseDType, TBaseScalar],
    chunk_grid: ChunkGridMetadata,
) -> None:
    """Validates that the codec configuration is compatible with the array metadata.
    Raises errors when the codec configuration is not compatible.

    Parameters
    ----------
    shape : tuple[int, ...]
        The array shape
    dtype : np.dtype[Any]
        The array data type
    chunk_grid : ChunkGridMetadata
        The array chunk grid metadata
    """

VLenUTF8Codec dataclass

Bases: ArrayBytesCodec

Variable-length UTF8 codec

Source code in zarr/codecs/vlen_utf8.py

@dataclass(frozen=True)
class VLenUTF8Codec(ArrayBytesCodec):
    """Variable-length UTF8 codec"""

    @classmethod
    def from_dict(cls, data: dict[str, JSON]) -> Self:
        _, configuration_parsed = parse_named_configuration(
            data, "vlen-utf8", require_configuration=False
        )
        configuration_parsed = configuration_parsed or {}
        return cls(**configuration_parsed)

    def to_dict(self) -> dict[str, JSON]:
        return {"name": "vlen-utf8", "configuration": {}}

    def evolve_from_array_spec(self, array_spec: ArraySpec) -> Self:
        return self

    def _decode_sync(
        self,
        chunk_bytes: Buffer,
        chunk_spec: ArraySpec,
    ) -> NDBuffer:
        assert isinstance(chunk_bytes, Buffer)

        raw_bytes = chunk_bytes.as_array_like()
        decoded = _vlen_utf8_codec.decode(raw_bytes)
        assert decoded.dtype == np.object_
        decoded = _reshape_view(decoded, chunk_spec.shape)
        as_string_dtype = decoded.astype(chunk_spec.dtype.to_native_dtype(), copy=False)
        return chunk_spec.prototype.nd_buffer.from_numpy_array(as_string_dtype)

    async def _decode_single(
        self,
        chunk_bytes: Buffer,
        chunk_spec: ArraySpec,
    ) -> NDBuffer:
        return self._decode_sync(chunk_bytes, chunk_spec)

    def _encode_sync(
        self,
        chunk_array: NDBuffer,
        chunk_spec: ArraySpec,
    ) -> Buffer | None:
        assert isinstance(chunk_array, NDBuffer)
        # numcodecs vlen codecs flatten with order="A", so an F-contiguous chunk
        # would be encoded in transposed element order (gh-3558)
        return chunk_spec.prototype.buffer.from_bytes(
            _vlen_utf8_codec.encode(np.ascontiguousarray(chunk_array.as_numpy_array()))
        )

    async def _encode_single(
        self,
        chunk_array: NDBuffer,
        chunk_spec: ArraySpec,
    ) -> Buffer | None:
        return self._encode_sync(chunk_array, chunk_spec)

    def compute_encoded_size(self, input_byte_length: int, _chunk_spec: ArraySpec) -> int:
        # what is input_byte_length for an object dtype?
        raise NotImplementedError("compute_encoded_size is not implemented for VLen codecs")

is_fixed_size class-attribute instance-attribute

is_fixed_size: bool = False

__init__

__init__() -> None

compute_encoded_size

compute_encoded_size(
    input_byte_length: int, _chunk_spec: ArraySpec
) -> int

Given an input byte length, this method returns the output byte length. Raises a NotImplementedError for codecs with variable-sized outputs (e.g. compressors).

Parameters:

• input_byte_length (int) –
• chunk_spec (ArraySpec) –

Returns:

• int –

Source code in zarr/codecs/vlen_utf8.py

def compute_encoded_size(self, input_byte_length: int, _chunk_spec: ArraySpec) -> int:
    # what is input_byte_length for an object dtype?
    raise NotImplementedError("compute_encoded_size is not implemented for VLen codecs")

decode async

decode(
    chunks_and_specs: Iterable[tuple[CO | None, ArraySpec]],
) -> Iterable[CI | None]

Decodes a batch of chunks. Chunks can be None in which case they are ignored by the codec.

Parameters:

• chunks_and_specs (Iterable[tuple[CodecOutput | None, ArraySpec]]) –

  Ordered set of encoded chunks with their accompanying chunk spec.

Returns:

• Iterable[CI | None] –

Source code in zarr/abc/codec.py

async def decode(
    self,
    chunks_and_specs: Iterable[tuple[CO | None, ArraySpec]],
) -> Iterable[CI | None]:
    """Decodes a batch of chunks.
    Chunks can be None in which case they are ignored by the codec.

    Parameters
    ----------
    chunks_and_specs : Iterable[tuple[CodecOutput | None, ArraySpec]]
        Ordered set of encoded chunks with their accompanying chunk spec.

    Returns
    -------
    Iterable[CI | None]
    """
    return await _batching_helper(self._decode_single, chunks_and_specs)

encode async

encode(
    chunks_and_specs: Iterable[tuple[CI | None, ArraySpec]],
) -> Iterable[CO | None]

Encodes a batch of chunks. Chunks can be None in which case they are ignored by the codec.

Parameters:

• chunks_and_specs (Iterable[tuple[CI | None, ArraySpec]]) –

  Ordered set of to-be-encoded chunks with their accompanying chunk spec.

Returns:

• Iterable[CodecOutput | None] –

Source code in zarr/abc/codec.py

async def encode(
    self,
    chunks_and_specs: Iterable[tuple[CI | None, ArraySpec]],
) -> Iterable[CO | None]:
    """Encodes a batch of chunks.
    Chunks can be None in which case they are ignored by the codec.

    Parameters
    ----------
    chunks_and_specs : Iterable[tuple[CI | None, ArraySpec]]
        Ordered set of to-be-encoded chunks with their accompanying chunk spec.

    Returns
    -------
    Iterable[CodecOutput | None]
    """
    return await _batching_helper(self._encode_single, chunks_and_specs)

evolve_from_array_spec

evolve_from_array_spec(array_spec: ArraySpec) -> Self

Fills in codec configuration parameters that can be automatically inferred from the array metadata.

Parameters:

• array_spec (ArraySpec) –

Returns:

• Self –

Source code in zarr/codecs/vlen_utf8.py

def evolve_from_array_spec(self, array_spec: ArraySpec) -> Self:
    return self

from_dict classmethod

from_dict(data: dict[str, JSON]) -> Self

Create an instance of the model from a dictionary

Source code in zarr/codecs/vlen_utf8.py

@classmethod
def from_dict(cls, data: dict[str, JSON]) -> Self:
    _, configuration_parsed = parse_named_configuration(
        data, "vlen-utf8", require_configuration=False
    )
    configuration_parsed = configuration_parsed or {}
    return cls(**configuration_parsed)

resolve_metadata

resolve_metadata(chunk_spec: ArraySpec) -> ArraySpec

Computed the spec of the chunk after it has been encoded by the codec. This is important for codecs that change the shape, data type or fill value of a chunk. The spec will then be used for subsequent codecs in the pipeline.

Parameters:

• chunk_spec (ArraySpec) –

Returns:

• ArraySpec –

Source code in zarr/abc/codec.py

def resolve_metadata(self, chunk_spec: ArraySpec) -> ArraySpec:
    """Computed the spec of the chunk after it has been encoded by the codec.
    This is important for codecs that change the shape, data type or fill value of a chunk.
    The spec will then be used for subsequent codecs in the pipeline.

    Parameters
    ----------
    chunk_spec : ArraySpec

    Returns
    -------
    ArraySpec
    """
    return chunk_spec

to_dict

to_dict() -> dict[str, JSON]

Recursively serialize this model to a dictionary. This method inspects the fields of self and calls x.to_dict() for any fields that are instances of Metadata. Sequences of Metadata are similarly recursed into, and the output of that recursion is collected in a list.

Source code in zarr/codecs/vlen_utf8.py

def to_dict(self) -> dict[str, JSON]:
    return {"name": "vlen-utf8", "configuration": {}}

validate

validate(
    *,
    shape: tuple[int, ...],
    dtype: ZDType[TBaseDType, TBaseScalar],
    chunk_grid: ChunkGridMetadata,
) -> None

Validates that the codec configuration is compatible with the array metadata. Raises errors when the codec configuration is not compatible.

Parameters:

• shape (tuple[int, ...]) –

  The array shape

• dtype (dtype[Any]) –

  The array data type

• chunk_grid (ChunkGridMetadata) –

  The array chunk grid metadata

Source code in zarr/abc/codec.py

def validate(
    self,
    *,
    shape: tuple[int, ...],
    dtype: ZDType[TBaseDType, TBaseScalar],
    chunk_grid: ChunkGridMetadata,
) -> None:
    """Validates that the codec configuration is compatible with the array metadata.
    Raises errors when the codec configuration is not compatible.

    Parameters
    ----------
    shape : tuple[int, ...]
        The array shape
    dtype : np.dtype[Any]
        The array data type
    chunk_grid : ChunkGridMetadata
        The array chunk grid metadata
    """

ZstdCodec dataclass

Bases: BytesBytesCodec

zstd codec

Source code in zarr/codecs/zstd.py

@dataclass(frozen=True)
class ZstdCodec(BytesBytesCodec):
    """zstd codec"""

    is_fixed_size = False

    level: int = 0
    checksum: bool = False

    def __init__(self, *, level: int = 0, checksum: bool = False) -> None:
        # numcodecs 0.13.0 introduces the checksum attribute for the zstd codec
        _numcodecs_version = Version(numcodecs.__version__)
        if _numcodecs_version < Version("0.13.0"):
            raise RuntimeError(
                "numcodecs version >= 0.13.0 is required to use the zstd codec. "
                f"Version {_numcodecs_version} is currently installed."
            )

        level_parsed = parse_zstd_level(level)
        checksum_parsed = parse_checksum(checksum)

        object.__setattr__(self, "level", level_parsed)
        object.__setattr__(self, "checksum", checksum_parsed)

    @classmethod
    def from_dict(cls, data: dict[str, JSON]) -> Self:
        _, configuration_parsed = parse_named_configuration(data, "zstd")
        return cls(**configuration_parsed)  # type: ignore[arg-type]

    def to_dict(self) -> dict[str, JSON]:
        return {"name": "zstd", "configuration": {"level": self.level, "checksum": self.checksum}}

    @cached_property
    def _zstd_codec(self) -> Zstd:
        config_dict = {"level": self.level, "checksum": self.checksum}
        return Zstd.from_config(config_dict)

    def _decode_sync(
        self,
        chunk_bytes: Buffer,
        chunk_spec: ArraySpec,
    ) -> Buffer:
        return as_numpy_array_wrapper(self._zstd_codec.decode, chunk_bytes, chunk_spec.prototype)

    async def _decode_single(
        self,
        chunk_bytes: Buffer,
        chunk_spec: ArraySpec,
    ) -> Buffer:
        return await asyncio.to_thread(self._decode_sync, chunk_bytes, chunk_spec)

    def _encode_sync(
        self,
        chunk_bytes: Buffer,
        chunk_spec: ArraySpec,
    ) -> Buffer | None:
        return as_numpy_array_wrapper(self._zstd_codec.encode, chunk_bytes, chunk_spec.prototype)

    async def _encode_single(
        self,
        chunk_bytes: Buffer,
        chunk_spec: ArraySpec,
    ) -> Buffer | None:
        return await asyncio.to_thread(self._encode_sync, chunk_bytes, chunk_spec)

    def compute_encoded_size(self, _input_byte_length: int, _chunk_spec: ArraySpec) -> int:
        raise NotImplementedError

checksum class-attribute instance-attribute

checksum: bool = False

is_fixed_size class-attribute instance-attribute

is_fixed_size = False

level class-attribute instance-attribute

level: int = 0

__init__

__init__(*, level: int = 0, checksum: bool = False) -> None

Source code in zarr/codecs/zstd.py

def __init__(self, *, level: int = 0, checksum: bool = False) -> None:
    # numcodecs 0.13.0 introduces the checksum attribute for the zstd codec
    _numcodecs_version = Version(numcodecs.__version__)
    if _numcodecs_version < Version("0.13.0"):
        raise RuntimeError(
            "numcodecs version >= 0.13.0 is required to use the zstd codec. "
            f"Version {_numcodecs_version} is currently installed."
        )

    level_parsed = parse_zstd_level(level)
    checksum_parsed = parse_checksum(checksum)

    object.__setattr__(self, "level", level_parsed)
    object.__setattr__(self, "checksum", checksum_parsed)

compute_encoded_size

compute_encoded_size(
    _input_byte_length: int, _chunk_spec: ArraySpec
) -> int

Given an input byte length, this method returns the output byte length. Raises a NotImplementedError for codecs with variable-sized outputs (e.g. compressors).

Parameters:

• input_byte_length (int) –
• chunk_spec (ArraySpec) –

Returns:

• int –

Source code in zarr/codecs/zstd.py

def compute_encoded_size(self, _input_byte_length: int, _chunk_spec: ArraySpec) -> int:
    raise NotImplementedError

decode async

decode(
    chunks_and_specs: Iterable[tuple[CO | None, ArraySpec]],
) -> Iterable[CI | None]

Decodes a batch of chunks. Chunks can be None in which case they are ignored by the codec.

Parameters:

• chunks_and_specs (Iterable[tuple[CodecOutput | None, ArraySpec]]) –

  Ordered set of encoded chunks with their accompanying chunk spec.

Returns:

• Iterable[CI | None] –

Source code in zarr/abc/codec.py

async def decode(
    self,
    chunks_and_specs: Iterable[tuple[CO | None, ArraySpec]],
) -> Iterable[CI | None]:
    """Decodes a batch of chunks.
    Chunks can be None in which case they are ignored by the codec.

    Parameters
    ----------
    chunks_and_specs : Iterable[tuple[CodecOutput | None, ArraySpec]]
        Ordered set of encoded chunks with their accompanying chunk spec.

    Returns
    -------
    Iterable[CI | None]
    """
    return await _batching_helper(self._decode_single, chunks_and_specs)

encode async

encode(
    chunks_and_specs: Iterable[tuple[CI | None, ArraySpec]],
) -> Iterable[CO | None]

Encodes a batch of chunks. Chunks can be None in which case they are ignored by the codec.

Parameters:

• chunks_and_specs (Iterable[tuple[CI | None, ArraySpec]]) –

  Ordered set of to-be-encoded chunks with their accompanying chunk spec.

Returns:

• Iterable[CodecOutput | None] –

Source code in zarr/abc/codec.py

async def encode(
    self,
    chunks_and_specs: Iterable[tuple[CI | None, ArraySpec]],
) -> Iterable[CO | None]:
    """Encodes a batch of chunks.
    Chunks can be None in which case they are ignored by the codec.

    Parameters
    ----------
    chunks_and_specs : Iterable[tuple[CI | None, ArraySpec]]
        Ordered set of to-be-encoded chunks with their accompanying chunk spec.

    Returns
    -------
    Iterable[CodecOutput | None]
    """
    return await _batching_helper(self._encode_single, chunks_and_specs)

evolve_from_array_spec

evolve_from_array_spec(array_spec: ArraySpec) -> Self

Fills in codec configuration parameters that can be automatically inferred from the array metadata.

Parameters:

• array_spec (ArraySpec) –

Returns:

• Self –

Source code in zarr/abc/codec.py

def evolve_from_array_spec(self, array_spec: ArraySpec) -> Self:
    """Fills in codec configuration parameters that can be automatically
    inferred from the array metadata.

    Parameters
    ----------
    array_spec : ArraySpec

    Returns
    -------
    Self
    """
    return self

from_dict classmethod

from_dict(data: dict[str, JSON]) -> Self

Create an instance of the model from a dictionary

Source code in zarr/codecs/zstd.py

@classmethod
def from_dict(cls, data: dict[str, JSON]) -> Self:
    _, configuration_parsed = parse_named_configuration(data, "zstd")
    return cls(**configuration_parsed)  # type: ignore[arg-type]

resolve_metadata

resolve_metadata(chunk_spec: ArraySpec) -> ArraySpec

Computed the spec of the chunk after it has been encoded by the codec. This is important for codecs that change the shape, data type or fill value of a chunk. The spec will then be used for subsequent codecs in the pipeline.

Parameters:

• chunk_spec (ArraySpec) –

Returns:

• ArraySpec –

Source code in zarr/abc/codec.py

def resolve_metadata(self, chunk_spec: ArraySpec) -> ArraySpec:
    """Computed the spec of the chunk after it has been encoded by the codec.
    This is important for codecs that change the shape, data type or fill value of a chunk.
    The spec will then be used for subsequent codecs in the pipeline.

    Parameters
    ----------
    chunk_spec : ArraySpec

    Returns
    -------
    ArraySpec
    """
    return chunk_spec

to_dict

to_dict() -> dict[str, JSON]

Recursively serialize this model to a dictionary. This method inspects the fields of self and calls x.to_dict() for any fields that are instances of Metadata. Sequences of Metadata are similarly recursed into, and the output of that recursion is collected in a list.

Source code in zarr/codecs/zstd.py

def to_dict(self) -> dict[str, JSON]:
    return {"name": "zstd", "configuration": {"level": self.level, "checksum": self.checksum}}

validate

validate(
    *,
    shape: tuple[int, ...],
    dtype: ZDType[TBaseDType, TBaseScalar],
    chunk_grid: ChunkGridMetadata,
) -> None

Validates that the codec configuration is compatible with the array metadata. Raises errors when the codec configuration is not compatible.

Parameters:

• shape (tuple[int, ...]) –

  The array shape

• dtype (dtype[Any]) –

  The array data type

• chunk_grid (ChunkGridMetadata) –

  The array chunk grid metadata

Source code in zarr/abc/codec.py

def validate(
    self,
    *,
    shape: tuple[int, ...],
    dtype: ZDType[TBaseDType, TBaseScalar],
    chunk_grid: ChunkGridMetadata,
) -> None:
    """Validates that the codec configuration is compatible with the array metadata.
    Raises errors when the codec configuration is not compatible.

    Parameters
    ----------
    shape : tuple[int, ...]
        The array shape
    dtype : np.dtype[Any]
        The array data type
    chunk_grid : ChunkGridMetadata
        The array chunk grid metadata
    """