zarr_indexing.transform

An IndexTransform is a function between coordinate spaces, and its field names follow the function, not the data:

the API says	in array terms
input space (domain, input_rank)	request coordinates — the result being built
output space (output, one map per dimension)	source coordinates — where values are read

output is not data: it is the rule, per source dimension, for producing coordinates. Values flow source → request, against the arrow. The guide demonstrates each output map form against its NumPy counterpart.

zarr_indexing.transform

Index transforms — composable, lazy coordinate mappings.

An IndexTransform pairs an input domain (the coordinates a user sees) with a tuple of output maps (the output coordinates those inputs map to). One output map per output dimension. See output_map.py for the three output map types.

Key operations:

• Indexing (transform[2:8], .oindex[idx], .vindex[idx]) — produces a new transform with a narrower input domain and adjusted output maps. No I/O occurs. This is how lazy slicing works.

• intersect(output_domain) — restrict to output coordinates within a region. This is chunk resolution: "which of my coordinates fall in this chunk?"

• translate(shift) — shift all output coordinates. This makes coordinates chunk-local: "express my coordinates relative to the chunk origin."

• transform.compose(inner) — chain two transforms into one.

The transform is the atomic unit that connects user-facing indexing to chunk-level I/O. A wrapper holds one — LazyArray starts from the identity — and .lazy[...] composes a new transform lazily rather than reading. Reading resolves the transform against the chunk grid via intersect + translate.

IndexTransform dataclass

A composable mapping from input coordinates to output coordinates.

An IndexTransform has:

• domain: an IndexDomain describing the valid input coordinates (the result's coordinate range, possibly with non-zero origin).
• output: a tuple of output maps (one per output dimension), each describing which output coordinates the inputs touch.

In array-indexing terms: domain describes the coordinates of the result array an indexing operation produces, and output is the rule relating each result coordinate to a coordinate in the source. Note the direction — the transform's input side is the result, its output side addresses the source; the coordinate mapping runs opposite to the data flow.

Indexing an existing transform composes a new one without I/O.

Examples:

The operation "every other element of a 100-element array, starting at index 0" — array[::2] — is a 50-cell domain whose cell i reads output coordinate 2 * i:

>>> domain = IndexDomain.from_shape((50,))
>>> output = (DimensionMap(input_dimension=0, offset=0, stride=2),)
>>> transform = IndexTransform(domain=domain, output=output)
>>> transform.apply((0,)), transform.apply((1,)), transform.apply((49,))
((0,), (2,), (98,))

The selection compiler derives the identical transform from the source's shape and the slice:

>>> transform == IndexTransform.from_shape((100,))[::2]
True

Source code in src/zarr_indexing/transform.py

@dataclass(frozen=True, slots=True)
class IndexTransform:
    """A composable mapping from input coordinates to output coordinates.

    An `IndexTransform` has:

    - `domain`: an `IndexDomain` describing the valid input coordinates
      (the result's coordinate range, possibly with non-zero origin).
    - `output`: a tuple of output maps (one per output dimension), each
      describing which output coordinates the inputs touch.

    In array-indexing terms: `domain` describes the coordinates of the result
    array an indexing operation produces, and `output` is the rule relating
    each result coordinate to a coordinate in the source. Note the direction —
    the transform's input side is the result, its output side addresses the
    source; the coordinate mapping runs opposite to the data flow.

    Indexing an existing transform composes a new one without I/O.

    Examples
    --------
    The operation "every other element of a 100-element array, starting at
    index 0" — `array[::2]` — is a 50-cell domain whose cell `i` reads
    output coordinate `2 * i`:

    >>> domain = IndexDomain.from_shape((50,))
    >>> output = (DimensionMap(input_dimension=0, offset=0, stride=2),)
    >>> transform = IndexTransform(domain=domain, output=output)
    >>> transform.apply((0,)), transform.apply((1,)), transform.apply((49,))
    ((0,), (2,), (98,))

    The selection compiler derives the identical transform from the source's
    shape and the slice:

    >>> transform == IndexTransform.from_shape((100,))[::2]
    True
    """

    domain: IndexDomain
    """The input domain: the request coordinates this transform accepts."""

    output: tuple[OutputIndexMap, ...]
    """One output map per output dimension, each producing that dimension's coordinate."""

    def __post_init__(self) -> None:
        for i, m in enumerate(self.output):
            if isinstance(m, DimensionMap):
                if m.input_dimension < 0 or m.input_dimension >= self.domain.ndim:
                    raise ValueError(
                        f"output[{i}].input_dimension = {m.input_dimension} "
                        f"is out of range for input rank {self.domain.ndim}"
                    )
            elif isinstance(m, ArrayMap):
                # An index array carries the transform's full input rank: the axis
                # a map varies over is full-sized, every other axis a singleton.
                # The rank is what makes the dependency axes readable from the
                # shape, so a mismatch is a bug rather than a spelling. External
                # JSON may use a lower-rank array that broadcasts against the
                # domain; `from_json` widens those on the way in,
                # so the invariant holds for every transform that exists.
                if m.index_array.ndim != self.domain.ndim:
                    raise ValueError(
                        f"output[{i}].index_array has {m.index_array.ndim} dims "
                        f"but input domain has {self.domain.ndim} dims"
                    )
                # Every axis is either the domain's extent or a singleton it
                # broadcasts over. Any other size addresses input coordinates the
                # array has no entry for, which reads as a smaller selection
                # rather than as the error it is.
                bad = [
                    (axis, size, extent)
                    for axis, (size, extent) in enumerate(
                        zip(m.index_array.shape, self.domain.shape, strict=True)
                    )
                    if size not in (1, extent)
                ]
                if len(bad) > 0:
                    axis, size, extent = bad[0]
                    raise ValueError(
                        f"output[{i}].index_array has {size} entries on axis {axis}, "
                        f"which is neither 1 nor the domain's extent of {extent} "
                        f"(index_array shape {m.index_array.shape}, "
                        f"domain shape {self.domain.shape})"
                    )

    def __eq__(self, other: object) -> bool:
        """Value equality. `ArrayMap` compares its index array element-wise, so
        a transform holding one can be compared at all — the generated `__eq__`
        raised `ValueError: the truth value of an array ... is ambiguous`."""
        if not isinstance(other, IndexTransform):
            return NotImplemented
        return self.domain == other.domain and self.output == other.output

    def __hash__(self) -> int:
        """Hashed by value, so a transform can key a cache or enter a set."""
        return hash((self.domain, self.output))

    @property
    def input_rank(self) -> int:
        """Number of input dimensions — the rank of `domain`."""
        return self.domain.ndim

    @property
    def output_rank(self) -> int:
        """Number of output dimensions — one per output map."""
        return len(self.output)

    @classmethod
    def _unchecked(cls, domain: IndexDomain, output: tuple[OutputIndexMap, ...]) -> IndexTransform:
        """Build a transform whose maps the caller has already fitted to `domain`.

        Skips `__post_init__`. For internal producers such as chunk resolution,
        which derive every map from an already-valid transform and build two
        transforms per chunk.
        """
        transform = object.__new__(cls)
        object.__setattr__(transform, "domain", domain)
        object.__setattr__(transform, "output", output)
        return transform

    @classmethod
    def identity(cls, domain: IndexDomain) -> IndexTransform:
        """The identity transform over `domain`: every result cell reads the source at its own address."""
        output = tuple(DimensionMap(input_dimension=i) for i in range(domain.ndim))
        return cls(domain=domain, output=output)

    @classmethod
    def from_shape(cls, shape: tuple[int, ...]) -> IndexTransform:
        """The identity transform over a zero-origin domain of the given `shape`."""
        return cls.identity(IndexDomain.from_shape(shape))

    def apply(self, point: Sequence[int]) -> tuple[int, ...]:
        """Map one coordinate of `domain` to the source coordinate that fills it.

        In array-indexing terms: `point` names a cell of the result array, and
        the returned tuple — each `output` map evaluated at `point` — names the
        source-array cell its value is read from: the coordinate arrow,
        running result to source.

        Parameters
        ----------
        point : Sequence[int]
            One literal coordinate for each input dimension.

        Returns
        -------
        tuple[int, ...]
            One coordinate for each output map.

        Raises
        ------
        ValueError
            If ``point`` does not have exactly one coordinate per input
            dimension.
        TypeError
            If the coordinates do not have an integer dtype.
        BoundsCheckError
            If a coordinate lies outside the input domain.
        OverflowError
            If a mapped output coordinate cannot be represented by
            ``np.intp``.

        Examples
        --------
        The `[::2]` transform reads result cell `i` from source coordinate
        `2 * i`, so cell 3 of the result holds `source[6]`:

        >>> transform = IndexTransform.from_shape((100,))[::2]
        >>> transform.apply((3,))
        (6,)
        """
        coordinates = np.asarray(point)
        expected_shape = (self.input_rank,)
        if coordinates.shape != expected_shape:
            raise ValueError(f"point must have shape {expected_shape}, got {coordinates.shape}")
        # An empty Python sequence has no elements from which NumPy can infer
        # an integer dtype, but it is the unique point in a rank-zero domain.
        if self.input_rank == 0 and isinstance(point, (list, tuple)):
            coordinates = coordinates.astype(np.intp)
        try:
            result = self._apply_points(coordinates)
        except _PointOutOfBounds as error:
            raise BoundsCheckError(
                f"coordinate {error.value} on input dimension {error.dimension} "
                f"is outside the domain [{error.lower}, {error.upper})"
            ) from None
        return tuple(int(value) for value in result)

    def apply_many(self, points: npt.ArrayLike) -> npt.NDArray[np.intp]:
        """Map a batch of `domain` coordinates to the source coordinates that fill them.

        The vectorized form of `apply`: each row of `points` names a result
        cell, and the corresponding output row names the source-array cell
        its value is read from.

        Parameters
        ----------
        points : numpy.typing.ArrayLike
            Integer coordinates with shape ``batch_shape + (input_rank,)``.

        Returns
        -------
        numpy.typing.NDArray[numpy.intp]
            An owned array with shape ``batch_shape + (output_rank,)``.

        Raises
        ------
        ValueError
            If ``points`` has no trailing coordinate axis or that axis does
            not contain exactly one coordinate per input dimension.
        TypeError
            If the coordinates do not have an integer dtype.
        BoundsCheckError
            If a coordinate lies outside the input domain.
        OverflowError
            If a mapped output coordinate cannot be represented by
            ``np.intp``.

        Examples
        --------
        Three result cells of the `[::2]` transform, located in one call:

        >>> transform = IndexTransform.from_shape((100,))[::2]
        >>> transform.apply_many(np.array([[0], [1], [49]])).tolist()
        [[0], [2], [98]]
        """
        coordinates = np.asarray(points)
        if coordinates.ndim == 0 or coordinates.shape[-1] != self.input_rank:
            raise ValueError(
                "points must have a trailing coordinate axis of size "
                f"{self.input_rank}, got shape {coordinates.shape}"
            )
        try:
            return self._apply_points(coordinates)
        except _PointOutOfBounds as error:
            raise BoundsCheckError(
                f"point at batch position {error.batch_position} has input dimension "
                f"{error.dimension} coordinate {error.value} outside "
                f"[{error.lower}, {error.upper})"
            ) from None

    def _apply_points(self, points: np.ndarray[Any, Any]) -> npt.NDArray[np.intp]:
        """Vectorized implementation shared by ``apply`` and ``apply_many``.

        Out-of-domain coordinates raise the internal `_PointOutOfBounds`
        signal; each public entry point formats it in its own vocabulary —
        `apply` never mentions a batch, `apply_many` names the batch position."""
        if not np.issubdtype(points.dtype, np.integer):
            raise TypeError(f"points must have an integer dtype, got {points.dtype}")

        invalid = np.zeros(points.shape, dtype=np.bool_)
        for dimension, (lower, upper) in enumerate(
            zip(self.domain.inclusive_min, self.domain.exclusive_max, strict=True)
        ):
            invalid[..., dimension] = (points[..., dimension] < lower) | (
                points[..., dimension] >= upper
            )
        invalid_positions = np.argwhere(invalid)
        if invalid_positions.size > 0:
            first = invalid_positions[0]
            dimension = int(first[-1])
            batch_position = tuple(int(position) for position in first[:-1])
            point_index = tuple(int(position) for position in first)
            value = int(points[point_index])
            lower = self.domain.inclusive_min[dimension]
            upper = self.domain.exclusive_max[dimension]
            raise _PointOutOfBounds(dimension, value, lower, upper, batch_position)

        batch_shape = points.shape[:-1]
        result = np.empty(batch_shape + (self.output_rank,), dtype=np.intp)
        for output_dimension, output_map in enumerate(self.output):
            if isinstance(output_map, ConstantMap):
                if result[..., output_dimension].size == 0:
                    continue
                result[..., output_dimension] = checked_affine(output_map.offset, 0, 0)
            elif isinstance(output_map, DimensionMap):
                result[..., output_dimension] = checked_affine(
                    output_map.offset,
                    output_map.stride,
                    points[..., output_map.input_dimension],
                )
            else:
                index = tuple(
                    np.zeros(batch_shape, dtype=np.intp)
                    if output_map.index_array.shape[axis] == 1
                    else _positions_from_origin(points[..., axis], self.domain.inclusive_min[axis])
                    for axis in range(self.input_rank)
                )
                result[..., output_dimension] = checked_affine(
                    output_map.offset,
                    output_map.stride,
                    np.asarray(output_map.index_array[index]),
                )
        return result

    def inverted(self) -> IndexTransform:
        """Return the restricted, exactly representable inverse transform.

        Inversion is defined for square transforms containing only constants
        and unique unit-stride dimension maps. Any input dimension not named by
        a dimension map must have singleton extent, so its coordinate can be
        recovered as a constant.

        Returns
        -------
        IndexTransform
            A new transform mapping output coordinates back to input
            coordinates.

        Raises
        ------
        ValueError
            If this transform does not have a representable inverse, including
            when input labels cannot be transferred to unlabeled output
            dimensions.
        """
        if self.domain.labels is not None:
            raise ValueError(
                "cannot invert transform: input labels cannot be represented "
                "because output dimensions do not carry labels"
            )
        if self.input_rank != self.output_rank:
            raise ValueError(
                "cannot invert transform: input rank must equal output rank, got "
                f"{self.input_rank} and {self.output_rank}"
            )

        referenced: set[int] = set()
        for output_dimension, output_map in enumerate(self.output):
            if isinstance(output_map, ArrayMap):
                raise ValueError(  # noqa: TRY004 - valid map, invalid inverse
                    f"cannot invert transform: output[{output_dimension}] is an ArrayMap"
                )
            if isinstance(output_map, DimensionMap):
                if output_map.stride not in (-1, 1):
                    raise ValueError(
                        "cannot invert transform: DimensionMap stride must be +1 or -1, "
                        f"got {output_map.stride} for output[{output_dimension}]"
                    )
                if output_map.input_dimension in referenced:
                    raise ValueError(
                        "cannot invert transform: input dimension "
                        f"{output_map.input_dimension} is referenced more than once"
                    )
                referenced.add(output_map.input_dimension)

        for input_dimension, extent in enumerate(self.domain.shape):
            if input_dimension not in referenced and extent != 1:
                raise ValueError(
                    "cannot invert transform: unreferenced input dimension "
                    f"{input_dimension} has extent {extent}, not 1"
                )

        inverse_min: list[int] = []
        inverse_max: list[int] = []
        inverse_output: dict[int, OutputIndexMap] = {}
        for output_dimension, output_map in enumerate(self.output):
            if isinstance(output_map, ConstantMap):
                inverse_min.append(output_map.offset)
                inverse_max.append(output_map.offset + 1)
                continue

            assert isinstance(output_map, DimensionMap)
            input_dimension = output_map.input_dimension
            lower = self.domain.inclusive_min[input_dimension]
            upper = self.domain.exclusive_max[input_dimension]
            if output_map.stride == 1:
                inverse_min.append(output_map.offset + lower)
                inverse_max.append(output_map.offset + upper)
                inverse_output[input_dimension] = DimensionMap(
                    output_dimension,
                    offset=-output_map.offset,
                )
            else:
                inverse_min.append(output_map.offset - upper + 1)
                inverse_max.append(output_map.offset - lower + 1)
                inverse_output[input_dimension] = DimensionMap(
                    output_dimension,
                    offset=output_map.offset,
                    stride=-1,
                )

        for input_dimension, lower in enumerate(self.domain.inclusive_min):
            if input_dimension not in referenced:
                inverse_output[input_dimension] = ConstantMap(lower)

        return IndexTransform(
            domain=IndexDomain(tuple(inverse_min), tuple(inverse_max)),
            output=tuple(inverse_output[dimension] for dimension in range(self.input_rank)),
        )

    @property
    def selection_repr(self) -> str:
        """Compact domain string, e.g. `'{ [2, 8), [0, 10) }'`.

        Follows TensorStore's IndexDomain notation: each dimension shown
        as `[inclusive_min, exclusive_max)` with stride annotation if not 1.
        Constant (integer-indexed) dimensions show as a single value.
        Array-indexed dimensions show the set of selected coordinates.
        """
        parts: list[str] = []
        for m in self.output:
            if isinstance(m, ConstantMap):
                parts.append(str(m.offset))
            elif isinstance(m, DimensionMap):
                d = m.input_dimension
                lo = self.domain.inclusive_min[d]
                hi = self.domain.exclusive_max[d]
                start = m.offset + m.stride * lo
                stop = m.offset + m.stride * hi
                if m.stride == 1:
                    parts.append(f"[{start}, {stop})")
                else:
                    parts.append(f"[{start}, {stop}) step {m.stride}")
            else:
                # m: ArrayMap (OutputIndexMap = ConstantMap | DimensionMap | ArrayMap)
                storage = m.offset + m.stride * m.index_array
                n = int(storage.size)  # .size, not len(): index_array may be 0-d
                if n <= 5:
                    vals = ", ".join(str(int(v)) for v in storage.ravel())
                    parts.append("{" + vals + "}")
                else:
                    parts.append("{" + f"array({n})" + "}")
        return "{ " + ", ".join(parts) + " }"

    def __repr__(self) -> str:
        maps: list[str] = []
        for i, m in enumerate(self.output):
            if isinstance(m, ConstantMap):
                maps.append(f"out[{i}] = {m.offset}")
            elif isinstance(m, DimensionMap):
                maps.append(f"out[{i}] = {m.offset} + {m.stride} * in[{m.input_dimension}]")
            else:
                # m: ArrayMap (OutputIndexMap = ConstantMap | DimensionMap | ArrayMap)
                maps.append(f"out[{i}] = {m.offset} + {m.stride} * arr{m.index_array.shape}[in]")
        maps_str = ", ".join(maps)
        return f"IndexTransform(domain={self.domain}, {maps_str})"

    def intersect(
        self, output_domain: IndexDomain
    ) -> (
        tuple[
            IndexTransform,
            dict[int, np.ndarray[Any, np.dtype[np.intp]]]
            | np.ndarray[Any, np.dtype[np.intp]]
            | None,
        ]
        | None
    ):
        """Keep only the cells whose source coordinates fall inside `output_domain`.

        Chunk resolution is the canonical caller: intersecting a request with
        one chunk's box keeps the cells that chunk can serve.

        Returns `(restricted_transform, out_indices)` or None if empty.

        `out_indices` carries the surviving output positions: `None` when all
        positions survive (ConstantMap/DimensionMap only), a single integer array
        for one ArrayMap (or correlated/vectorized ArrayMaps), or a dict keyed by
        output dimension for >= 2 orthogonal ArrayMaps (an outer product).
        """
        return _intersect(self, output_domain)

    def translate(self, shift: tuple[int, ...]) -> IndexTransform:
        """Shift the source coordinates every cell reads by `shift`, per dimension.

        The domain is untouched: the result keeps its cells, and each one
        reads from a shifted source address — for example, making a chunk's
        global addresses chunk-local by translating by the chunk's negated
        origin.
        """
        if len(shift) != self.output_rank:
            raise ValueError(f"shift must have length {self.output_rank}, got {len(shift)}")
        new_output: list[OutputIndexMap] = []
        for m, s in zip(self.output, shift, strict=True):
            if isinstance(m, ConstantMap):
                new_output.append(ConstantMap(offset=m.offset + s))
            elif isinstance(m, DimensionMap):
                new_output.append(
                    DimensionMap(
                        input_dimension=m.input_dimension,
                        offset=m.offset + s,
                        stride=m.stride,
                    )
                )
            else:
                # m: ArrayMap (OutputIndexMap = ConstantMap | DimensionMap | ArrayMap)
                new_output.append(
                    m._with_affine(m.offset + s, m.stride)  # pyright: ignore[reportPrivateUsage]
                )
        return IndexTransform(domain=self.domain, output=tuple(new_output))

    def __getitem__(self, selection: Any) -> IndexTransform:
        """Compose a basic selection (int, slice, ellipsis, newaxis) into a new transform.

        No I/O occurs. Integers and slice bounds are literal domain coordinates
        (TensorStore convention): negative values are not counted from the end,
        and out-of-domain values raise `BoundsCheckError`. Integer indices drop
        their input dimension; `None` inserts a size-1 dimension.
        """
        return _apply_basic_indexing(self, selection)

    def translate_domain_by(self, shift: tuple[int, ...]) -> IndexTransform:
        """Shift the *input* domain by `shift`, preserving which cells are addressed.

        TensorStore's `translate_by`: the domain moves, and every output map is
        re-offset so that new coordinate `c` addresses the cell that `c - shift`
        addressed before. ArrayMaps are indexed positionally over the domain, so
        their index arrays are unchanged.
        """
        if len(shift) != self.input_rank:
            raise ValueError(f"shift must have length {self.input_rank}, got {len(shift)}")
        new_domain = self.domain.translate(shift)
        new_output: list[OutputIndexMap] = []
        for m in self.output:
            if isinstance(m, DimensionMap):
                s = shift[m.input_dimension]
                new_output.append(
                    DimensionMap(
                        input_dimension=m.input_dimension,
                        offset=m.offset - m.stride * s,
                        stride=m.stride,
                    )
                )
            else:
                # ConstantMap: no input dependence. ArrayMap: positional over
                # the domain, invariant under domain translation.
                new_output.append(m)
        return IndexTransform(domain=new_domain, output=tuple(new_output))

    def translate_domain_to(self, origins: tuple[int, ...]) -> IndexTransform:
        """Move the input domain so its per-dimension origins equal `origins`.

        TensorStore's `translate_to`; `translate_domain_to((0,) * rank)`
        re-zeros a view's coordinate system without changing which cells it
        addresses.
        """
        if len(origins) != self.input_rank:
            raise ValueError(f"origins must have length {self.input_rank}, got {len(origins)}")
        shift = tuple(o - m for o, m in zip(origins, self.domain.inclusive_min, strict=True))
        return self.translate_domain_by(shift)

    @property
    def oindex(self) -> _OIndexHelper:
        """Accessor for the orthogonal (outer-product) indexing dialect.

        `transform.oindex[sel]` applies each index array independently per
        dimension and returns a new transform.
        """
        return _OIndexHelper(self)

    @property
    def vindex(self) -> _VIndexHelper:
        """Accessor for the vectorized (coordinate/mask) indexing dialect.

        `transform.vindex[sel]` broadcasts all index arrays together, NumPy
        fancy-indexing style, and returns a new transform.
        """
        return _VIndexHelper(self)

    @property
    def index_array_structure(self) -> Literal["none", "orthogonal", "general"]:
        """Classify how a transform's index arrays relate to its input axes.

        Returns
        -------
        `"none"` when no output map is an `ArrayMap`; `"orthogonal"` when every
        `ArrayMap` varies over exactly one input axis, each its own (an outer
        product, one independent gather per axis); `"general"` otherwise —
        correlated (`vindex`) maps sharing their non-singleton axes, maps produced
        by composing fancy steps, maps sharing an input axis (a diagonal gather),
        and empty or hand-built all-singleton maps whose shape names no axis. The
        orthogonal resolvers narrow one axis at a time and are only sound for
        `"orthogonal"`; everything else takes the pointwise path that collapses
        the joint block. Everything is read off the index arrays' shapes.

        Examples
        --------
        >>> t = IndexTransform.from_shape((4, 5))
        >>> t.index_array_structure
        'none'

        `oindex` arrays each vary over their own axis (an outer product):

        >>> t.oindex[[0, 2], [1, 3]].index_array_structure
        'orthogonal'

        `vindex` arrays are correlated — they share the broadcast axis:

        >>> t.vindex[np.array([0, 2]), np.array([1, 3])].index_array_structure
        'general'
        """
        seen: set[int] = set()
        has_array = False
        for m in self.output:
            if not isinstance(m, ArrayMap):
                continue
            has_array = True
            dep = m.dependency_axes
            if len(dep) != 1 or dep[0] in seen:
                return "general"
            seen.add(dep[0])
        return "orthogonal" if has_array else "none"

    def select(
        self,
        selection: Any,
        mode: Literal["basic", "orthogonal", "vectorized"] = "basic",
    ) -> IndexTransform:
        """Convert a user selection into a composed IndexTransform.

        Negative indices are treated as literal coordinates (TensorStore convention).
        The caller (Array layer) is responsible for converting numpy-style negative
        indices before calling this function.

        Examples
        --------
        The `mode` picks the dialect; the result is the composed self the
        corresponding accessor builds:

        >>> t = IndexTransform.from_shape((10,))
        >>> t.select(slice(2, 8)) == t[2:8]
        True
        >>> s = t.select(([9, 0, 0],), mode="orthogonal")
        >>> s.apply((0,)), s.apply((1,)), s.apply((2,))
        ((9,), (0,), (0,))
        """
        if mode == "basic":
            _validate_basic_selection(selection)
            return self[selection]
        elif mode == "orthogonal":
            _validate_array_selection(selection, self.domain.shape, mode)
            return self.oindex[selection]
        elif mode == "vectorized":
            _validate_array_selection(selection, self.domain.shape, mode)
            return self.vindex[selection]
        else:
            raise ValueError(f"Unknown mode: {mode!r}")

    def compose(self, inner: IndexTransform) -> IndexTransform:
        """Chain `inner` onto this transform, yielding one direct transform.

        This transform maps its own input coordinates to `inner`'s input
        coordinates, and `inner` maps those onward; the result maps this
        transform's input coordinates straight to `inner`'s output
        coordinates. Composition is what keeps a view of a view a single
        description rather than a stack of layers, and it is exact: index
        arrays are evaluated at the new coordinates rather than accumulated.

        The precondition is that this transform's output rank equals `inner`'s
        input rank; a mismatch, or coordinates leaving `inner`'s domain, raises.

        Examples
        --------
        Chained indexing — `source[2:5]`, then `[::-1]` on the result —
        collapses to one transform (a reversed axis keeps literal coordinates,
        so the composed domain is `[-4, -1)`):

        >>> inner = IndexTransform.from_shape((10,))[2:5]
        >>> outer = IndexTransform.identity(inner.domain)[::-1]
        >>> chained = outer.compose(inner)
        >>> chained == inner[::-1]
        True
        >>> [chained.apply((i,)) for i in (-4, -3, -2)]
        [(4,), (3,), (2,)]
        """
        from zarr_indexing._composition import compose

        return compose(self, inner)

    # -- serialization ------------------------------------------------------

    def to_json(self) -> IndexTransformJSON:
        """Convert to the canonical ndsel transform body (spec section 4.3).

        The result is fully explicit: `input_rank`, fully written bounds and
        labels, and an `output` carrying `offset`/`stride` on every affine and
        array map. It is field-for-field a TensorStore `IndexTransform` minus
        the `kind` discriminator, so it loads directly into
        `tensorstore.IndexTransform(json=...)`.

        Examples
        --------
        >>> body = IndexTransform.from_shape((6,))[1:5:2].to_json()
        >>> (body["input_inclusive_min"], body["input_exclusive_max"])
        ([0], [2])
        >>> body["output"]
        [{'offset': 1, 'stride': 2, 'input_dimension': 0}]
        """
        from zarr_indexing._wire import emit_labels

        return {
            "input_rank": self.domain.ndim,
            "input_inclusive_min": list(self.domain.inclusive_min),
            "input_exclusive_max": list(self.domain.exclusive_max),
            "input_labels": emit_labels(self.domain.labels, self.domain.ndim),
            "output": [m.to_json() for m in self.output],
        }

    @classmethod
    def from_json(cls, data: IndexTransformJSON) -> IndexTransform:
        """Construct from a canonical (or canonicalizable) ndsel transform body.

        The body is first run through the message layer (`normalize_ndsel`) so
        that omitted fields — identity `output`, default bounds and labels —
        are filled and validated, then lowered to the engine representation.
        Lower-rank `index_array`s are widened to the full input rank on the way
        in.

        Examples
        --------
        >>> body = IndexTransform.from_shape((6,))[1:5:2].to_json()
        >>> transform = IndexTransform.from_json(body)
        >>> transform.domain.shape
        (2,)
        >>> transform.to_json() == body  # the round trip is exact
        True
        """
        from zarr_indexing._wire import (
            full_rank_index_array,
            lower_bound,
            lower_index_array,
            lower_labels,
        )
        from zarr_indexing.messages import NdselError, normalize_ndsel

        if not isinstance(data, dict):  # pyright: ignore[reportUnnecessaryIsInstance]
            raise NdselError(
                "invalid_json", f"a transform body must be a JSON object, got {data!r}"
            )
        kind = data.get("kind", "transform")
        if kind != "transform":
            # Spelled before normalization so a body carrying its own `kind`
            # cannot reinterpret the document as some other message and return
            # a selection this constructor never promised.
            raise NdselError("invalid_json", f"a transform body cannot carry kind {kind!r}")
        body = normalize_ndsel({**data, "kind": "transform"})

        domain = IndexDomain(
            inclusive_min=tuple(
                lower_bound(b, f"input_inclusive_min[{i}]")
                for i, b in enumerate(body["input_inclusive_min"])
            ),
            exclusive_max=tuple(
                lower_bound(b, f"input_exclusive_max[{i}]")
                for i, b in enumerate(body["input_exclusive_max"])
            ),
            labels=lower_labels(body["input_labels"]),
        )

        output: list[OutputIndexMap] = []
        for i, om in enumerate(body["output"]):
            if "index_array" in om:
                where = f"output[{i}]"
                arr = lower_index_array(om["index_array"], f"{where}.index_array")
                # ndsel leaves index-array rank unvalidated, so an external
                # producer may send an array of lower rank that broadcasts
                # against the domain. Widen it here, on the way in, so every
                # transform that exists holds the full-rank invariant the
                # engine reads dependency axes from.
                output.append(
                    ArrayMap(
                        index_array=full_rank_index_array(arr, domain, where),
                        offset=om.get("offset", 0),
                        stride=om.get("stride", 1),
                    )
                )
            elif "input_dimension" in om:
                output.append(
                    DimensionMap(
                        input_dimension=om["input_dimension"],
                        offset=om.get("offset", 0),
                        stride=om.get("stride", 1),
                    )
                )
            else:
                output.append(ConstantMap(offset=om.get("offset", 0)))

        try:
            return cls(domain=domain, output=tuple(output))
        except ValueError as exc:
            # The engine's invariants are the last gate a document passes, and
            # they speak in the engine's vocabulary. A document that fails them
            # is invalid input, so it leaves here as one — with the engine's
            # account of what was wrong kept, since it names the offending
            # output map and axis.
            raise NdselError("rank_mismatch", str(exc)) from exc

domain instance-attribute

domain: IndexDomain

The input domain: the request coordinates this transform accepts.

index_array_structure property

index_array_structure: Literal[
    "none", "orthogonal", "general"
]

Classify how a transform's index arrays relate to its input axes.

Returns:

• `"none"` when no output map is an `ArrayMap`; `"orthogonal"` when every –
• `ArrayMap` varies over exactly one input axis, each its own (an outer –
• product, one independent gather per axis); `"general"` otherwise — –
• correlated (`vindex`) maps sharing their non-singleton axes, maps produced –
• by composing fancy steps, maps sharing an input axis (a diagonal gather), –
• and empty or hand-built all-singleton maps whose shape names no axis. The –
• orthogonal resolvers narrow one axis at a time and are only sound for –
• `"orthogonal"`; everything else takes the pointwise path that collapses –
• the joint block. Everything is read off the index arrays' shapes. –

Examples:

>>> t = IndexTransform.from_shape((4, 5))
>>> t.index_array_structure
'none'

oindex arrays each vary over their own axis (an outer product):

>>> t.oindex[[0, 2], [1, 3]].index_array_structure
'orthogonal'

vindex arrays are correlated — they share the broadcast axis:

>>> t.vindex[np.array([0, 2]), np.array([1, 3])].index_array_structure
'general'

input_rank property

input_rank: int

Number of input dimensions — the rank of domain.

oindex property

oindex: _OIndexHelper

Accessor for the orthogonal (outer-product) indexing dialect.

transform.oindex[sel] applies each index array independently per dimension and returns a new transform.

output instance-attribute

output: tuple[OutputIndexMap, ...]

One output map per output dimension, each producing that dimension's coordinate.

output_rank property

output_rank: int

Number of output dimensions — one per output map.

selection_repr property

selection_repr: str

Compact domain string, e.g. '{ [2, 8), [0, 10) }'.

Follows TensorStore's IndexDomain notation: each dimension shown as [inclusive_min, exclusive_max) with stride annotation if not 1. Constant (integer-indexed) dimensions show as a single value. Array-indexed dimensions show the set of selected coordinates.

vindex property

vindex: _VIndexHelper

Accessor for the vectorized (coordinate/mask) indexing dialect.

transform.vindex[sel] broadcasts all index arrays together, NumPy fancy-indexing style, and returns a new transform.

__eq__

__eq__(other: object) -> bool

Value equality. ArrayMap compares its index array element-wise, so a transform holding one can be compared at all — the generated __eq__ raised ValueError: the truth value of an array ... is ambiguous.

Source code in src/zarr_indexing/transform.py

def __eq__(self, other: object) -> bool:
    """Value equality. `ArrayMap` compares its index array element-wise, so
    a transform holding one can be compared at all — the generated `__eq__`
    raised `ValueError: the truth value of an array ... is ambiguous`."""
    if not isinstance(other, IndexTransform):
        return NotImplemented
    return self.domain == other.domain and self.output == other.output

__getitem__

__getitem__(selection: Any) -> IndexTransform

Compose a basic selection (int, slice, ellipsis, newaxis) into a new transform.

No I/O occurs. Integers and slice bounds are literal domain coordinates (TensorStore convention): negative values are not counted from the end, and out-of-domain values raise BoundsCheckError. Integer indices drop their input dimension; None inserts a size-1 dimension.

Source code in src/zarr_indexing/transform.py

def __getitem__(self, selection: Any) -> IndexTransform:
    """Compose a basic selection (int, slice, ellipsis, newaxis) into a new transform.

    No I/O occurs. Integers and slice bounds are literal domain coordinates
    (TensorStore convention): negative values are not counted from the end,
    and out-of-domain values raise `BoundsCheckError`. Integer indices drop
    their input dimension; `None` inserts a size-1 dimension.
    """
    return _apply_basic_indexing(self, selection)

__hash__

__hash__() -> int

Hashed by value, so a transform can key a cache or enter a set.

Source code in src/zarr_indexing/transform.py

def __hash__(self) -> int:
    """Hashed by value, so a transform can key a cache or enter a set."""
    return hash((self.domain, self.output))

__init__

__init__(
    domain: IndexDomain, output: tuple[OutputIndexMap, ...]
) -> None

__post_init__

__post_init__() -> None

Source code in src/zarr_indexing/transform.py

def __post_init__(self) -> None:
    for i, m in enumerate(self.output):
        if isinstance(m, DimensionMap):
            if m.input_dimension < 0 or m.input_dimension >= self.domain.ndim:
                raise ValueError(
                    f"output[{i}].input_dimension = {m.input_dimension} "
                    f"is out of range for input rank {self.domain.ndim}"
                )
        elif isinstance(m, ArrayMap):
            # An index array carries the transform's full input rank: the axis
            # a map varies over is full-sized, every other axis a singleton.
            # The rank is what makes the dependency axes readable from the
            # shape, so a mismatch is a bug rather than a spelling. External
            # JSON may use a lower-rank array that broadcasts against the
            # domain; `from_json` widens those on the way in,
            # so the invariant holds for every transform that exists.
            if m.index_array.ndim != self.domain.ndim:
                raise ValueError(
                    f"output[{i}].index_array has {m.index_array.ndim} dims "
                    f"but input domain has {self.domain.ndim} dims"
                )
            # Every axis is either the domain's extent or a singleton it
            # broadcasts over. Any other size addresses input coordinates the
            # array has no entry for, which reads as a smaller selection
            # rather than as the error it is.
            bad = [
                (axis, size, extent)
                for axis, (size, extent) in enumerate(
                    zip(m.index_array.shape, self.domain.shape, strict=True)
                )
                if size not in (1, extent)
            ]
            if len(bad) > 0:
                axis, size, extent = bad[0]
                raise ValueError(
                    f"output[{i}].index_array has {size} entries on axis {axis}, "
                    f"which is neither 1 nor the domain's extent of {extent} "
                    f"(index_array shape {m.index_array.shape}, "
                    f"domain shape {self.domain.shape})"
                )

__repr__

__repr__() -> str

Source code in src/zarr_indexing/transform.py

def __repr__(self) -> str:
    maps: list[str] = []
    for i, m in enumerate(self.output):
        if isinstance(m, ConstantMap):
            maps.append(f"out[{i}] = {m.offset}")
        elif isinstance(m, DimensionMap):
            maps.append(f"out[{i}] = {m.offset} + {m.stride} * in[{m.input_dimension}]")
        else:
            # m: ArrayMap (OutputIndexMap = ConstantMap | DimensionMap | ArrayMap)
            maps.append(f"out[{i}] = {m.offset} + {m.stride} * arr{m.index_array.shape}[in]")
    maps_str = ", ".join(maps)
    return f"IndexTransform(domain={self.domain}, {maps_str})"

apply

apply(point: Sequence[int]) -> tuple[int, ...]

Map one coordinate of domain to the source coordinate that fills it.

In array-indexing terms: point names a cell of the result array, and the returned tuple — each output map evaluated at point — names the source-array cell its value is read from: the coordinate arrow, running result to source.

Parameters:

• point (Sequence[int]) –

  One literal coordinate for each input dimension.

Returns:

• tuple[int, ...] –

  One coordinate for each output map.

Raises:

• ValueError –

  If point does not have exactly one coordinate per input dimension.

• TypeError –

  If the coordinates do not have an integer dtype.

• BoundsCheckError –

  If a coordinate lies outside the input domain.

• OverflowError –

  If a mapped output coordinate cannot be represented by np.intp.

Examples:

The [::2] transform reads result cell i from source coordinate 2 * i, so cell 3 of the result holds source[6]:

>>> transform = IndexTransform.from_shape((100,))[::2]
>>> transform.apply((3,))
(6,)

Source code in src/zarr_indexing/transform.py

def apply(self, point: Sequence[int]) -> tuple[int, ...]:
    """Map one coordinate of `domain` to the source coordinate that fills it.

    In array-indexing terms: `point` names a cell of the result array, and
    the returned tuple — each `output` map evaluated at `point` — names the
    source-array cell its value is read from: the coordinate arrow,
    running result to source.

    Parameters
    ----------
    point : Sequence[int]
        One literal coordinate for each input dimension.

    Returns
    -------
    tuple[int, ...]
        One coordinate for each output map.

    Raises
    ------
    ValueError
        If ``point`` does not have exactly one coordinate per input
        dimension.
    TypeError
        If the coordinates do not have an integer dtype.
    BoundsCheckError
        If a coordinate lies outside the input domain.
    OverflowError
        If a mapped output coordinate cannot be represented by
        ``np.intp``.

    Examples
    --------
    The `[::2]` transform reads result cell `i` from source coordinate
    `2 * i`, so cell 3 of the result holds `source[6]`:

    >>> transform = IndexTransform.from_shape((100,))[::2]
    >>> transform.apply((3,))
    (6,)
    """
    coordinates = np.asarray(point)
    expected_shape = (self.input_rank,)
    if coordinates.shape != expected_shape:
        raise ValueError(f"point must have shape {expected_shape}, got {coordinates.shape}")
    # An empty Python sequence has no elements from which NumPy can infer
    # an integer dtype, but it is the unique point in a rank-zero domain.
    if self.input_rank == 0 and isinstance(point, (list, tuple)):
        coordinates = coordinates.astype(np.intp)
    try:
        result = self._apply_points(coordinates)
    except _PointOutOfBounds as error:
        raise BoundsCheckError(
            f"coordinate {error.value} on input dimension {error.dimension} "
            f"is outside the domain [{error.lower}, {error.upper})"
        ) from None
    return tuple(int(value) for value in result)

apply_many

apply_many(points: ArrayLike) -> NDArray[intp]

Map a batch of domain coordinates to the source coordinates that fill them.

The vectorized form of apply: each row of points names a result cell, and the corresponding output row names the source-array cell its value is read from.

Parameters:

• points (ArrayLike) –

  Integer coordinates with shape batch_shape + (input_rank,).

Returns:

• NDArray[intp] –

  An owned array with shape batch_shape + (output_rank,).

Raises:

• ValueError –

  If points has no trailing coordinate axis or that axis does not contain exactly one coordinate per input dimension.

• TypeError –

  If the coordinates do not have an integer dtype.

• BoundsCheckError –

  If a coordinate lies outside the input domain.

• OverflowError –

  If a mapped output coordinate cannot be represented by np.intp.

Examples:

Three result cells of the [::2] transform, located in one call:

>>> transform = IndexTransform.from_shape((100,))[::2]
>>> transform.apply_many(np.array([[0], [1], [49]])).tolist()
[[0], [2], [98]]

Source code in src/zarr_indexing/transform.py

def apply_many(self, points: npt.ArrayLike) -> npt.NDArray[np.intp]:
    """Map a batch of `domain` coordinates to the source coordinates that fill them.

    The vectorized form of `apply`: each row of `points` names a result
    cell, and the corresponding output row names the source-array cell
    its value is read from.

    Parameters
    ----------
    points : numpy.typing.ArrayLike
        Integer coordinates with shape ``batch_shape + (input_rank,)``.

    Returns
    -------
    numpy.typing.NDArray[numpy.intp]
        An owned array with shape ``batch_shape + (output_rank,)``.

    Raises
    ------
    ValueError
        If ``points`` has no trailing coordinate axis or that axis does
        not contain exactly one coordinate per input dimension.
    TypeError
        If the coordinates do not have an integer dtype.
    BoundsCheckError
        If a coordinate lies outside the input domain.
    OverflowError
        If a mapped output coordinate cannot be represented by
        ``np.intp``.

    Examples
    --------
    Three result cells of the `[::2]` transform, located in one call:

    >>> transform = IndexTransform.from_shape((100,))[::2]
    >>> transform.apply_many(np.array([[0], [1], [49]])).tolist()
    [[0], [2], [98]]
    """
    coordinates = np.asarray(points)
    if coordinates.ndim == 0 or coordinates.shape[-1] != self.input_rank:
        raise ValueError(
            "points must have a trailing coordinate axis of size "
            f"{self.input_rank}, got shape {coordinates.shape}"
        )
    try:
        return self._apply_points(coordinates)
    except _PointOutOfBounds as error:
        raise BoundsCheckError(
            f"point at batch position {error.batch_position} has input dimension "
            f"{error.dimension} coordinate {error.value} outside "
            f"[{error.lower}, {error.upper})"
        ) from None

compose

compose(inner: IndexTransform) -> IndexTransform

Chain inner onto this transform, yielding one direct transform.

This transform maps its own input coordinates to inner's input coordinates, and inner maps those onward; the result maps this transform's input coordinates straight to inner's output coordinates. Composition is what keeps a view of a view a single description rather than a stack of layers, and it is exact: index arrays are evaluated at the new coordinates rather than accumulated.

The precondition is that this transform's output rank equals inner's input rank; a mismatch, or coordinates leaving inner's domain, raises.

Examples:

Chained indexing — source[2:5], then [::-1] on the result — collapses to one transform (a reversed axis keeps literal coordinates, so the composed domain is [-4, -1)):

>>> inner = IndexTransform.from_shape((10,))[2:5]
>>> outer = IndexTransform.identity(inner.domain)[::-1]
>>> chained = outer.compose(inner)
>>> chained == inner[::-1]
True
>>> [chained.apply((i,)) for i in (-4, -3, -2)]
[(4,), (3,), (2,)]

Source code in src/zarr_indexing/transform.py

def compose(self, inner: IndexTransform) -> IndexTransform:
    """Chain `inner` onto this transform, yielding one direct transform.

    This transform maps its own input coordinates to `inner`'s input
    coordinates, and `inner` maps those onward; the result maps this
    transform's input coordinates straight to `inner`'s output
    coordinates. Composition is what keeps a view of a view a single
    description rather than a stack of layers, and it is exact: index
    arrays are evaluated at the new coordinates rather than accumulated.

    The precondition is that this transform's output rank equals `inner`'s
    input rank; a mismatch, or coordinates leaving `inner`'s domain, raises.

    Examples
    --------
    Chained indexing — `source[2:5]`, then `[::-1]` on the result —
    collapses to one transform (a reversed axis keeps literal coordinates,
    so the composed domain is `[-4, -1)`):

    >>> inner = IndexTransform.from_shape((10,))[2:5]
    >>> outer = IndexTransform.identity(inner.domain)[::-1]
    >>> chained = outer.compose(inner)
    >>> chained == inner[::-1]
    True
    >>> [chained.apply((i,)) for i in (-4, -3, -2)]
    [(4,), (3,), (2,)]
    """
    from zarr_indexing._composition import compose

    return compose(self, inner)

from_json classmethod

from_json(data: IndexTransformJSON) -> IndexTransform

Construct from a canonical (or canonicalizable) ndsel transform body.

The body is first run through the message layer (normalize_ndsel) so that omitted fields — identity output, default bounds and labels — are filled and validated, then lowered to the engine representation. Lower-rank index_arrays are widened to the full input rank on the way in.

Examples:

>>> body = IndexTransform.from_shape((6,))[1:5:2].to_json()
>>> transform = IndexTransform.from_json(body)
>>> transform.domain.shape
(2,)
>>> transform.to_json() == body  # the round trip is exact
True

Source code in src/zarr_indexing/transform.py

@classmethod
def from_json(cls, data: IndexTransformJSON) -> IndexTransform:
    """Construct from a canonical (or canonicalizable) ndsel transform body.

    The body is first run through the message layer (`normalize_ndsel`) so
    that omitted fields — identity `output`, default bounds and labels —
    are filled and validated, then lowered to the engine representation.
    Lower-rank `index_array`s are widened to the full input rank on the way
    in.

    Examples
    --------
    >>> body = IndexTransform.from_shape((6,))[1:5:2].to_json()
    >>> transform = IndexTransform.from_json(body)
    >>> transform.domain.shape
    (2,)
    >>> transform.to_json() == body  # the round trip is exact
    True
    """
    from zarr_indexing._wire import (
        full_rank_index_array,
        lower_bound,
        lower_index_array,
        lower_labels,
    )
    from zarr_indexing.messages import NdselError, normalize_ndsel

    if not isinstance(data, dict):  # pyright: ignore[reportUnnecessaryIsInstance]
        raise NdselError(
            "invalid_json", f"a transform body must be a JSON object, got {data!r}"
        )
    kind = data.get("kind", "transform")
    if kind != "transform":
        # Spelled before normalization so a body carrying its own `kind`
        # cannot reinterpret the document as some other message and return
        # a selection this constructor never promised.
        raise NdselError("invalid_json", f"a transform body cannot carry kind {kind!r}")
    body = normalize_ndsel({**data, "kind": "transform"})

    domain = IndexDomain(
        inclusive_min=tuple(
            lower_bound(b, f"input_inclusive_min[{i}]")
            for i, b in enumerate(body["input_inclusive_min"])
        ),
        exclusive_max=tuple(
            lower_bound(b, f"input_exclusive_max[{i}]")
            for i, b in enumerate(body["input_exclusive_max"])
        ),
        labels=lower_labels(body["input_labels"]),
    )

    output: list[OutputIndexMap] = []
    for i, om in enumerate(body["output"]):
        if "index_array" in om:
            where = f"output[{i}]"
            arr = lower_index_array(om["index_array"], f"{where}.index_array")
            # ndsel leaves index-array rank unvalidated, so an external
            # producer may send an array of lower rank that broadcasts
            # against the domain. Widen it here, on the way in, so every
            # transform that exists holds the full-rank invariant the
            # engine reads dependency axes from.
            output.append(
                ArrayMap(
                    index_array=full_rank_index_array(arr, domain, where),
                    offset=om.get("offset", 0),
                    stride=om.get("stride", 1),
                )
            )
        elif "input_dimension" in om:
            output.append(
                DimensionMap(
                    input_dimension=om["input_dimension"],
                    offset=om.get("offset", 0),
                    stride=om.get("stride", 1),
                )
            )
        else:
            output.append(ConstantMap(offset=om.get("offset", 0)))

    try:
        return cls(domain=domain, output=tuple(output))
    except ValueError as exc:
        # The engine's invariants are the last gate a document passes, and
        # they speak in the engine's vocabulary. A document that fails them
        # is invalid input, so it leaves here as one — with the engine's
        # account of what was wrong kept, since it names the offending
        # output map and axis.
        raise NdselError("rank_mismatch", str(exc)) from exc

from_shape classmethod

from_shape(shape: tuple[int, ...]) -> IndexTransform

The identity transform over a zero-origin domain of the given shape.

Source code in src/zarr_indexing/transform.py

@classmethod
def from_shape(cls, shape: tuple[int, ...]) -> IndexTransform:
    """The identity transform over a zero-origin domain of the given `shape`."""
    return cls.identity(IndexDomain.from_shape(shape))

identity classmethod

identity(domain: IndexDomain) -> IndexTransform

The identity transform over domain: every result cell reads the source at its own address.

Source code in src/zarr_indexing/transform.py

@classmethod
def identity(cls, domain: IndexDomain) -> IndexTransform:
    """The identity transform over `domain`: every result cell reads the source at its own address."""
    output = tuple(DimensionMap(input_dimension=i) for i in range(domain.ndim))
    return cls(domain=domain, output=output)

intersect

intersect(
    output_domain: IndexDomain,
) -> (
    tuple[
        IndexTransform,
        dict[int, ndarray[Any, dtype[intp]]]
        | ndarray[Any, dtype[intp]]
        | None,
    ]
    | None
)

Keep only the cells whose source coordinates fall inside output_domain.

Chunk resolution is the canonical caller: intersecting a request with one chunk's box keeps the cells that chunk can serve.

Returns (restricted_transform, out_indices) or None if empty.

out_indices carries the surviving output positions: None when all positions survive (ConstantMap/DimensionMap only), a single integer array for one ArrayMap (or correlated/vectorized ArrayMaps), or a dict keyed by output dimension for >= 2 orthogonal ArrayMaps (an outer product).

Source code in src/zarr_indexing/transform.py

def intersect(
    self, output_domain: IndexDomain
) -> (
    tuple[
        IndexTransform,
        dict[int, np.ndarray[Any, np.dtype[np.intp]]]
        | np.ndarray[Any, np.dtype[np.intp]]
        | None,
    ]
    | None
):
    """Keep only the cells whose source coordinates fall inside `output_domain`.

    Chunk resolution is the canonical caller: intersecting a request with
    one chunk's box keeps the cells that chunk can serve.

    Returns `(restricted_transform, out_indices)` or None if empty.

    `out_indices` carries the surviving output positions: `None` when all
    positions survive (ConstantMap/DimensionMap only), a single integer array
    for one ArrayMap (or correlated/vectorized ArrayMaps), or a dict keyed by
    output dimension for >= 2 orthogonal ArrayMaps (an outer product).
    """
    return _intersect(self, output_domain)

inverted

inverted() -> IndexTransform

Return the restricted, exactly representable inverse transform.

Inversion is defined for square transforms containing only constants and unique unit-stride dimension maps. Any input dimension not named by a dimension map must have singleton extent, so its coordinate can be recovered as a constant.

Returns:

• IndexTransform –

  A new transform mapping output coordinates back to input coordinates.

Raises:

• ValueError –

  If this transform does not have a representable inverse, including when input labels cannot be transferred to unlabeled output dimensions.

Source code in src/zarr_indexing/transform.py

def inverted(self) -> IndexTransform:
    """Return the restricted, exactly representable inverse transform.

    Inversion is defined for square transforms containing only constants
    and unique unit-stride dimension maps. Any input dimension not named by
    a dimension map must have singleton extent, so its coordinate can be
    recovered as a constant.

    Returns
    -------
    IndexTransform
        A new transform mapping output coordinates back to input
        coordinates.

    Raises
    ------
    ValueError
        If this transform does not have a representable inverse, including
        when input labels cannot be transferred to unlabeled output
        dimensions.
    """
    if self.domain.labels is not None:
        raise ValueError(
            "cannot invert transform: input labels cannot be represented "
            "because output dimensions do not carry labels"
        )
    if self.input_rank != self.output_rank:
        raise ValueError(
            "cannot invert transform: input rank must equal output rank, got "
            f"{self.input_rank} and {self.output_rank}"
        )

    referenced: set[int] = set()
    for output_dimension, output_map in enumerate(self.output):
        if isinstance(output_map, ArrayMap):
            raise ValueError(  # noqa: TRY004 - valid map, invalid inverse
                f"cannot invert transform: output[{output_dimension}] is an ArrayMap"
            )
        if isinstance(output_map, DimensionMap):
            if output_map.stride not in (-1, 1):
                raise ValueError(
                    "cannot invert transform: DimensionMap stride must be +1 or -1, "
                    f"got {output_map.stride} for output[{output_dimension}]"
                )
            if output_map.input_dimension in referenced:
                raise ValueError(
                    "cannot invert transform: input dimension "
                    f"{output_map.input_dimension} is referenced more than once"
                )
            referenced.add(output_map.input_dimension)

    for input_dimension, extent in enumerate(self.domain.shape):
        if input_dimension not in referenced and extent != 1:
            raise ValueError(
                "cannot invert transform: unreferenced input dimension "
                f"{input_dimension} has extent {extent}, not 1"
            )

    inverse_min: list[int] = []
    inverse_max: list[int] = []
    inverse_output: dict[int, OutputIndexMap] = {}
    for output_dimension, output_map in enumerate(self.output):
        if isinstance(output_map, ConstantMap):
            inverse_min.append(output_map.offset)
            inverse_max.append(output_map.offset + 1)
            continue

        assert isinstance(output_map, DimensionMap)
        input_dimension = output_map.input_dimension
        lower = self.domain.inclusive_min[input_dimension]
        upper = self.domain.exclusive_max[input_dimension]
        if output_map.stride == 1:
            inverse_min.append(output_map.offset + lower)
            inverse_max.append(output_map.offset + upper)
            inverse_output[input_dimension] = DimensionMap(
                output_dimension,
                offset=-output_map.offset,
            )
        else:
            inverse_min.append(output_map.offset - upper + 1)
            inverse_max.append(output_map.offset - lower + 1)
            inverse_output[input_dimension] = DimensionMap(
                output_dimension,
                offset=output_map.offset,
                stride=-1,
            )

    for input_dimension, lower in enumerate(self.domain.inclusive_min):
        if input_dimension not in referenced:
            inverse_output[input_dimension] = ConstantMap(lower)

    return IndexTransform(
        domain=IndexDomain(tuple(inverse_min), tuple(inverse_max)),
        output=tuple(inverse_output[dimension] for dimension in range(self.input_rank)),
    )

select

select(
    selection: Any,
    mode: Literal[
        "basic", "orthogonal", "vectorized"
    ] = "basic",
) -> IndexTransform

Convert a user selection into a composed IndexTransform.

Negative indices are treated as literal coordinates (TensorStore convention). The caller (Array layer) is responsible for converting numpy-style negative indices before calling this function.

Examples:

The mode picks the dialect; the result is the composed self the corresponding accessor builds:

>>> t = IndexTransform.from_shape((10,))
>>> t.select(slice(2, 8)) == t[2:8]
True
>>> s = t.select(([9, 0, 0],), mode="orthogonal")
>>> s.apply((0,)), s.apply((1,)), s.apply((2,))
((9,), (0,), (0,))

Source code in src/zarr_indexing/transform.py

def select(
    self,
    selection: Any,
    mode: Literal["basic", "orthogonal", "vectorized"] = "basic",
) -> IndexTransform:
    """Convert a user selection into a composed IndexTransform.

    Negative indices are treated as literal coordinates (TensorStore convention).
    The caller (Array layer) is responsible for converting numpy-style negative
    indices before calling this function.

    Examples
    --------
    The `mode` picks the dialect; the result is the composed self the
    corresponding accessor builds:

    >>> t = IndexTransform.from_shape((10,))
    >>> t.select(slice(2, 8)) == t[2:8]
    True
    >>> s = t.select(([9, 0, 0],), mode="orthogonal")
    >>> s.apply((0,)), s.apply((1,)), s.apply((2,))
    ((9,), (0,), (0,))
    """
    if mode == "basic":
        _validate_basic_selection(selection)
        return self[selection]
    elif mode == "orthogonal":
        _validate_array_selection(selection, self.domain.shape, mode)
        return self.oindex[selection]
    elif mode == "vectorized":
        _validate_array_selection(selection, self.domain.shape, mode)
        return self.vindex[selection]
    else:
        raise ValueError(f"Unknown mode: {mode!r}")

to_json

to_json() -> IndexTransformJSON

Convert to the canonical ndsel transform body (spec section 4.3).

The result is fully explicit: input_rank, fully written bounds and labels, and an output carrying offset/stride on every affine and array map. It is field-for-field a TensorStore IndexTransform minus the kind discriminator, so it loads directly into tensorstore.IndexTransform(json=...).

Examples:

>>> body = IndexTransform.from_shape((6,))[1:5:2].to_json()
>>> (body["input_inclusive_min"], body["input_exclusive_max"])
([0], [2])
>>> body["output"]
[{'offset': 1, 'stride': 2, 'input_dimension': 0}]

Source code in src/zarr_indexing/transform.py

def to_json(self) -> IndexTransformJSON:
    """Convert to the canonical ndsel transform body (spec section 4.3).

    The result is fully explicit: `input_rank`, fully written bounds and
    labels, and an `output` carrying `offset`/`stride` on every affine and
    array map. It is field-for-field a TensorStore `IndexTransform` minus
    the `kind` discriminator, so it loads directly into
    `tensorstore.IndexTransform(json=...)`.

    Examples
    --------
    >>> body = IndexTransform.from_shape((6,))[1:5:2].to_json()
    >>> (body["input_inclusive_min"], body["input_exclusive_max"])
    ([0], [2])
    >>> body["output"]
    [{'offset': 1, 'stride': 2, 'input_dimension': 0}]
    """
    from zarr_indexing._wire import emit_labels

    return {
        "input_rank": self.domain.ndim,
        "input_inclusive_min": list(self.domain.inclusive_min),
        "input_exclusive_max": list(self.domain.exclusive_max),
        "input_labels": emit_labels(self.domain.labels, self.domain.ndim),
        "output": [m.to_json() for m in self.output],
    }

translate

translate(shift: tuple[int, ...]) -> IndexTransform

Shift the source coordinates every cell reads by shift, per dimension.

The domain is untouched: the result keeps its cells, and each one reads from a shifted source address — for example, making a chunk's global addresses chunk-local by translating by the chunk's negated origin.

Source code in src/zarr_indexing/transform.py

def translate(self, shift: tuple[int, ...]) -> IndexTransform:
    """Shift the source coordinates every cell reads by `shift`, per dimension.

    The domain is untouched: the result keeps its cells, and each one
    reads from a shifted source address — for example, making a chunk's
    global addresses chunk-local by translating by the chunk's negated
    origin.
    """
    if len(shift) != self.output_rank:
        raise ValueError(f"shift must have length {self.output_rank}, got {len(shift)}")
    new_output: list[OutputIndexMap] = []
    for m, s in zip(self.output, shift, strict=True):
        if isinstance(m, ConstantMap):
            new_output.append(ConstantMap(offset=m.offset + s))
        elif isinstance(m, DimensionMap):
            new_output.append(
                DimensionMap(
                    input_dimension=m.input_dimension,
                    offset=m.offset + s,
                    stride=m.stride,
                )
            )
        else:
            # m: ArrayMap (OutputIndexMap = ConstantMap | DimensionMap | ArrayMap)
            new_output.append(
                m._with_affine(m.offset + s, m.stride)  # pyright: ignore[reportPrivateUsage]
            )
    return IndexTransform(domain=self.domain, output=tuple(new_output))

translate_domain_by

translate_domain_by(
    shift: tuple[int, ...],
) -> IndexTransform

Shift the input domain by shift, preserving which cells are addressed.

TensorStore's translate_by: the domain moves, and every output map is re-offset so that new coordinate c addresses the cell that c - shift addressed before. ArrayMaps are indexed positionally over the domain, so their index arrays are unchanged.

Source code in src/zarr_indexing/transform.py

def translate_domain_by(self, shift: tuple[int, ...]) -> IndexTransform:
    """Shift the *input* domain by `shift`, preserving which cells are addressed.

    TensorStore's `translate_by`: the domain moves, and every output map is
    re-offset so that new coordinate `c` addresses the cell that `c - shift`
    addressed before. ArrayMaps are indexed positionally over the domain, so
    their index arrays are unchanged.
    """
    if len(shift) != self.input_rank:
        raise ValueError(f"shift must have length {self.input_rank}, got {len(shift)}")
    new_domain = self.domain.translate(shift)
    new_output: list[OutputIndexMap] = []
    for m in self.output:
        if isinstance(m, DimensionMap):
            s = shift[m.input_dimension]
            new_output.append(
                DimensionMap(
                    input_dimension=m.input_dimension,
                    offset=m.offset - m.stride * s,
                    stride=m.stride,
                )
            )
        else:
            # ConstantMap: no input dependence. ArrayMap: positional over
            # the domain, invariant under domain translation.
            new_output.append(m)
    return IndexTransform(domain=new_domain, output=tuple(new_output))

translate_domain_to

translate_domain_to(
    origins: tuple[int, ...],
) -> IndexTransform

Move the input domain so its per-dimension origins equal origins.

TensorStore's translate_to; translate_domain_to((0,) * rank) re-zeros a view's coordinate system without changing which cells it addresses.

Source code in src/zarr_indexing/transform.py

def translate_domain_to(self, origins: tuple[int, ...]) -> IndexTransform:
    """Move the input domain so its per-dimension origins equal `origins`.

    TensorStore's `translate_to`; `translate_domain_to((0,) * rank)`
    re-zeros a view's coordinate system without changing which cells it
    addresses.
    """
    if len(origins) != self.input_rank:
        raise ValueError(f"origins must have length {self.input_rank}, got {len(origins)}")
    shift = tuple(o - m for o, m in zip(origins, self.domain.inclusive_min, strict=True))
    return self.translate_domain_by(shift)