Merge pull request #193 from scverse/fix/modules

moved 2 functions in wrong module

Author: LucaMarconato

src/spatialdata/_core/operations/transform.py

@@ -2,7 +2,7 @@
 
 import itertools
 from functools import singledispatch
-from typing import TYPE_CHECKING, Any, Optional, Union
+from typing import TYPE_CHECKING, Any, Optional
 
 import dask.array as da
 import dask_image.ndinterp
@@ -11,7 +11,6 @@
 from dask.dataframe.core import DataFrame as DaskDataFrame
 from geopandas import GeoDataFrame
 from multiscale_spatial_image import MultiscaleSpatialImage
-from skimage.transform import estimate_transform
 from spatial_image import SpatialImage
 from xarray import DataArray
 
@@ -21,14 +20,10 @@
 from spatialdata.models import SpatialElement, get_axis_names, get_model
 from spatialdata.models._utils import DEFAULT_COORDINATE_SYSTEM
 from spatialdata.transformations._utils import _get_scale, compute_coordinates
-from spatialdata.transformations.operations import (
-    get_transformation,
-    set_transformation,
-)
+from spatialdata.transformations.operations import set_transformation
 
 if TYPE_CHECKING:
     from spatialdata.transformations.transformations import (
-        Affine,
         BaseTransformation,
         Translation,
     )
@@ -384,156 +379,3 @@ def _(data: GeoDataFrame, transformation: BaseTransformation, maintain_positioni
     )
     ShapesModel.validate(transformed_data)
     return transformed_data
-
-
-def get_transformation_between_landmarks(
-    references_coords: Union[GeoDataFrame, DaskDataFrame],
-    moving_coords: Union[GeoDataFrame, DaskDataFrame],
-) -> Affine:
-    """
-    Get a similarity transformation between two lists of (n >= 3) landmarks. Landmarks are assumed to be in the same space.
-
-    Parameters
-    ----------
-    references_coords
-        landmarks annotating the reference element. Must be a valid element describing points or circles.
-    moving_coords
-        landmarks annotating the moving element. Must be a valid element describing points or circles.
-
-    Returns
-    -------
-    The Affine transformation that maps the moving element to the reference element.
-
-    Examples
-    --------
-    If you save the landmark points using napari_spatialdata, they will be alredy saved as circles. Here is an
-    example on how to call this function on two sets of numpy arrays describing x, y coordinates.
-    >>> import numpy as np
-    >>> from spatialdata.models import PointsModel
-    >>> from spatialdata.transform import get_transformation_between_landmarks
-    >>> points_moving = np.array([[0, 0], [1, 1], [2, 2]])
-    >>> points_reference = np.array([[0, 0], [10, 10], [20, 20]])
-    >>> moving_coords = PointsModel(points_moving)
-    >>> references_coords = PointsModel(points_reference)
-    >>> transformation = get_transformation_between_landmarks(references_coords, moving_coords)
-    """
-    from spatialdata.transformations.transformations import (
-        Affine,
-        BaseTransformation,
-        Sequence,
-    )
-
-    assert get_axis_names(references_coords) == ("x", "y")
-    assert get_axis_names(moving_coords) == ("x", "y")
-
-    if isinstance(references_coords, GeoDataFrame):
-        references_xy = np.stack([references_coords.geometry.x, references_coords.geometry.y], axis=1)
-        moving_xy = np.stack([moving_coords.geometry.x, moving_coords.geometry.y], axis=1)
-    elif isinstance(references_coords, DaskDataFrame):
-        references_xy = references_coords[["x", "y"]].to_dask_array().compute()
-        moving_xy = moving_coords[["x", "y"]].to_dask_array().compute()
-    else:
-        raise TypeError("references_coords must be either an GeoDataFrame or a DaskDataFrame")
-
-    model = estimate_transform("affine", src=moving_xy, dst=references_xy)
-    transform_matrix = model.params
-    a = transform_matrix[:2, :2]
-    d = np.linalg.det(a)
-    final: BaseTransformation
-    if d < 0:
-        m = (moving_xy[:, 0].max() - moving_xy[:, 0].min()) / 2
-        flip = Affine(
-            np.array(
-                [
-                    [-1, 0, 2 * m],
-                    [0, 1, 0],
-                    [0, 0, 1],
-                ]
-            ),
-            input_axes=("x", "y"),
-            output_axes=("x", "y"),
-        )
-        flipped_moving = transform(moving_coords, flip, maintain_positioning=False)
-        if isinstance(flipped_moving, GeoDataFrame):
-            flipped_moving_xy = np.stack([flipped_moving.geometry.x, flipped_moving.geometry.y], axis=1)
-        elif isinstance(flipped_moving, DaskDataFrame):
-            flipped_moving_xy = flipped_moving[["x", "y"]].to_dask_array().compute()
-        else:
-            raise TypeError("flipped_moving must be either an GeoDataFrame or a DaskDataFrame")
-        model = estimate_transform("similarity", src=flipped_moving_xy, dst=references_xy)
-        final = Sequence([flip, Affine(model.params, input_axes=("x", "y"), output_axes=("x", "y"))])
-    else:
-        model = estimate_transform("similarity", src=moving_xy, dst=references_xy)
-        final = Affine(model.params, input_axes=("x", "y"), output_axes=("x", "y"))
-
-    affine = Affine(
-        final.to_affine_matrix(input_axes=("x", "y"), output_axes=("x", "y")),
-        input_axes=("x", "y"),
-        output_axes=("x", "y"),
-    )
-    return affine
-
-
-def align_elements_using_landmarks(
-    references_coords: Union[GeoDataFrame | DaskDataFrame],
-    moving_coords: Union[GeoDataFrame | DaskDataFrame],
-    reference_element: SpatialElement,
-    moving_element: SpatialElement,
-    reference_coordinate_system: str = "global",
-    moving_coordinate_system: str = "global",
-    new_coordinate_system: Optional[str] = None,
-    write_to_sdata: Optional[SpatialData] = None,
-) -> BaseTransformation:
-    """
-    Maps a moving object into a reference object using two lists of (n >= 3) landmarks; returns the transformations that enable this
-    mapping and optinally saves them, to map to a new shared coordinate system.
-
-    Parameters
-    ----------
-    references_coords
-        landmarks annotating the reference element. Must be a valid element describing points or circles.
-    moving_coords
-        landmarks annotating the moving element. Must be a valid element describing points or circles.
-    reference_element
-        the reference element.
-    moving_element
-        the moving element.
-    reference_coordinate_system
-        the coordinate system of the reference element that have been used to annotate the landmarks.
-    moving_coordinate_system
-        the coordinate system of the moving element that have been used to annotate the landmarks.
-    new_coordinate_system
-        If provided, both elements will be mapped to this new coordinate system with the new transformations just
-        computed.
-    write_to_sdata
-        If provided, the transformations will be saved to disk in the specified SpatialData object. The SpatialData
-        object must be backed and must contain both the reference and moving elements.
-
-    Returns
-    -------
-    A similarity transformation that maps the moving element to the same coordinate of reference element in the
-    coordinate system specified by reference_coordinate_system.
-    """
-    from spatialdata.transformations.transformations import BaseTransformation, Sequence
-
-    affine = get_transformation_between_landmarks(references_coords, moving_coords)
-
-    # get the old transformations of the visium and xenium data
-    old_moving_transformation = get_transformation(moving_element, moving_coordinate_system)
-    old_reference_transformation = get_transformation(reference_element, reference_coordinate_system)
-    assert isinstance(old_moving_transformation, BaseTransformation)
-    assert isinstance(old_reference_transformation, BaseTransformation)
-
-    # compute the new transformations
-    new_moving_transformation = Sequence([old_moving_transformation, affine])
-    new_reference_transformation = old_reference_transformation
-
-    if new_coordinate_system is not None:
-        # this allows to work on singleton objects, not embedded in a SpatialData object
-        set_transformation(
-            moving_element, new_moving_transformation, new_coordinate_system, write_to_sdata=write_to_sdata
-        )
-        set_transformation(
-            reference_element, new_reference_transformation, new_coordinate_system, write_to_sdata=write_to_sdata
-        )
-    return new_moving_transformation

src/spatialdata/transformations/__init__.py

@@ -1,6 +1,8 @@
 from spatialdata.transformations.operations import (
+    align_elements_using_landmarks,
     get_transformation,
     get_transformation_between_coordinate_systems,
+    get_transformation_between_landmarks,
     remove_transformation,
     set_transformation,
 )
@@ -26,4 +28,6 @@
     "set_transformation",
     "remove_transformation",
     "get_transformation_between_coordinate_systems",
+    "get_transformation_between_landmarks",
+    "align_elements_using_landmarks",
 ]

src/spatialdata/transformations/operations.py

@@ -3,7 +3,11 @@
 from typing import TYPE_CHECKING, Optional, Union
 
 import networkx as nx
+import numpy
 import numpy as np
+from dask.dataframe import DataFrame as DaskDataFrame
+from geopandas import GeoDataFrame
+from skimage.transform import estimate_transform
 
 from spatialdata.transformations._utils import (
     _get_transformations,
@@ -13,7 +17,7 @@
 if TYPE_CHECKING:
     from spatialdata import SpatialData
     from spatialdata.models import SpatialElement
-    from spatialdata.transformations import BaseTransformation
+    from spatialdata.transformations import Affine, BaseTransformation
 
 
 def set_transformation(
@@ -291,3 +295,158 @@ def _describe_paths(paths: list[list[Union[int, str]]]) -> str:
             transformations.append(g[path[i]][path[i + 1]]["transformation"])
         sequence = Sequence(transformations)
         return sequence
+
+
+def get_transformation_between_landmarks(
+    references_coords: Union[GeoDataFrame, DaskDataFrame],
+    moving_coords: Union[GeoDataFrame, DaskDataFrame],
+) -> Affine:
+    """
+    Get a similarity transformation between two lists of (n >= 3) landmarks. Landmarks are assumed to be in the same space.
+
+    Parameters
+    ----------
+    references_coords
+        landmarks annotating the reference element. Must be a valid element describing points or circles.
+    moving_coords
+        landmarks annotating the moving element. Must be a valid element describing points or circles.
+
+    Returns
+    -------
+    The Affine transformation that maps the moving element to the reference element.
+
+    Examples
+    --------
+    If you save the landmark points using napari_spatialdata, they will be alredy saved as circles. Here is an
+    example on how to call this function on two sets of numpy arrays describing x, y coordinates.
+    >>> import numpy as np
+    >>> from spatialdata.models import PointsModel
+    >>> from spatialdata.transform import get_transformation_between_landmarks
+    >>> points_moving = np.array([[0, 0], [1, 1], [2, 2]])
+    >>> points_reference = np.array([[0, 0], [10, 10], [20, 20]])
+    >>> moving_coords = PointsModel(points_moving)
+    >>> references_coords = PointsModel(points_reference)
+    >>> transformation = get_transformation_between_landmarks(references_coords, moving_coords)
+    """
+    from spatialdata import transform
+    from spatialdata.models import get_axis_names
+    from spatialdata.transformations.transformations import (
+        Affine,
+        BaseTransformation,
+        Sequence,
+    )
+
+    assert get_axis_names(references_coords) == ("x", "y")
+    assert get_axis_names(moving_coords) == ("x", "y")
+
+    if isinstance(references_coords, GeoDataFrame):
+        references_xy = np.stack([references_coords.geometry.x, references_coords.geometry.y], axis=1)
+        moving_xy = np.stack([moving_coords.geometry.x, moving_coords.geometry.y], axis=1)
+    elif isinstance(references_coords, DaskDataFrame):
+        references_xy = references_coords[["x", "y"]].to_dask_array().compute()
+        moving_xy = moving_coords[["x", "y"]].to_dask_array().compute()
+    else:
+        raise TypeError("references_coords must be either an GeoDataFrame or a DaskDataFrame")
+
+    model = estimate_transform("affine", src=moving_xy, dst=references_xy)
+    transform_matrix = model.params
+    a = transform_matrix[:2, :2]
+    d = np.linalg.det(a)
+    final: BaseTransformation
+    if d < 0:
+        m = (moving_xy[:, 0].max() - moving_xy[:, 0].min()) / 2
+        flip = Affine(
+            np.array(
+                [
+                    [-1, 0, 2 * m],
+                    [0, 1, 0],
+                    [0, 0, 1],
+                ]
+            ),
+            input_axes=("x", "y"),
+            output_axes=("x", "y"),
+        )
+        flipped_moving = transform(moving_coords, flip, maintain_positioning=False)
+        if isinstance(flipped_moving, GeoDataFrame):
+            flipped_moving_xy = np.stack([flipped_moving.geometry.x, flipped_moving.geometry.y], axis=1)
+        elif isinstance(flipped_moving, DaskDataFrame):
+            flipped_moving_xy = flipped_moving[["x", "y"]].to_dask_array().compute()
+        else:
+            raise TypeError("flipped_moving must be either an GeoDataFrame or a DaskDataFrame")
+        model = estimate_transform("similarity", src=flipped_moving_xy, dst=references_xy)
+        final = Sequence([flip, Affine(model.params, input_axes=("x", "y"), output_axes=("x", "y"))])
+    else:
+        model = estimate_transform("similarity", src=moving_xy, dst=references_xy)
+        final = Affine(model.params, input_axes=("x", "y"), output_axes=("x", "y"))
+
+    affine = Affine(
+        final.to_affine_matrix(input_axes=("x", "y"), output_axes=("x", "y")),
+        input_axes=("x", "y"),
+        output_axes=("x", "y"),
+    )
+    return affine
+
+
+def align_elements_using_landmarks(
+    references_coords: Union[GeoDataFrame | DaskDataFrame],
+    moving_coords: Union[GeoDataFrame | DaskDataFrame],
+    reference_element: SpatialElement,
+    moving_element: SpatialElement,
+    reference_coordinate_system: str = "global",
+    moving_coordinate_system: str = "global",
+    new_coordinate_system: Optional[str] = None,
+    write_to_sdata: Optional[SpatialData] = None,
+) -> BaseTransformation:
+    """
+    Maps a moving object into a reference object using two lists of (n >= 3) landmarks; returns the transformations that enable this
+    mapping and optinally saves them, to map to a new shared coordinate system.
+
+    Parameters
+    ----------
+    references_coords
+        landmarks annotating the reference element. Must be a valid element describing points or circles.
+    moving_coords
+        landmarks annotating the moving element. Must be a valid element describing points or circles.
+    reference_element
+        the reference element.
+    moving_element
+        the moving element.
+    reference_coordinate_system
+        the coordinate system of the reference element that have been used to annotate the landmarks.
+    moving_coordinate_system
+        the coordinate system of the moving element that have been used to annotate the landmarks.
+    new_coordinate_system
+        If provided, both elements will be mapped to this new coordinate system with the new transformations just
+        computed.
+    write_to_sdata
+        If provided, the transformations will be saved to disk in the specified SpatialData object. The SpatialData
+        object must be backed and must contain both the reference and moving elements.
+
+    Returns
+    -------
+    A similarity transformation that maps the moving element to the same coordinate of reference element in the
+    coordinate system specified by reference_coordinate_system.
+    """
+    from spatialdata.transformations.transformations import BaseTransformation, Sequence
+
+    affine = get_transformation_between_landmarks(references_coords, moving_coords)
+
+    # get the old transformations of the visium and xenium data
+    old_moving_transformation = get_transformation(moving_element, moving_coordinate_system)
+    old_reference_transformation = get_transformation(reference_element, reference_coordinate_system)
+    assert isinstance(old_moving_transformation, BaseTransformation)
+    assert isinstance(old_reference_transformation, BaseTransformation)
+
+    # compute the new transformations
+    new_moving_transformation = Sequence([old_moving_transformation, affine])
+    new_reference_transformation = old_reference_transformation
+
+    if new_coordinate_system is not None:
+        # this allows to work on singleton objects, not embedded in a SpatialData object
+        set_transformation(
+            moving_element, new_moving_transformation, new_coordinate_system, write_to_sdata=write_to_sdata
+        )
+        set_transformation(
+            reference_element, new_reference_transformation, new_coordinate_system, write_to_sdata=write_to_sdata
+        )
+    return new_moving_transformation