feat(api): refresh gripper jaw state from firmware (#13506)

* update gripper state from firmware

* add can message defs

* add skip to home and only use it when stopping hardware controller

* add gripper error

* check gripper jaw state before actually labware movement

api/src/opentrons/hardware_control/backends/ot3controller.py

@@ -45,6 +45,7 @@
     LIMIT_SWITCH_OVERTRAVEL_DISTANCE,
     map_pipette_type_to_sensor_id,
     moving_axes_in_move_group,
+    gripper_jaw_state_from_fw,
 )
 
 try:
@@ -127,6 +128,7 @@
     TipStateType,
     FailedTipStateCheck,
     EstopState,
+    GripperJawState,
 )
 from opentrons.hardware_control.errors import (
     InvalidPipetteName,
@@ -152,6 +154,9 @@
     set_deck_light,
     get_deck_light_state,
 )
+from opentrons_hardware.hardware_control.gripper_settings import (
+    get_gripper_jaw_state,
+)
 
 from opentrons_hardware.drivers.gpio import OT3GPIO, RemoteOT3GPIO
 from opentrons_shared_data.pipette.dev_types import PipetteName
@@ -732,6 +737,10 @@ async def gripper_home_jaw(self, duty_cycle: float) -> None:
         positions = await runner.run(can_messenger=self._messenger)
         self._handle_motor_status_response(positions)
 
+    async def get_jaw_state(self) -> GripperJawState:
+        res = await get_gripper_jaw_state(self._messenger)
+        return gripper_jaw_state_from_fw(res)
+
     @staticmethod
     def _lookup_serial_key(pipette_name: FirmwarePipetteName) -> str:
         lookup_name = {

api/src/opentrons/hardware_control/backends/ot3simulator.py

@@ -59,6 +59,7 @@
     SubSystem,
     SubSystemState,
     TipStateType,
+    GripperJawState,
 )
 from opentrons_hardware.hardware_control.motion import MoveStopCondition
 from opentrons_hardware.hardware_control import status_bar
@@ -391,6 +392,10 @@ async def get_tip_present_state(self, mount: OT3Mount) -> int:
         """Get the state of the tip ejector flag for a given mount."""
         pass
 
+    async def get_jaw_state(self) -> GripperJawState:
+        """Get the state of the gripper jaw."""
+        return GripperJawState.HOMED_READY
+
     async def tip_action(
         self,
         moves: Optional[List[Move[Axis]]] = None,

api/src/opentrons/hardware_control/backends/ot3utils.py

@@ -15,6 +15,7 @@
     PipetteSubType,
     UpdateState,
     UpdateStatus,
+    GripperJawState,
 )
 import numpy as np
 
@@ -25,6 +26,7 @@
     SensorId,
     PipetteTipActionType,
     USBTarget,
+    GripperJawState as FirmwareGripperjawState,
 )
 from opentrons_hardware.firmware_update.types import FirmwareUpdateStatus, StatusElement
 from opentrons_hardware.hardware_control import network
@@ -631,3 +633,15 @@ def update(
         progress = int(progress * 100)
         self._tracker[target] = UpdateStatus(subsystem, state, progress)
         return set(self._tracker.values())
+
+
+_gripper_jaw_state_lookup: Dict[FirmwareGripperjawState, GripperJawState] = {
+    FirmwareGripperjawState.unhomed: GripperJawState.UNHOMED,
+    FirmwareGripperjawState.force_controlling_home: GripperJawState.HOMED_READY,
+    FirmwareGripperjawState.force_controlling: GripperJawState.GRIPPING,
+    FirmwareGripperjawState.position_controlling: GripperJawState.HOLDING,
+}
+
+
+def gripper_jaw_state_from_fw(state: FirmwareGripperjawState) -> GripperJawState:
+    return _gripper_jaw_state_lookup[state]

api/src/opentrons/hardware_control/instruments/ot3/gripper_handler.py

@@ -1,5 +1,6 @@
 from typing import Optional
 import logging
+import math
 
 from opentrons.types import Point
 from .instrument_calibration import (
@@ -137,6 +138,15 @@ def check_ready_for_jaw_move(self) -> None:
         if gripper.state == GripperJawState.UNHOMED:
             raise GripError("Gripper jaw must be homed before moving")
 
+    def is_ready_for_jaw_home(self) -> bool:
+        """Raise an exception if it is not currently valid to home the jaw."""
+        gripper = self.get_gripper()
+        if gripper.state == GripperJawState.GRIPPING and not math.isclose(
+            gripper.jaw_width, gripper.geometry.jaw_width["min"], abs_tol=5.0
+        ):
+            return False
+        return True
+
     def set_jaw_state(self, state: GripperJawState) -> None:
         self.get_gripper().state = state
 

api/src/opentrons/hardware_control/ot3api.py

@@ -746,7 +746,7 @@ async def halt(self, disengage_before_stopping: bool = False) -> None:
         """Immediately disengage all present motors and clear motor and module tasks."""
         if disengage_before_stopping:
             await self.disengage_axes(
-                [ax for ax in Axis if self._backend.axis_is_present(ax)]
+                [ax for ax in Axis if self._backend.axis_is_present(ax) if ax != Axis.G]
             )
         await self._stop_motors()
 
@@ -757,7 +757,13 @@ async def stop(self, home_after: bool = True) -> None:
         self._log.info("Resetting OT3API")
         await self.reset()
         if home_after:
-            await self.home()
+            skip = []
+            if (
+                self._gripper_handler.has_gripper()
+                and not self._gripper_handler.is_ready_for_jaw_home()
+            ):
+                skip.append(Axis.G)
+            await self.home(skip=skip)
 
     async def reset(self) -> None:
         """Reset the stored state of the system."""
@@ -790,11 +796,11 @@ async def home_gripper_jaw(self) -> None:
         try:
             gripper = self._gripper_handler.get_gripper()
             self._log.info("Homing gripper jaw.")
+
             dc = self._gripper_handler.get_duty_cycle_by_grip_force(
                 gripper.default_home_force
             )
             await self._ungrip(duty_cycle=dc)
-            gripper.state = GripperJawState.HOMED_READY
         except GripperNotAttachedError:
             pass
 
@@ -860,6 +866,14 @@ async def refresh_positions(self) -> None:
             await self._backend.update_motor_status()
             await self._cache_current_position()
             await self._cache_encoder_position()
+            await self._refresh_jaw_state()
+
+    async def _refresh_jaw_state(self) -> None:
+        try:
+            gripper = self._gripper_handler.get_gripper()
+            gripper.state = await self._backend.get_jaw_state()
+        except GripperNotAttachedError:
+            pass
 
     async def _cache_current_position(self) -> Dict[Axis, float]:
         """Cache current position from backend and return in absolute deck coords."""
@@ -1325,7 +1339,11 @@ async def _home(self, axes: Sequence[Axis]) -> None:
                     await self._cache_encoder_position()
 
     @ExecutionManagerProvider.wait_for_running
-    async def home(self, axes: Optional[List[Axis]] = None) -> None:
+    async def home(
+        self,
+        axes: Optional[List[Axis]] = None,
+        skip: Optional[List[Axis]] = None,
+    ) -> None:
         """
         Worker function to home the robot by axis or list of
         desired axes.
@@ -1339,6 +1357,9 @@ async def home(self, axes: Optional[List[Axis]] = None) -> None:
             checked_axes = [ax for ax in Axis if ax != Axis.Q]
         if self.gantry_load == GantryLoad.HIGH_THROUGHPUT:
             checked_axes.append(Axis.Q)
+
+        if skip:
+            checked_axes = [ax for ax in checked_axes if ax not in skip]
         self._log.info(f"Homing {axes}")
 
         home_seq = [
@@ -1449,6 +1470,7 @@ async def _grip(self, duty_cycle: float, stay_engaged: bool = True) -> None:
                 duty_cycle=duty_cycle, stay_engaged=stay_engaged
             )
             await self._cache_encoder_position()
+            self._gripper_handler.set_jaw_state(await self._backend.get_jaw_state())
         except Exception:
             self._log.exception(
                 f"Gripper grip failed, encoder pos: {self._encoder_position[Axis.G]}"
@@ -1461,6 +1483,7 @@ async def _ungrip(self, duty_cycle: float) -> None:
         try:
             await self._backend.gripper_home_jaw(duty_cycle=duty_cycle)
             await self._cache_encoder_position()
+            self._gripper_handler.set_jaw_state(await self._backend.get_jaw_state())
         except Exception:
             self._log.exception("Gripper home failed")
             raise
@@ -1476,6 +1499,7 @@ async def _hold_jaw_width(self, jaw_width_mm: float) -> None:
             jaw_displacement_mm = (width_max - jaw_width_mm) / 2.0
             await self._backend.gripper_hold_jaw(int(1000 * jaw_displacement_mm))
             await self._cache_encoder_position()
+            self._gripper_handler.set_jaw_state(await self._backend.get_jaw_state())
         except Exception:
             self._log.exception("Gripper set width failed")
             raise
@@ -1488,21 +1512,24 @@ async def grip(
             force_newtons or self._gripper_handler.get_gripper().default_grip_force
         )
         await self._grip(duty_cycle=dc, stay_engaged=stay_engaged)
-        self._gripper_handler.set_jaw_state(GripperJawState.GRIPPING)
 
     async def ungrip(self, force_newtons: Optional[float] = None) -> None:
+        """
+        Release gripped object.
+
+        To simply open the jaw, use `home_gripper_jaw` instead.
+        """
         # get default grip force for release if not provided
         self._gripper_handler.check_ready_for_jaw_move()
+        # TODO: check jaw width to make sure it is actually gripping something
         dc = self._gripper_handler.get_duty_cycle_by_grip_force(
             force_newtons or self._gripper_handler.get_gripper().default_home_force
         )
         await self._ungrip(duty_cycle=dc)
-        self._gripper_handler.set_jaw_state(GripperJawState.HOMED_READY)
 
     async def hold_jaw_width(self, jaw_width_mm: int) -> None:
         self._gripper_handler.check_ready_for_jaw_move()
         await self._hold_jaw_width(jaw_width_mm)
-        self._gripper_handler.set_jaw_state(GripperJawState.HOLDING_CLOSED)
 
     async def _move_to_plunger_bottom(
         self, mount: OT3Mount, rate: float, acquire_lock: bool = True

api/src/opentrons/hardware_control/types.py

@@ -575,11 +575,8 @@ class GripperJawState(enum.Enum):
     #: the gripper has been homed and is at its fully-open homed position
     GRIPPING = enum.auto()
     #: the gripper is actively force-control gripping something
-    HOLDING_CLOSED = enum.auto()
-    #: the gripper is in position-control mode somewhere other than its
-    #: open position and probably should be opened before gripping something
-    HOLDING_OPENED = enum.auto()
-    #: the gripper is holding itself open but not quite at its homed position
+    HOLDING = enum.auto()
+    #: the gripper is in position-control mode
 
 
 class InstrumentProbeType(enum.Enum):

api/src/opentrons/protocol_engine/errors/__init__.py

@@ -48,6 +48,7 @@
     CannotPerformModuleAction,
     PauseNotAllowedError,
     GripperNotAttachedError,
+    CannotPerformGripperAction,
     HardwareNotSupportedError,
     LabwareMovementNotAllowedError,
     LocationIsOccupiedError,
@@ -107,6 +108,7 @@
     "PauseNotAllowedError",
     "ProtocolCommandFailedError",
     "GripperNotAttachedError",
+    "CannotPerformGripperAction",
     "HardwareNotSupportedError",
     "LabwareMovementNotAllowedError",
     "LocationIsOccupiedError",

api/src/opentrons/protocol_engine/errors/exceptions.py

@@ -676,6 +676,19 @@ def __init__(
         super().__init__(ErrorCodes.GRIPPER_NOT_PRESENT, message, details, wrapping)
 
 
+class CannotPerformGripperAction(ProtocolEngineError):
+    """Raised when trying to perform an illegal gripper action."""
+
+    def __init__(
+        self,
+        message: Optional[str] = None,
+        details: Optional[Dict[str, Any]] = None,
+        wrapping: Optional[Sequence[EnumeratedError]] = None,
+    ) -> None:
+        """Build a CannotPerformGripperAction."""
+        super().__init__(ErrorCodes.GENERAL_ERROR, message, details, wrapping)
+
+
 class LabwareMovementNotAllowedError(ProtocolEngineError):
     """Raised when attempting an illegal labware movement."""
 

api/src/opentrons/protocol_engine/execution/labware_movement.py

@@ -20,6 +20,7 @@
     LabwareMovementNotAllowedError,
     ThermocyclerNotOpenError,
     HeaterShakerLabwareLatchNotOpenError,
+    CannotPerformGripperAction,
 )
 
 from ..types import (
@@ -97,6 +98,10 @@ async def move_labware_with_gripper(
             raise GripperNotAttachedError(
                 "No gripper found for performing labware movements."
             )
+        if not ot3api._gripper_handler.is_ready_for_jaw_home():
+            raise CannotPerformGripperAction(
+                "Cannot pick up labware when gripper is already gripping."
+            )
 
         gripper_mount = OT3Mount.GRIPPER
 
@@ -130,7 +135,7 @@ async def move_labware_with_gripper(
 
             for waypoint_data in movement_waypoints:
                 if waypoint_data.jaw_open:
-                    await ot3api.ungrip()
+                    await ot3api.home_gripper_jaw()
                 else:
                     await ot3api.grip(force_newtons=labware_grip_force)
                 await ot3api.move_to(

api/src/opentrons/protocol_runner/protocol_runner.py

@@ -155,6 +155,7 @@ async def load(
         initial_home_command = pe_commands.HomeCreate(
             params=pe_commands.HomeParams(axes=None)
         )
+        # this command homes all axes, including pipette plugner and gripper jaw
         self._protocol_engine.add_command(request=initial_home_command)
 
         self._task_queue.set_run_func(
@@ -248,6 +249,7 @@ async def load(self, protocol_source: ProtocolSource) -> None:
         initial_home_command = pe_commands.HomeCreate(
             params=pe_commands.HomeParams(axes=None)
         )
+        # this command homes all axes, including pipette plugner and gripper jaw
         self._protocol_engine.add_command(request=initial_home_command)
 
         for command in commands: