feat(api): Addition of Evotip specific commands (#17351)

Covers EXEC-907
Introduces Evotip (or resin-tip) specific commands to the instrument context. These commands achieve the desired sealing, pressurization and unsealing steps for a resin tip protocol.

Author: CaseyBatten

api/src/opentrons/hardware_control/api.py

@@ -778,6 +778,7 @@ async def move_axes(
         position: Mapping[Axis, float],
         speed: Optional[float] = None,
         max_speeds: Optional[Dict[Axis, float]] = None,
+        expect_stalls: bool = False,
     ) -> None:
         """Moves the effectors of the specified axis to the specified position.
         The effector of the x,y axis is the center of the carriage.
@@ -1248,7 +1249,10 @@ async def pick_up_tip(
             await self.prepare_for_aspirate(mount)
 
     async def tip_drop_moves(
-        self, mount: top_types.Mount, home_after: bool = True
+        self,
+        mount: top_types.Mount,
+        home_after: bool = True,
+        ignore_plunger: bool = False,
     ) -> None:
         spec, _ = self.plan_check_drop_tip(mount, home_after)
 

api/src/opentrons/hardware_control/ot3api.py

@@ -1189,7 +1189,7 @@ async def move_to(
         speed: Optional[float] = None,
         critical_point: Optional[CriticalPoint] = None,
         max_speeds: Union[None, Dict[Axis, float], OT3AxisMap[float]] = None,
-        _expect_stalls: bool = False,
+        expect_stalls: bool = False,
     ) -> None:
         """Move the critical point of the specified mount to a location
         relative to the deck, at the specified speed."""
@@ -1233,14 +1233,15 @@ async def move_to(
             target_position,
             speed=speed,
             max_speeds=checked_max,
-            expect_stalls=_expect_stalls,
+            expect_stalls=expect_stalls,
         )
 
     async def move_axes(  # noqa: C901
         self,
         position: Mapping[Axis, float],
         speed: Optional[float] = None,
         max_speeds: Optional[Dict[Axis, float]] = None,
+        expect_stalls: bool = False,
     ) -> None:
         """Moves the effectors of the specified axis to the specified position.
         The effector of the x,y axis is the center of the carriage.
@@ -1296,7 +1297,11 @@ async def move_axes(  # noqa: C901
             if axis not in absolute_positions:
                 absolute_positions[axis] = position_value
 
-        await self._move(target_position=absolute_positions, speed=speed)
+        await self._move(
+            target_position=absolute_positions,
+            speed=speed,
+            expect_stalls=expect_stalls,
+        )
 
     async def move_rel(
         self,
@@ -1306,7 +1311,7 @@ async def move_rel(
         max_speeds: Union[None, Dict[Axis, float], OT3AxisMap[float]] = None,
         check_bounds: MotionChecks = MotionChecks.NONE,
         fail_on_not_homed: bool = False,
-        _expect_stalls: bool = False,
+        expect_stalls: bool = False,
     ) -> None:
         """Move the critical point of the specified mount by a specified
         displacement in a specified direction, at the specified speed."""
@@ -1348,7 +1353,7 @@ async def move_rel(
             speed=speed,
             max_speeds=checked_max,
             check_bounds=check_bounds,
-            expect_stalls=_expect_stalls,
+            expect_stalls=expect_stalls,
         )
 
     async def _cache_and_maybe_retract_mount(self, mount: OT3Mount) -> None:
@@ -2320,11 +2325,16 @@ def set_working_volume(
         instrument.working_volume = tip_volume
 
     async def tip_drop_moves(
-        self, mount: Union[top_types.Mount, OT3Mount], home_after: bool = False
+        self,
+        mount: Union[top_types.Mount, OT3Mount],
+        home_after: bool = False,
+        ignore_plunger: bool = False,
     ) -> None:
         realmount = OT3Mount.from_mount(mount)
-
-        await self._move_to_plunger_bottom(realmount, rate=1.0, check_current_vol=False)
+        if ignore_plunger is False:
+            await self._move_to_plunger_bottom(
+                realmount, rate=1.0, check_current_vol=False
+            )
 
         if self.gantry_load == GantryLoad.HIGH_THROUGHPUT:
             spec = self._pipette_handler.plan_ht_drop_tip()

api/src/opentrons/hardware_control/protocols/liquid_handler.py

@@ -183,7 +183,10 @@ async def pick_up_tip(
         ...
 
     async def tip_drop_moves(
-        self, mount: MountArgType, home_after: bool = True
+        self,
+        mount: MountArgType,
+        home_after: bool = True,
+        ignore_plunger: bool = False,
     ) -> None:
         ...
 

api/src/opentrons/hardware_control/protocols/motion_controller.py

@@ -171,6 +171,7 @@ async def move_axes(
         position: Mapping[Axis, float],
         speed: Optional[float] = None,
         max_speeds: Optional[Dict[Axis, float]] = None,
+        expect_stalls: bool = False,
     ) -> None:
         """Moves the effectors of the specified axis to the specified position.
         The effector of the x,y axis is the center of the carriage.

api/src/opentrons/legacy_commands/commands.py

@@ -299,3 +299,40 @@ def move_to_disposal_location(
         "name": command_types.MOVE_TO_DISPOSAL_LOCATION,
         "payload": {"instrument": instrument, "location": location, "text": text},
     }
+
+
+def seal(
+    instrument: InstrumentContext,
+    location: Well,
+) -> command_types.SealCommand:
+    location_text = stringify_location(location)
+    text = f"Sealing to {location_text}"
+    return {
+        "name": command_types.SEAL,
+        "payload": {"instrument": instrument, "location": location, "text": text},
+    }
+
+
+def unseal(
+    instrument: InstrumentContext,
+    location: Well,
+) -> command_types.UnsealCommand:
+    location_text = stringify_location(location)
+    text = f"Unsealing from {location_text}"
+    return {
+        "name": command_types.UNSEAL,
+        "payload": {"instrument": instrument, "location": location, "text": text},
+    }
+
+
+def resin_tip_dispense(
+    instrument: InstrumentContext,
+    flow_rate: float | None,
+) -> command_types.PressurizeCommand:
+    if flow_rate is None:
+        flow_rate = 10  # The Protocol Engine default for Resin Tip Dispense
+    text = f"Pressurize pipette to dispense from resin tip at {flow_rate}uL/s."
+    return {
+        "name": command_types.PRESSURIZE,
+        "payload": {"instrument": instrument, "text": text},
+    }

api/src/opentrons/legacy_commands/types.py

@@ -43,6 +43,10 @@
 RETURN_TIP: Final = "command.RETURN_TIP"
 MOVE_TO: Final = "command.MOVE_TO"
 MOVE_TO_DISPOSAL_LOCATION: Final = "command.MOVE_TO_DISPOSAL_LOCATION"
+SEAL: Final = "command.SEAL"
+UNSEAL: Final = "command.UNSEAL"
+PRESSURIZE: Final = "command.PRESSURIZE"
+
 
 # Modules #
 
@@ -535,11 +539,40 @@ class MoveLabwareCommandPayload(TextOnlyPayload):
     pass
 
 
+class SealCommandPayload(TextOnlyPayload):
+    instrument: InstrumentContext
+    location: Union[None, Location, Well]
+
+
+class UnsealCommandPayload(TextOnlyPayload):
+    instrument: InstrumentContext
+    location: Union[None, Location, Well]
+
+
+class PressurizeCommandPayload(TextOnlyPayload):
+    instrument: InstrumentContext
+
+
 class MoveLabwareCommand(TypedDict):
     name: Literal["command.MOVE_LABWARE"]
     payload: MoveLabwareCommandPayload
 
 
+class SealCommand(TypedDict):
+    name: Literal["command.SEAL"]
+    payload: SealCommandPayload
+
+
+class UnsealCommand(TypedDict):
+    name: Literal["command.UNSEAL"]
+    payload: UnsealCommandPayload
+
+
+class PressurizeCommand(TypedDict):
+    name: Literal["command.PRESSURIZE"]
+    payload: PressurizeCommandPayload
+
+
 Command = Union[
     DropTipCommand,
     DropTipInDisposalLocationCommand,
@@ -588,6 +621,9 @@ class MoveLabwareCommand(TypedDict):
     MoveToCommand,
     MoveToDisposalLocationCommand,
     MoveLabwareCommand,
+    SealCommand,
+    UnsealCommand,
+    PressurizeCommand,
 ]
 
 
@@ -637,6 +673,9 @@ class MoveLabwareCommand(TypedDict):
     MoveToCommandPayload,
     MoveToDisposalLocationCommandPayload,
     MoveLabwareCommandPayload,
+    SealCommandPayload,
+    UnsealCommandPayload,
+    PressurizeCommandPayload,
 ]
 
 

api/src/opentrons/protocol_api/core/engine/instrument.py

@@ -48,6 +48,8 @@
     from opentrons.protocol_api._liquid import LiquidClass
 
 _DISPENSE_VOLUME_VALIDATION_ADDED_IN = APIVersion(2, 17)
+_RESIN_TIP_DEFAULT_VOLUME = 400
+_RESIN_TIP_DEFAULT_FLOW_RATE = 10.0
 
 
 class InstrumentCore(AbstractInstrument[WellCore]):
@@ -678,6 +680,113 @@ def move_to(
                 location=location, mount=self.get_mount()
             )
 
+    def resin_tip_seal(
+        self, location: Location, well_core: WellCore, in_place: Optional[bool] = False
+    ) -> None:
+        labware_id = well_core.labware_id
+        well_name = well_core.get_name()
+        well_location = (
+            self._engine_client.state.geometry.get_relative_pick_up_tip_well_location(
+                labware_id=labware_id,
+                well_name=well_name,
+                absolute_point=location.point,
+            )
+        )
+
+        self._engine_client.execute_command(
+            cmd.EvotipSealPipetteParams(
+                pipetteId=self._pipette_id,
+                labwareId=labware_id,
+                wellName=well_name,
+                wellLocation=well_location,
+            )
+        )
+
+    def resin_tip_unseal(self, location: Location, well_core: WellCore) -> None:
+        well_name = well_core.get_name()
+        labware_id = well_core.labware_id
+
+        if location is not None:
+            relative_well_location = (
+                self._engine_client.state.geometry.get_relative_well_location(
+                    labware_id=labware_id,
+                    well_name=well_name,
+                    absolute_point=location.point,
+                )
+            )
+
+            well_location = DropTipWellLocation(
+                origin=DropTipWellOrigin(relative_well_location.origin.value),
+                offset=relative_well_location.offset,
+            )
+        else:
+            well_location = DropTipWellLocation()
+
+        pipette_movement_conflict.check_safe_for_pipette_movement(
+            engine_state=self._engine_client.state,
+            pipette_id=self._pipette_id,
+            labware_id=labware_id,
+            well_name=well_name,
+            well_location=well_location,
+        )
+        self._engine_client.execute_command(
+            cmd.EvotipUnsealPipetteParams(
+                pipetteId=self._pipette_id,
+                labwareId=labware_id,
+                wellName=well_name,
+                wellLocation=well_location,
+            )
+        )
+
+        self._protocol_core.set_last_location(location=location, mount=self.get_mount())
+
+    def resin_tip_dispense(
+        self,
+        location: Location,
+        well_core: WellCore,
+        volume: Optional[float] = None,
+        flow_rate: Optional[float] = None,
+    ) -> None:
+        """
+        Args:
+            volume: The volume of liquid to dispense, in microliters. Defaults to 400uL.
+            location: The exact location to dispense to.
+            well_core: The well to dispense to, if applicable.
+            flow_rate: The flow rate in µL/s to dispense at. Defaults to 10.0uL/S.
+        """
+        if isinstance(location, (TrashBin, WasteChute)):
+            raise ValueError("Trash Bin and Waste Chute have no Wells.")
+        well_name = well_core.get_name()
+        labware_id = well_core.labware_id
+        if volume is None:
+            volume = _RESIN_TIP_DEFAULT_VOLUME
+        if flow_rate is None:
+            flow_rate = _RESIN_TIP_DEFAULT_FLOW_RATE
+
+        well_location = self._engine_client.state.geometry.get_relative_liquid_handling_well_location(
+            labware_id=labware_id,
+            well_name=well_name,
+            absolute_point=location.point,
+            is_meniscus=None,
+        )
+        pipette_movement_conflict.check_safe_for_pipette_movement(
+            engine_state=self._engine_client.state,
+            pipette_id=self._pipette_id,
+            labware_id=labware_id,
+            well_name=well_name,
+            well_location=well_location,
+        )
+        self._engine_client.execute_command(
+            cmd.EvotipDispenseParams(
+                pipetteId=self._pipette_id,
+                labwareId=labware_id,
+                wellName=well_name,
+                wellLocation=well_location,
+                volume=volume,
+                flowRate=flow_rate,
+            )
+        )
+
     def get_mount(self) -> Mount:
         """Get the mount the pipette is attached to."""
         return self._engine_client.state.pipettes.get(

api/src/opentrons/protocol_api/core/instrument.py

@@ -180,6 +180,33 @@ def move_to(
     ) -> None:
         ...
 
+    @abstractmethod
+    def resin_tip_seal(
+        self,
+        location: types.Location,
+        well_core: WellCoreType,
+        in_place: Optional[bool] = False,
+    ) -> None:
+        ...
+
+    @abstractmethod
+    def resin_tip_unseal(
+        self,
+        location: types.Location,
+        well_core: WellCoreType,
+    ) -> None:
+        ...
+
+    @abstractmethod
+    def resin_tip_dispense(
+        self,
+        location: types.Location,
+        well_core: WellCoreType,
+        volume: Optional[float] = None,
+        flow_rate: Optional[float] = None,
+    ) -> None:
+        ...
+
     @abstractmethod
     def get_mount(self) -> types.Mount:
         ...