Skip to content

platform_handler

Manipulator platform handler.

Responsible for performing the various manipulator commands. Instantiates the appropriate bindings based on the platform type and uses them to perform the commands.

Usage

Instantiate PlatformHandler with the platform type and call the desired command.

PlatformHandler

Handler for platform commands.

Source code in src/ephys_link/back_end/platform_handler.py 
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
@final
class PlatformHandler:
    """Handler for platform commands."""

    def __init__(self, options: EphysLinkOptions, console: Console) -> None:
        """Initialize platform handler.

        Args:
            options: CLI options.
            console: Console instance.
        """
        # Store the CLI options.
        self._options = options

        # Store the console.
        self._console = console

        # Define bindings based on platform type.
        self._bindings = self._get_binding_instance(options)

        # Record which IDs are inside the brain.
        self._inside_brain: set[str] = set()

        # Generate a Pinpoint ID for proxy usage.
        self._pinpoint_id = str(uuid4())[:8]

    def _get_binding_instance(self, options: EphysLinkOptions) -> BaseBinding:
        """Match the platform type to the appropriate bindings.

        Args:
            options: CLI options.

        Raises:
            ValueError: If the platform type is not recognized.

        Returns:
            Bindings for the specified platform type.
        """
        for binding_type in get_bindings():
            binding_cli_name = binding_type.get_cli_name()

            if binding_cli_name == options.type:
                # Pass in HTTP port for Pathfinder MPM.
                if binding_cli_name == "pathfinder-mpm":
                    return MPMBinding(options.mpm_port)

                # Otherwise just return the binding.
                return binding_type()

        # Raise an error if the platform type is not recognized.
        error_message = f'Platform type "{options.type}" not recognized.'
        self._console.critical_print(error_message)
        raise ValueError(error_message)

    # Platform metadata.

    def get_display_name(self) -> str:
        """Get the display name for the platform.

        Returns:
            Display name for the platform.
        """
        return self._bindings.get_display_name()

    async def get_platform_info(self) -> PlatformInfo:
        """Get the manipulator platform type connected to Ephys Link.

        Returns:
            Platform type config identifier (see CLI options for examples).
        """
        return PlatformInfo(
            name=self._bindings.get_display_name(),
            cli_name=self._bindings.get_cli_name(),
            axes_count=await self._bindings.get_axes_count(),
            dimensions=self._bindings.get_dimensions(),
        )

    # Manipulator commands.

    async def get_manipulators(self) -> GetManipulatorsResponse:
        """Get a list of available manipulators on the current handler.

        Returns:
            List of manipulator IDs or an error message if any.
        """
        try:
            manipulators = await self._bindings.get_manipulators()
        except Exception as e:  # noqa: BLE001
            self._console.exception_error_print("Get Manipulators", e)
            return GetManipulatorsResponse(error=self._console.pretty_exception(e))
        else:
            return GetManipulatorsResponse(manipulators=manipulators)

    async def get_position(self, manipulator_id: str) -> PositionalResponse:
        """Get the current translation position of a manipulator in unified coordinates (mm).

        Args:
            manipulator_id: Manipulator ID.

        Returns:
            Current position of the manipulator and an error message if any.
        """
        try:
            unified_position = self._bindings.platform_space_to_unified_space(
                await self._bindings.get_position(manipulator_id)
            )
        except Exception as e:  # noqa: BLE001
            self._console.exception_error_print("Get Position", e)
            return PositionalResponse(error=str(e))
        else:
            return PositionalResponse(position=unified_position)

    async def get_angles(self, manipulator_id: str) -> AngularResponse:
        """Get the current rotation angles of a manipulator in Yaw, Pitch, Roll (degrees).

        Args:
            manipulator_id: Manipulator ID.

        Returns:
            Current angles of the manipulator and an error message if any.
        """
        try:
            angles = await self._bindings.get_angles(manipulator_id)
        except Exception as e:  # noqa: BLE001
            self._console.exception_error_print("Get Angles", e)
            return AngularResponse(error=self._console.pretty_exception(e))
        else:
            return AngularResponse(angles=angles)

    async def get_shank_count(self, manipulator_id: str) -> ShankCountResponse:
        """Get the number of shanks on a manipulator.

        Args:
            manipulator_id: Manipulator ID.

        Returns:
            Number of shanks on the manipulator and an error message if any.
        """
        try:
            shank_count = await self._bindings.get_shank_count(manipulator_id)
        except Exception as e:  # noqa: BLE001
            self._console.exception_error_print("Get Shank Count", e)
            return ShankCountResponse(error=self._console.pretty_exception(e))
        else:
            return ShankCountResponse(shank_count=shank_count)

    async def set_position(self, request: SetPositionRequest) -> PositionalResponse:
        """Move a manipulator to a specified translation position in unified coordinates (mm).

        Args:
            request: Request to move a manipulator to a specified position.

        Returns:
            Final position of the manipulator and an error message if any.
        """
        try:
            # Disallow setting manipulator position while inside the brain.
            if request.manipulator_id in self._inside_brain:
                error_message = 'Can not move manipulator while inside the brain. Set the depth ("set_depth") instead.'
                self._console.error_print("Set Position", error_message)
                return PositionalResponse(error=error_message)

            # Move to the new position.
            final_platform_position = await self._bindings.set_position(
                manipulator_id=request.manipulator_id,
                position=self._bindings.unified_space_to_platform_space(request.position),
                speed=request.speed,
            )
            final_unified_position = self._bindings.platform_space_to_unified_space(final_platform_position)

            # Return error if movement did not reach target within tolerance.
            for index, axis in enumerate(vector4_to_array(final_unified_position - request.position)):
                # End once index is the number of axes.
                if index == await self._bindings.get_axes_count():
                    break

                # Check if the axis is within the movement tolerance.
                if abs(axis) > self._bindings.get_movement_tolerance():
                    error_message = (
                        f"Manipulator {request.manipulator_id} did not reach target"
                        f" position on axis {list(Vector4.model_fields.keys())[index]}."
                        f" Requested: {request.position}, got: {final_unified_position}."
                    )
                    self._console.error_print("Set Position", error_message)
                    return PositionalResponse(error=error_message)
        except Exception as e:  # noqa: BLE001
            self._console.exception_error_print("Set Position", e)
            return PositionalResponse(error=self._console.pretty_exception(e))
        else:
            return PositionalResponse(position=final_unified_position)

    async def set_depth(self, request: SetDepthRequest) -> SetDepthResponse:
        """Move a manipulator's depth translation stage to a specific value (mm).

        Args:
            request: Request to move a manipulator to a specified depth.

        Returns:
            Final depth of the manipulator and an error message if any.
        """
        try:
            # Move to the new depth.
            final_platform_depth = await self._bindings.set_depth(
                manipulator_id=request.manipulator_id,
                depth=self._bindings.unified_space_to_platform_space(Vector4(w=request.depth)).w,
                speed=request.speed,
            )
            final_unified_depth = self._bindings.platform_space_to_unified_space(Vector4(w=final_platform_depth)).w

            # Return error if movement did not reach target within tolerance.
            if abs(final_unified_depth - request.depth) > self._bindings.get_movement_tolerance():
                error_message = (
                    f"Manipulator {request.manipulator_id} did not reach target depth."
                    f" Requested: {request.depth}, got: {final_unified_depth}."
                )
                self._console.error_print("Set Depth", error_message)
                return SetDepthResponse(error=error_message)
        except Exception as e:  # noqa: BLE001
            self._console.exception_error_print("Set Depth", e)
            return SetDepthResponse(error=self._console.pretty_exception(e))
        else:
            return SetDepthResponse(depth=final_unified_depth)

    async def set_inside_brain(self, request: SetInsideBrainRequest) -> BooleanStateResponse:
        """Mark a manipulator as inside the brain or not.

        This should restrict the manipulator's movement to just the depth axis.

        Args:
            request: Request to set

        Returns:
            Inside brain state of the manipulator and an error message if any.
        """
        try:
            if request.inside:
                self._inside_brain.add(request.manipulator_id)
            else:
                self._inside_brain.discard(request.manipulator_id)
        except Exception as e:  # noqa: BLE001
            self._console.exception_error_print("Set Inside Brain", e)
            return BooleanStateResponse(error=self._console.pretty_exception(e))
        else:
            return BooleanStateResponse(state=request.inside)

    async def stop(self, manipulator_id: str) -> str:
        """Stop a manipulator.

        Args:
            manipulator_id: Manipulator ID.

        Returns:
            Error message if any.
        """
        try:
            await self._bindings.stop(manipulator_id)
        except Exception as e:  # noqa: BLE001
            self._console.exception_error_print("Stop", e)
            return self._console.pretty_exception(e)
        else:
            return ""

    async def stop_all(self) -> str:
        """Stop all manipulators.

        Returns:
            Error message if any.
        """
        try:
            for manipulator_id in await self._bindings.get_manipulators():
                await self._bindings.stop(manipulator_id)
        except Exception as e:  # noqa: BLE001
            self._console.exception_error_print("Stop", e)
            return self._console.pretty_exception(e)
        else:
            return ""

    async def emergency_stop(self) -> None:
        """Stops all manipulators with a message."""
        self._console.critical_print("Emergency Stopping All Manipulators...")
        _ = await self.stop_all()

__init__(options, console)

Initialize platform handler.

Parameters:

Name Type Description Default
options EphysLinkOptions

CLI options.

 required
console Console

Console instance.

 required
Source code in src/ephys_link/back_end/platform_handler.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def __init__(self, options: EphysLinkOptions, console: Console) -> None:
    """Initialize platform handler.

    Args:
        options: CLI options.
        console: Console instance.
    """
    # Store the CLI options.
    self._options = options

    # Store the console.
    self._console = console

    # Define bindings based on platform type.
    self._bindings = self._get_binding_instance(options)

    # Record which IDs are inside the brain.
    self._inside_brain: set[str] = set()

    # Generate a Pinpoint ID for proxy usage.
    self._pinpoint_id = str(uuid4())[:8]

emergency_stop() async

Stops all manipulators with a message.

Source code in src/ephys_link/back_end/platform_handler.py 
312
313
314
315
async def emergency_stop(self) -> None:
    """Stops all manipulators with a message."""
    self._console.critical_print("Emergency Stopping All Manipulators...")
    _ = await self.stop_all()

get_angles(manipulator_id) async

Get the current rotation angles of a manipulator in Yaw, Pitch, Roll (degrees).

Parameters:

Name Type Description Default
manipulator_id str

Manipulator ID.

 required

Returns:

Type Description
AngularResponse

Current angles of the manipulator and an error message if any.
Source code in src/ephys_link/back_end/platform_handler.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
async def get_angles(self, manipulator_id: str) -> AngularResponse:
    """Get the current rotation angles of a manipulator in Yaw, Pitch, Roll (degrees).

    Args:
        manipulator_id: Manipulator ID.

    Returns:
        Current angles of the manipulator and an error message if any.
    """
    try:
        angles = await self._bindings.get_angles(manipulator_id)
    except Exception as e:  # noqa: BLE001
        self._console.exception_error_print("Get Angles", e)
        return AngularResponse(error=self._console.pretty_exception(e))
    else:
        return AngularResponse(angles=angles)

get_display_name()

Get the display name for the platform.

Returns:

Type Description
str

Display name for the platform.
Source code in src/ephys_link/back_end/platform_handler.py 
91
92
93
94
95
96
97
def get_display_name(self) -> str:
    """Get the display name for the platform.

    Returns:
        Display name for the platform.
    """
    return self._bindings.get_display_name()

get_manipulators() async

Get a list of available manipulators on the current handler.

Returns:

Type Description
GetManipulatorsResponse

List of manipulator IDs or an error message if any.
Source code in src/ephys_link/back_end/platform_handler.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
async def get_manipulators(self) -> GetManipulatorsResponse:
    """Get a list of available manipulators on the current handler.

    Returns:
        List of manipulator IDs or an error message if any.
    """
    try:
        manipulators = await self._bindings.get_manipulators()
    except Exception as e:  # noqa: BLE001
        self._console.exception_error_print("Get Manipulators", e)
        return GetManipulatorsResponse(error=self._console.pretty_exception(e))
    else:
        return GetManipulatorsResponse(manipulators=manipulators)

get_platform_info() async

Get the manipulator platform type connected to Ephys Link.

Returns:

Type Description
PlatformInfo

Platform type config identifier (see CLI options for examples).
Source code in src/ephys_link/back_end/platform_handler.py 
 99
100
101
102
103
104
105
106
107
108
109
110
async def get_platform_info(self) -> PlatformInfo:
    """Get the manipulator platform type connected to Ephys Link.

    Returns:
        Platform type config identifier (see CLI options for examples).
    """
    return PlatformInfo(
        name=self._bindings.get_display_name(),
        cli_name=self._bindings.get_cli_name(),
        axes_count=await self._bindings.get_axes_count(),
        dimensions=self._bindings.get_dimensions(),
    )

get_position(manipulator_id) async

Get the current translation position of a manipulator in unified coordinates (mm).

Parameters:

Name Type Description Default
manipulator_id str

Manipulator ID.

 required

Returns:

Type Description
PositionalResponse

Current position of the manipulator and an error message if any.
Source code in src/ephys_link/back_end/platform_handler.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
async def get_position(self, manipulator_id: str) -> PositionalResponse:
    """Get the current translation position of a manipulator in unified coordinates (mm).

    Args:
        manipulator_id: Manipulator ID.

    Returns:
        Current position of the manipulator and an error message if any.
    """
    try:
        unified_position = self._bindings.platform_space_to_unified_space(
            await self._bindings.get_position(manipulator_id)
        )
    except Exception as e:  # noqa: BLE001
        self._console.exception_error_print("Get Position", e)
        return PositionalResponse(error=str(e))
    else:
        return PositionalResponse(position=unified_position)

get_shank_count(manipulator_id) async

Get the number of shanks on a manipulator.

Parameters:

Name Type Description Default
manipulator_id str

Manipulator ID.

 required

Returns:

Type Description
ShankCountResponse

Number of shanks on the manipulator and an error message if any.
Source code in src/ephys_link/back_end/platform_handler.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
async def get_shank_count(self, manipulator_id: str) -> ShankCountResponse:
    """Get the number of shanks on a manipulator.

    Args:
        manipulator_id: Manipulator ID.

    Returns:
        Number of shanks on the manipulator and an error message if any.
    """
    try:
        shank_count = await self._bindings.get_shank_count(manipulator_id)
    except Exception as e:  # noqa: BLE001
        self._console.exception_error_print("Get Shank Count", e)
        return ShankCountResponse(error=self._console.pretty_exception(e))
    else:
        return ShankCountResponse(shank_count=shank_count)

set_depth(request) async

Move a manipulator's depth translation stage to a specific value (mm).

Parameters:

Name Type Description Default
request SetDepthRequest

Request to move a manipulator to a specified depth.

 required

Returns:

Type Description
SetDepthResponse

Final depth of the manipulator and an error message if any.
Source code in src/ephys_link/back_end/platform_handler.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
async def set_depth(self, request: SetDepthRequest) -> SetDepthResponse:
    """Move a manipulator's depth translation stage to a specific value (mm).

    Args:
        request: Request to move a manipulator to a specified depth.

    Returns:
        Final depth of the manipulator and an error message if any.
    """
    try:
        # Move to the new depth.
        final_platform_depth = await self._bindings.set_depth(
            manipulator_id=request.manipulator_id,
            depth=self._bindings.unified_space_to_platform_space(Vector4(w=request.depth)).w,
            speed=request.speed,
        )
        final_unified_depth = self._bindings.platform_space_to_unified_space(Vector4(w=final_platform_depth)).w

        # Return error if movement did not reach target within tolerance.
        if abs(final_unified_depth - request.depth) > self._bindings.get_movement_tolerance():
            error_message = (
                f"Manipulator {request.manipulator_id} did not reach target depth."
                f" Requested: {request.depth}, got: {final_unified_depth}."
            )
            self._console.error_print("Set Depth", error_message)
            return SetDepthResponse(error=error_message)
    except Exception as e:  # noqa: BLE001
        self._console.exception_error_print("Set Depth", e)
        return SetDepthResponse(error=self._console.pretty_exception(e))
    else:
        return SetDepthResponse(depth=final_unified_depth)

set_inside_brain(request) async

Mark a manipulator as inside the brain or not.

This should restrict the manipulator's movement to just the depth axis.

Parameters:

Name Type Description Default
request SetInsideBrainRequest

Request to set

 required

Returns:

Type Description
BooleanStateResponse

Inside brain state of the manipulator and an error message if any.
Source code in src/ephys_link/back_end/platform_handler.py 
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
async def set_inside_brain(self, request: SetInsideBrainRequest) -> BooleanStateResponse:
    """Mark a manipulator as inside the brain or not.

    This should restrict the manipulator's movement to just the depth axis.

    Args:
        request: Request to set

    Returns:
        Inside brain state of the manipulator and an error message if any.
    """
    try:
        if request.inside:
            self._inside_brain.add(request.manipulator_id)
        else:
            self._inside_brain.discard(request.manipulator_id)
    except Exception as e:  # noqa: BLE001
        self._console.exception_error_print("Set Inside Brain", e)
        return BooleanStateResponse(error=self._console.pretty_exception(e))
    else:
        return BooleanStateResponse(state=request.inside)

set_position(request) async

Move a manipulator to a specified translation position in unified coordinates (mm).

Parameters:

Name Type Description Default
request SetPositionRequest

Request to move a manipulator to a specified position.

 required

Returns:

Type Description
PositionalResponse

Final position of the manipulator and an error message if any.
Source code in src/ephys_link/back_end/platform_handler.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
async def set_position(self, request: SetPositionRequest) -> PositionalResponse:
    """Move a manipulator to a specified translation position in unified coordinates (mm).

    Args:
        request: Request to move a manipulator to a specified position.

    Returns:
        Final position of the manipulator and an error message if any.
    """
    try:
        # Disallow setting manipulator position while inside the brain.
        if request.manipulator_id in self._inside_brain:
            error_message = 'Can not move manipulator while inside the brain. Set the depth ("set_depth") instead.'
            self._console.error_print("Set Position", error_message)
            return PositionalResponse(error=error_message)

        # Move to the new position.
        final_platform_position = await self._bindings.set_position(
            manipulator_id=request.manipulator_id,
            position=self._bindings.unified_space_to_platform_space(request.position),
            speed=request.speed,
        )
        final_unified_position = self._bindings.platform_space_to_unified_space(final_platform_position)

        # Return error if movement did not reach target within tolerance.
        for index, axis in enumerate(vector4_to_array(final_unified_position - request.position)):
            # End once index is the number of axes.
            if index == await self._bindings.get_axes_count():
                break

            # Check if the axis is within the movement tolerance.
            if abs(axis) > self._bindings.get_movement_tolerance():
                error_message = (
                    f"Manipulator {request.manipulator_id} did not reach target"
                    f" position on axis {list(Vector4.model_fields.keys())[index]}."
                    f" Requested: {request.position}, got: {final_unified_position}."
                )
                self._console.error_print("Set Position", error_message)
                return PositionalResponse(error=error_message)
    except Exception as e:  # noqa: BLE001
        self._console.exception_error_print("Set Position", e)
        return PositionalResponse(error=self._console.pretty_exception(e))
    else:
        return PositionalResponse(position=final_unified_position)

stop(manipulator_id) async

Stop a manipulator.

Parameters:

Name Type Description Default
manipulator_id str

Manipulator ID.

 required

Returns:

Type Description
str

Error message if any.
Source code in src/ephys_link/back_end/platform_handler.py 
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
async def stop(self, manipulator_id: str) -> str:
    """Stop a manipulator.

    Args:
        manipulator_id: Manipulator ID.

    Returns:
        Error message if any.
    """
    try:
        await self._bindings.stop(manipulator_id)
    except Exception as e:  # noqa: BLE001
        self._console.exception_error_print("Stop", e)
        return self._console.pretty_exception(e)
    else:
        return ""

stop_all() async

Stop all manipulators.

Returns:

Type Description
str

Error message if any.
Source code in src/ephys_link/back_end/platform_handler.py 
297
298
299
300
301
302
303
304
305
306
307
308
309
310
async def stop_all(self) -> str:
    """Stop all manipulators.

    Returns:
        Error message if any.
    """
    try:
        for manipulator_id in await self._bindings.get_manipulators():
            await self._bindings.stop(manipulator_id)
    except Exception as e:  # noqa: BLE001
        self._console.exception_error_print("Stop", e)
        return self._console.pretty_exception(e)
    else:
        return ""