165 transform manipulators into unified manipulator space (#242)

* Version bump

* Create unified space converters

* Defined p to u for new scale, defined dimensions in platform handler

* p -> u and u -> p for Sensapex

* use invert dimensions instead of negate

* Correctly defined p -> u and u -> p for Sensapex and New Scale

* Implemented converters to Sensapex

* Changed version number

* Ran Black, fixed version number

* styling

* Fixed sensapex handler

* WIP using virtual depth axis

* Revert "WIP using virtual depth axis"

This reverts commit a2b765f.

* Fixed comment typo

* Set upper python version limit

* Switch type to axis count, send dimensions with get

* Change to "dimensions"

* Specify it is 3D

* Reformat, set pathfinder to -1

* Pylama fix

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "ephys-link"
-version = "0.9.14"
+version = "0.9.15"
 license = { file = "LICENSE" }
 
 authors = [{ name = "Kenneth Yang", email = "[email protected]" }]
@@ -24,7 +24,7 @@ classifiers = [
     "Topic :: Scientific/Engineering :: Medical Science Apps.",
 ]
 
-requires-python = ">=3.8"
+requires-python = ">=3.8, <3.12"
 dependencies = [
     "aiohttp==3.8.6",
     "pyserial==3.5",

src/ephys_link/common.py

@@ -70,19 +70,26 @@ class GetManipulatorsOutputData(dict):
 
     :param manipulators: Tuple of manipulator IDs (as strings)
     :type manipulators: list
-    :param manipulator_type: Type of the output data (temporary solution until #165 is implemented)
-    :type manipulator_type: str
+    :param num_axes: Number of axes this manipulator has
+    :type num_axes: int
+    :param dimensions: Size of the movement space in mm (first 3 axes)
+    :type dimensions: list
     :param error: Error message
     :type error: str
 
     :example: Example generated dictionary
         :code:`{"manipulators": ["1", "2"], "error": ""}`
     """
 
-    def __init__(self, manipulators: list, manipulator_type: str, error: str) -> None:
+    def __init__(
+        self, manipulators: list, num_axes: int, dimensions: list, error: str
+    ) -> None:
         """Constructor"""
         super(GetManipulatorsOutputData, self).__init__(
-            manipulators=manipulators, type=manipulator_type, error=error
+            manipulators=manipulators,
+            num_axes=num_axes,
+            dimensions=dimensions,
+            error=error,
         )
 
 

src/ephys_link/platform_handler.py

@@ -29,7 +29,10 @@ def __init__(self):
         # Registered manipulators are stored as a dictionary of IDs (string) to
         # manipulator objects
         self.manipulators = {}
-        self.type = "sensapex"  # Remove this after #165
+        self.num_axes = 4
+
+        # Platform axes dimensions in mm
+        self.dimensions = [20, 20, 20, 20]
 
     # Platform Handler Methods
 
@@ -73,7 +76,9 @@ def get_manipulators(self) -> com.GetManipulatorsOutputData:
         except Exception as e:
             print(f"[ERROR]\t\t Getting manipulators: {type(e)}: {e}\n")
         finally:
-            return com.GetManipulatorsOutputData(devices, self.type, error)
+            return com.GetManipulatorsOutputData(
+                devices, self.num_axes, self.dimensions, error
+            )
 
     def register_manipulator(self, manipulator_id: str) -> str:
         """Register a manipulator
@@ -147,8 +152,13 @@ def get_pos(self, manipulator_id: str) -> com.PositionalOutputData:
                 print(f"[ERROR]\t\t Calibration not complete: {manipulator_id}\n")
                 return com.PositionalOutputData([], "Manipulator not calibrated")
 
-            # Get position
-            return self._get_pos(manipulator_id)
+            # Get position and convert to unified space
+            manipulator_pos = self._get_pos(manipulator_id)
+            if manipulator_pos["error"] != "":
+                return manipulator_pos
+            return com.PositionalOutputData(
+                self._platform_space_to_unified_space(manipulator_pos["position"]), ""
+            )
 
         except KeyError:
             # Manipulator not found in registered manipulators
@@ -207,7 +217,16 @@ async def goto_pos(
                 print(f"[ERROR]\t\t Cannot write to manipulator: {manipulator_id}")
                 return com.PositionalOutputData([], "Cannot write to manipulator")
 
-            return await self._goto_pos(manipulator_id, position, speed)
+            # Convert position to platform space, move, and convert final position back to
+            # unified space
+            end_position = await self._goto_pos(
+                manipulator_id, self._unified_space_to_platform_space(position), speed
+            )
+            if end_position["error"] != "":
+                return end_position
+            return com.PositionalOutputData(
+                self._platform_space_to_unified_space(end_position["position"]), ""
+            )
 
         except KeyError:
             # Manipulator not found in registered manipulators
@@ -240,7 +259,17 @@ async def drive_to_depth(
                 print(f"[ERROR]\t\t Cannot write to manipulator: {manipulator_id}")
                 return com.DriveToDepthOutputData(0, "Cannot write to manipulator")
 
-            return await self._drive_to_depth(manipulator_id, depth, speed)
+            end_depth = await self._drive_to_depth(
+                manipulator_id,
+                self._unified_space_to_platform_space([0, 0, 0, depth])[3],
+                speed,
+            )
+            if end_depth["error"] != "":
+                return end_depth
+            return com.DriveToDepthOutputData(
+                self._platform_space_to_unified_space([0, 0, 0, end_depth["depth"]])[3],
+                "",
+            )
 
         except KeyError:
             # Manipulator not found in registered manipulators
@@ -377,7 +406,6 @@ def _get_manipulators(self) -> list:
         :return: List of manipulator IDs
         :rtype: list
         """
-        pass
 
     @abstractmethod
     def _register_manipulator(self, manipulator_id: str) -> None:
@@ -387,7 +415,6 @@ def _register_manipulator(self, manipulator_id: str) -> None:
         :type manipulator_id: str
         :return: None
         """
-        pass
 
     @abstractmethod
     def _unregister_manipulator(self, manipulator_id: str) -> None:
@@ -397,7 +424,6 @@ def _unregister_manipulator(self, manipulator_id: str) -> None:
         :type manipulator_id: str
         :return: None
         """
-        pass
 
     @abstractmethod
     def _get_pos(self, manipulator_id: str) -> com.PositionalOutputData:
@@ -409,7 +435,6 @@ def _get_pos(self, manipulator_id: str) -> com.PositionalOutputData:
             empty array on error) in mm, error message)
         :rtype: :class:`ephys_link.common.PositionalOutputData`
         """
-        pass
 
     @abstractmethod
     def _get_angles(self, manipulator_id: str) -> com.AngularOutputData:
@@ -421,7 +446,6 @@ def _get_angles(self, manipulator_id: str) -> com.AngularOutputData:
             empty array on error) in degrees, error message)
         :rtype: :class:`ephys_link.common.AngularOutputData`
         """
-        pass
 
     @abstractmethod
     async def _goto_pos(
@@ -439,7 +463,6 @@ async def _goto_pos(
                  empty array on error) in mm, error message)
         :rtype: :class:`ephys_link.common.PositionalOutputData`
         """
-        pass
 
     @abstractmethod
     async def _drive_to_depth(
@@ -457,7 +480,6 @@ async def _drive_to_depth(
                  message)
         :rtype: :class:`ephys_link.common.DriveToDepthOutputData`
         """
-        pass
 
     @abstractmethod
     def _set_inside_brain(
@@ -472,7 +494,6 @@ def _set_inside_brain(
         :return: Callback parameters (manipulator ID, inside, error message)
         :rtype: :class:`ephys_link.common.StateOutputData`
         """
-        pass
 
     @abstractmethod
     async def _calibrate(self, manipulator_id: str, sio: socketio.AsyncServer) -> str:
@@ -485,7 +506,6 @@ async def _calibrate(self, manipulator_id: str, sio: socketio.AsyncServer) -> st
         :return: Callback parameters (manipulator ID, error message)
         :rtype: str
         """
-        pass
 
     @abstractmethod
     def _bypass_calibration(self, manipulator_id: str) -> str:
@@ -496,7 +516,6 @@ def _bypass_calibration(self, manipulator_id: str) -> str:
         :return: Callback parameters (manipulator ID, error message)
         :rtype: str
         """
-        pass
 
     @abstractmethod
     def _set_can_write(
@@ -519,4 +538,27 @@ def _set_can_write(
         :return: Callback parameters (manipulator ID, can_write, error message)
         :rtype: :class:`ephys_link.common.StateOutputData`
         """
-        pass
+
+    @abstractmethod
+    def _platform_space_to_unified_space(
+        self, platform_position: list[float]
+    ) -> list[float]:
+        """Convert position in platform space to position in unified manipulator space
+
+        :param platform_position: Position in platform space (x, y, z, w) in mm
+        :type platform_position: list[float]
+        :return: Position in unified manipulator space (x, y, z, w) in mm
+        :rtype: list[float]
+        """
+
+    @abstractmethod
+    def _unified_space_to_platform_space(
+        self, unified_position: list[float]
+    ) -> list[float]:
+        """Convert position in unified manipulator space to position in platform space
+
+        :param unified_position: Position in unified manipulator space (x, y, z, w) in mm
+        :type unified_position: list[float]
+        :return: Position in platform space (x, y, z, w) in mm
+        :rtype: list[float]
+        """

src/ephys_link/platforms/new_scale_handler.py

@@ -24,7 +24,8 @@ def __init__(self) -> None:
         """Initialize New Scale handler"""
         super().__init__()
 
-        self.type = "new_scale"
+        self.num_axes = 3
+        self.dimensions = [15, 15, 15]
 
         # Load New Scale API
         # noinspection PyUnresolvedReferences
@@ -120,3 +121,35 @@ def _set_can_write(
             f"[SUCCESS]\t Set can_write state for manipulator" f" {manipulator_id}\n"
         )
         return com.StateOutputData(can_write, "")
+
+    def _platform_space_to_unified_space(
+        self, platform_position: list[float]
+    ) -> list[float]:
+        # unified   <-  platform
+        # +x        <-  -x
+        # +y        <-  +z
+        # +z        <-  +y
+        # +d        <-  -d
+
+        return [
+            self.dimensions[0] - platform_position[0],
+            platform_position[2],
+            platform_position[1],
+            self.dimensions[3] - platform_position[3],
+        ]
+
+    def _unified_space_to_platform_space(
+        self, unified_position: list[float]
+    ) -> list[float]:
+        # platform  <-  unified
+        # +x        <-  -x
+        # +y        <-  +z
+        # +z        <-  +y
+        # +d        <-  -d
+
+        return [
+            self.dimensions[0] - unified_position[0],
+            unified_position[2],
+            unified_position[1],
+            self.dimensions[3] - unified_position[3],
+        ]

src/ephys_link/platforms/new_scale_pathfinder_handler.py

@@ -71,7 +71,7 @@ def __init__(self, port: int = 8080) -> None:
         """
         super().__init__()
 
-        self.type = "new_scale_pathfinder"
+        self.num_axes = -1
 
         self.port = port
 
@@ -219,3 +219,13 @@ def _set_can_write(
         sio: socketio.AsyncServer,
     ) -> com.StateOutputData:
         pass
+
+    def _unified_space_to_platform_space(
+        self, unified_position: list[float]
+    ) -> list[float]:
+        pass
+
+    def _platform_space_to_unified_space(
+        self, platform_position: list[float]
+    ) -> list[float]:
+        pass

src/ephys_link/platforms/sensapex_handler.py

@@ -124,3 +124,35 @@ def _set_can_write(
             f"[SUCCESS]\t Set can_write state for manipulator" f" {manipulator_id}\n"
         )
         return com.StateOutputData(can_write, "")
+
+    def _platform_space_to_unified_space(
+        self, platform_position: list[float]
+    ) -> list[float]:
+        # unified   <-  platform
+        # +x        <-  +y
+        # +y        <-  -z
+        # +z        <-  +x
+        # +d        <-  +d
+
+        return [
+            platform_position[1],
+            self.dimensions[2] - platform_position[2],
+            platform_position[0],
+            platform_position[3],
+        ]
+
+    def _unified_space_to_platform_space(
+        self, unified_position: list[float]
+    ) -> list[float]:
+        # platform  <-  unified
+        # +x        <-  +z
+        # +y        <-  +x
+        # +z        <-  -y
+        # +d        <-  +d
+
+        return [
+            unified_position[2],
+            unified_position[0],
+            self.dimensions[2] - unified_position[1],
+            unified_position[3],
+        ]

src/ephys_link/platforms/sensapex_manipulator.py

@@ -69,8 +69,8 @@ async def goto_pos(
         """
         # Check if able to write
         if not self._can_write:
-            print(f"[ERROR]\t\t Manipulator {self._id} movement " f"canceled")
-            return com.PositionalOutputData([], "Manipulator " "movement canceled")
+            print(f"[ERROR]\t\t Manipulator {self._id} movement canceled")
+            return com.PositionalOutputData([], "Manipulator movement canceled")
 
         # Stop current movement
         if self._is_moving: