feat: model the remaining RFC-5 transforms in both ports

docs/rfc5.md

@@ -4,9 +4,10 @@
 
 [RFC-5] extends OME-NGFF with named **coordinate systems** and a richer set of
 **coordinate transformations** — identity, scale, translation, rotation,
-affine, transformation sequences, and array-backed *displacement* and
-*coordinate* fields. This is the OME-Zarr v0.6 data model. `ngff-zarr` reads and
-writes it in both the Python and TypeScript packages.
+affine, axis permutations, transformation sequences, per-dimension and
+invertible wrappers, and array-backed *displacement* and *coordinate* fields.
+This is the OME-Zarr v0.6 data model. `ngff-zarr` reads and writes it in both
+the Python and TypeScript packages.
 
 ## Overview
 
@@ -76,6 +77,23 @@ The transformation data classes live in `ngff_zarr.v06.zarr_metadata`:
 | `Displacements` | `displacements` | `path: str`, `interpolation: str` |
 | `Coordinates` | `coordinates` | `path: str`, `interpolation: str` |
 | `TransformSequence` | `sequence` | `transformations: list[Transform]` |
+| `MapAxis` | `mapAxis` | `mapAxis: list[int]` |
+| `ByDimension` | `byDimension` | `transformations: list[ByDimensionItem]` |
+| `Bijection` | `bijection` | `forward: Transform`, `inverse: Transform` |
+
+`MapAxis` stores an axis permutation as a transpose vector: the value at
+position `i` is the input axis that becomes the `i`-th output axis, and every
+zero-based input axis index appears exactly once. `ByDimension` builds a high
+dimensional transform from lower dimensional ones; each `ByDimensionItem`
+wraps a transformation with the `input_axes` and `output_axes` (zero-based
+indices into the parent's coordinate systems) it applies to, and every output
+axis is produced by exactly one item. `Bijection` pairs an explicit `forward`
+transformation with its `inverse`. Constraints that follow from a transform's
+parameters alone are enforced when it is constructed, so an invalid instance
+cannot exist; the ones that depend on the resolved input and output
+coordinate systems are checked on read, and by
+`ngff_zarr.v06.zarr_metadata.validate_transform` (or the transform's own
+`validate` method) for programmatically built transforms.
 
 Every transform has an `input` and `output`, each a
 `CoordinateSystemIdentifier` naming a coordinate system (`name=`) or referencing

py/ngff_zarr/spec/0.6/schemas/coordinate_transformations.schema

@@ -322,16 +322,18 @@
               "input_axes": {
                 "type": "array",
                 "items": {
-                  "type": "number"
+                  "type": "integer",
+                  "minimum": 0
                 },
-                "description": "Names of the input axes for this transformation."
+                "description": "Zero-based axis indices into the parent byDimension transformation's input coordinate system."
               },
               "output_axes": {
                 "type": "array",
                 "items": {
-                  "type": "number"
+                  "type": "integer",
+                  "minimum": 0
                 },
-                "description": "Names of the output axes for this transformation."
+                "description": "Zero-based axis indices into the parent byDimension transformation's output coordinate system."
               }
             },
             "required": [

py/ngff_zarr/v06/zarr_metadata.py

@@ -56,6 +56,36 @@ class CoordinateSystemIdentifier:
     path: str | None = None
     name: str | None = None
 
+    def axis_count(
+        self, coordinateSystems: list[CoordinateSystem] | None
+    ) -> int | None:
+        """Number of axes of the referenced coordinate system, when resolvable.
+
+        Returns ``None`` when this identifier names no system in
+        ``coordinateSystems``, as for the wrapped transforms that omit
+        ``input`` and ``output``.
+        """
+        if self.name is None or not coordinateSystems:
+            return None
+        for coordinate_system in coordinateSystems:
+            if coordinate_system.name == self.name:
+                return len(coordinate_system.axes)
+        return None
+
+
+def _resolved_axis_count(
+    identifier: CoordinateSystemIdentifier | None,
+    coordinateSystems: list[CoordinateSystem] | None,
+) -> int | None:
+    return None if identifier is None else identifier.axis_count(coordinateSystems)
+
+
+def _require_integer_axes(axes: list, context: str) -> None:
+    """Axis indices are zero-based integer positions; reject anything else."""
+    for axis in axes:
+        if isinstance(axis, bool) or not isinstance(axis, int):
+            raise ValueError(f"{context} axis indices must be integers; got {axis!r}")
+
 
 @dataclass(kw_only=True)
 class BaseTransform(ABC):  # noqa: B024
@@ -73,6 +103,23 @@ def to_dict(self) -> dict:
     def from_dict(cls, data: dict) -> "BaseTransform":
         return cls(**data)
 
+    def validate(  # noqa: B027
+        self, coordinateSystems: list[CoordinateSystem] | None = None
+    ) -> None:
+        """Check this transform's RFC-5 constraints.
+
+        Constraints that follow from the parameters alone are enforced by
+        ``__post_init__``, so an invalid instance cannot be constructed;
+        this method checks them again, which covers instances mutated after
+        construction. Constraints against the ``input`` and ``output``
+        coordinate systems run when those resolve in ``coordinateSystems``
+        and are skipped otherwise. Raises ``ValueError`` on the first
+        violation.
+
+        Transform types without constraints beyond their field types, such
+        as ``Scale`` or ``Identity``, inherit this no-op deliberately.
+        """
+
 
 @dataclass(kw_only=True)
 class Identity(BaseTransform):
@@ -119,6 +166,45 @@ class Displacements(BaseTransform):
     type: str = "displacements"
 
 
+@dataclass(kw_only=True)
+class MapAxis(BaseTransform):
+    """An axis permutation stored as a transpose vector of integer indices.
+
+    The value at position ``i`` names which input axis becomes the ``i``-th
+    output axis. Every zero-based input axis index appears exactly once.
+    """
+
+    mapAxis: list[int]
+    type: str = "mapAxis"
+
+    def __post_init__(self) -> None:
+        self._check_intrinsic()
+
+    def _check_intrinsic(self) -> None:
+        indices = self.mapAxis
+        _require_integer_axes(indices, "mapAxis")
+        if not (2 <= len(indices) <= 5):
+            raise ValueError(
+                "mapAxis must hold between 2 and 5 indices, one per axis "
+                f"of the coordinate systems it permutes; got {indices}"
+            )
+        if sorted(indices) != list(range(len(indices))):
+            raise ValueError(
+                "mapAxis must be a permutation holding every zero-based "
+                f"input axis index exactly once; got {indices}"
+            )
+
+    def validate(self, coordinateSystems: list[CoordinateSystem] | None = None) -> None:
+        self._check_intrinsic()
+        for identifier in (self.input, self.output):
+            count = _resolved_axis_count(identifier, coordinateSystems)
+            if count is not None and count != len(self.mapAxis):
+                raise ValueError(
+                    f"mapAxis length {len(self.mapAxis)} does not match the "
+                    f"{count} axes of coordinate system '{identifier.name}'"
+                )
+
+
 Transform = Union[
     Identity,
     Scale,
@@ -127,16 +213,201 @@ class Displacements(BaseTransform):
     Affine,
     Coordinates,
     Displacements,
+    MapAxis,
+    "ByDimension",
+    "Bijection",
     "TransformSequence",
 ]
 
 
+def _item_dimensions(transformation: "Transform") -> int | None:
+    """Dimensionality a byDimension item's axis lists must have, if knowable."""
+    if isinstance(transformation, Scale):
+        return len(transformation.scale)
+    if isinstance(transformation, Translation):
+        return len(transformation.translation)
+    if isinstance(transformation, MapAxis):
+        return len(transformation.mapAxis)
+    return None
+
+
+@dataclass
+class ByDimensionItem:
+    """One lower-dimensional transformation of a byDimension transform.
+
+    ``input_axes`` and ``output_axes`` hold zero-based axis indices into the
+    parent byDimension's input and output coordinate systems.
+    """
+
+    transformation: Transform
+    input_axes: list[int]
+    output_axes: list[int]
+
+    def __post_init__(self) -> None:
+        self._check_intrinsic()
+
+    def _check_intrinsic(self) -> None:
+        axes = list(self.input_axes) + list(self.output_axes)
+        _require_integer_axes(axes, "byDimension")
+        if any(axis < 0 for axis in axes):
+            raise ValueError(
+                f"byDimension axis indices must be non-negative; got {axes}"
+            )
+        if len(set(self.output_axes)) != len(self.output_axes):
+            raise ValueError(
+                "byDimension output axes must each be produced by exactly "
+                f"one transformation; {self.output_axes} repeats an axis"
+            )
+        dimensions = _item_dimensions(self.transformation)
+        if dimensions is not None and (
+            len(self.input_axes) != dimensions or len(self.output_axes) != dimensions
+        ):
+            raise ValueError(
+                f"byDimension item of type '{self.transformation.type}' is "
+                f"{dimensions}-dimensional but maps {len(self.input_axes)} "
+                f"input axes to {len(self.output_axes)} output axes"
+            )
+
+    @classmethod
+    def from_dict(
+        cls, data: dict, coordinateSystems: list[CoordinateSystem] | None = None
+    ) -> "ByDimensionItem":
+        (transformation,) = Metadata._parse_transforms(
+            [data["transformation"]], coordinateSystems or []
+        )
+        return cls(
+            transformation=transformation,
+            input_axes=list(data["input_axes"]),
+            output_axes=list(data["output_axes"]),
+        )
+
+
+@dataclass(kw_only=True)
+class ByDimension(BaseTransform):
+    """A high dimensional transform built from lower dimensional ones.
+
+    Every axis index of the output coordinate system appears in exactly one
+    item's ``output_axes``.
+    """
+
+    transformations: list[ByDimensionItem]
+    type: str = "byDimension"
+
+    def __post_init__(self) -> None:
+        self._check_intrinsic()
+
+    def _check_intrinsic(self) -> None:
+        seen_output_axes: set[int] = set()
+        for item in self.transformations:
+            duplicated = seen_output_axes.intersection(item.output_axes)
+            if duplicated:
+                raise ValueError(
+                    "byDimension output axes must each be produced by exactly "
+                    f"one transformation; axis {sorted(duplicated)} appears "
+                    "more than once"
+                )
+            seen_output_axes.update(item.output_axes)
+
+    @property
+    def produced_output_axes(self) -> set[int]:
+        """The output axis indices the items produce, taken together."""
+        axes: set[int] = set()
+        for item in self.transformations:
+            axes.update(item.output_axes)
+        return axes
+
+    def validate(self, coordinateSystems: list[CoordinateSystem] | None = None) -> None:
+        for item in self.transformations:
+            item._check_intrinsic()
+            item.transformation.validate(coordinateSystems)
+        self._check_intrinsic()
+        input_count = _resolved_axis_count(self.input, coordinateSystems)
+        if input_count is not None:
+            for item in self.transformations:
+                if any(axis >= input_count for axis in item.input_axes):
+                    raise ValueError(
+                        f"byDimension input axes {item.input_axes} exceed the "
+                        f"{input_count} axes of coordinate system "
+                        f"'{self.input.name}'"
+                    )
+        output_count = _resolved_axis_count(self.output, coordinateSystems)
+        if output_count is not None:
+            produced = self.produced_output_axes
+            if produced != set(range(output_count)):
+                raise ValueError(
+                    "byDimension items must cover every output axis exactly "
+                    f"once; coordinate system '{self.output.name}' has "
+                    f"{output_count} axes but the items produce {sorted(produced)}"
+                )
+
+    @classmethod
+    def from_dict(
+        cls, data: dict, coordinateSystems: list[CoordinateSystem] | None = None
+    ) -> "ByDimension":
+        return cls(
+            transformations=[
+                ByDimensionItem.from_dict(item, coordinateSystems)
+                for item in data["transformations"]
+            ]
+        )
+
+
+@dataclass(kw_only=True)
+class Bijection(BaseTransform):
+    """An invertible transform with explicit forward and inverse directions."""
+
+    forward: Transform
+    inverse: Transform
+    type: str = "bijection"
+
+    def validate(self, coordinateSystems: list[CoordinateSystem] | None = None) -> None:
+        self.forward.validate(coordinateSystems)
+        self.inverse.validate(coordinateSystems)
+        input_count = _resolved_axis_count(self.input, coordinateSystems)
+        output_count = _resolved_axis_count(self.output, coordinateSystems)
+        if (
+            input_count is not None
+            and output_count is not None
+            and input_count != output_count
+        ):
+            raise ValueError(
+                "bijection input and output coordinate systems must have the "
+                f"same dimensionality; got {input_count} and {output_count}"
+            )
+
+    @classmethod
+    def from_dict(
+        cls, data: dict, coordinateSystems: list[CoordinateSystem] | None = None
+    ) -> "Bijection":
+        systems = coordinateSystems or []
+        (forward,) = Metadata._parse_transforms([data["forward"]], systems)
+        (inverse,) = Metadata._parse_transforms([data["inverse"]], systems)
+        return cls(forward=forward, inverse=inverse)
+
+
 @dataclass(kw_only=True)
 class TransformSequence(BaseTransform):
     transformations: list[Transform]
     name: str | None = "transformSequence"
     type: str = "sequence"
 
+    def validate(self, coordinateSystems: list[CoordinateSystem] | None = None) -> None:
+        for transformation in self.transformations:
+            transformation.validate(coordinateSystems)
+
+
+def validate_transform(
+    transformation: Transform,
+    coordinateSystems: list[CoordinateSystem] | None = None,
+) -> None:
+    """Check a transform's RFC-5 constraints; see ``BaseTransform.validate``.
+
+    Constraints that follow from the parameters alone hold by construction.
+    This also checks the ones that need the ``input`` and ``output``
+    coordinate systems, when those resolve in ``coordinateSystems``.
+    """
+    transformation.validate(coordinateSystems)
+
 
 @dataclass
 class Dataset:
@@ -536,6 +807,12 @@ def _parse_transforms(
                 transformation = Coordinates.from_dict(transform)
             elif transform["type"] == "displacements":
                 transformation = Displacements.from_dict(transform)
+            elif transform["type"] == "mapAxis":
+                transformation = MapAxis(mapAxis=list(transform["mapAxis"]))
+            elif transform["type"] == "byDimension":
+                transformation = ByDimension.from_dict(transform, coordinateSystems)
+            elif transform["type"] == "bijection":
+                transformation = Bijection.from_dict(transform, coordinateSystems)
             elif transform["type"] == "sequence":
                 # TODO: Undo nested sequences on import?
                 sub_transforms = cls._parse_transforms(
@@ -570,6 +847,12 @@ def _parse_transforms(
 
             transformation.input = input
             transformation.output = output
+            # The wrapper branches above construct their transform from the
+            # payload fields alone, so the optional name is restored here for
+            # every type.
+            if transform.get("name") is not None:
+                transformation.name = transform["name"]
+            validate_transform(transformation, coordinateSystems)
             parsed_transforms.append(transformation)
 
         return parsed_transforms