diff --git a/.gitignore b/.gitignore index 23fa4b3d2..421f25d39 100644 --- a/.gitignore +++ b/.gitignore @@ -61,3 +61,14 @@ node_modules/ uv.lock pixi.lock +# local hatch config +hatch.toml + +# Kilo and plans +.kilo/ +plans/ + +# test coverage +.coverage +htmlcov/ + diff --git a/pyproject.toml b/pyproject.toml index 03181eadb..f25a5fbb3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -64,6 +64,7 @@ extra = [ [dependency-groups] dev = [ "bump2version", + "pandas-stubs>=3.0.3.260530", ] test = [ "pytest", diff --git a/src/spatialdata/__init__.py b/src/spatialdata/__init__.py index 7ba66e710..81e63f6c6 100644 --- a/src/spatialdata/__init__.py +++ b/src/spatialdata/__init__.py @@ -59,7 +59,6 @@ "polygon_query": "spatialdata._core.query.spatial_query", # _core.spatialdata "SpatialData": "spatialdata._core.spatialdata", - # _io._utils "get_dask_backing_files": "spatialdata._io._utils", # _io.format "SpatialDataFormatType": "spatialdata._io.format", @@ -208,6 +207,9 @@ def __dir__() -> list[str]: # _core.spatialdata from spatialdata._core.spatialdata import SpatialData + # _core.transformation_manager + from spatialdata._core.transformation_manager import TransformationManager + # _io._utils from spatialdata._io._utils import get_dask_backing_files diff --git a/src/spatialdata/_core/spatialdata.py b/src/spatialdata/_core/spatialdata.py index fb55ab086..8f81a8d7c 100644 --- a/src/spatialdata/_core/spatialdata.py +++ b/src/spatialdata/_core/spatialdata.py @@ -22,6 +22,7 @@ from zarr.errors import GroupNotFoundError from spatialdata._core._elements import Images, Labels, Points, Shapes, Tables +from spatialdata._core.transformation_manager import TransformationManager from spatialdata._core.validation import ( check_all_keys_case_insensitively_unique, check_target_region_column_symmetry, @@ -130,6 +131,7 @@ def __init__( self._shapes: Shapes = Shapes(shared_keys=self._shared_keys) self._tables: Tables = Tables(shared_keys=self._shared_keys) self.attrs = attrs if attrs else {} # type: ignore[assignment] + self.transformation_manager = TransformationManager() # Initialize the TransformationManager element_names = list(chain.from_iterable([e.keys() for e in [images, labels, points, shapes] if e is not None])) diff --git a/src/spatialdata/_core/transformation_manager/__init__.py b/src/spatialdata/_core/transformation_manager/__init__.py new file mode 100644 index 000000000..6e8f227d0 --- /dev/null +++ b/src/spatialdata/_core/transformation_manager/__init__.py @@ -0,0 +1,623 @@ +from __future__ import annotations + +from collections.abc import Sequence + +import networkx as nx + +from spatialdata._core.transformation_manager.exceptions import ( + CannotRemoveCoordinateSystemError, + CoordinateSystemAlreadyExistsError, + CoordinateSystemHasElementsError, + CoordinateSystemHasTransformationsError, + CoordinateSystemNotFoundError, + ElementAlreadyExistsError, + ElementNotRegisteredToAnyCoordinateSystemError, + InvalidPathError, + TransformationNotFoundError, + TransformationPathAmbiguousError, + TransformationPathAmbiguousMultipleEdgeExpectedError, + TransformationPathAmbiguousNoEdgeExpectedError, + TransformationPathNotFoundError, + TransformationPathNotSimple, +) +from spatialdata.transformations import transformations as sd_transforms +from spatialdata.transformations.ngff.ngff_coordinate_system import NgffCoordinateSystem + +TRANSFORM_KEY = "transformation" +EDGE_DEF = tuple[NgffCoordinateSystem, NgffCoordinateSystem, sd_transforms.BaseTransformation] + + +class TransformationManager: + def __init__(self) -> None: + """Initialize a TransformationManager with empty graph and mappings.""" + self._graph: nx.MultiDiGraph[NgffCoordinateSystem] = nx.MultiDiGraph() + # MultiDiGraph with NgffCoordinateSystem objects as nodes and transforms as edge attributes + self._element_to_cs_mapping: dict[str, NgffCoordinateSystem] = {} + # mapping element_name to the coordinate system to which the element belongs + + def assert_element_exists(self, element_name: str) -> None: + """ + Assert that an element exists in the transformation manager. + + Parameters + ---------- + element_name + The name of the element to check. + + Raises + ------ + ElementNotFoundError + If the element does not exist. + """ + if element_name not in self._element_to_cs_mapping: + raise ElementNotRegisteredToAnyCoordinateSystemError(element_name) + + def assert_element_does_not_exist(self, element_name: str) -> None: + """ + Assert than an element doesn't exist in the transformation manager. + + Parameters + ---------- + element_name + The name of the element to check. + + Raises + ------ + ElementAlreadyExistsError + If the element already exists. + """ + if element_name in self._element_to_cs_mapping: + raise ElementAlreadyExistsError(element_name) + + def assert_coordinate_system_exists(self, cs: NgffCoordinateSystem) -> None: + """ + Assert that coordinate system exists in the graph. + + Parameters + ---------- + cs + The coordinate system to check. + + Raises + ------ + CoordinateSystemNotFoundError + If the coordinate system does not exist. + """ + if cs not in self._graph: + raise CoordinateSystemNotFoundError(cs.name) + + def assert_an_edge_exists_between_coordinate_systems( + self, source_cs: NgffCoordinateSystem, target_cs: NgffCoordinateSystem, edge_key: str | None = None + ) -> None: + """ + Assert that an edge exists between coordinate systems. + + Parameters + ---------- + source_cs + The input coordinate system. + target_cs + The output coordinate system. + edge_key + If specified, asserts that the specific edge exists between coordinate systems + + + Raises + ------ + CoordinateSystemNotFoundError + If either coordinate system does not exist. + TransformationNotFoundError + If the edge does not exist. + """ + self.assert_coordinate_system_exists(source_cs) + self.assert_coordinate_system_exists(target_cs) + if not self._graph.has_edge(source_cs, target_cs): + raise TransformationNotFoundError(source_cs.name, target_cs.name, edge_key) + + def assert_coordinate_system_has_no_transformations(self, cs: NgffCoordinateSystem) -> None: + """ + Assert that coordinate system has no associated transformations. + + Parameters + ---------- + cs + The coordinate system to check. + + Raises + ------ + CoordinateSystemNotFoundError + If either coordinate system does not exist. + CoordinateSystemHasTransformationsError + If the coordinate system has associated transformations. + + """ + self.assert_coordinate_system_exists(cs) + has_successors = next(self._graph.successors(cs), None) is not None + has_predecessors = next(self._graph.predecessors(cs), None) is not None + + if has_successors or has_predecessors: + raise CoordinateSystemHasTransformationsError(cs.name) + + def assert_coordinate_system_has_no_elements(self, cs: NgffCoordinateSystem) -> None: + """ + Assert that a coordinate system doesn't have elements belonging to it. + + Parameters + ---------- + cs + The coordinate system to check + + Raises + ------ + Co + + """ + elements = self._get_elements_belonging_to_cs(cs) + # also checks if cs exists + if elements: + raise CoordinateSystemHasElementsError(cs.name, elements) + + def add_coordinate_system(self, cs: NgffCoordinateSystem) -> None: + """ + Add a coordinate system to the transformation manager. + + Parameters + ---------- + cs + The coordinate system to add. + + Raises + ------ + CoordinateSystemAlreadyExistsError + If the coordinate system already exists. + """ + try: + self.assert_coordinate_system_exists(cs) + except CoordinateSystemNotFoundError: + self._graph.add_node(cs) + return + + raise CoordinateSystemAlreadyExistsError(cs.name) + + def remove_coordinate_system(self, cs: NgffCoordinateSystem) -> None: + """ + Remove a coordinate system from the transformation manager. + + Parameters + ---------- + cs + The coordinate system to remove. + + Raises + ------ + CoordinateSystemNotFoundError + If the coordinate system is not found + CannotRemoveCoordinateSystemError + If the coordinate system cannot be removed. + Chained from + CoordinateSystemHasTransformationsError or + CoordinateSystemHasElementsError when the system has dependencies. + """ + try: + self.assert_coordinate_system_has_no_transformations(cs) + # also asserts that cs exists + self.assert_coordinate_system_has_no_elements(cs) + except (CoordinateSystemHasTransformationsError, CoordinateSystemHasElementsError) as err: + raise CannotRemoveCoordinateSystemError(cs.name) from err + + self._graph.remove_node(cs) + + def list_coordinate_systems(self) -> list[NgffCoordinateSystem]: + """ + List all registered coordinate systems. + + Returns + ------- + A list of coordinate system objects. + """ + return list(self._graph.nodes()) + + def add_element(self, element_name: str, coordinate_system: NgffCoordinateSystem) -> None: + """ + Register an element and associate it with a coordinate system. + + Parameters + ---------- + element_name + The name of the element. + coordinate_system + The coordinate system to which the element belongs. + + Raises + ------ + CoordinateSystemNotFoundError + If the coordinate system is not found. + """ + self.assert_element_does_not_exist(element_name) + self.assert_coordinate_system_exists(coordinate_system) + self._element_to_cs_mapping[element_name] = coordinate_system + + def get_element_coordinate_system(self, element_name: str) -> NgffCoordinateSystem: + """ + Get the coordinate system to which an element belongs. + + Parameters + ---------- + element_name + The name of the element. + + Returns + ------- + The coordinate system + + Raises + ------ + ElementNotFoundError + If the element does not exist. + """ + self.assert_element_exists(element_name) + return self._element_to_cs_mapping[element_name] + + def _unset_element(self, element_name: str) -> None: + """ + Unregister an element from the coordinate system to which it belongs. + + Parameters + ---------- + element_name + The name of the element. + + Raises + ------ + ElementNotFoundError + If the element has not been registered to any coordinate system. + """ + self.assert_element_exists(element_name) + del self._element_to_cs_mapping[element_name] + + @staticmethod + def _get_edge_key(edge_def: EDGE_DEF) -> str: + """ + Encode the definition of an edge into a string to be used as graph-wide edge identifier. + + Parameters + ---------- + edge_def + tuple, (source_cs, target_cs, transformation) + + Returns + ------- + edge_key + str, the encoded edge identifier + """ + source_cs, target_cs, transform = edge_def + return f"{source_cs.name}_{target_cs.name}_{id(transform)}" + + def add_transformation( + self, + source_cs: NgffCoordinateSystem, + target_cs: NgffCoordinateSystem, + transformation: sd_transforms.BaseTransformation, + ) -> None: + """ + Add a transformation between coordinate systems. + + Parameters + ---------- + source_cs + The input coordinate system. + target_cs + The output coordinate system. + transformation + The transformation to add. + + Raises + ------ + CoordinateSystemNotFoundError + If either coordinate system does not exist. + """ + self.assert_coordinate_system_exists(source_cs) + self.assert_coordinate_system_exists(target_cs) + + edge_key = self._get_edge_key((source_cs, target_cs, transformation)) + edge_attributes = {TRANSFORM_KEY: transformation} + self._graph.add_edge(source_cs, target_cs, key=edge_key, **edge_attributes) + + def get_direct_transformations( + self, source_cs: NgffCoordinateSystem, target_cs: NgffCoordinateSystem + ) -> list[sd_transforms.BaseTransformation]: + """ + Retrieve transformations directly defined between coordinate systems. + + Parameters + ---------- + source_cs + The input coordinate system. + target_cs + The output coordinate system. + + Returns + ------- + List of transformations (empty if there are no direct transformations). + + Raises + ------ + CoordinateSystemNotFoundError + If either coordinate system does not exist. + """ + self.assert_coordinate_system_exists(source_cs) + self.assert_coordinate_system_exists(target_cs) + + transforms = [] + if self._graph.has_edge(source_cs, target_cs): + for edge_data in self._graph[source_cs][target_cs].values(): + transform: sd_transforms.BaseTransformation = edge_data[TRANSFORM_KEY] + transforms.append(transform) + return transforms + + def remove_specific_transformation( + self, + source_cs: NgffCoordinateSystem, + target_cs: NgffCoordinateSystem, + transformation: sd_transforms.BaseTransformation, + ) -> None: + """ + Remove a specific transformation between coordinate systems. + + Parameters + ---------- + source_cs + The input coordinate system. + target_cs + The output coordinate system. + transformation + The transformation to remove. + (mainly useful for cases with multiple transformations between the same coordinate systems). + + Raises + ------ + CoordinateSystemNotFoundError + If either coordinate system does not exist. + TransformationNotFoundError + If the transformation does not exist. + """ + expected_edge_key = self._get_edge_key((source_cs, target_cs, transformation)) + self.assert_an_edge_exists_between_coordinate_systems(source_cs, target_cs, expected_edge_key) + # also checks if source_cs and target_cs exist + self._graph.remove_edge(source_cs, target_cs, key=expected_edge_key) + + def remove_all_transformations_between_coordinate_systems( + self, + source_cs: NgffCoordinateSystem, + target_cs: NgffCoordinateSystem, + ) -> None: + """ + Remove all transformation between coordinate systems. + + Parameters + ---------- + source_cs + The input coordinate system. + target_cs + The output coordinate system. + + Raises + ------ + CoordinateSystemNotFoundError + If either coordinate system does not exist. + TransformationNotFoundError + If no transformation exists between the coordiante systems + """ + self.assert_coordinate_system_exists(source_cs) + self.assert_coordinate_system_exists(target_cs) + if self._graph.has_edge(source_cs, target_cs): + for edge_key in list(self._graph[source_cs][target_cs].keys()): + # need to covert keys() to list to freeze it, else it will change during the following removal + self._graph.remove_edge(source_cs, target_cs, key=edge_key) + + def _get_transformation_sequences_from_simple_paths_after_disambiguation( + self, + paths: list[list[NgffCoordinateSystem]], + expected_intermediate_edges: Sequence[EDGE_DEF], + ) -> list[sd_transforms.Sequence]: + """ + Traverses paths to form sequence of Transformations. + + In case of ambiguity looks into `expected_intermediate_transformations` to disambiguate. + + Parameters + ---------- + paths: + sequence of list of nodes + expected_intermediate_edges: + list of edge definitions, for use when multiple edges are found between two coordinate systems in a path + An edge is defined as a tuple, (source_cs, target_cs, transform). + + Returns + ------- + list of transformations, each an instance of the transformation class Sequence + + Raises + ------ + TransformationPathNotSimple + if any path in `paths` is not simple, i.e., has recurring coordinate systems + """ + for path in paths: + if len(set(path)) != len(path): + raise TransformationPathNotSimple(path) + + expected_intermediate_transformation_edge_keys = set() + if expected_intermediate_edges is not None: + for edge_def in expected_intermediate_edges: + edge_key = self._get_edge_key(edge_def) + expected_intermediate_transformation_edge_keys |= {edge_key} + + all_sequences = [] + deduplicated_paths = list({repr(x): x for x in paths}.values()) + for path in deduplicated_paths: + if len(path) <= 1: + raise InvalidPathError(path) + + transformation_list = [] + for i in range(len(path) - 1): + source_cs_here, target_cs_here = path[i], path[i + 1] + edge_data = self._graph[source_cs_here][target_cs_here] + if len(edge_data) > 1: + # when there are multiple edges between a pair of coordinate systems in the path + # find the expected edge based on key + expected_intermediate_transformation_key_here = ( + expected_intermediate_transformation_edge_keys & set(edge_data.keys()) + ) + if len(expected_intermediate_transformation_key_here) == 0: + # transformation was not specified in `expected_intermediate_transformations` for disambiguation + raise TransformationPathAmbiguousNoEdgeExpectedError(source_cs_here.name, target_cs_here.name) + if len(expected_intermediate_transformation_key_here) > 1: + # multiple transformations were specified in `expected_intermediate_transformations` + # for disambiguation + raise TransformationPathAmbiguousMultipleEdgeExpectedError( + source_cs_here.name, target_cs_here.name, len(expected_intermediate_transformation_key_here) + ) + edge_key_to_use = list(expected_intermediate_transformation_key_here)[0] + transformation_list.append(edge_data[edge_key_to_use][TRANSFORM_KEY]) + else: + # Only one edge, no ambiguity + edge_key = next(iter(edge_data.keys())) + transformation_list.append(edge_data[edge_key][TRANSFORM_KEY]) + all_sequences.append(sd_transforms.Sequence(transformation_list)) + return all_sequences + + def get_all_shortest_transformation_sequences( + self, + source_cs: NgffCoordinateSystem, + target_cs: NgffCoordinateSystem, + expected_intermediate_edges: Sequence[EDGE_DEF] = (), + ) -> list[sd_transforms.Sequence]: + """ + Get all shortest sequences of transformations between two coordinate systems. + + Parameters + ---------- + source_cs + The source coordinate system. + target_cs + The target coordinate system. + expected_intermediate_edges + list of intermediate edges expected. + An edge is defined as a tuple, (source_cs, target_cs, transform). + Used to choose an edge when multiple edges are found between the same coordinate systems. + + Returns + ------- + list[Sequence] + All shortest sequences of transformations from source_cs to target_cs. + When multiple transformations are defined between the same coordinate systems, only those containing + a transformation among intermediate transformations are included. + + Raises + ------ + CoordinateSystemNotFoundError + If either coordinate system does not exist. + TransformationPathNotFoundError + If no path exists between the source and target coordinate systems. + TransformationPathAmbiguousError + When multiple transformations are defined between the same coordinate systems and transformations + are not specified in `expeceted_intermediate_transformations` for disambiguation. + """ + try: + paths = list(nx.all_shortest_paths(self._graph, source=source_cs, target=target_cs)) + except nx.NetworkXNoPath as nxe: + raise TransformationPathNotFoundError(source_cs.name, target_cs.name) from nxe + + try: + return self._get_transformation_sequences_from_simple_paths_after_disambiguation( + paths, expected_intermediate_edges + ) + except TransformationPathAmbiguousError as tpae: + raise TransformationPathAmbiguousError(source_cs.name, target_cs.name) from tpae + + def get_all_transformation_sequences( + self, + source_cs: NgffCoordinateSystem, + target_cs: NgffCoordinateSystem, + expected_intermediate_edges: Sequence[EDGE_DEF] = (), + ) -> list[sd_transforms.Sequence]: + """ + Get all existing sequences of transformations between two coordinate systems. + + Parameters + ---------- + source_cs + The source coordinate system. + target_cs + The target coordinate system. + expected_intermediate_edges + list of intermediate edges expected. + An edge is defined as a tuple, (source_cs, target_cs, transform). + Used to choose an edge when multiple edges are found between the same coordinate systems. + + Returns + ------- + list[Sequence] + All existing sequences of transformations from source_cs to target_cs. + When multiple transformations are defined between the same coordinate systems, only those containing + a transformation among intermediate transformations are included. + + Raises + ------ + CoordinateSystemNotFoundError + If either coordinate system does not exist. + TransformationPathNotFoundError + If no path exists between the source and target coordinate systems. + TransformationPathAmbiguousError + When multiple transformations are defined between the same coordinate systems and transformations + are not specified in `expected_intermediate_transformations` for disambiguation. + """ + try: + paths = list(nx.all_simple_paths(self._graph, source=source_cs, target=target_cs)) + except nx.NetworkXNoPath as nxe: + raise TransformationPathNotFoundError(source_cs.name, target_cs.name) from nxe + + try: + return self._get_transformation_sequences_from_simple_paths_after_disambiguation( + paths, expected_intermediate_edges + ) + except TransformationPathAmbiguousError as tpae: + raise tpae from TransformationPathAmbiguousError(source_cs.name, target_cs.name) + + def _get_elements_belonging_to_cs(self, cs: NgffCoordinateSystem) -> list[str]: + """ + Get all elements belonging to a coordinate system. + + Parameters + ---------- + cs + The coordinate system to check. + + Returns + ------- + List of element names belonging to the coordinate system. + + Raises + ------ + CoordinateSystemNotFoundError + If the coordinate system does not exist. + + """ + self.assert_coordinate_system_exists(cs) + + elements = [] + for element_name, element_cs in self._element_to_cs_mapping.items(): + if element_cs == cs: + elements.append(element_name) + + return elements + + def __repr__(self) -> str: + """Return a string representation of the TransformationManager.""" + return ( + f"TransformationManager(" + f" coordinate_systems={list(self._graph.nodes())}, " + f" coordinate_transforms={[x[TRANSFORM_KEY] for *_, x in self._graph.edges(data=True)]}, " + f" elements={list(self._element_to_cs_mapping.keys())}" + f")" + ) diff --git a/src/spatialdata/_core/transformation_manager/exceptions.py b/src/spatialdata/_core/transformation_manager/exceptions.py new file mode 100644 index 000000000..a85ee2544 --- /dev/null +++ b/src/spatialdata/_core/transformation_manager/exceptions.py @@ -0,0 +1,229 @@ +from __future__ import annotations + +from spatialdata.transformations.ngff.ngff_coordinate_system import NgffCoordinateSystem + + +class CoordinateSystemNotFoundError(ValueError): + """ + Exception raised when a coordinate system is not found in the transformation manager. + + Attributes + ---------- + name : str + The name of the coordinate system that was not found. + """ + + def __init__(self, name: str) -> None: + self.name = name + super().__init__(f"Coordinate system '{name}' not found in the transformation manager.") + + +class ElementNotRegisteredToAnyCoordinateSystemError(KeyError): + """ + Exception raised when an element is not found in the transformation manager. + + Attributes + ---------- + element_name : str + The name of the element that was not found. + """ + + def __init__(self, element_name: str) -> None: + self.element_name = element_name + super().__init__(f"Element '{element_name}' has not been registered to any coordinate system.") + + +class TransformationNotFoundError(KeyError): + """ + Exception raised when a transformation is not found between coordinate systems. + + Attributes + ---------- + input_cs_name : str + The name of the input coordinate system. + output_cs_name : str + The name of the output coordinate system. + edge_key: str or None + key used when adding transformation + """ + + def __init__(self, source_cs_name: str, target_cs_name: str, edge_key: str | None = None) -> None: + self.input_cs_name = source_cs_name + self.output_cs_name = target_cs_name + self.edge_key = edge_key + msg = f"Transformation from '{source_cs_name}' to '{target_cs_name}' not found" + if edge_key is not None: + msg += f" with key '{edge_key}'" + super().__init__(msg) + + +class CoordinateSystemAlreadyExistsError(ValueError): + """ + Exception raised when coordinate system already exists. + + Attributes + ---------- + name : str + The name of the coordinate system that already exists. + """ + + def __init__(self, name: str) -> None: + self.name = name + super().__init__(f"Coordinate system '{name}' already exists") + + +class ElementAlreadyExistsError(ValueError): + """ + Exception raised when trying to add an Element that already exists. + + Attributes + ---------- + name : str + The name of the Element that already exists. + """ + + def __init__(self, name: str) -> None: + self.name = name + super().__init__(f"Element '{name}' already exists in the transformation manager") + + +class InvalidPathError(ValueError): + """ + Exception raised when a path is defined with less than 2 nodes. + + Attributes + ---------- + invalid_path : list[NgffCoordinateSystem] + """ + + def __init__(self, invalid_path: list[NgffCoordinateSystem]) -> None: + super().__init__(f"Found an invalid path with less than 2 nodes: {invalid_path}") + + +class TransformationPathNotFoundError(ValueError): + """ + Exception raised when no transformation path exists between coordinate systems. + + Attributes + ---------- + source_cs_name : str + The name of the source coordinate system. + target_cs_name : str + The name of the target coordinate system. + """ + + def __init__(self, source_cs_name: str, target_cs_name: str) -> None: + self.source_cs_name = source_cs_name + self.target_cs_name = target_cs_name + super().__init__(f"No transformation path found from {source_cs_name} to {target_cs_name}") + + +class TransformationPathAmbiguousError(ValueError): + """ + Exception raised when multiple transformation path exists between coordinate systems. + + Attributes + ---------- + source_cs_name : str + The name of the source coordinate system. + target_cs_name : str + The name of the target coordinate system. + """ + + def __init__(self, source_cs_name: str, target_cs_name: str) -> None: + self.source_cs_name = source_cs_name + self.target_cs_name = target_cs_name + base_msg = f"Transformation Path ambiguous from {source_cs_name} to {target_cs_name}." + cause_of_confusion = self.cause_of_confusion() + msg = f"{base_msg} {cause_of_confusion}" if cause_of_confusion else base_msg + super().__init__(msg) + + def cause_of_confusion(self) -> str: + return "" + + +class TransformationPathAmbiguousNoEdgeExpectedError(TransformationPathAmbiguousError): + def __init__(self, source_cs_name: str, target_cs_name: str) -> None: + super().__init__(source_cs_name, target_cs_name) + + def cause_of_confusion(self) -> str: + + return "Multiple edges found. None of them were specified to be expected" + + +class TransformationPathAmbiguousMultipleEdgeExpectedError(TransformationPathAmbiguousError): + def __init__(self, source_cs_name: str, target_cs_name: str, number_of_edges_expected: int) -> None: + self.number_of_edges_expected = number_of_edges_expected + super().__init__(source_cs_name, target_cs_name) + + def cause_of_confusion(self) -> str: + return f"Multiple ({self.number_of_edges_expected}) edges were specified to be expected" + + +class TransformationPathNotSimple(ValueError): + """ + Exception raised when a path represented by a list of coordinate systems is not simple. + + A simple path is one in which each coordinate system appears only once + """ + + def __init__(self, path: list[NgffCoordinateSystem]) -> None: + self.path = path + css_formatted_one_per_line = "\n".join(repr(cs) for cs in path) + super().__init__( + f"Transformation Path not simple, i.e., some coordinate systems appear multiple times:\n" + f"{css_formatted_one_per_line}" + ) + + +class CannotRemoveCoordinateSystemError(ValueError): + """Exception raised when trying to remove a coordinate system. + + Attributes + ---------- + name: str + name of the coordinate system + """ + + def __init__(self, name: str) -> None: + self.name = name + super().__init__(f"Cannot remove coordinate system with name {name}.") + + +class CoordinateSystemHasTransformationsError(ValueError): + """ + Exception raised when trying to remove a coordinate system that has associated transformations. + + Attributes + ---------- + name : str + The name of the coordinate system that has transformations. + """ + + def __init__(self, name: str) -> None: + self.name = name + super().__init__(f"Coordinate System ('{name}') has transformations.") + + +class CoordinateSystemHasElementsError(ValueError): + """ + Exception raised when trying to remove a coordinate system that has associated elements. + + Attributes + ---------- + name : str + The name of the coordinate system that has elements. + associated_elements : list[str] + List of element names associated with the coordinate system. + """ + + def __init__(self, name: str, associated_elements: list[str]) -> None: + self.name = name + self.associated_elements = associated_elements + super().__init__(f"Coordinate system '{name}' has elements belonging to it: {associated_elements}") + + +class TransformationManagerWarning(UserWarning): + """Base warning category for TransformationManager.""" + + pass diff --git a/src/spatialdata/_io/io_raster.py b/src/spatialdata/_io/io_raster.py index 276f016bd..29973ad20 100644 --- a/src/spatialdata/_io/io_raster.py +++ b/src/spatialdata/_io/io_raster.py @@ -2,7 +2,7 @@ from collections.abc import Sequence from pathlib import Path -from typing import Any, Literal, TypeGuard, cast +from typing import Any, Literal, TypeGuard, cast, get_args import dask.array as da import numpy as np @@ -28,6 +28,7 @@ RasterFormatType, get_ome_zarr_format, ) +from spatialdata._types import ELEMENT_TYPE, ELEMENT_TYPE_RASTER, GROUP_NAME from spatialdata._utils import get_pyramid_levels from spatialdata.models.models import ATTRS_KEY from spatialdata.models.pyramids_utils import dask_arrays_to_datatree @@ -160,10 +161,12 @@ def _prepare_storage_options( def _read_multiscale( - store: str | Path, raster_type: Literal["image", "labels"], reader_format: Format + store: str | Path, raster_type: ELEMENT_TYPE_RASTER, reader_format: Format ) -> DataArray | DataTree: assert isinstance(store, str | Path) - assert raster_type in ["image", "labels"] + assert raster_type in get_args(ELEMENT_TYPE_RASTER), ValueError( + f"{raster_type} is not a valid raster type. Must be one of {get_args(ELEMENT_TYPE_RASTER)}." + ) nodes: list[Node] = [] image_loc = ZarrLocation(store, fmt=reader_format) @@ -208,7 +211,7 @@ def _read_multiscale( transformations = _get_transformations_from_ngff_dict(encoded_ngff_transformations) # if image, read channels metadata channels: list[Any] | None = None - if raster_type == "image": + if raster_type == ELEMENT_TYPE.IMAGE: if legacy_channels_metadata is not None: channels = [d["label"] for d in legacy_channels_metadata["channels"]] if omero_metadata is not None: @@ -223,7 +226,7 @@ def _read_multiscale( data = node.load(Multiscales).array(resolution=datasets[0]) si = DataArray( data, - name="image", + name=ELEMENT_TYPE.IMAGE, dims=axes, coords={"c": channels} if channels is not None else {}, ) @@ -259,7 +262,7 @@ def _get_multiscale_nodes(image_nodes: list[Node], nodes: list[Node]) -> list[No def _write_raster( - raster_type: Literal["image", "labels"], + raster_type: ELEMENT_TYPE_RASTER, raster_data: DataArray | DataTree, group: zarr.Group, name: str, @@ -292,12 +295,13 @@ def _write_raster( metadata Additional metadata for the raster element """ - if raster_type not in ["image", "labels"]: - raise ValueError(f"{raster_type} is not a valid raster type. Must be 'image' or 'labels'.") + assert raster_type in get_args(ELEMENT_TYPE_RASTER), ValueError( + f"{raster_type} is not a valid raster type. Must be one of {get_args(ELEMENT_TYPE_RASTER)}." + ) # "name" and "label_metadata" are only used for labels. "name" is written in write_multiscale_ngff() but ignored in # write_image_ngff() (possibly an ome-zarr-py bug). We only use "name" to ensure correct group access in the # ome-zarr API. - if raster_type == "labels": + if raster_type == ELEMENT_TYPE.LABELS: metadata["name"] = name metadata["label_metadata"] = label_metadata @@ -326,8 +330,8 @@ def _write_raster( else: raise ValueError("Not a valid labels object") - group = group["labels"][name] if raster_type == "labels" else group - if raster_type == "image": + group = group[GROUP_NAME.LABELS][name] if raster_type == ELEMENT_TYPE.LABELS else group + if raster_type == ELEMENT_TYPE.IMAGE: # ome-zarr-py >= 0.18 no longer writes the omero channel metadata, so we write it ourselves. overwrite_channel_names(group, raster_data) if ATTRS_KEY not in group.attrs: @@ -418,7 +422,7 @@ def _update_dict_v3(d: dict[str, Any]) -> None: def _write_raster_dataarray( - raster_type: Literal["image", "labels"], + raster_type: ELEMENT_TYPE_RASTER, group: zarr.Group, element_name: str, raster_data: DataArray, @@ -448,7 +452,7 @@ def _write_raster_dataarray( metadata Additional metadata for the raster element """ - write_single_scale_ngff = write_image_ngff if raster_type == "image" else write_labels_ngff + write_single_scale_ngff = write_image_ngff if raster_type == ELEMENT_TYPE.IMAGE else write_labels_ngff data = raster_data.data transformations = _get_transformations(raster_data) @@ -479,7 +483,7 @@ def _write_raster_dataarray( **metadata, ) - trans_group = group["labels"][element_name] if raster_type == "labels" else group + trans_group = group[GROUP_NAME.LABELS][element_name] if raster_type == ELEMENT_TYPE.LABELS else group overwrite_coordinate_transformations_raster( group=trans_group, transformations=transformations, @@ -489,7 +493,7 @@ def _write_raster_dataarray( def _write_raster_datatree( - raster_type: Literal["image", "labels"], + raster_type: ELEMENT_TYPE_RASTER, group: zarr.Group, element_name: str, raster_data: DataTree, @@ -519,7 +523,9 @@ def _write_raster_datatree( metadata Additional metadata for the raster element """ - write_multi_scale_ngff = write_multiscale_ngff if raster_type == "image" else write_multiscale_labels_ngff + write_multi_scale_ngff = ( + write_multiscale_ngff if raster_type == ELEMENT_TYPE.IMAGE else write_multiscale_labels_ngff + ) data = get_pyramid_levels(raster_data, attr="data") list_of_input_axes: list[Any] = get_pyramid_levels(raster_data, attr="dims") assert len(set(list_of_input_axes)) == 1 @@ -562,7 +568,7 @@ def _write_raster_datatree( # This workaround should not be needed once https://github.com/ome/ome-zarr-py/issues/580 is fixed. group = zarr.open_group(store=group.store, path=group.path, mode="r+", use_consolidated=False) - trans_group = group["labels"][element_name] if raster_type == "labels" else group + trans_group = group[GROUP_NAME.LABELS][element_name] if raster_type == ELEMENT_TYPE.LABELS else group overwrite_coordinate_transformations_raster( group=trans_group, transformations=transformations, @@ -582,7 +588,7 @@ def write_image( **metadata: str | JSONDict | list[JSONDict], ) -> None: _write_raster( - raster_type="image", + raster_type=ELEMENT_TYPE.IMAGE, raster_data=image, group=group, name=name, @@ -604,7 +610,7 @@ def write_labels( **metadata: JSONDict, ) -> None: _write_raster( - raster_type="labels", + raster_type=ELEMENT_TYPE.LABELS, raster_data=labels, group=group, name=name, diff --git a/src/spatialdata/_io/io_zarr.py b/src/spatialdata/_io/io_zarr.py index 4c410fab0..10ec2a480 100644 --- a/src/spatialdata/_io/io_zarr.py +++ b/src/spatialdata/_io/io_zarr.py @@ -5,7 +5,7 @@ from collections.abc import Callable from json import JSONDecodeError from pathlib import Path -from typing import Any, Literal, cast +from typing import Any, Literal, cast, get_args import zarr.storage from anndata import AnnData @@ -27,7 +27,13 @@ from spatialdata._io.io_shapes import _read_shapes from spatialdata._io.io_table import _read_table from spatialdata._logging import logger -from spatialdata._types import Raster_T +from spatialdata._types import ( + ELEMENT_TYPE, + ELEMENT_TYPE_RASTER, + ELEMENT_TYPE_VECTOR, + GROUP_NAME, + Raster_T, +) def _read_zarr_group_spatialdata_element( @@ -36,8 +42,8 @@ def _read_zarr_group_spatialdata_element( sdata_version: Literal["0.1", "0.2"], selector: set[str], read_func: Callable[..., Any], - group_name: Literal["images", "labels", "shapes", "points", "tables"], - element_type: Literal["image", "labels", "shapes", "points", "tables"], + group_name: GROUP_NAME, + element_type: ELEMENT_TYPE, element_container: (dict[str, Raster_T] | dict[str, DaskDataFrame] | dict[str, GeoDataFrame] | dict[str, AnnData]), on_bad_files: Literal[BadFileHandleMethod.ERROR, BadFileHandleMethod.WARN], ) -> None: @@ -67,14 +73,14 @@ def _read_zarr_group_spatialdata_element( ValueError, ), ): - if element_type in ["image", "labels"]: + if element_type in get_args(ELEMENT_TYPE_RASTER): reader_format = get_raster_format_for_read(elem_group, sdata_version) element = read_func( elem_group_path, - cast(Literal["image", "labels"], element_type), + cast(ELEMENT_TYPE_RASTER, element_type), reader_format, ) - elif element_type in ["shapes", "points", "tables"]: + elif element_type in get_args(ELEMENT_TYPE_VECTOR) + (ELEMENT_TYPE.TABLES,): element = read_func(elem_group_path) else: raise ValueError(f"Unknown element type {element_type}") @@ -183,19 +189,19 @@ def read_zarr( # we could make this more readable. One can get lost when looking at this dict and iteration over the items group_readers: dict[ - Literal["images", "labels", "shapes", "points", "tables"], + GROUP_NAME, tuple[ Callable[..., Any], - Literal["image", "labels", "shapes", "points", "tables"], + ELEMENT_TYPE, dict[str, Raster_T] | dict[str, DaskDataFrame] | dict[str, GeoDataFrame] | dict[str, AnnData], ], ] = { - # ome-zarr-py needs a kwargs that has "image" has key. So here we have "image" and not "images" - "images": (_read_multiscale, "image", images), - "labels": (_read_multiscale, "labels", labels), - "points": (_read_points, "points", points), - "shapes": (_read_shapes, "shapes", shapes), - "tables": (_read_table, "tables", tables), + # ome-zarr-py needs a kwargs that has "image" as key. So here we have "image" and not "images" + GROUP_NAME.IMAGES: (_read_multiscale, ELEMENT_TYPE.IMAGE, images), + GROUP_NAME.LABELS: (_read_multiscale, ELEMENT_TYPE.LABELS, labels), + GROUP_NAME.POINTS: (_read_points, ELEMENT_TYPE.POINTS, points), + GROUP_NAME.SHAPES: (_read_shapes, ELEMENT_TYPE.SHAPES, shapes), + GROUP_NAME.TABLES: (_read_table, ELEMENT_TYPE.TABLES, tables), } for group_name, ( read_func, diff --git a/src/spatialdata/_types.py b/src/spatialdata/_types.py index da4443afc..48fb5de5c 100644 --- a/src/spatialdata/_types.py +++ b/src/spatialdata/_types.py @@ -1,11 +1,21 @@ from __future__ import annotations -from typing import Any +from enum import StrEnum +from typing import Any, Literal import numpy as np from xarray import DataArray, DataTree -__all__ = ["ArrayLike", "ColorLike", "DTypeLike", "Raster_T"] +__all__ = [ + "ArrayLike", + "ColorLike", + "DTypeLike", + "Raster_T", + "ELEMENT_TYPE", + "ELEMENT_TYPE_RASTER", + "ELEMENT_TYPE_VECTOR", + "GROUP_NAME", +] from numpy.typing import DTypeLike, NDArray @@ -14,3 +24,23 @@ type Raster_T = DataArray | DataTree ColorLike = tuple[float, ...] | str + + +class ELEMENT_TYPE(StrEnum): + IMAGE = "image" + LABELS = "labels" + SHAPES = "shapes" + POINTS = "points" + TABLES = "tables" + + +class GROUP_NAME(StrEnum): + IMAGES = "images" + LABELS = "labels" + SHAPES = "shapes" + POINTS = "points" + TABLES = "tables" + + +ELEMENT_TYPE_RASTER = Literal[ELEMENT_TYPE.IMAGE, ELEMENT_TYPE.LABELS] +ELEMENT_TYPE_VECTOR = Literal[ELEMENT_TYPE.POINTS, ELEMENT_TYPE.SHAPES] diff --git a/src/spatialdata/_utils.py b/src/spatialdata/_utils.py index 609cd0403..e7515dc22 100644 --- a/src/spatialdata/_utils.py +++ b/src/spatialdata/_utils.py @@ -35,7 +35,7 @@ def disable_dask_tune_optimization() -> Generator[None, None, None]: config.set({"optimization.tune.active": old_setting}) -def _parse_list_into_array(array: list[Number] | ArrayLike) -> ArrayLike: +def _parse_list_into_array(array: list[list[Number]] | list[Number] | ArrayLike) -> ArrayLike: if isinstance(array, list): array = np.array(array) if array.dtype != float: diff --git a/src/spatialdata/transformations/transformations.py b/src/spatialdata/transformations/transformations.py index deace0ae1..5fdf3fbf0 100644 --- a/src/spatialdata/transformations/transformations.py +++ b/src/spatialdata/transformations/transformations.py @@ -509,7 +509,7 @@ def __eq__(self, other: Any) -> bool: class Affine(BaseTransformation): def __init__( self, - matrix: list[Number] | ArrayLike, + matrix: list[list[Number]] | ArrayLike, input_axes: tuple[ValidAxis_t, ...], output_axes: tuple[ValidAxis_t, ...], ) -> None: diff --git a/tests/core/transformation_manager/conftest.py b/tests/core/transformation_manager/conftest.py new file mode 100644 index 000000000..696935996 --- /dev/null +++ b/tests/core/transformation_manager/conftest.py @@ -0,0 +1,92 @@ +#!/usr/bin/env python3 + +""" +Fixtures for transformation manager tests. +""" + +from __future__ import annotations + +import pytest + +from spatialdata.transformations import Affine, Scale, Translation +from spatialdata.transformations.ngff.ngff_coordinate_system import NgffAxis, NgffCoordinateSystem + + +def get_ngff_coodinate_system(cs_name: str) -> NgffCoordinateSystem: + return NgffCoordinateSystem( + name=cs_name, + axes=[NgffAxis(name="x", type="space", unit="micrometer"), NgffAxis(name="y", type="space", unit="micrometer")], + ) + + +@pytest.fixture +def one_point_graph() -> tuple[list[NgffCoordinateSystem], list[Translation]]: + """Fixture providing a single point graph with one coordinate system and one transformation.""" + coordinate_systems = [get_ngff_coodinate_system("cs1")] + transformations = [Translation(translation=[1.0, 2.0], axes=("x", "y"))] + return coordinate_systems, transformations + + +@pytest.fixture +def fully_connected_two_point_graph() -> tuple[list[NgffCoordinateSystem], list[Translation]]: + """Fixture providing a fully connected two-point graph with two coordinate systems and transformations.""" + coordinate_systems = [get_ngff_coodinate_system("cs1"), get_ngff_coodinate_system("cs2")] + transformations = [ + Translation(translation=[-1.0, -2.0], axes=("x", "y")), + ] + return coordinate_systems, transformations + + +@pytest.fixture +def four_point_graph() -> tuple[list[NgffCoordinateSystem], list[Scale | Translation]]: + """Fixture providing a four-point graph with four coordinate systems and transformations.""" + coordinate_systems = [ + get_ngff_coodinate_system("cs1"), + get_ngff_coodinate_system("cs2"), + get_ngff_coodinate_system("cs3"), + get_ngff_coodinate_system("cs4"), + ] + transformations = [ + Translation(translation=[1.0, 2.0], axes=("x", "y")), # cs1 -> cs2 + Translation(translation=[3.0, 4.0], axes=("x", "y")), # cs2 -> cs3 + Scale( + scale=[ + 2, + 2, + ], + axes=("x", "y"), + ), # cs3 -> cs4; + Translation(translation=[4.0, 6.0], axes=("x", "y")), # cs1 -> cs3 (consistent with cs1->cs2 and cs2->cs3) + ] + return coordinate_systems, transformations + + +@pytest.fixture +def five_point_graph() -> tuple[list[NgffCoordinateSystem], list[Scale | Translation | Affine]]: + """Fixture providing a five-point graph with five coordinate systems and five transformations.""" + + coordinate_systems = [ + get_ngff_coodinate_system("cs1"), + get_ngff_coodinate_system("cs2"), + get_ngff_coodinate_system("cs3"), + get_ngff_coodinate_system("cs4"), + get_ngff_coodinate_system("cs5"), + ] + transformations = [ + Translation(translation=[1.0, 2.0], axes=("x", "y")), # cs1 -> cs2, cs4 -> cs3 + Translation(translation=[3.0, 4.0], axes=("x", "y")), # cs2 -> cs3, cs1 -> cs4 + Translation(translation=[4.0, 6.0], axes=("x", "y")), # cs1 -> cs3 (consistent with cs1->cs2 and cs2->cs3) + Scale( + scale=[ + 2, + 2, + ], + axes=("x", "y"), + ), # cs3 -> cs5; + Affine( + matrix=[[2, 0, 0], [0, 2, 0], [0, 0, 1]], + input_axes=("x", "y"), + output_axes=("x", "y"), + ), # cs3 -> cs5 + ] + return coordinate_systems, transformations diff --git a/tests/core/transformation_manager/test_transformation_manager.py b/tests/core/transformation_manager/test_transformation_manager.py new file mode 100644 index 000000000..c1f5d8300 --- /dev/null +++ b/tests/core/transformation_manager/test_transformation_manager.py @@ -0,0 +1,643 @@ +#!/usr/bin/env python3 + +""" +Unit tests for the TransformationManager class in _core/transformation_manager/__init__.py. + +This module tests all the functionality of the TransformationManager class to achieve 100% coverage. +""" + +from __future__ import annotations + +import pytest + +from spatialdata._core.transformation_manager import TRANSFORM_KEY, TransformationManager +from spatialdata._core.transformation_manager.exceptions import ( + CannotRemoveCoordinateSystemError, + CoordinateSystemAlreadyExistsError, + CoordinateSystemNotFoundError, + ElementAlreadyExistsError, + ElementNotRegisteredToAnyCoordinateSystemError, + InvalidPathError, + TransformationNotFoundError, + TransformationPathAmbiguousError, + TransformationPathAmbiguousMultipleEdgeExpectedError, + TransformationPathAmbiguousNoEdgeExpectedError, + TransformationPathNotFoundError, + TransformationPathNotSimple, +) +from spatialdata.transformations.transformations import Sequence + + +def test_initialization(): + """Test that TransformationManager initializes correctly.""" + + tm = TransformationManager() + assert len(tm._graph.nodes()) == 0 + assert len(tm._graph.edges()) == 0 + assert tm._element_to_cs_mapping == {} + + +def test_add_coordinate_system(one_point_graph): + """Test adding a coordinate system.""" + + tm = TransformationManager() + [cs1], _ = one_point_graph + + tm.add_coordinate_system(cs1) + assert cs1 in tm._graph.nodes() + + +def test_add_coordinate_system_duplicate(one_point_graph): + """Test adding an already existing coordinate system""" + tm = TransformationManager() + [cs1], _ = one_point_graph + + tm.add_coordinate_system(cs1) + with pytest.raises(CoordinateSystemAlreadyExistsError, match=f"Coordinate system '{cs1.name}' already exists"): + tm.add_coordinate_system(cs1) + + +def test_remove_coordinate_system(one_point_graph): + """Test removing a coordinate system.""" + tm = TransformationManager() + [cs1], _ = one_point_graph + + tm.add_coordinate_system(cs1) + tm.remove_coordinate_system(cs1) + + assert cs1 not in tm._graph.nodes() + + +def test_remove_coordinate_system_nonexistent(one_point_graph): + """Test that removing a non-existent coordinate system raises CoordinateSystemNotFoundError.""" + tm = TransformationManager() + [cs1], _ = one_point_graph + + # Add the coordinate system first + tm.add_coordinate_system(cs1) + + # Remove it + tm.remove_coordinate_system(cs1) + + # Try to remove it again - should raise CoordinateSystemNotFoundError + with pytest.raises(CoordinateSystemNotFoundError): + tm.remove_coordinate_system(cs1) + + +def test_remove_coordinate_system_with_associated_transformations(fully_connected_two_point_graph): + """ + Test that removing a coordinate system with associated transformations raises CannotRemoveCoordinateSystemError. + """ + tm = TransformationManager() + [cs1, cs2], [transformation] = fully_connected_two_point_graph + + tm.add_coordinate_system(cs1) + tm.add_coordinate_system(cs2) + + tm.add_transformation(cs1, cs2, transformation) + + # Should raise CannotRemoveCoordinateSystem when trying to remove a coordinate system with transformations + with pytest.raises(CannotRemoveCoordinateSystemError, match="Cannot remove coordinate system"): + tm.remove_coordinate_system(cs1) + + # Should raise CannotRemoveCoordinateSystem when trying to remove a coordinate system with transformations + with pytest.raises(CannotRemoveCoordinateSystemError, match="Cannot remove coordinate system"): + tm.remove_coordinate_system(cs2) + + +def test_remove_coordinate_system_with_belonging_elements(one_point_graph): + """Test that removing a coordinate system with element associations raises CannotRemoveCoordinateSystemError.""" + tm = TransformationManager() + [cs1], _ = one_point_graph + tm.add_coordinate_system(cs1) + tm.add_element("image1", cs1) + + with pytest.raises(CannotRemoveCoordinateSystemError, match="Cannot remove coordinate system"): + tm.remove_coordinate_system(cs1) + + +def test_list_coordinate_systems(fully_connected_two_point_graph): + """Test listing all coordinate systems.""" + tm = TransformationManager() + [cs1, cs2], _ = fully_connected_two_point_graph + tm.add_coordinate_system(cs1) + tm.add_coordinate_system(cs2) + + systems = tm.list_coordinate_systems() + assert len(systems) == 2 + assert cs1 in systems + assert cs2 in systems + + +def test_add_element(one_point_graph): + """Test adding an element with an existing coordinate system.""" + tm = TransformationManager() + [cs1], _ = one_point_graph + tm.add_coordinate_system(cs1) + tm.add_element("image1", cs1) + + mapping = tm._element_to_cs_mapping + assert "image1" in mapping + assert mapping["image1"] == cs1 + + +def test_add_element_duplicate(one_point_graph): + """Test that adding a duplicate element raises ElementAlreadyExistsError.""" + tm = TransformationManager() + [cs1], _ = one_point_graph + tm.add_coordinate_system(cs1) + element_name = "image1" + tm.add_element(element_name, cs1) + + # Try to add the same element again - should raise ElementAlreadyExistsError + with pytest.raises( + ElementAlreadyExistsError, match=f"Element '{element_name}' already exists in the transformation manager" + ): + tm.add_element(element_name, cs1) + + +def test_unset_element(one_point_graph): + """Test unsetting an element.""" + tm = TransformationManager() + [cs1], _ = one_point_graph + tm.add_coordinate_system(cs1) + tm.add_element("image1", cs1) + tm._unset_element("image1") + + assert "image1" not in tm._element_to_cs_mapping + + +def test_unset_element_nonexistent(one_point_graph): + """Test that unsetting a non-existent element raises ElementNotFoundError.""" + tm = TransformationManager() + [cs1], _ = one_point_graph + tm.add_coordinate_system(cs1) + element_name = "image1" + + # Try to unset non-existent element + with pytest.raises(ElementNotRegisteredToAnyCoordinateSystemError): + tm._unset_element(element_name) + + +def test_add_transformation(fully_connected_two_point_graph): + """Test adding a transformation between coordinate systems.""" + + tm = TransformationManager() + [cs1, cs2], [transform] = fully_connected_two_point_graph + + tm.add_coordinate_system(cs1) + tm.add_coordinate_system(cs2) + + tm.add_transformation(cs1, cs2, transform) + + assert tm._graph.has_edge(cs1, cs2) + # Get the first edge key and check the transformation + edge_key = tm._get_edge_key((cs1, cs2, transform)) + assert tm._graph[cs1][cs2][edge_key][TRANSFORM_KEY] == transform + + +def test_add_transformation_existing_edge(fully_connected_two_point_graph): + """Test adding a transformation between coordinate systems that already have an edge.""" + tm = TransformationManager() + [cs1, cs2], [transform] = fully_connected_two_point_graph + tm.add_coordinate_system(cs1) + tm.add_coordinate_system(cs2) + + tm.add_transformation(cs1, cs2, transform) + assert tm._graph.has_edge(cs1, cs2) + + # Add same transformation again between the same coordinate systems + tm.add_transformation(cs1, cs2, transform) + # Should have just rewritten the edge + assert tm._graph.has_edge(cs1, cs2) + assert len(tm._graph[cs1][cs2]) == 1 + + +def test_add_transformation_nonexistent_cs(fully_connected_two_point_graph): + """Test that adding a transformation with non-existent coordinate systems raises CoordinateSystemNotFoundError.""" + tm = TransformationManager() + [cs1, cs2], [transform] = fully_connected_two_point_graph + + expected_msg = "Coordinate system .* not found in the transformation manager." + with pytest.raises(CoordinateSystemNotFoundError, match=expected_msg): + tm.add_transformation(cs1, cs2, transform) + + # Add one coordinate system + tm.add_coordinate_system(cs1) + + with pytest.raises(CoordinateSystemNotFoundError, match=expected_msg): + tm.add_transformation(cs1, cs2, transform) + + +def test_get_direct_transformations(fully_connected_two_point_graph): + """Test getting an existing transformation.""" + tm = TransformationManager() + [cs1, cs2], [transform] = fully_connected_two_point_graph + tm.add_coordinate_system(cs1) + tm.add_coordinate_system(cs2) + + tm.add_transformation(cs1, cs2, transform) + + retrieved = tm.get_direct_transformations(cs1, cs2) + assert retrieved == [transform] + + +def test_get_direct_transformations_nonexistent(fully_connected_two_point_graph): + """Test getting a non-existent transformation.""" + tm = TransformationManager() + coordinate_systems, _transformations = fully_connected_two_point_graph + cs1, cs2 = coordinate_systems + tm.add_coordinate_system(cs1) + tm.add_coordinate_system(cs2) + + retrieved = tm.get_direct_transformations(cs1, cs2) + assert retrieved == [] + + +def test_remove_all_transformations_between_coordinate_systems(fully_connected_two_point_graph, mocker): + """Test removing all transformations between coordinate systems.""" + + tm = TransformationManager() + [cs1, cs2], [transform] = fully_connected_two_point_graph + tm.add_coordinate_system(cs1) + tm.add_coordinate_system(cs2) + + tm.add_transformation(cs1, cs2, transform) + + spy = mocker.spy(tm._graph, "remove_edge") + tm.remove_all_transformations_between_coordinate_systems(cs1, cs2) + assert spy.call_count == 1 + assert not tm._graph.has_edge(cs1, cs2) + + +def test_remove_all_transformation_nonexistent(fully_connected_two_point_graph, mocker): + """Test that removing non-existent transformations between coordinate systems raises TransformationNotFoundError.""" + + tm = TransformationManager() + [cs1, cs2], _ = fully_connected_two_point_graph + tm._graph.add_node(cs1) + tm._graph.add_node(cs2) + + spy = mocker.spy(tm._graph, "remove_edge") + tm.remove_all_transformations_between_coordinate_systems(cs1, cs2) + assert spy.call_count == 0 + + +def test_remove_specific_transformation_between_coordinate_systems(five_point_graph): + """Test removing specific transformation between coordinate systems.""" + + tm = TransformationManager() + [_cs1, _cs2, cs3, _cs4, cs5], [_transform1, _transform2, _transform3, transform4, transform5] = five_point_graph + tm.add_coordinate_system(cs3) + tm.add_coordinate_system(cs5) + + tm.add_transformation(cs3, cs5, transform4) + tm.add_transformation(cs3, cs5, transform5) + + tm.remove_specific_transformation(cs3, cs5, transform4) + edge_key = tm._get_edge_key((cs3, cs5, transform4)) + assert not tm._graph.has_edge(cs3, cs5, key=edge_key) + + +def test_remove_specific_transformation_between_coordinate_systems_non_existent(five_point_graph): + """ + Test that removing non-existent specific transformation between coordinate systems raises + TransformationNotFoundError. + """ + + tm = TransformationManager() + [_cs1, _cs2, cs3, _cs4, cs5], [_transform1, _transform2, _transform3, transform4, _transform5] = five_point_graph + tm.add_coordinate_system(cs3) + tm.add_coordinate_system(cs5) + + with pytest.raises( + TransformationNotFoundError, match=f"Transformation from '{cs3.name}' to '{cs5.name}' not found" + ): + tm.remove_specific_transformation(cs3, cs5, transform4) + + +def test_get_all_shortest_transformation_sequences_direct(four_point_graph): + """Test getting all shortest transformation sequences for a direct transformation path.""" + tm = TransformationManager() + [cs1, cs2, cs3, cs4], [transform1, transform2, _transform3, transform4] = four_point_graph + + tm.add_coordinate_system(cs1) + tm.add_coordinate_system(cs2) + tm.add_coordinate_system(cs3) + tm.add_coordinate_system(cs4) + + tm.add_transformation(cs1, cs2, transform1) + tm.add_transformation(cs2, cs3, transform2) + tm.add_transformation(cs1, cs3, transform4) + + sequences = tm.get_all_shortest_transformation_sequences(cs1, cs3) + assert len(sequences) == 1 + assert Sequence([transform4]) in sequences + + +def test_get_all_shortest_transformation_sequences_indirect_one_path(four_point_graph): + """Test getting all shortest transformation sequences for one indirect transformation path.""" + tm = TransformationManager() + [cs1, cs2, cs3, _cs4], [transform1, transform2, _transform3, _transform4] = four_point_graph + tm.add_coordinate_system(cs1) + tm.add_coordinate_system(cs2) + tm.add_coordinate_system(cs3) + + tm.add_transformation(cs1, cs2, transform1) + tm.add_transformation(cs2, cs3, transform2) + + sequences = tm.get_all_shortest_transformation_sequences(cs1, cs3) + assert len(sequences) == 1 + assert Sequence([transform1, transform2]) in sequences + + +def test_get_all_shortest_transformation_sequences_indirect_two_paths(five_point_graph): + """Test getting all shortest transformation sequences for two indirect transformation paths.""" + tm = TransformationManager() + ([cs1, cs2, cs3, cs4, _cs5], [transform1, transform2, _transform3, _transform4, _transform5]) = five_point_graph + tm.add_coordinate_system(cs1) + tm.add_coordinate_system(cs2) + tm.add_coordinate_system(cs3) + tm.add_coordinate_system(cs4) + + tm.add_transformation(cs1, cs2, transform1) + tm.add_transformation(cs2, cs3, transform2) + tm.add_transformation(cs1, cs4, transform2) + tm.add_transformation(cs4, cs3, transform1) + + sequences = tm.get_all_shortest_transformation_sequences(cs1, cs3) + assert len(sequences) == 2 + assert Sequence([transform1, transform2]) in sequences + assert Sequence([transform2, transform1]) in sequences + + +def test_get_all_shortest_transformation_sequences_no_path(four_point_graph): + """Test that getting all shortest transformation sequences with no path raises TransformationPathNotFoundError.""" + tm = TransformationManager() + [cs1, cs2, cs3, cs4], [transform1, transform2, _transform3, _transform4] = four_point_graph + tm.add_coordinate_system(cs1) + tm.add_coordinate_system(cs2) + tm.add_coordinate_system(cs3) + tm.add_coordinate_system(cs4) + + tm.add_transformation(cs1, cs2, transform1) + tm.add_transformation(cs2, cs3, transform2) + + expected_error_msg = f"No transformation path found from {cs1.name} to {cs4.name}" + with pytest.raises(TransformationPathNotFoundError, match=expected_error_msg): + tm.get_all_shortest_transformation_sequences(cs1, cs4) + + +def test_get_all_shortest_transformation_sequences_multiple_paths_multiple_edges_success(five_point_graph): + """Test getting all shortest transformation sequences, with multiple path and multiple edges between nodes.""" + tm = TransformationManager() + [cs1, cs2, cs3, cs4, cs5], [transform1, transform2, transform3, transform4, transform5] = five_point_graph + tm.add_coordinate_system(cs1) + tm.add_coordinate_system(cs2) + tm.add_coordinate_system(cs3) + tm.add_coordinate_system(cs4) + tm.add_coordinate_system(cs5) + + tm.add_transformation(cs1, cs2, transform1) + tm.add_transformation(cs2, cs3, transform2) + tm.add_transformation(cs1, cs4, transform2) + tm.add_transformation(cs4, cs3, transform1) + # tm.add_transformation(cs1, cs3, transform3) + tm.add_transformation(cs3, cs5, transform4) + tm.add_transformation(cs3, cs5, transform5) + + expected_intermediate_edges = [(cs3, cs5, transform4)] + + sequences = tm.get_all_shortest_transformation_sequences( + cs1, cs5, expected_intermediate_edges=expected_intermediate_edges + ) + assert len(sequences) == 2 + assert Sequence([transform1, transform2, transform4]) in sequences + assert Sequence([transform2, transform1, transform4]) in sequences + + +def test_get_all_shortest_transformation_sequences_multiple_paths_multiple_edges_failure(five_point_graph): + """Test getting all shortest transformation sequences, with multiple path and multiple edges between nodes.""" + tm = TransformationManager() + [cs1, cs2, cs3, cs4, cs5], [transform1, transform2, transform3, transform4, transform5] = five_point_graph + tm.add_coordinate_system(cs1) + tm.add_coordinate_system(cs2) + tm.add_coordinate_system(cs3) + tm.add_coordinate_system(cs4) + tm.add_coordinate_system(cs5) + + tm.add_transformation(cs1, cs2, transform1) + tm.add_transformation(cs2, cs3, transform2) + tm.add_transformation(cs1, cs4, transform2) + tm.add_transformation(cs4, cs3, transform1) + tm.add_transformation(cs1, cs3, transform3) + tm.add_transformation(cs3, cs5, transform4) + tm.add_transformation(cs3, cs5, transform5) + + expected_msg = f"Transformation Path ambiguous from {cs1.name} to {cs5.name}." + with pytest.raises(TransformationPathAmbiguousError, match=expected_msg): + tm.get_all_shortest_transformation_sequences(cs1, cs5) + + +def test_get_all_transformation_sequences(four_point_graph): + """Test getting all transformation sequences.""" + tm = TransformationManager() + [cs1, cs2, cs3, cs4], [transform1, transform2, transform3, transform4] = four_point_graph + tm.add_coordinate_system(cs1) + tm.add_coordinate_system(cs2) + tm.add_coordinate_system(cs3) + tm.add_coordinate_system(cs4) + + tm.add_transformation(cs1, cs2, transform1) + tm.add_transformation(cs2, cs3, transform2) + tm.add_transformation(cs3, cs4, transform3) + tm.add_transformation(cs1, cs3, transform4) + + sequences = tm.get_all_transformation_sequences(cs1, cs4) + assert len(sequences) == 2 + assert Sequence([transform1, transform2, transform3]) in sequences + assert Sequence([transform4, transform3]) in sequences + + +def test_get_all_transformation_sequences_multiple_paths(five_point_graph): + """Test getting all transformation sequences, with multiple path and multiple edges between nodes.""" + tm = TransformationManager() + [cs1, cs2, cs3, cs4, _cs5], [transform1, transform2, transform3, _transform4, _transform5] = five_point_graph + tm.add_coordinate_system(cs1) + tm.add_coordinate_system(cs2) + tm.add_coordinate_system(cs3) + tm.add_coordinate_system(cs4) + + tm.add_transformation(cs1, cs2, transform1) + tm.add_transformation(cs2, cs3, transform2) + tm.add_transformation(cs1, cs4, transform2) + tm.add_transformation(cs4, cs3, transform1) + tm.add_transformation(cs1, cs3, transform3) + + sequences = tm.get_all_transformation_sequences(cs1, cs3) + assert len(sequences) == 3 + assert Sequence([transform1, transform2]) in sequences + assert Sequence([transform2, transform1]) in sequences + assert Sequence([transform3]) in sequences + + +def test_get_all_transformation_sequences_multiple_paths_multiple_edges_success(five_point_graph): + """Test getting all transformation sequences, with multiple paths and multiple edges between nodes.""" + tm = TransformationManager() + [cs1, cs2, cs3, cs4, cs5], [transform1, transform2, transform3, transform4, transform5] = five_point_graph + tm.add_coordinate_system(cs1) + tm.add_coordinate_system(cs2) + tm.add_coordinate_system(cs3) + tm.add_coordinate_system(cs4) + tm.add_coordinate_system(cs5) + + tm.add_transformation(cs1, cs2, transform1) + tm.add_transformation(cs2, cs3, transform2) + tm.add_transformation(cs1, cs4, transform2) + tm.add_transformation(cs4, cs3, transform1) + tm.add_transformation(cs1, cs3, transform3) + tm.add_transformation(cs3, cs5, transform4) + tm.add_transformation(cs3, cs5, transform5) + + expected_intermediate_edges = [(cs3, cs5, transform4)] + sequences = tm.get_all_transformation_sequences(cs1, cs5, expected_intermediate_edges=expected_intermediate_edges) + assert len(sequences) == 3 + assert Sequence([transform1, transform2, transform4]) in sequences + assert Sequence([transform2, transform1, transform4]) in sequences + assert Sequence([transform3, transform4]) in sequences + + +def test_get_all_transformation_sequences_multiple_paths_multiple_edges_no_edge_expected(five_point_graph): + """Test getting all transformation sequences with multiple paths and multiple edges, no edge expected.""" + tm = TransformationManager() + [cs1, cs2, cs3, cs4, cs5], [transform1, transform2, transform3, transform4, transform5] = five_point_graph + tm.add_coordinate_system(cs1) + tm.add_coordinate_system(cs2) + tm.add_coordinate_system(cs3) + tm.add_coordinate_system(cs4) + tm.add_coordinate_system(cs5) + + tm.add_transformation(cs1, cs2, transform1) + tm.add_transformation(cs2, cs3, transform2) + tm.add_transformation(cs1, cs4, transform2) + tm.add_transformation(cs4, cs3, transform1) + tm.add_transformation(cs1, cs3, transform3) + tm.add_transformation(cs3, cs5, transform4) + tm.add_transformation(cs3, cs5, transform5) + + expected_msg = f"Transformation Path ambiguous from {cs3.name} to {cs5.name}." + with pytest.raises(TransformationPathAmbiguousNoEdgeExpectedError, match=expected_msg): + tm.get_all_transformation_sequences(cs1, cs5) + + +def test_get_all_transformation_sequences_multiple_paths_multiple_edges_multiple_expected(five_point_graph): + """Test getting all transformation sequences with multiple paths and multiple edges, multiple edges expected.""" + tm = TransformationManager() + [cs1, cs2, cs3, cs4, cs5], [transform1, transform2, transform3, transform4, transform5] = five_point_graph + tm.add_coordinate_system(cs1) + tm.add_coordinate_system(cs2) + tm.add_coordinate_system(cs3) + tm.add_coordinate_system(cs4) + tm.add_coordinate_system(cs5) + + tm.add_transformation(cs1, cs2, transform1) + tm.add_transformation(cs2, cs3, transform2) + tm.add_transformation(cs1, cs4, transform2) + tm.add_transformation(cs4, cs3, transform1) + tm.add_transformation(cs1, cs3, transform3) + tm.add_transformation(cs3, cs5, transform4) + tm.add_transformation(cs3, cs5, transform5) + + expected_intermediate_edges = [ + (cs3, cs5, transform4), + (cs3, cs5, transform5), + ] + + expected_msg = f"Transformation Path ambiguous from {cs3.name} to {cs5.name}." + with pytest.raises(TransformationPathAmbiguousMultipleEdgeExpectedError, match=expected_msg): + tm.get_all_transformation_sequences(cs1, cs5, expected_intermediate_edges=expected_intermediate_edges) + + +def test_get_transformation_sequences_from_simple_paths_after_disambiguation_with_non_simple_paths( + fully_connected_two_point_graph, +): + """Test _get_transformation_sequences_from_simple_paths_after_disambiguation with non-simple paths.""" + tm = TransformationManager() + [cs1, cs2], _ = fully_connected_two_point_graph + + tm.add_coordinate_system(cs1) + + paths = [[cs1, cs2, cs1, cs2]] + + with pytest.raises(TransformationPathNotSimple): + tm._get_transformation_sequences_from_simple_paths_after_disambiguation(paths, ()) + + +def test_get_transformation_sequences_from_simple_paths_after_disambiguation_with_single_node_path( + fully_connected_two_point_graph, +): + """Test _get_transformation_sequences_from_simple_paths_after_disambiguation with a path with only one node.""" + tm = TransformationManager() + [cs1, cs2], [transformation] = fully_connected_two_point_graph + + tm.add_coordinate_system(cs1) + + paths = [[cs1]] + with pytest.raises(InvalidPathError): + tm._get_transformation_sequences_from_simple_paths_after_disambiguation(paths, ()) + + +def test_get_all_transformation_sequences_no_path(four_point_graph): + """Test getting all transformation sequences.""" + tm = TransformationManager() + [cs1, cs2, cs3, cs4], [transform1, transform2, transform3, transform4] = four_point_graph + tm.add_coordinate_system(cs1) + tm.add_coordinate_system(cs2) + tm.add_coordinate_system(cs3) + tm.add_coordinate_system(cs4) + + expected_error_msg = f"No transformation path found from {cs1.name} to {cs4.name}" + with pytest.raises(TransformationPathNotFoundError, match=expected_error_msg): + tm.get_all_shortest_transformation_sequences(cs1, cs4) + + +def test_get_element_coordinate_system_success(one_point_graph): + """Test successfully getting an element's coordinate system.""" + tm = TransformationManager() + [cs1], _ = one_point_graph + tm.add_coordinate_system(cs1) + tm.add_element("image1", cs1) + + # This should successfully return the coordinate system + result = tm.get_element_coordinate_system("image1") + assert result == cs1 + + +def test_repr_method_comprehensive(fully_connected_two_point_graph): + """Test TransformationManager.__repr__ method comprehensively.""" + + # Test empty TransformationManager + tm_empty = TransformationManager() + repr_empty = repr(tm_empty) + assert "TransformationManager" in repr_empty + assert "coordinate_systems=[]" in repr_empty + assert "coordinate_transforms=[]" in repr_empty + assert "elements=[]" in repr_empty + + # Test TransformationManager with data + tm = TransformationManager() + [cs1, cs2], [transform] = fully_connected_two_point_graph + tm.add_coordinate_system(cs1) + tm.add_coordinate_system(cs2) + + tm.add_transformation(cs1, cs2, transform) + tm.add_element("image1", cs1) + tm.add_element("image2", cs2) + + repr_str = repr(tm) + assert "TransformationManager" in repr_str + assert "cs1" in repr_str + assert "cs2" in repr_str + assert repr_str.find(repr(transform)) + assert "image1" in repr_str + assert "image2" in repr_str