(WIP) Add spatial support to a standard ExperimentAxisQuery

python-spec/src/somacore/__init__.py

@@ -16,6 +16,9 @@
 
 from .base import SOMAObject
 from .collection import Collection
+from .coordinates import Axis
+from .coordinates import CoordinateSystem
+from .coordinates import CoordinateTransform
 from .data import DataFrame
 from .data import DenseNDArray
 from .data import NDArray
@@ -63,4 +66,7 @@
     "AxisQuery",
     "ExperimentAxisQuery",
     "ContextBase",
+    "Axis",
+    "CoordinateSystem",
+    "CoordinateTransform",
 )

python-spec/src/somacore/coordinates.py

@@ -0,0 +1,50 @@
+"""Definitions of types related to coordinate systems."""
+
+
+import abc
+from typing import Optional, Tuple
+
+import numpy as np
+import numpy.typing as npt
+
+
+class Axis(metaclass=abc.ABCMeta):
+    """A description of an axis of a coordinate system
+
+    Lifecycle: experimental
+    """
+
+    @property
+    @abc.abstractmethod
+    def name(self) -> str:
+        """TODO: Add docstring for Axis.name"""
+        raise NotImplementedError()
+
+    @property
+    @abc.abstractmethod
+    def type(self) -> Optional[str]:
+        """TODO: Add docstring for Axis.type"""
+        raise NotImplementedError()
+
+    @property
+    @abc.abstractmethod
+    def unit(self) -> Optional[str]:
+        """TODO: Add docstring for Axis.unit"""
+        raise NotImplementedError()
+
+
+class CoordinateSystem(metaclass=abc.ABCMeta):
+    """A coordinate system for spatial data."""
+
+    @property
+    @abc.abstractmethod
+    def axes(self) -> Tuple[Axis, ...]:
+        """TODO: Add docstring for CoordinateSystem.axes"""
+        raise NotImplementedError()
+
+
+class CoordinateTransform(metaclass=abc.ABCMeta):
+    @abc.abstractmethod
+    def to_numpy(self) -> npt.NDArray[np.float64]:
+        """TODO: Add docstring for Transformation.to_numpy"""
+        raise NotImplementedError()

python-spec/src/somacore/ephemeral/collections.py

@@ -1,9 +1,10 @@
-from typing import Any, Dict, Iterator, NoReturn, Optional, TypeVar
+from typing import Any, Dict, Iterator, MutableMapping, NoReturn, Optional, TypeVar
 
 from typing_extensions import Literal, Self
 
 from .. import base
 from .. import collection
+from .. import coordinates
 from .. import data
 from .. import experiment
 from .. import measurement
@@ -142,6 +143,16 @@ class Scene(  # type: ignore[misc]   # __eq__ false positive
 
     __slots__ = ()
 
+    @property
+    def local_coordinate_system(self) -> coordinates.CoordinateSystem:
+        """Coordinate system for this scene."""
+        raise NotImplementedError()
+
+    @property
+    def transformations(self) -> MutableMapping[str, coordinates.CoordinateTransform]:
+        """Transformations saved for this scene."""
+        raise NotImplementedError()
+
 
 class Experiment(  # type: ignore[misc]  # __eq__ false positive
     BaseCollection[base.SOMAObject],

python-spec/src/somacore/query/query.py

@@ -25,6 +25,7 @@
 from .. import data
 from .. import measurement
 from .. import options
+from .. import scene
 from .. import types as base_types
 from . import _fast_csr
 from . import axis
@@ -265,6 +266,21 @@ def varm(self, layer: str) -> data.SparseRead:
         """
         return self._axism_inner(_Axis.VAR, layer)
 
+    def scenes(self) -> pa.Array:
+        """Returns ``scene_names`` of scenes matching this query as an Arrow array.
+
+        Lifecycle: experimental
+        """
+        raise NotImplementedError()
+
+    def scene_query(self, scene_name) -> "SceneQuery":
+        """Returns a ``SceneQuery`` for the requested Scene with the filters
+        provided by the experiment.
+
+        Lifecycle: experimental
+        """
+        return SceneQuery(self, scene_name)
+
     def to_anndata(
         self,
         X_name: str,
@@ -604,6 +620,76 @@ def _threadpool(self) -> futures.ThreadPoolExecutor:
         return self._threadpool_
 
 
+class SceneQuery:
+    """TODO: Add docstring"""
+
+    def __init__(
+        self,
+        experiment_or_query: Union[_Exp, ExperimentAxisQuery],
+        scene_name: str,
+    ):
+        if isinstance(experiment_or_query, ExperimentAxisQuery):
+            self.experiment = experiment_or_query.experiment
+            self._axis_query: Optional[ExperimentAxisQuery] = experiment_or_query
+        else:
+            self.experiment = experiment_or_query
+            self._axis_query = None
+
+        if scene_name not in self.experiment.spatial:
+            raise ValueError("Scene does not exist in the experiment")
+
+        # TODO: Add this if we open the scene on query
+        # if not isinstance(self.scene, scene.Scene):
+        #    raise TypeError(
+        #        f"Unexpected SOMA type {type(m_scene).__name__} stored in spatial"
+        #    )
+
+        self.scene_name = scene_name
+
+    def img(self, layer, *, coord_system=None):
+        """Returns a full image."""
+        raise NotImplementedError()
+
+    def img_by_var(self, layer, *, apply_mask=False, crop=False, coord_system=None):
+        """Returns an image with a mask for the var locations."""
+        raise NotImplementedError()
+
+    def img_by_obs(self, layer, *, apply_mask=False, crop=False, coord_system=None):
+        """Returns an image with a mask for the obs locations."""
+        raise NotImplementedError()
+
+    def obsl(self, layer):
+        """TODO: Add docstring"""
+        # Get the requested layer inside obsl. Check it is a geometry dataframe.
+        try:
+            _obsl = self.experiment.spatial[self.scene_name].obsl
+        except KeyError as ke:
+            raise ValueError(
+                f"Scene '{self.scene_name}' does not conatin contain obsl data"
+            ) from ke
+        if not isinstance(
+            _obsl, data.DataFrame
+        ):  # TODO: Update type when GeometryDataFrame is implemented.
+            raise TypeError(
+                f"Unexpected SOMA type stored 'obsl' in Scene '{self.scene_name}'"
+            )
+
+        if self._axis_query is None:
+            # TODO: Add directly query for all obs.
+            raise NotImplementedError()
+
+        # Query the obsl by obs soma_joinid.
+        return _obsl.read((self._axis_query.obs_joinids(),))
+
+    def varl(self, layer, *, measurement_name: Optional[str] = None):
+        """TODO: Add docstring"""
+        if measurement_name is None and self._axis_query is None:
+            raise ValueError(
+                "No ExperimentAxisQuery is set. A measurement name must be provided."
+            )
+        raise NotImplementedError()
+
+
 # Private internal data structures
 
 
@@ -805,6 +891,10 @@ def ms(self) -> Mapping[str, measurement.Measurement]:
     def obs(self) -> data.DataFrame:
         ...
 
+    @property
+    def spatial(self) -> Mapping[str, scene.Scene]:
+        ...
+
     @property
     def context(self) -> Optional[base_types.ContextBase]:
         ...

python-spec/src/somacore/scene.py

@@ -1,12 +1,14 @@
 """Implementation of the SOMA scene collection for spatial data"""
 
-from typing import Generic, TypeVar
+import abc
+from typing import Generic, MutableMapping, TypeVar
 
 from typing_extensions import Final
 
 from . import _mixin
 from . import base
 from . import collection
+from . import coordinates
 from . import data
 
 _SpatialDF = TypeVar(
@@ -102,3 +104,15 @@ class Scene(
     This collection exists to store any spatial data in the scene that joins on the var
     ``soma_joinid``.
     """
+
+    @property
+    @abc.abstractmethod
+    def local_coordinate_system(self) -> coordinates.CoordinateSystem:
+        """Coordinate system for this scene."""
+        raise NotImplementedError()
+
+    @property
+    @abc.abstractmethod
+    def transformations(self) -> MutableMapping[str, coordinates.CoordinateTransform]:
+        """Transformations saved for this scene."""
+        raise NotImplementedError()