remove codecchain

Author: d-v-b

src/zarr/core/codec_pipeline.py

@@ -1,8 +1,8 @@
 from __future__ import annotations
 
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from itertools import islice, pairwise
-from typing import TYPE_CHECKING, Any, TypeVar, cast
+from typing import TYPE_CHECKING, Any, TypeVar
 from warnings import warn
 
 from zarr.abc.codec import (
@@ -13,7 +13,6 @@
     BytesBytesCodec,
     Codec,
     CodecPipeline,
-    SupportsSyncCodec,
 )
 from zarr.core.common import concurrent_map
 from zarr.core.config import config
@@ -69,106 +68,6 @@ def fill_value_or_default(chunk_spec: ArraySpec) -> Any:
         return fill_value
 
 
-@dataclass(frozen=True, slots=True)
-class CodecChain:
-    """Codec chain with pre-resolved metadata specs.
-
-    Constructed from an iterable of codecs and a chunk ArraySpec.
-    Resolves each codec against the spec so that encode/decode can
-    run without re-resolving.
-    """
-
-    codecs: tuple[Codec, ...]
-    chunk_spec: ArraySpec
-
-    _aa_codecs: tuple[ArrayArrayCodec, ...] = field(init=False, repr=False, compare=False)
-    _aa_specs: tuple[ArraySpec, ...] = field(init=False, repr=False, compare=False)
-    _ab_codec: ArrayBytesCodec = field(init=False, repr=False, compare=False)
-    _ab_spec: ArraySpec = field(init=False, repr=False, compare=False)
-    _bb_codecs: tuple[BytesBytesCodec, ...] = field(init=False, repr=False, compare=False)
-    _bb_spec: ArraySpec = field(init=False, repr=False, compare=False)
-    _all_sync: bool = field(init=False, repr=False, compare=False)
-
-    def __post_init__(self) -> None:
-        aa, ab, bb = codecs_from_list(list(self.codecs))
-
-        aa_specs: list[ArraySpec] = []
-        spec = self.chunk_spec
-        for aa_codec in aa:
-            aa_specs.append(spec)
-            spec = aa_codec.resolve_metadata(spec)
-
-        object.__setattr__(self, "_aa_codecs", aa)
-        object.__setattr__(self, "_aa_specs", tuple(aa_specs))
-        object.__setattr__(self, "_ab_codec", ab)
-        object.__setattr__(self, "_ab_spec", spec)
-
-        spec = ab.resolve_metadata(spec)
-        object.__setattr__(self, "_bb_codecs", bb)
-        object.__setattr__(self, "_bb_spec", spec)
-
-        object.__setattr__(
-            self,
-            "_all_sync",
-            all(isinstance(c, SupportsSyncCodec) for c in self.codecs),
-        )
-
-    @property
-    def all_sync(self) -> bool:
-        return self._all_sync
-
-    def decode_chunk(
-        self,
-        chunk_bytes: Buffer,
-    ) -> NDBuffer:
-        """Decode a single chunk through the full codec chain, synchronously.
-
-        Pure compute -- no IO. Only callable when all codecs support sync.
-        """
-        bb_out: Any = chunk_bytes
-        for bb_codec in reversed(self._bb_codecs):
-            bb_out = cast("SupportsSyncCodec", bb_codec)._decode_sync(bb_out, self._bb_spec)
-
-        ab_out: Any = cast("SupportsSyncCodec", self._ab_codec)._decode_sync(bb_out, self._ab_spec)
-
-        for aa_codec, spec in zip(reversed(self._aa_codecs), reversed(self._aa_specs), strict=True):
-            ab_out = cast("SupportsSyncCodec", aa_codec)._decode_sync(ab_out, spec)
-
-        return ab_out  # type: ignore[no-any-return]
-
-    def encode_chunk(
-        self,
-        chunk_array: NDBuffer,
-    ) -> Buffer | None:
-        """Encode a single chunk through the full codec chain, synchronously.
-
-        Pure compute -- no IO. Only callable when all codecs support sync.
-        """
-        aa_out: Any = chunk_array
-
-        for aa_codec, spec in zip(self._aa_codecs, self._aa_specs, strict=True):
-            if aa_out is None:
-                return None
-            aa_out = cast("SupportsSyncCodec", aa_codec)._encode_sync(aa_out, spec)
-
-        if aa_out is None:
-            return None
-        bb_out: Any = cast("SupportsSyncCodec", self._ab_codec)._encode_sync(aa_out, self._ab_spec)
-
-        for bb_codec in self._bb_codecs:
-            if bb_out is None:
-                return None
-            bb_out = cast("SupportsSyncCodec", bb_codec)._encode_sync(bb_out, self._bb_spec)
-
-        return bb_out  # type: ignore[no-any-return]
-
-    def compute_encoded_size(self, byte_length: int, array_spec: ArraySpec) -> int:
-        for codec in self.codecs:
-            byte_length = codec.compute_encoded_size(byte_length, array_spec)
-            array_spec = codec.resolve_metadata(array_spec)
-        return byte_length
-
-
 @dataclass(frozen=True)
 class BatchedCodecPipeline(CodecPipeline):
     """Default codec pipeline.