From 3d78c1f4fff687ac2363739a4804423d48319c5b Mon Sep 17 00:00:00 2001 From: CaseyBatten Date: Thu, 30 Jan 2025 17:33:58 -0500 Subject: [PATCH] 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. --- api/src/opentrons/hardware_control/api.py | 6 +- api/src/opentrons/hardware_control/ot3api.py | 26 +- .../protocols/liquid_handler.py | 5 +- .../protocols/motion_controller.py | 1 + api/src/opentrons/legacy_commands/commands.py | 37 ++ api/src/opentrons/legacy_commands/types.py | 39 +++ .../protocol_api/core/engine/instrument.py | 109 ++++++ .../opentrons/protocol_api/core/instrument.py | 27 ++ .../core/legacy/legacy_instrument_core.py | 50 +++ .../legacy_instrument_core.py | 24 ++ .../protocol_api/instrument_context.py | 141 ++++++++ .../protocol_engine/commands/__init__.py | 40 +++ .../commands/command_unions.py | 39 +++ .../commands/evotip_dispense.py | 156 +++++++++ .../commands/evotip_seal_pipette.py | 331 ++++++++++++++++++ .../commands/evotip_unseal_pipette.py | 160 +++++++++ .../protocol_engine/errors/__init__.py | 2 + .../protocol_engine/errors/exceptions.py | 13 + .../protocol_engine/execution/gantry_mover.py | 5 + .../protocol_engine/execution/tip_handler.py | 39 ++- .../resources/labware_validation.py | 5 + .../protocol_engine/state/pipettes.py | 5 + .../hardware_control/test_ot3_api.py | 8 +- .../commands/test_evotip_dispense.py | 133 +++++++ .../commands/test_evotip_seal_pipette.py | 300 ++++++++++++++++ .../commands/test_evotip_unseal_pipette.py | 330 +++++++++++++++++ .../execution/test_gantry_mover.py | 4 +- .../execution/test_tip_handler.py | 4 +- .../gripper_assembly_qc_ot3/test_mount.py | 4 +- .../robot_assembly_qc_ot3/test_instruments.py | 4 +- .../production_qc/z_stage_qc_ot3.py | 2 +- shared-data/command/schemas/11.json | 226 ++++++++++++ .../3/evotips_opentrons_96_labware/1.json | 14 + 33 files changed, 2260 insertions(+), 29 deletions(-) create mode 100644 api/src/opentrons/protocol_engine/commands/evotip_dispense.py create mode 100644 api/src/opentrons/protocol_engine/commands/evotip_seal_pipette.py create mode 100644 api/src/opentrons/protocol_engine/commands/evotip_unseal_pipette.py create mode 100644 api/tests/opentrons/protocol_engine/commands/test_evotip_dispense.py create mode 100644 api/tests/opentrons/protocol_engine/commands/test_evotip_seal_pipette.py create mode 100644 api/tests/opentrons/protocol_engine/commands/test_evotip_unseal_pipette.py diff --git a/api/src/opentrons/hardware_control/api.py b/api/src/opentrons/hardware_control/api.py index c52fae64131..175c89dda7e 100644 --- a/api/src/opentrons/hardware_control/api.py +++ b/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) diff --git a/api/src/opentrons/hardware_control/ot3api.py b/api/src/opentrons/hardware_control/ot3api.py index 6295757e7ab..038843e23ac 100644 --- a/api/src/opentrons/hardware_control/ot3api.py +++ b/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,7 +1233,7 @@ 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 @@ -1241,6 +1241,7 @@ async def move_axes( # noqa: C901 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() diff --git a/api/src/opentrons/hardware_control/protocols/liquid_handler.py b/api/src/opentrons/hardware_control/protocols/liquid_handler.py index 2aea15bd55b..090b7dfec93 100644 --- a/api/src/opentrons/hardware_control/protocols/liquid_handler.py +++ b/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: ... diff --git a/api/src/opentrons/hardware_control/protocols/motion_controller.py b/api/src/opentrons/hardware_control/protocols/motion_controller.py index e95a9d2e24f..77f78506506 100644 --- a/api/src/opentrons/hardware_control/protocols/motion_controller.py +++ b/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. diff --git a/api/src/opentrons/legacy_commands/commands.py b/api/src/opentrons/legacy_commands/commands.py index 68b6f1a0595..fbbb14d7fc4 100755 --- a/api/src/opentrons/legacy_commands/commands.py +++ b/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}, + } diff --git a/api/src/opentrons/legacy_commands/types.py b/api/src/opentrons/legacy_commands/types.py index 5aaa72b8e09..61302985c2c 100755 --- a/api/src/opentrons/legacy_commands/types.py +++ b/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, ] diff --git a/api/src/opentrons/protocol_api/core/engine/instrument.py b/api/src/opentrons/protocol_api/core/engine/instrument.py index 010f3110fdb..cd6548202ff 100644 --- a/api/src/opentrons/protocol_api/core/engine/instrument.py +++ b/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( diff --git a/api/src/opentrons/protocol_api/core/instrument.py b/api/src/opentrons/protocol_api/core/instrument.py index bc1ec3669df..7f0fa4d72a7 100644 --- a/api/src/opentrons/protocol_api/core/instrument.py +++ b/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: ... diff --git a/api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py b/api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py index d2d25051d49..20d0b862e53 100644 --- a/api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py +++ b/api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py @@ -308,6 +308,30 @@ def drop_tip_in_disposal_location( ) -> None: raise APIVersionError(api_element="Dropping tips in a trash bin or waste chute") + def resin_tip_seal( + self, + location: types.Location, + well_core: WellCore, + in_place: Optional[bool] = False, + ) -> None: + raise APIVersionError(api_element="Sealing resin tips.") + + def resin_tip_unseal( + self, + location: types.Location, + well_core: WellCore, + ) -> None: + raise APIVersionError(api_element="Unsealing resin tips.") + + def resin_tip_dispense( + self, + location: types.Location, + well_core: WellCore, + volume: Optional[float] = None, + flow_rate: Optional[float] = None, + ) -> None: + raise APIVersionError(api_element="Dispensing liquid from resin tips.") + def home(self) -> None: """Home the mount""" self._protocol_interface.get_hardware().home_z( @@ -400,6 +424,32 @@ def move_to( location=location, mount=location_cache_mount ) + def evotip_seal( + self, + location: types.Location, + well_core: LegacyWellCore, + in_place: Optional[bool] = False, + ) -> None: + """This will never be called because it was added in API 2.22.""" + assert False, "evotip_seal only supported in API 2.22 & later" + + def evotip_unseal( + self, location: types.Location, well_core: WellCore, home_after: Optional[bool] + ) -> None: + """This will never be called because it was added in API 2.22.""" + assert False, "evotip_unseal only supported in API 2.22 & later" + + def evotip_dispense( + self, + location: types.Location, + well_core: WellCore, + volume: Optional[float] = None, + flow_rate: Optional[float] = None, + push_out: Optional[float] = None, + ) -> None: + """This will never be called because it was added in API 2.22.""" + assert False, "evotip_dispense only supported in API 2.22 & later" + def get_mount(self) -> types.Mount: """Get the mount this pipette is attached to.""" return self._mount diff --git a/api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py b/api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py index ec194874528..54c43a90f8c 100644 --- a/api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py +++ b/api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py @@ -276,6 +276,30 @@ def drop_tip_in_disposal_location( ) -> None: raise APIVersionError(api_element="Dropping tips in a trash bin or waste chute") + def resin_tip_seal( + self, + location: types.Location, + well_core: WellCore, + in_place: Optional[bool] = False, + ) -> None: + raise APIVersionError(api_element="Sealing resin tips.") + + def resin_tip_unseal( + self, + location: types.Location, + well_core: WellCore, + ) -> None: + raise APIVersionError(api_element="Unsealing resin tips.") + + def resin_tip_dispense( + self, + location: types.Location, + well_core: WellCore, + volume: Optional[float] = None, + flow_rate: Optional[float] = None, + ) -> None: + raise APIVersionError(api_element="Dispensing liquid from resin tips.") + def home(self) -> None: self._protocol_interface.set_last_location(None) diff --git a/api/src/opentrons/protocol_api/instrument_context.py b/api/src/opentrons/protocol_api/instrument_context.py index 9c6338270c7..bc2e072b671 100644 --- a/api/src/opentrons/protocol_api/instrument_context.py +++ b/api/src/opentrons/protocol_api/instrument_context.py @@ -1712,6 +1712,147 @@ def move_to( return self + @requires_version(2, 22) + def resin_tip_seal( + self, + location: Union[labware.Well, labware.Labware], + ) -> InstrumentContext: + """Seal resin tips onto the pipette. + + The location provided should contain resin tips. Sealing the + tip will perform a `pick up` action but there will be no tip tracking + associated with the pipette. + + :param location: A location containing resin tips, must be a Labware or a Well. + + :type location: :py:class:`~.types.Location` + """ + if isinstance(location, labware.Labware): + well = location.wells()[0] + else: + well = location + + with publisher.publish_context( + broker=self.broker, + command=cmds.seal( + instrument=self, + location=well, + ), + ): + self._core.resin_tip_seal( + location=well.top(), well_core=well._core, in_place=False + ) + return self + + @requires_version(2, 22) + def resin_tip_unseal( + self, + location: Union[labware.Well, labware.Labware], + ) -> InstrumentContext: + """Release resin tips from the pipette. + + The location provided should be a valid location to drop resin tips. + + :param location: A location containing that can accept tips. + + :type location: :py:class:`~.types.Location` + + :param home_after: + Whether to home the pipette after dropping the tip. If not specified + defaults to ``True`` on a Flex. The plunger will not home on an unseal. + + When ``False``, the pipette does not home its plunger. This can save a few + seconds, but is not recommended. Homing helps the robot track the pipette's + position. + + """ + if isinstance(location, labware.Labware): + well = location.wells()[0] + else: + well = location + + with publisher.publish_context( + broker=self.broker, + command=cmds.unseal( + instrument=self, + location=well, + ), + ): + self._core.resin_tip_unseal(location=well.top(), well_core=well._core) + + return self + + @requires_version(2, 22) + def resin_tip_dispense( + self, + location: types.Location, + volume: Optional[float] = None, + rate: Optional[float] = None, + ) -> InstrumentContext: + """Dispense a volume from resin tips into a labware. + + The location provided should contain resin tips labware as well as a + receptical for dispensed liquid. Dispensing from tip will perform a + `dispense` action of the specified volume at a desired flow rate. + + :param location: A location containing resin tips. + :type location: :py:class:`~.types.Location` + + :param volume: Will default to maximum, recommended to use the default. + The volume, in µL, that the pipette will prepare to handle. + :type volume: float + + :param rate: Will default to 10.0, recommended to use the default. How quickly + a pipette dispenses liquid. The speed in µL/s is calculated as + ``rate`` multiplied by :py:attr:`flow_rate.dispense`. + :type rate: float + + """ + well: Optional[labware.Well] = None + last_location = self._get_last_location_by_api_version() + + try: + target = validation.validate_location( + location=location, last_location=last_location + ) + except validation.NoLocationError as e: + raise RuntimeError( + "If dispense is called without an explicit location, another" + " method that moves to a location (such as move_to or " + "aspirate) must previously have been called so the robot " + "knows where it is." + ) from e + + if isinstance(target, validation.WellTarget): + well = target.well + if target.location: + move_to_location = target.location + elif well.parent._core.is_fixed_trash(): + move_to_location = target.well.top() + else: + move_to_location = target.well.bottom( + z=self._well_bottom_clearances.dispense + ) + else: + raise RuntimeError( + "A well must be specified when using `resin_tip_dispense`." + ) + + with publisher.publish_context( + broker=self.broker, + command=cmds.resin_tip_dispense( + instrument=self, + flow_rate=rate, + ), + ): + self._core.resin_tip_dispense( + move_to_location, + well_core=well._core, + volume=volume, + flow_rate=rate, + ) + return self + @requires_version(2, 18) def _retract( self, diff --git a/api/src/opentrons/protocol_engine/commands/__init__.py b/api/src/opentrons/protocol_engine/commands/__init__.py index 4ad91012b11..6f6c08c35e3 100644 --- a/api/src/opentrons/protocol_engine/commands/__init__.py +++ b/api/src/opentrons/protocol_engine/commands/__init__.py @@ -380,6 +380,28 @@ TryLiquidProbeCommandType, ) +from .evotip_seal_pipette import ( + EvotipSealPipette, + EvotipSealPipetteParams, + EvotipSealPipetteCreate, + EvotipSealPipetteResult, + EvotipSealPipetteCommandType, +) +from .evotip_unseal_pipette import ( + EvotipUnsealPipette, + EvotipUnsealPipetteParams, + EvotipUnsealPipetteCreate, + EvotipUnsealPipetteResult, + EvotipUnsealPipetteCommandType, +) +from .evotip_dispense import ( + EvotipDispense, + EvotipDispenseParams, + EvotipDispenseCreate, + EvotipDispenseResult, + EvotipDispenseCommandType, +) + __all__ = [ # command type unions "Command", @@ -670,4 +692,22 @@ "TryLiquidProbeCreate", "TryLiquidProbeResult", "TryLiquidProbeCommandType", + # evotip seal command bundle + "EvotipSealPipette", + "EvotipSealPipetteParams", + "EvotipSealPipetteCreate", + "EvotipSealPipetteResult", + "EvotipSealPipetteCommandType", + # evotip unseal command bundle + "EvotipUnsealPipette", + "EvotipUnsealPipetteParams", + "EvotipUnsealPipetteCreate", + "EvotipUnsealPipetteResult", + "EvotipUnsealPipetteCommandType", + # evotip dispense command bundle + "EvotipDispense", + "EvotipDispenseParams", + "EvotipDispenseCreate", + "EvotipDispenseResult", + "EvotipDispenseCommandType", ] diff --git a/api/src/opentrons/protocol_engine/commands/command_unions.py b/api/src/opentrons/protocol_engine/commands/command_unions.py index b04b381ae6b..d5d1b0a3fc9 100644 --- a/api/src/opentrons/protocol_engine/commands/command_unions.py +++ b/api/src/opentrons/protocol_engine/commands/command_unions.py @@ -360,6 +360,30 @@ TryLiquidProbeCommandType, ) +from .evotip_seal_pipette import ( + EvotipSealPipette, + EvotipSealPipetteParams, + EvotipSealPipetteCreate, + EvotipSealPipetteResult, + EvotipSealPipetteCommandType, +) + +from .evotip_dispense import ( + EvotipDispense, + EvotipDispenseParams, + EvotipDispenseCreate, + EvotipDispenseResult, + EvotipDispenseCommandType, +) + +from .evotip_unseal_pipette import ( + EvotipUnsealPipette, + EvotipUnsealPipetteParams, + EvotipUnsealPipetteCreate, + EvotipUnsealPipetteResult, + EvotipUnsealPipetteCommandType, +) + Command = Annotated[ Union[ AirGapInPlace, @@ -404,6 +428,9 @@ GetNextTip, LiquidProbe, TryLiquidProbe, + EvotipSealPipette, + EvotipDispense, + EvotipUnsealPipette, heater_shaker.WaitForTemperature, heater_shaker.SetTargetTemperature, heater_shaker.DeactivateHeater, @@ -492,6 +519,9 @@ GetNextTipParams, LiquidProbeParams, TryLiquidProbeParams, + EvotipSealPipetteParams, + EvotipDispenseParams, + EvotipUnsealPipetteParams, heater_shaker.WaitForTemperatureParams, heater_shaker.SetTargetTemperatureParams, heater_shaker.DeactivateHeaterParams, @@ -578,6 +608,9 @@ GetNextTipCommandType, LiquidProbeCommandType, TryLiquidProbeCommandType, + EvotipSealPipetteCommandType, + EvotipDispenseCommandType, + EvotipUnsealPipetteCommandType, heater_shaker.WaitForTemperatureCommandType, heater_shaker.SetTargetTemperatureCommandType, heater_shaker.DeactivateHeaterCommandType, @@ -665,6 +698,9 @@ GetNextTipCreate, LiquidProbeCreate, TryLiquidProbeCreate, + EvotipSealPipetteCreate, + EvotipDispenseCreate, + EvotipUnsealPipetteCreate, heater_shaker.WaitForTemperatureCreate, heater_shaker.SetTargetTemperatureCreate, heater_shaker.DeactivateHeaterCreate, @@ -760,6 +796,9 @@ GetNextTipResult, LiquidProbeResult, TryLiquidProbeResult, + EvotipSealPipetteResult, + EvotipDispenseResult, + EvotipUnsealPipetteResult, heater_shaker.WaitForTemperatureResult, heater_shaker.SetTargetTemperatureResult, heater_shaker.DeactivateHeaterResult, diff --git a/api/src/opentrons/protocol_engine/commands/evotip_dispense.py b/api/src/opentrons/protocol_engine/commands/evotip_dispense.py new file mode 100644 index 00000000000..e0053262295 --- /dev/null +++ b/api/src/opentrons/protocol_engine/commands/evotip_dispense.py @@ -0,0 +1,156 @@ +"""Evotip Dispense-in-place command request, result, and implementation models.""" + +from __future__ import annotations +from typing import TYPE_CHECKING, Optional, Type, Union +from typing_extensions import Literal + +from opentrons.protocol_engine.errors import UnsupportedLabwareForActionError +from .pipetting_common import ( + PipetteIdMixin, + FlowRateMixin, + DispenseVolumeMixin, + BaseLiquidHandlingResult, + dispense_in_place, +) +from .movement_common import ( + LiquidHandlingWellLocationMixin, + StallOrCollisionError, + move_to_well, +) + +from .command import ( + AbstractCommandImpl, + BaseCommand, + BaseCommandCreate, + SuccessData, + DefinedErrorData, +) +from ..state.update_types import StateUpdate +from ..resources import labware_validation +from ..errors import ProtocolEngineError + +if TYPE_CHECKING: + from ..execution import PipettingHandler, GantryMover, MovementHandler + from ..resources import ModelUtils + from ..state.state import StateView + + +EvotipDispenseCommandType = Literal["evotipDispense"] + + +class EvotipDispenseParams( + PipetteIdMixin, DispenseVolumeMixin, FlowRateMixin, LiquidHandlingWellLocationMixin +): + """Payload required to dispense in place.""" + + pass + + +class EvotipDispenseResult(BaseLiquidHandlingResult): + """Result data from the execution of a DispenseInPlace command.""" + + pass + + +_ExecuteReturn = Union[ + SuccessData[EvotipDispenseResult], + DefinedErrorData[StallOrCollisionError], +] + + +class EvotipDispenseImplementation( + AbstractCommandImpl[EvotipDispenseParams, _ExecuteReturn] +): + """DispenseInPlace command implementation.""" + + def __init__( + self, + pipetting: PipettingHandler, + state_view: StateView, + gantry_mover: GantryMover, + model_utils: ModelUtils, + movement: MovementHandler, + **kwargs: object, + ) -> None: + self._pipetting = pipetting + self._state_view = state_view + self._gantry_mover = gantry_mover + self._model_utils = model_utils + self._movement = movement + + async def execute(self, params: EvotipDispenseParams) -> _ExecuteReturn: + """Move to and dispense to the requested well.""" + well_location = params.wellLocation + labware_id = params.labwareId + well_name = params.wellName + + labware_definition = self._state_view.labware.get_definition(params.labwareId) + if not labware_validation.is_evotips(labware_definition.parameters.loadName): + raise UnsupportedLabwareForActionError( + f"Cannot use command: `EvotipDispense` with labware: {labware_definition.parameters.loadName}" + ) + move_result = await move_to_well( + movement=self._movement, + model_utils=self._model_utils, + pipette_id=params.pipetteId, + labware_id=labware_id, + well_name=well_name, + well_location=well_location, + ) + if isinstance(move_result, DefinedErrorData): + return move_result + + current_position = await self._gantry_mover.get_position(params.pipetteId) + result = await dispense_in_place( + pipette_id=params.pipetteId, + volume=params.volume, + flow_rate=params.flowRate, + push_out=None, + location_if_error={ + "retryLocation": ( + current_position.x, + current_position.y, + current_position.z, + ) + }, + pipetting=self._pipetting, + model_utils=self._model_utils, + ) + if isinstance(result, DefinedErrorData): + # TODO (chb, 2025-01-29): Remove this and the OverpressureError returns once disabled for this function + raise ProtocolEngineError( + message="Overpressure Error during Resin Tip Dispense Command." + ) + return SuccessData( + public=EvotipDispenseResult(volume=result.public.volume), + state_update=StateUpdate.reduce( + move_result.state_update, result.state_update + ), + ) + + +class EvotipDispense( + BaseCommand[ + EvotipDispenseParams, + EvotipDispenseResult, + StallOrCollisionError, + ] +): + """DispenseInPlace command model.""" + + commandType: EvotipDispenseCommandType = "evotipDispense" + params: EvotipDispenseParams + result: Optional[EvotipDispenseResult] = None + + _ImplementationCls: Type[ + EvotipDispenseImplementation + ] = EvotipDispenseImplementation + + +class EvotipDispenseCreate(BaseCommandCreate[EvotipDispenseParams]): + """DispenseInPlace command request model.""" + + commandType: EvotipDispenseCommandType = "evotipDispense" + params: EvotipDispenseParams + + _CommandCls: Type[EvotipDispense] = EvotipDispense diff --git a/api/src/opentrons/protocol_engine/commands/evotip_seal_pipette.py b/api/src/opentrons/protocol_engine/commands/evotip_seal_pipette.py new file mode 100644 index 00000000000..0e67e8fc2c0 --- /dev/null +++ b/api/src/opentrons/protocol_engine/commands/evotip_seal_pipette.py @@ -0,0 +1,331 @@ +"""Seal evotip resin tip command request, result, and implementation models.""" + +from __future__ import annotations +from pydantic import Field, BaseModel +from typing import TYPE_CHECKING, Optional, Type, Union +from opentrons.types import MountType +from opentrons.protocol_engine.types import MotorAxis +from typing_extensions import Literal + +from opentrons.protocol_engine.errors import UnsupportedLabwareForActionError +from ..resources import ModelUtils, labware_validation +from ..types import PickUpTipWellLocation, FluidKind, AspiratedFluid +from .pipetting_common import ( + PipetteIdMixin, +) +from .movement_common import ( + DestinationPositionResult, + StallOrCollisionError, + move_to_well, +) +from .command import ( + AbstractCommandImpl, + BaseCommand, + BaseCommandCreate, + DefinedErrorData, + SuccessData, +) + +from opentrons.hardware_control import HardwareControlAPI +from opentrons.hardware_control.types import Axis +from ..state.update_types import StateUpdate + +if TYPE_CHECKING: + from ..state.state import StateView + from ..execution import ( + MovementHandler, + TipHandler, + GantryMover, + PipettingHandler, + ) + + +EvotipSealPipetteCommandType = Literal["evotipSealPipette"] +_PREP_DISTANCE_DEFAULT = 8.25 +_PRESS_DISTANCE_DEFAULT = 3.5 +_EJECTOR_PUSH_MM_DEFAULT = 7.0 + + +class TipPickUpParams(BaseModel): + """Payload used to specify press-tip parameters for a seal command.""" + + prepDistance: float = Field( + default=0, description="The distance to move down to fit the tips on." + ) + pressDistance: float = Field( + default=0, description="The distance to press on tips." + ) + ejectorPushMm: float = Field( + default=0, + description="The distance to back off to ensure that the tip presence sensors are not triggered.", + ) + + +class EvotipSealPipetteParams(PipetteIdMixin): + """Payload needed to seal resin tips to a pipette.""" + + labwareId: str = Field(..., description="Identifier of labware to use.") + wellName: str = Field(..., description="Name of well to use in labware.") + wellLocation: PickUpTipWellLocation = Field( + default_factory=PickUpTipWellLocation, + description="Relative well location at which to pick up the tip.", + ) + tipPickUpParams: Optional[TipPickUpParams] = Field( + default=None, description="Specific parameters for " + ) + + +class EvotipSealPipetteResult(DestinationPositionResult): + """Result data from the execution of a EvotipSealPipette.""" + + tipVolume: float = Field( + 0, + description="Maximum volume of liquid that the picked up tip can hold, in µL.", + ge=0, + ) + + tipLength: float = Field( + 0, + description="The length of the tip in mm.", + ge=0, + ) + + tipDiameter: float = Field( + 0, + description="The diameter of the tip in mm.", + ge=0, + ) + + +_ExecuteReturn = Union[ + SuccessData[EvotipSealPipetteResult], + DefinedErrorData[StallOrCollisionError], +] + + +class EvotipSealPipetteImplementation( + AbstractCommandImpl[EvotipSealPipetteParams, _ExecuteReturn] +): + """Evotip seal pipette command implementation.""" + + def __init__( + self, + state_view: StateView, + tip_handler: TipHandler, + model_utils: ModelUtils, + movement: MovementHandler, + hardware_api: HardwareControlAPI, + gantry_mover: GantryMover, + pipetting: PipettingHandler, + **kwargs: object, + ) -> None: + self._state_view = state_view + self._tip_handler = tip_handler + self._model_utils = model_utils + self._movement = movement + self._gantry_mover = gantry_mover + self._pipetting = pipetting + self._hardware_api = hardware_api + + async def relative_pickup_tip( + self, + tip_pick_up_params: TipPickUpParams, + mount: MountType, + ) -> None: + """A relative press-fit pick up command using gantry moves.""" + prep_distance = tip_pick_up_params.prepDistance + press_distance = tip_pick_up_params.pressDistance + retract_distance = -1 * (prep_distance + press_distance) + + mount_axis = MotorAxis.LEFT_Z if mount == MountType.LEFT else MotorAxis.RIGHT_Z + + # TODO chb, 2025-01-29): Factor out the movement constants and relocate this logic into the hardware controller + await self._gantry_mover.move_axes( + axis_map={mount_axis: prep_distance}, speed=10, relative_move=True + ) + + # Drive mount down for press-fit + await self._gantry_mover.move_axes( + axis_map={mount_axis: press_distance}, + speed=10.0, + relative_move=True, + expect_stalls=True, + ) + # retract cam : 11.05 + await self._gantry_mover.move_axes( + axis_map={mount_axis: retract_distance}, speed=5.5, relative_move=True + ) + + async def cam_action_relative_pickup_tip( + self, + tip_pick_up_params: TipPickUpParams, + mount: MountType, + ) -> None: + """A cam action pick up command using gantry moves.""" + prep_distance = tip_pick_up_params.prepDistance + press_distance = tip_pick_up_params.pressDistance + ejector_push_mm = tip_pick_up_params.ejectorPushMm + retract_distance = -1 * (prep_distance + press_distance) + + mount_axis = MotorAxis.LEFT_Z if mount == MountType.LEFT else MotorAxis.RIGHT_Z + + # TODO chb, 2025-01-29): Factor out the movement constants and relocate this logic into the hardware controller + await self._gantry_mover.move_axes( + axis_map={mount_axis: -6}, speed=10, relative_move=True + ) + + # Drive Q down 3mm at fast speed - look into the pick up tip fuinction to find slow and fast: 10.0 + await self._gantry_mover.move_axes( + axis_map={MotorAxis.AXIS_96_CHANNEL_CAM: prep_distance}, + speed=10.0, + relative_move=True, + ) + # 2.8mm at slow speed - cam action pickup speed: 5.5 + await self._gantry_mover.move_axes( + axis_map={MotorAxis.AXIS_96_CHANNEL_CAM: press_distance}, + speed=5.5, + relative_move=True, + ) + # retract cam : 11.05 + await self._gantry_mover.move_axes( + axis_map={MotorAxis.AXIS_96_CHANNEL_CAM: retract_distance}, + speed=5.5, + relative_move=True, + ) + + # Lower tip presence + await self._gantry_mover.move_axes( + axis_map={mount_axis: 2}, speed=10, relative_move=True + ) + await self._gantry_mover.move_axes( + axis_map={MotorAxis.AXIS_96_CHANNEL_CAM: ejector_push_mm}, + speed=5.5, + relative_move=True, + ) + await self._gantry_mover.move_axes( + axis_map={MotorAxis.AXIS_96_CHANNEL_CAM: -1 * ejector_push_mm}, + speed=5.5, + relative_move=True, + ) + + async def execute( + self, params: EvotipSealPipetteParams + ) -> Union[SuccessData[EvotipSealPipetteResult], _ExecuteReturn]: + """Move to and pick up a tip using the requested pipette.""" + pipette_id = params.pipetteId + labware_id = params.labwareId + well_name = params.wellName + + labware_definition = self._state_view.labware.get_definition(params.labwareId) + if not labware_validation.is_evotips(labware_definition.parameters.loadName): + raise UnsupportedLabwareForActionError( + f"Cannot use command: `EvotipSealPipette` with labware: {labware_definition.parameters.loadName}" + ) + + well_location = self._state_view.geometry.convert_pick_up_tip_well_location( + well_location=params.wellLocation + ) + move_result = await move_to_well( + movement=self._movement, + model_utils=self._model_utils, + pipette_id=pipette_id, + labware_id=labware_id, + well_name=well_name, + well_location=well_location, + ) + if isinstance(move_result, DefinedErrorData): + return move_result + + # Aspirate to move plunger to a maximum volume position per pipette type + tip_geometry = self._state_view.geometry.get_nominal_tip_geometry( + pipette_id, labware_id, well_name + ) + maximum_volume = self._state_view.pipettes.get_maximum_volume(pipette_id) + if self._state_view.pipettes.get_mount(pipette_id) == MountType.LEFT: + await self._hardware_api.home(axes=[Axis.P_L]) + else: + await self._hardware_api.home(axes=[Axis.P_R]) + + # Begin relative pickup steps for the resin tips + + channels = self._state_view.tips.get_pipette_active_channels(pipette_id) + mount = self._state_view.pipettes.get_mount(pipette_id) + tip_pick_up_params = params.tipPickUpParams + if tip_pick_up_params is None: + tip_pick_up_params = TipPickUpParams( + prepDistance=_PREP_DISTANCE_DEFAULT, + pressDistance=_PRESS_DISTANCE_DEFAULT, + ejectorPushMm=_EJECTOR_PUSH_MM_DEFAULT, + ) + + if channels != 96: + await self.relative_pickup_tip( + tip_pick_up_params=tip_pick_up_params, + mount=mount, + ) + elif channels == 96: + await self.cam_action_relative_pickup_tip( + tip_pick_up_params=tip_pick_up_params, + mount=mount, + ) + else: + tip_geometry = await self._tip_handler.pick_up_tip( + pipette_id=pipette_id, + labware_id=labware_id, + well_name=well_name, + do_not_ignore_tip_presence=True, + ) + + # cache_tip + if self._state_view.config.use_virtual_pipettes is False: + self._tip_handler.cache_tip(pipette_id, tip_geometry) + hw_instr = self._hardware_api.hardware_instruments[mount.to_hw_mount()] + if hw_instr is not None: + hw_instr.set_current_volume(maximum_volume) + + state_update = StateUpdate() + state_update.update_pipette_tip_state( + pipette_id=pipette_id, + tip_geometry=tip_geometry, + ) + + state_update.set_fluid_aspirated( + pipette_id=pipette_id, + fluid=AspiratedFluid(kind=FluidKind.LIQUID, volume=maximum_volume), + ) + return SuccessData( + public=EvotipSealPipetteResult( + tipVolume=tip_geometry.volume, + tipLength=tip_geometry.length, + tipDiameter=tip_geometry.diameter, + position=move_result.public.position, + ), + state_update=state_update, + ) + + +class EvotipSealPipette( + BaseCommand[ + EvotipSealPipetteParams, + EvotipSealPipetteResult, + StallOrCollisionError, + ] +): + """Seal evotip resin tip command model.""" + + commandType: EvotipSealPipetteCommandType = "evotipSealPipette" + params: EvotipSealPipetteParams + result: Optional[EvotipSealPipetteResult] = None + + _ImplementationCls: Type[ + EvotipSealPipetteImplementation + ] = EvotipSealPipetteImplementation + + +class EvotipSealPipetteCreate(BaseCommandCreate[EvotipSealPipetteParams]): + """Seal evotip resin tip command creation request model.""" + + commandType: EvotipSealPipetteCommandType = "evotipSealPipette" + params: EvotipSealPipetteParams + + _CommandCls: Type[EvotipSealPipette] = EvotipSealPipette diff --git a/api/src/opentrons/protocol_engine/commands/evotip_unseal_pipette.py b/api/src/opentrons/protocol_engine/commands/evotip_unseal_pipette.py new file mode 100644 index 00000000000..e8cde34fd5c --- /dev/null +++ b/api/src/opentrons/protocol_engine/commands/evotip_unseal_pipette.py @@ -0,0 +1,160 @@ +"""Unseal evotip resin tip command request, result, and implementation models.""" + +from __future__ import annotations + +from pydantic import Field +from typing import TYPE_CHECKING, Optional, Type +from typing_extensions import Literal + +from opentrons.protocol_engine.resources.model_utils import ModelUtils +from opentrons.protocol_engine.errors import UnsupportedLabwareForActionError +from opentrons.protocol_engine.types import MotorAxis +from opentrons.types import MountType + +from ..types import DropTipWellLocation +from .pipetting_common import ( + PipetteIdMixin, +) +from .movement_common import ( + DestinationPositionResult, + move_to_well, + StallOrCollisionError, +) +from .command import ( + AbstractCommandImpl, + BaseCommand, + BaseCommandCreate, + DefinedErrorData, + SuccessData, +) +from ..resources import labware_validation + +if TYPE_CHECKING: + from ..state.state import StateView + from ..execution import MovementHandler, TipHandler, GantryMover + + +EvotipUnsealPipetteCommandType = Literal["evotipUnsealPipette"] + + +class EvotipUnsealPipetteParams(PipetteIdMixin): + """Payload required to drop a tip in a specific well.""" + + labwareId: str = Field(..., description="Identifier of labware to use.") + wellName: str = Field(..., description="Name of well to use in labware.") + wellLocation: DropTipWellLocation = Field( + default_factory=DropTipWellLocation, + description="Relative well location at which to drop the tip.", + ) + + +class EvotipUnsealPipetteResult(DestinationPositionResult): + """Result data from the execution of a DropTip command.""" + + pass + + +_ExecuteReturn = ( + SuccessData[EvotipUnsealPipetteResult] | DefinedErrorData[StallOrCollisionError] +) + + +class EvotipUnsealPipetteImplementation( + AbstractCommandImpl[EvotipUnsealPipetteParams, _ExecuteReturn] +): + """Drop tip command implementation.""" + + def __init__( + self, + state_view: StateView, + tip_handler: TipHandler, + movement: MovementHandler, + model_utils: ModelUtils, + gantry_mover: GantryMover, + **kwargs: object, + ) -> None: + self._state_view = state_view + self._tip_handler = tip_handler + self._movement_handler = movement + self._model_utils = model_utils + self._gantry_mover = gantry_mover + + async def execute(self, params: EvotipUnsealPipetteParams) -> _ExecuteReturn: + """Move to and drop a tip using the requested pipette.""" + pipette_id = params.pipetteId + labware_id = params.labwareId + well_name = params.wellName + + well_location = params.wellLocation + labware_definition = self._state_view.labware.get_definition(params.labwareId) + if not labware_validation.is_evotips(labware_definition.parameters.loadName): + raise UnsupportedLabwareForActionError( + f"Cannot use command: `EvotipUnsealPipette` with labware: {labware_definition.parameters.loadName}" + ) + is_partially_configured = self._state_view.pipettes.get_is_partially_configured( + pipette_id=pipette_id + ) + tip_drop_location = self._state_view.geometry.get_checked_tip_drop_location( + pipette_id=pipette_id, + labware_id=labware_id, + well_location=well_location, + partially_configured=is_partially_configured, + ) + + move_result = await move_to_well( + movement=self._movement_handler, + model_utils=self._model_utils, + pipette_id=pipette_id, + labware_id=labware_id, + well_name=well_name, + well_location=tip_drop_location, + ) + if isinstance(move_result, DefinedErrorData): + return move_result + + # Move to an appropriate position + mount = self._state_view.pipettes.get_mount(pipette_id) + + mount_axis = MotorAxis.LEFT_Z if mount == MountType.LEFT else MotorAxis.RIGHT_Z + await self._gantry_mover.move_axes( + axis_map={mount_axis: -14}, speed=10, relative_move=True + ) + + await self._tip_handler.drop_tip( + pipette_id=pipette_id, + home_after=None, + do_not_ignore_tip_presence=False, + ignore_plunger=True, + ) + + return SuccessData( + public=EvotipUnsealPipetteResult(position=move_result.public.position), + state_update=move_result.state_update.set_fluid_unknown( + pipette_id=pipette_id + ).update_pipette_tip_state(pipette_id=params.pipetteId, tip_geometry=None), + ) + + +class EvotipUnsealPipette( + BaseCommand[ + EvotipUnsealPipetteParams, EvotipUnsealPipetteResult, StallOrCollisionError + ] +): + """Evotip unseal command model.""" + + commandType: EvotipUnsealPipetteCommandType = "evotipUnsealPipette" + params: EvotipUnsealPipetteParams + result: Optional[EvotipUnsealPipetteResult] = None + + _ImplementationCls: Type[ + EvotipUnsealPipetteImplementation + ] = EvotipUnsealPipetteImplementation + + +class EvotipUnsealPipetteCreate(BaseCommandCreate[EvotipUnsealPipetteParams]): + """Evotip unseal command creation request model.""" + + commandType: EvotipUnsealPipetteCommandType = "evotipUnsealPipette" + params: EvotipUnsealPipetteParams + + _CommandCls: Type[EvotipUnsealPipette] = EvotipUnsealPipette diff --git a/api/src/opentrons/protocol_engine/errors/__init__.py b/api/src/opentrons/protocol_engine/errors/__init__.py index 419043120a6..d3dcc5abaac 100644 --- a/api/src/opentrons/protocol_engine/errors/__init__.py +++ b/api/src/opentrons/protocol_engine/errors/__init__.py @@ -11,6 +11,7 @@ PickUpTipTipNotAttachedError, TipAttachedError, CommandDoesNotExistError, + UnsupportedLabwareForActionError, LabwareNotLoadedError, LabwareNotLoadedOnModuleError, LabwareNotLoadedOnLabwareError, @@ -104,6 +105,7 @@ "LabwareNotLoadedOnLabwareError", "LabwareNotOnDeckError", "LiquidDoesNotExistError", + "UnsupportedLabwareForActionError", "LabwareDefinitionDoesNotExistError", "LabwareCannotBeStackedError", "LabwareCannotSitOnDeckError", diff --git a/api/src/opentrons/protocol_engine/errors/exceptions.py b/api/src/opentrons/protocol_engine/errors/exceptions.py index 2f7e4b07e56..d8c96d8dd31 100644 --- a/api/src/opentrons/protocol_engine/errors/exceptions.py +++ b/api/src/opentrons/protocol_engine/errors/exceptions.py @@ -387,6 +387,19 @@ def __init__( super().__init__(ErrorCodes.GENERAL_ERROR, message, details, wrapping) +class UnsupportedLabwareForActionError(ProtocolEngineError): + """Raised when trying to use an unsupported labware for a command.""" + + def __init__( + self, + message: Optional[str] = None, + details: Optional[Dict[str, Any]] = None, + wrapping: Optional[Sequence[EnumeratedError]] = None, + ) -> None: + """Build a UnsupportedLabwareForActionError.""" + super().__init__(ErrorCodes.GENERAL_ERROR, message, details, wrapping) + + class WellDoesNotExistError(ProtocolEngineError): """Raised when referencing a well that does not exist.""" diff --git a/api/src/opentrons/protocol_engine/execution/gantry_mover.py b/api/src/opentrons/protocol_engine/execution/gantry_mover.py index 5413de8741c..ca90d0a12cb 100644 --- a/api/src/opentrons/protocol_engine/execution/gantry_mover.py +++ b/api/src/opentrons/protocol_engine/execution/gantry_mover.py @@ -114,6 +114,7 @@ async def move_axes( critical_point: Optional[Dict[MotorAxis, float]] = None, speed: Optional[float] = None, relative_move: bool = False, + expect_stalls: bool = False, ) -> Dict[MotorAxis, float]: """Move a set of axes a given distance.""" ... @@ -341,6 +342,7 @@ async def move_axes( critical_point: Optional[Dict[MotorAxis, float]] = None, speed: Optional[float] = None, relative_move: bool = False, + expect_stalls: bool = False, ) -> Dict[MotorAxis, float]: """Move a set of axes a given distance. @@ -349,6 +351,7 @@ async def move_axes( critical_point: A critical point override for axes speed: Optional speed parameter for the move. relative_move: Whether the axis map needs to be converted from a relative to absolute move. + expect_stalls: Whether it is expected that the move triggers a stall error. """ try: pos_hw = self._convert_axis_map_for_hw(axis_map) @@ -385,6 +388,7 @@ async def move_axes( await self._hardware_api.move_axes( position=absolute_pos, speed=speed, + expect_stalls=expect_stalls, ) except PositionUnknownError as e: @@ -588,6 +592,7 @@ async def move_axes( critical_point: Optional[Dict[MotorAxis, float]] = None, speed: Optional[float] = None, relative_move: bool = False, + expect_stalls: bool = False, ) -> Dict[MotorAxis, float]: """Move the give axes map. No-op in virtual implementation.""" mount = self.pick_mount_from_axis_map(axis_map) diff --git a/api/src/opentrons/protocol_engine/execution/tip_handler.py b/api/src/opentrons/protocol_engine/execution/tip_handler.py index 8a58536c10b..d27925e00fe 100644 --- a/api/src/opentrons/protocol_engine/execution/tip_handler.py +++ b/api/src/opentrons/protocol_engine/execution/tip_handler.py @@ -62,6 +62,7 @@ async def pick_up_tip( pipette_id: str, labware_id: str, well_name: str, + do_not_ignore_tip_presence: bool = True, ) -> TipGeometry: """Pick up the named tip. @@ -75,7 +76,13 @@ async def pick_up_tip( """ ... - async def drop_tip(self, pipette_id: str, home_after: Optional[bool]) -> None: + async def drop_tip( + self, + pipette_id: str, + home_after: Optional[bool], + do_not_ignore_tip_presence: bool = True, + ignore_plunger: bool = False, + ) -> None: """Drop the attached tip into the current location. Pipette should be in place over the destination prior to calling this method. @@ -230,6 +237,7 @@ async def pick_up_tip( pipette_id: str, labware_id: str, well_name: str, + do_not_ignore_tip_presence: bool = True, ) -> TipGeometry: """See documentation on abstract base class.""" hw_mount = self._get_hw_mount(pipette_id) @@ -253,10 +261,11 @@ async def pick_up_tip( await self._hardware_api.tip_pickup_moves( mount=hw_mount, presses=None, increment=None ) - try: - await self.verify_tip_presence(pipette_id, TipPresenceStatus.PRESENT) - except TipNotAttachedError as e: - raise PickUpTipTipNotAttachedError(tip_geometry=tip_geometry) from e + if do_not_ignore_tip_presence: + try: + await self.verify_tip_presence(pipette_id, TipPresenceStatus.PRESENT) + except TipNotAttachedError as e: + raise PickUpTipTipNotAttachedError(tip_geometry=tip_geometry) from e self.cache_tip(pipette_id, tip_geometry) @@ -264,7 +273,13 @@ async def pick_up_tip( return tip_geometry - async def drop_tip(self, pipette_id: str, home_after: Optional[bool]) -> None: + async def drop_tip( + self, + pipette_id: str, + home_after: Optional[bool], + do_not_ignore_tip_presence: bool = True, + ignore_plunger: bool = False, + ) -> None: """See documentation on abstract base class.""" hw_mount = self._get_hw_mount(pipette_id) @@ -275,10 +290,13 @@ async def drop_tip(self, pipette_id: str, home_after: Optional[bool]) -> None: else: kwargs = {} - await self._hardware_api.tip_drop_moves(mount=hw_mount, **kwargs) + await self._hardware_api.tip_drop_moves( + mount=hw_mount, ignore_plunger=ignore_plunger, **kwargs + ) - # Allow TipNotAttachedError to propagate. - await self.verify_tip_presence(pipette_id, TipPresenceStatus.ABSENT) + if do_not_ignore_tip_presence: + # Allow TipNotAttachedError to propagate. + await self.verify_tip_presence(pipette_id, TipPresenceStatus.ABSENT) self.remove_tip(pipette_id) @@ -383,6 +401,7 @@ async def pick_up_tip( pipette_id: str, labware_id: str, well_name: str, + do_not_ignore_tip_presence: bool = True, ) -> TipGeometry: """Pick up a tip at the current location using a virtual pipette. @@ -424,6 +443,8 @@ async def drop_tip( self, pipette_id: str, home_after: Optional[bool], + do_not_ignore_tip_presence: bool = True, + ignore_plunger: bool = False, ) -> None: """Pick up a tip at the current location using a virtual pipette. diff --git a/api/src/opentrons/protocol_engine/resources/labware_validation.py b/api/src/opentrons/protocol_engine/resources/labware_validation.py index 451f3afbfab..b33650e65be 100644 --- a/api/src/opentrons/protocol_engine/resources/labware_validation.py +++ b/api/src/opentrons/protocol_engine/resources/labware_validation.py @@ -14,6 +14,11 @@ def is_absorbance_reader_lid(load_name: str) -> bool: return load_name == "opentrons_flex_lid_absorbance_plate_reader_module" +def is_evotips(load_name: str) -> bool: + """Check if a labware is an evotips tiprack.""" + return load_name == "evotips_opentrons_96_labware" + + def validate_definition_is_labware(definition: LabwareDefinition) -> bool: """Validate that one of the definition's allowed roles is `labware`. diff --git a/api/src/opentrons/protocol_engine/state/pipettes.py b/api/src/opentrons/protocol_engine/state/pipettes.py index 9b7d289e890..fc7b1da20ac 100644 --- a/api/src/opentrons/protocol_engine/state/pipettes.py +++ b/api/src/opentrons/protocol_engine/state/pipettes.py @@ -331,6 +331,11 @@ def _update_volumes(self, state_update: update_types.StateUpdate) -> None: def _update_aspirated( self, update: update_types.PipetteAspiratedFluidUpdate ) -> None: + if self._state.pipette_contents_by_id[update.pipette_id] is None: + self._state.pipette_contents_by_id[ + update.pipette_id + ] = fluid_stack.FluidStack() + self._fluid_stack_log_if_empty(update.pipette_id).add_fluid(update.fluid) def _update_ejected(self, update: update_types.PipetteEjectedFluidUpdate) -> None: diff --git a/api/tests/opentrons/hardware_control/test_ot3_api.py b/api/tests/opentrons/hardware_control/test_ot3_api.py index 200549108db..cbb5838c266 100644 --- a/api/tests/opentrons/hardware_control/test_ot3_api.py +++ b/api/tests/opentrons/hardware_control/test_ot3_api.py @@ -1895,7 +1895,9 @@ async def test_move_axes( await ot3_hardware.move_axes(position=input_position) mock_check_motor.return_value = True - mock_move.assert_called_once_with(target_position=expected_move_pos, speed=None) + mock_move.assert_called_once_with( + target_position=expected_move_pos, speed=None, expect_stalls=False + ) async def test_move_gripper_mount_without_gripper_attached( @@ -1913,14 +1915,14 @@ async def test_move_expect_stall_flag( expected = HWStopCondition.stall if expect_stalls else HWStopCondition.none - await ot3_hardware.move_to(Mount.LEFT, Point(0, 0, 0), _expect_stalls=expect_stalls) + await ot3_hardware.move_to(Mount.LEFT, Point(0, 0, 0), expect_stalls=expect_stalls) mock_backend_move.assert_called_once() _, _, _, condition = mock_backend_move.call_args_list[0][0] assert condition == expected mock_backend_move.reset_mock() await ot3_hardware.move_rel( - Mount.LEFT, Point(10, 0, 0), _expect_stalls=expect_stalls + Mount.LEFT, Point(10, 0, 0), expect_stalls=expect_stalls ) mock_backend_move.assert_called_once() _, _, _, condition = mock_backend_move.call_args_list[0][0] diff --git a/api/tests/opentrons/protocol_engine/commands/test_evotip_dispense.py b/api/tests/opentrons/protocol_engine/commands/test_evotip_dispense.py new file mode 100644 index 00000000000..568ddda83c1 --- /dev/null +++ b/api/tests/opentrons/protocol_engine/commands/test_evotip_dispense.py @@ -0,0 +1,133 @@ +"""Test evotip dispense in place commands.""" +import json + +import pytest +from decoy import Decoy + +from opentrons.protocol_engine import ( + LiquidHandlingWellLocation, + WellOrigin, + WellOffset, + DeckPoint, +) +from opentrons.types import Point +from opentrons.protocol_engine.execution import ( + PipettingHandler, + GantryMover, + MovementHandler, +) + +from opentrons.protocols.models import LabwareDefinition +from opentrons.protocol_engine.commands.command import SuccessData +from opentrons.protocol_engine.commands.evotip_dispense import ( + EvotipDispenseParams, + EvotipDispenseResult, + EvotipDispenseImplementation, +) +from opentrons.protocol_engine.resources import ModelUtils +from opentrons.protocol_engine.state.state import StateView +from opentrons.protocol_engine.state import update_types + +from opentrons_shared_data import load_shared_data + + +@pytest.fixture +def evotips_definition() -> LabwareDefinition: + """A fixturee of the evotips definition.""" + # TODO (chb 2025-01-29): When we migrate all labware to v3 we can clean this up + return LabwareDefinition.model_validate( + json.loads( + load_shared_data( + "labware/definitions/3/evotips_opentrons_96_labware/1.json" + ) + ) + ) + + +@pytest.fixture +def subject( + pipetting: PipettingHandler, + state_view: StateView, + gantry_mover: GantryMover, + model_utils: ModelUtils, + movement: MovementHandler, + **kwargs: object, +) -> EvotipDispenseImplementation: + """Build a command implementation.""" + return EvotipDispenseImplementation( + pipetting=pipetting, + state_view=state_view, + gantry_mover=gantry_mover, + model_utils=model_utils, + movement=movement, + ) + + +async def test_evotip_dispense_implementation( + decoy: Decoy, + movement: MovementHandler, + gantry_mover: GantryMover, + pipetting: PipettingHandler, + state_view: StateView, + subject: EvotipDispenseImplementation, + evotips_definition: LabwareDefinition, +) -> None: + """It should dispense in place.""" + well_location = LiquidHandlingWellLocation( + origin=WellOrigin.TOP, offset=WellOffset(x=0, y=0, z=0) + ) + + data = EvotipDispenseParams( + pipetteId="pipette-id-abc123", + labwareId="labware-id-abc123", + wellName="A3", + volume=100, + flowRate=456, + ) + + decoy.when( + await movement.move_to_well( + pipette_id="pipette-id-abc123", + labware_id="labware-id-abc123", + well_name="A3", + well_location=well_location, + current_well=None, + force_direct=False, + minimum_z_height=None, + speed=None, + operation_volume=None, + ) + ).then_return(Point(x=1, y=2, z=3)) + + decoy.when(state_view.labware.get_definition("labware-id-abc123")).then_return( + evotips_definition + ) + + decoy.when( + await pipetting.dispense_in_place( + pipette_id="pipette-id-abc123", volume=100.0, flow_rate=456.0, push_out=None + ) + ).then_return(100) + + decoy.when(await gantry_mover.get_position("pipette-id-abc123")).then_return( + Point(1, 2, 3) + ) + + result = await subject.execute(data) + + assert result == SuccessData( + public=EvotipDispenseResult(volume=100), + state_update=update_types.StateUpdate( + pipette_location=update_types.PipetteLocationUpdate( + pipette_id="pipette-id-abc123", + new_location=update_types.Well( + labware_id="labware-id-abc123", + well_name="A3", + ), + new_deck_point=DeckPoint.model_construct(x=1, y=2, z=3), + ), + pipette_aspirated_fluid=update_types.PipetteEjectedFluidUpdate( + pipette_id="pipette-id-abc123", volume=100 + ), + ), + ) diff --git a/api/tests/opentrons/protocol_engine/commands/test_evotip_seal_pipette.py b/api/tests/opentrons/protocol_engine/commands/test_evotip_seal_pipette.py new file mode 100644 index 00000000000..3d1bbeb1406 --- /dev/null +++ b/api/tests/opentrons/protocol_engine/commands/test_evotip_seal_pipette.py @@ -0,0 +1,300 @@ +"""Test evotip seal commands.""" + +import pytest +from datetime import datetime + +from decoy import Decoy, matchers +from unittest.mock import sentinel + +from opentrons.protocols.models import LabwareDefinition + +from opentrons_shared_data.errors.exceptions import StallOrCollisionDetectedError + +from opentrons.types import MountType, Point + +from opentrons.protocol_engine import ( + WellLocation, + PickUpTipWellLocation, + WellOffset, + DeckPoint, +) +from opentrons.protocol_engine.errors import PickUpTipTipNotAttachedError +from opentrons.protocol_engine.execution import MovementHandler, GantryMover, TipHandler +from opentrons.protocol_engine.resources import ModelUtils +from opentrons.protocol_engine.state import update_types +from opentrons.protocol_engine.state.state import StateView +from opentrons.protocol_engine.types import TipGeometry, FluidKind, AspiratedFluid + +from opentrons.protocol_engine.commands.movement_common import StallOrCollisionError +from opentrons.protocol_engine.commands.command import DefinedErrorData, SuccessData +from opentrons.protocol_engine.commands.evotip_seal_pipette import ( + EvotipSealPipetteParams, + EvotipSealPipetteResult, + EvotipSealPipetteImplementation, +) +from opentrons.protocol_engine.execution import ( + PipettingHandler, +) +from opentrons.hardware_control import HardwareControlAPI +import json +from opentrons_shared_data import load_shared_data + + +@pytest.fixture +def evotips_definition() -> LabwareDefinition: + """A fixturee of the evotips definition.""" + # TODO (chb 2025-01-29): When we migrate all labware to v3 we can clean this up + return LabwareDefinition.model_validate( + json.loads( + load_shared_data( + "labware/definitions/3/evotips_opentrons_96_labware/1.json" + ) + ) + ) + + +async def test_success( + decoy: Decoy, + state_view: StateView, + movement: MovementHandler, + tip_handler: TipHandler, + model_utils: ModelUtils, + gantry_mover: GantryMover, + evotips_definition: LabwareDefinition, + hardware_api: HardwareControlAPI, + pipetting: PipettingHandler, +) -> None: + """A PickUpTip command should have an execution implementation.""" + subject = EvotipSealPipetteImplementation( + state_view=state_view, + movement=movement, + tip_handler=tip_handler, + model_utils=model_utils, + gantry_mover=gantry_mover, + hardware_api=hardware_api, + pipetting=pipetting, + ) + + decoy.when(state_view.pipettes.get_mount("pipette-id")).then_return(MountType.LEFT) + + decoy.when( + state_view.geometry.convert_pick_up_tip_well_location( + well_location=PickUpTipWellLocation(offset=WellOffset(x=1, y=2, z=3)) + ) + ).then_return(WellLocation(offset=WellOffset(x=1, y=2, z=3))) + + decoy.when(state_view.labware.get_definition("labware-id")).then_return( + evotips_definition + ) + decoy.when( + await movement.move_to_well( + pipette_id="pipette-id", + labware_id="labware-id", + well_name="A3", + well_location=WellLocation(offset=WellOffset(x=1, y=2, z=3)), + current_well=None, + force_direct=False, + minimum_z_height=None, + speed=None, + operation_volume=None, + ) + ).then_return(Point(x=111, y=222, z=333)) + decoy.when( + state_view.geometry.get_nominal_tip_geometry("pipette-id", "labware-id", "A3") + ).then_return(TipGeometry(length=42, diameter=5, volume=300)) + decoy.when(state_view.pipettes.get_maximum_volume("pipette-id")).then_return(1000) + + decoy.when( + await tip_handler.pick_up_tip( + pipette_id="pipette-id", + labware_id="labware-id", + well_name="A3", + ) + ).then_return(TipGeometry(length=42, diameter=5, volume=300)) + + result = await subject.execute( + EvotipSealPipetteParams( + pipetteId="pipette-id", + labwareId="labware-id", + wellName="A3", + wellLocation=PickUpTipWellLocation(offset=WellOffset(x=1, y=2, z=3)), + ) + ) + + assert result == SuccessData( + public=EvotipSealPipetteResult( + tipLength=42, + tipVolume=300, + tipDiameter=5, + position=DeckPoint(x=111, y=222, z=333), + ), + state_update=update_types.StateUpdate( + pipette_tip_state=update_types.PipetteTipStateUpdate( + pipette_id="pipette-id", + tip_geometry=TipGeometry(length=42, diameter=5, volume=300), + ), + pipette_aspirated_fluid=update_types.PipetteAspiratedFluidUpdate( + pipette_id="pipette-id", + fluid=AspiratedFluid(kind=FluidKind.LIQUID, volume=1000), + ), + ), + ) + + +async def test_no_tip_physically_missing_error( + decoy: Decoy, + state_view: StateView, + movement: MovementHandler, + tip_handler: TipHandler, + model_utils: ModelUtils, + gantry_mover: GantryMover, + hardware_api: HardwareControlAPI, + pipetting: PipettingHandler, + evotips_definition: LabwareDefinition, +) -> None: + """It should not return a TipPhysicallyMissingError even though evotips do not sit high enough on the pipette to be detected by the tip sensor.""" + subject = EvotipSealPipetteImplementation( + state_view=state_view, + movement=movement, + tip_handler=tip_handler, + model_utils=model_utils, + gantry_mover=gantry_mover, + hardware_api=hardware_api, + pipetting=pipetting, + ) + + pipette_id = "pipette-id" + labware_id = "labware-id" + well_name = "well-name" + error_id = "error-id" + error_created_at = datetime(1234, 5, 6) + + decoy.when( + state_view.geometry.convert_pick_up_tip_well_location( + well_location=PickUpTipWellLocation(offset=WellOffset()) + ) + ).then_return(WellLocation(offset=WellOffset())) + + decoy.when( + await movement.move_to_well( + pipette_id="pipette-id", + labware_id="labware-id", + well_name="well-name", + well_location=WellLocation(offset=WellOffset()), + current_well=None, + force_direct=False, + minimum_z_height=None, + speed=None, + operation_volume=None, + ) + ).then_return(Point(x=111, y=222, z=333)) + decoy.when( + state_view.geometry.get_nominal_tip_geometry(pipette_id, labware_id, well_name) + ).then_return(TipGeometry(length=42, diameter=5, volume=300)) + + decoy.when( + await tip_handler.pick_up_tip( + pipette_id=pipette_id, labware_id=labware_id, well_name=well_name + ) + ).then_raise(PickUpTipTipNotAttachedError(tip_geometry=sentinel.tip_geometry)) + decoy.when(model_utils.generate_id()).then_return(error_id) + decoy.when(model_utils.get_timestamp()).then_return(error_created_at) + decoy.when(state_view.labware.get_definition(labware_id)).then_return( + evotips_definition + ) + decoy.when(state_view.pipettes.get_maximum_volume(pipette_id)).then_return(1000) + + result = await subject.execute( + EvotipSealPipetteParams( + pipetteId=pipette_id, labwareId=labware_id, wellName=well_name + ) + ) + + assert result == SuccessData( + public=EvotipSealPipetteResult( + tipLength=42, + tipVolume=300, + tipDiameter=5, + position=DeckPoint(x=111, y=222, z=333), + ), + state_update=update_types.StateUpdate( + pipette_tip_state=update_types.PipetteTipStateUpdate( + pipette_id="pipette-id", + tip_geometry=TipGeometry(length=42, diameter=5, volume=300), + ), + pipette_aspirated_fluid=update_types.PipetteAspiratedFluidUpdate( + pipette_id="pipette-id", + fluid=AspiratedFluid(kind=FluidKind.LIQUID, volume=1000), + ), + ), + ) + + +async def test_stall_error( + decoy: Decoy, + state_view: StateView, + movement: MovementHandler, + tip_handler: TipHandler, + model_utils: ModelUtils, + gantry_mover: GantryMover, + hardware_api: HardwareControlAPI, + pipetting: PipettingHandler, + evotips_definition: LabwareDefinition, +) -> None: + """It should return a TipPhysicallyMissingError if the HW API indicates that.""" + subject = EvotipSealPipetteImplementation( + state_view=state_view, + movement=movement, + tip_handler=tip_handler, + model_utils=model_utils, + gantry_mover=gantry_mover, + hardware_api=hardware_api, + pipetting=pipetting, + ) + + pipette_id = "pipette-id" + labware_id = "labware-id" + well_name = "well-name" + error_id = "error-id" + error_created_at = datetime(1234, 5, 6) + + decoy.when( + state_view.geometry.convert_pick_up_tip_well_location( + well_location=PickUpTipWellLocation(offset=WellOffset()) + ) + ).then_return(WellLocation(offset=WellOffset())) + + decoy.when( + await movement.move_to_well( + pipette_id="pipette-id", + labware_id="labware-id", + well_name="well-name", + well_location=WellLocation(offset=WellOffset()), + current_well=None, + force_direct=False, + minimum_z_height=None, + speed=None, + operation_volume=None, + ) + ).then_raise(StallOrCollisionDetectedError()) + + decoy.when(model_utils.generate_id()).then_return(error_id) + decoy.when(model_utils.get_timestamp()).then_return(error_created_at) + decoy.when(state_view.labware.get_definition(labware_id)).then_return( + evotips_definition + ) + + result = await subject.execute( + EvotipSealPipetteParams( + pipetteId=pipette_id, labwareId=labware_id, wellName=well_name + ) + ) + + assert result == DefinedErrorData( + public=StallOrCollisionError.model_construct( + id=error_id, createdAt=error_created_at, wrappedErrors=[matchers.Anything()] + ), + state_update=update_types.StateUpdate( + pipette_location=update_types.CLEAR, + ), + ) diff --git a/api/tests/opentrons/protocol_engine/commands/test_evotip_unseal_pipette.py b/api/tests/opentrons/protocol_engine/commands/test_evotip_unseal_pipette.py new file mode 100644 index 00000000000..5f1c94c3dd6 --- /dev/null +++ b/api/tests/opentrons/protocol_engine/commands/test_evotip_unseal_pipette.py @@ -0,0 +1,330 @@ +"""Test evotip unseal commands.""" + +from datetime import datetime + +import pytest +from decoy import Decoy, matchers + +from opentrons_shared_data.errors.exceptions import StallOrCollisionDetectedError + +from opentrons.protocol_engine import ( + DropTipWellLocation, + DropTipWellOrigin, + WellLocation, + WellOffset, + DeckPoint, +) +from opentrons.protocol_engine.commands.command import DefinedErrorData, SuccessData +from opentrons.protocol_engine.commands.evotip_unseal_pipette import ( + EvotipUnsealPipetteParams, + EvotipUnsealPipetteResult, + EvotipUnsealPipetteImplementation, +) +from opentrons.protocol_engine.commands.movement_common import StallOrCollisionError +from opentrons.protocol_engine.errors.exceptions import TipAttachedError +from opentrons.protocol_engine.resources.model_utils import ModelUtils +from opentrons.protocol_engine.state import update_types +from opentrons.protocol_engine.state.state import StateView +from opentrons.protocol_engine.execution import MovementHandler, GantryMover, TipHandler +from opentrons.protocols.models import LabwareDefinition +import json +from opentrons_shared_data import load_shared_data + +from opentrons.types import Point + + +@pytest.fixture +def mock_state_view(decoy: Decoy) -> StateView: + """Get a mock StateView.""" + return decoy.mock(cls=StateView) + + +@pytest.fixture +def mock_movement_handler(decoy: Decoy) -> MovementHandler: + """Get a mock MovementHandler.""" + return decoy.mock(cls=MovementHandler) + + +@pytest.fixture +def mock_tip_handler(decoy: Decoy) -> TipHandler: + """Get a mock TipHandler.""" + return decoy.mock(cls=TipHandler) + + +@pytest.fixture +def mock_model_utils(decoy: Decoy) -> ModelUtils: + """Get a mock ModelUtils.""" + return decoy.mock(cls=ModelUtils) + + +def test_drop_tip_params_defaults() -> None: + """A drop tip should use a `WellOrigin.DROP_TIP` by default.""" + default_params = EvotipUnsealPipetteParams.model_validate( + {"pipetteId": "abc", "labwareId": "def", "wellName": "ghj"} + ) + + assert default_params.wellLocation == DropTipWellLocation( + origin=DropTipWellOrigin.DEFAULT, offset=WellOffset(x=0, y=0, z=0) + ) + + +def test_drop_tip_params_default_origin() -> None: + """A drop tip should drop a `WellOrigin.DROP_TIP` by default even if an offset is given.""" + default_params = EvotipUnsealPipetteParams.model_validate( + { + "pipetteId": "abc", + "labwareId": "def", + "wellName": "ghj", + "wellLocation": {"offset": {"x": 1, "y": 2, "z": 3}}, + } + ) + + assert default_params.wellLocation == DropTipWellLocation( + origin=DropTipWellOrigin.DEFAULT, offset=WellOffset(x=1, y=2, z=3) + ) + + +@pytest.fixture +def evotips_definition() -> LabwareDefinition: + """A fixturee of the evotips definition.""" + # TODO (chb 2025-01-29): When we migrate all labware to v3 we can clean this up + return LabwareDefinition.model_validate( + json.loads( + load_shared_data( + "labware/definitions/3/evotips_opentrons_96_labware/1.json" + ) + ) + ) + + +async def test_drop_tip_implementation( + decoy: Decoy, + mock_state_view: StateView, + mock_movement_handler: MovementHandler, + mock_tip_handler: TipHandler, + mock_model_utils: ModelUtils, + gantry_mover: GantryMover, + evotips_definition: LabwareDefinition, +) -> None: + """A DropTip command should have an execution implementation.""" + subject = EvotipUnsealPipetteImplementation( + state_view=mock_state_view, + movement=mock_movement_handler, + tip_handler=mock_tip_handler, + model_utils=mock_model_utils, + gantry_mover=gantry_mover, + ) + + params = EvotipUnsealPipetteParams( + pipetteId="abc", + labwareId="123", + wellName="A3", + wellLocation=DropTipWellLocation(offset=WellOffset(x=1, y=2, z=3)), + ) + decoy.when(mock_state_view.labware.get_definition("123")).then_return( + evotips_definition + ) + + decoy.when( + mock_state_view.pipettes.get_is_partially_configured(pipette_id="abc") + ).then_return(False) + + decoy.when( + mock_state_view.geometry.get_checked_tip_drop_location( + pipette_id="abc", + labware_id="123", + well_location=DropTipWellLocation(offset=WellOffset(x=1, y=2, z=3)), + partially_configured=False, + ) + ).then_return(WellLocation(offset=WellOffset(x=4, y=5, z=6))) + + decoy.when( + await mock_movement_handler.move_to_well( + pipette_id="abc", + labware_id="123", + well_name="A3", + well_location=WellLocation(offset=WellOffset(x=4, y=5, z=6)), + current_well=None, + force_direct=False, + minimum_z_height=None, + speed=None, + operation_volume=None, + ) + ).then_return(Point(x=111, y=222, z=333)) + + result = await subject.execute(params) + + assert result == SuccessData( + public=EvotipUnsealPipetteResult(position=DeckPoint(x=111, y=222, z=333)), + state_update=update_types.StateUpdate( + pipette_location=update_types.PipetteLocationUpdate( + pipette_id="abc", + new_location=update_types.Well( + labware_id="123", + well_name="A3", + ), + new_deck_point=DeckPoint(x=111, y=222, z=333), + ), + pipette_tip_state=update_types.PipetteTipStateUpdate( + pipette_id="abc", tip_geometry=None + ), + pipette_aspirated_fluid=update_types.PipetteUnknownFluidUpdate( + pipette_id="abc" + ), + ), + ) + decoy.verify( + await mock_tip_handler.drop_tip( + pipette_id="abc", + home_after=None, + do_not_ignore_tip_presence=False, + ignore_plunger=True, + ), + times=1, + ) + + +async def test_tip_attached_error( + decoy: Decoy, + mock_state_view: StateView, + mock_movement_handler: MovementHandler, + mock_tip_handler: TipHandler, + mock_model_utils: ModelUtils, + gantry_mover: GantryMover, + evotips_definition: LabwareDefinition, +) -> None: + """A Evotip Unseal command should have an execution implementation.""" + subject = EvotipUnsealPipetteImplementation( + state_view=mock_state_view, + movement=mock_movement_handler, + tip_handler=mock_tip_handler, + model_utils=mock_model_utils, + gantry_mover=gantry_mover, + ) + + params = EvotipUnsealPipetteParams( + pipetteId="abc", + labwareId="123", + wellName="A3", + wellLocation=DropTipWellLocation(offset=WellOffset(x=1, y=2, z=3)), + ) + decoy.when(mock_state_view.labware.get_definition("123")).then_return( + evotips_definition + ) + + decoy.when( + mock_state_view.pipettes.get_is_partially_configured(pipette_id="abc") + ).then_return(False) + + decoy.when( + mock_state_view.geometry.get_checked_tip_drop_location( + pipette_id="abc", + labware_id="123", + well_location=DropTipWellLocation(offset=WellOffset(x=1, y=2, z=3)), + partially_configured=False, + ) + ).then_return(WellLocation(offset=WellOffset(x=4, y=5, z=6))) + + decoy.when( + await mock_movement_handler.move_to_well( + pipette_id="abc", + labware_id="123", + well_name="A3", + well_location=WellLocation(offset=WellOffset(x=4, y=5, z=6)), + current_well=None, + force_direct=False, + minimum_z_height=None, + speed=None, + operation_volume=None, + ) + ).then_return(Point(x=111, y=222, z=333)) + decoy.when( + await mock_tip_handler.drop_tip( + pipette_id="abc", + home_after=None, + do_not_ignore_tip_presence=False, + ignore_plunger=True, + ) + ).then_raise(TipAttachedError("Egads!")) + + decoy.when(mock_model_utils.generate_id()).then_return("error-id") + decoy.when(mock_model_utils.get_timestamp()).then_return( + datetime(year=1, month=2, day=3) + ) + + with pytest.raises(TipAttachedError): + await subject.execute(params) + + +async def test_stall_error( + decoy: Decoy, + mock_state_view: StateView, + mock_movement_handler: MovementHandler, + mock_tip_handler: TipHandler, + mock_model_utils: ModelUtils, + gantry_mover: GantryMover, + evotips_definition: LabwareDefinition, +) -> None: + """A DropTip command should have an execution implementation.""" + subject = EvotipUnsealPipetteImplementation( + state_view=mock_state_view, + movement=mock_movement_handler, + tip_handler=mock_tip_handler, + model_utils=mock_model_utils, + gantry_mover=gantry_mover, + ) + + params = EvotipUnsealPipetteParams( + pipetteId="abc", + labwareId="123", + wellName="A3", + wellLocation=DropTipWellLocation(offset=WellOffset(x=1, y=2, z=3)), + ) + decoy.when(mock_state_view.labware.get_definition("123")).then_return( + evotips_definition + ) + + decoy.when( + mock_state_view.pipettes.get_is_partially_configured(pipette_id="abc") + ).then_return(False) + + decoy.when( + mock_state_view.geometry.get_checked_tip_drop_location( + pipette_id="abc", + labware_id="123", + well_location=DropTipWellLocation(offset=WellOffset(x=1, y=2, z=3)), + partially_configured=False, + ) + ).then_return(WellLocation(offset=WellOffset(x=4, y=5, z=6))) + + decoy.when( + await mock_movement_handler.move_to_well( + pipette_id="abc", + labware_id="123", + well_name="A3", + well_location=WellLocation(offset=WellOffset(x=4, y=5, z=6)), + current_well=None, + force_direct=False, + minimum_z_height=None, + speed=None, + operation_volume=None, + ) + ).then_raise(StallOrCollisionDetectedError()) + + decoy.when(mock_model_utils.generate_id()).then_return("error-id") + decoy.when(mock_model_utils.get_timestamp()).then_return( + datetime(year=1, month=2, day=3) + ) + + result = await subject.execute(params) + + assert result == DefinedErrorData( + public=StallOrCollisionError.model_construct( + id="error-id", + createdAt=datetime(year=1, month=2, day=3), + wrappedErrors=[matchers.Anything()], + ), + state_update=update_types.StateUpdate( + pipette_location=update_types.CLEAR, + ), + ) diff --git a/api/tests/opentrons/protocol_engine/execution/test_gantry_mover.py b/api/tests/opentrons/protocol_engine/execution/test_gantry_mover.py index 2c872c003d1..caa8d0cc3b6 100644 --- a/api/tests/opentrons/protocol_engine/execution/test_gantry_mover.py +++ b/api/tests/opentrons/protocol_engine/execution/test_gantry_mover.py @@ -568,7 +568,9 @@ def _current_position(mount: Mount, refresh: bool) -> Dict[HardwareAxis, float]: pos = await subject.move_axes(axis_map, critical_point, 100, relative_move) decoy.verify( - await ot3_hardware_api.move_axes(position=call_to_hw, speed=100), + await ot3_hardware_api.move_axes( + position=call_to_hw, speed=100, expect_stalls=False + ), times=1, ) assert pos == { diff --git a/api/tests/opentrons/protocol_engine/execution/test_tip_handler.py b/api/tests/opentrons/protocol_engine/execution/test_tip_handler.py index 23f701db80b..b0f5d618361 100644 --- a/api/tests/opentrons/protocol_engine/execution/test_tip_handler.py +++ b/api/tests/opentrons/protocol_engine/execution/test_tip_handler.py @@ -257,7 +257,9 @@ async def test_drop_tip( await subject.drop_tip(pipette_id="pipette-id", home_after=True) decoy.verify( - await mock_hardware_api.tip_drop_moves(mount=Mount.RIGHT, home_after=True) + await mock_hardware_api.tip_drop_moves( + mount=Mount.RIGHT, ignore_plunger=False, home_after=True + ) ) decoy.verify(mock_hardware_api.remove_tip(mount=Mount.RIGHT)) decoy.verify( diff --git a/hardware-testing/hardware_testing/production_qc/gripper_assembly_qc_ot3/test_mount.py b/hardware-testing/hardware_testing/production_qc/gripper_assembly_qc_ot3/test_mount.py index 7d0855e54b4..df40b60e61f 100644 --- a/hardware-testing/hardware_testing/production_qc/gripper_assembly_qc_ot3/test_mount.py +++ b/hardware-testing/hardware_testing/production_qc/gripper_assembly_qc_ot3/test_mount.py @@ -120,7 +120,7 @@ async def _save_result(tag: str, target_z: float, include_pass_fail: bool) -> bo mount, Point(z=-Z_AXIS_TRAVEL_DISTANCE), speed=speed, - _expect_stalls=True, + expect_stalls=True, ) down_end_passed = await _save_result( _get_test_tag(current, speed, "down", "end"), @@ -139,7 +139,7 @@ async def _save_result(tag: str, target_z: float, include_pass_fail: bool) -> bo mount, Point(z=Z_AXIS_TRAVEL_DISTANCE), speed=speed, - _expect_stalls=True, + expect_stalls=True, ) up_end_passed = await _save_result( _get_test_tag(current, speed, "up", "end"), diff --git a/hardware-testing/hardware_testing/production_qc/robot_assembly_qc_ot3/test_instruments.py b/hardware-testing/hardware_testing/production_qc/robot_assembly_qc_ot3/test_instruments.py index a9a86cc6d9b..ef169914ba9 100644 --- a/hardware-testing/hardware_testing/production_qc/robot_assembly_qc_ot3/test_instruments.py +++ b/hardware-testing/hardware_testing/production_qc/robot_assembly_qc_ot3/test_instruments.py @@ -201,10 +201,10 @@ async def _test_gripper(api: OT3API, report: CSVReport, section: str) -> None: target_z = 100 await api.home([z_ax, Axis.G]) start_pos = await api.gantry_position(OT3Mount.GRIPPER) - await api.move_to(mount, start_pos._replace(z=target_z), _expect_stalls=True) + await api.move_to(mount, start_pos._replace(z=target_z), expect_stalls=True) enc_pos = await api.encoder_current_position_ot3(OT3Mount.GRIPPER) if abs(enc_pos[Axis.Z_G] - target_z) < 0.25: - await api.move_to(mount, start_pos, _expect_stalls=True) + await api.move_to(mount, start_pos, expect_stalls=True) if abs(enc_pos[Axis.Z_G] - target_z) < 0.25: result = CSVResult.PASS await api.home([z_ax]) diff --git a/hardware-testing/hardware_testing/production_qc/z_stage_qc_ot3.py b/hardware-testing/hardware_testing/production_qc/z_stage_qc_ot3.py index 8a55a831c45..3ec704686c1 100644 --- a/hardware-testing/hardware_testing/production_qc/z_stage_qc_ot3.py +++ b/hardware-testing/hardware_testing/production_qc/z_stage_qc_ot3.py @@ -259,7 +259,7 @@ async def _force_gauge( mount=mount, abs_position=press_pos, speed=FORCE_SPEED, - _expect_stalls=True, + expect_stalls=True, ) finally: thread_sensor = False diff --git a/shared-data/command/schemas/11.json b/shared-data/command/schemas/11.json index ce6a1575062..8b13c42d01b 100644 --- a/shared-data/command/schemas/11.json +++ b/shared-data/command/schemas/11.json @@ -1589,6 +1589,195 @@ "title": "EngageParams", "type": "object" }, + "EvotipDispenseCreate": { + "description": "DispenseInPlace command request model.", + "properties": { + "commandType": { + "const": "evotipDispense", + "default": "evotipDispense", + "enum": ["evotipDispense"], + "title": "Commandtype", + "type": "string" + }, + "intent": { + "$ref": "#/$defs/CommandIntent", + "description": "The reason the command was added. If not specified or `protocol`, the command will be treated as part of the protocol run itself, and added to the end of the existing command queue.\n\nIf `setup`, the command will be treated as part of run setup. A setup command may only be enqueued if the run has not started.\n\nUse setup commands for activities like pre-run calibration checks and module setup, like pre-heating.", + "title": "Intent" + }, + "key": { + "description": "A key value, unique in this run, that can be used to track the same logical command across multiple runs of the same protocol. If a value is not provided, one will be generated.", + "title": "Key", + "type": "string" + }, + "params": { + "$ref": "#/$defs/EvotipDispenseParams" + } + }, + "required": ["params"], + "title": "EvotipDispenseCreate", + "type": "object" + }, + "EvotipDispenseParams": { + "description": "Payload required to dispense in place.", + "properties": { + "flowRate": { + "description": "Speed in \u00b5L/s configured for the pipette", + "exclusiveMinimum": 0.0, + "title": "Flowrate", + "type": "number" + }, + "labwareId": { + "description": "Identifier of labware to use.", + "title": "Labwareid", + "type": "string" + }, + "pipetteId": { + "description": "Identifier of pipette to use for liquid handling.", + "title": "Pipetteid", + "type": "string" + }, + "volume": { + "description": "The amount of liquid to dispense, in \u00b5L. Must not be greater than the currently aspirated volume. There is some tolerance for floating point rounding errors.", + "minimum": 0.0, + "title": "Volume", + "type": "number" + }, + "wellLocation": { + "$ref": "#/$defs/LiquidHandlingWellLocation", + "description": "Relative well location at which to perform the operation" + }, + "wellName": { + "description": "Name of well to use in labware.", + "title": "Wellname", + "type": "string" + } + }, + "required": ["labwareId", "wellName", "flowRate", "volume", "pipetteId"], + "title": "EvotipDispenseParams", + "type": "object" + }, + "EvotipSealPipetteCreate": { + "description": "Seal evotip resin tip command creation request model.", + "properties": { + "commandType": { + "const": "evotipSealPipette", + "default": "evotipSealPipette", + "enum": ["evotipSealPipette"], + "title": "Commandtype", + "type": "string" + }, + "intent": { + "$ref": "#/$defs/CommandIntent", + "description": "The reason the command was added. If not specified or `protocol`, the command will be treated as part of the protocol run itself, and added to the end of the existing command queue.\n\nIf `setup`, the command will be treated as part of run setup. A setup command may only be enqueued if the run has not started.\n\nUse setup commands for activities like pre-run calibration checks and module setup, like pre-heating.", + "title": "Intent" + }, + "key": { + "description": "A key value, unique in this run, that can be used to track the same logical command across multiple runs of the same protocol. If a value is not provided, one will be generated.", + "title": "Key", + "type": "string" + }, + "params": { + "$ref": "#/$defs/EvotipSealPipetteParams" + } + }, + "required": ["params"], + "title": "EvotipSealPipetteCreate", + "type": "object" + }, + "EvotipSealPipetteParams": { + "description": "Payload needed to seal resin tips to a pipette.", + "properties": { + "labwareId": { + "description": "Identifier of labware to use.", + "title": "Labwareid", + "type": "string" + }, + "pipetteId": { + "description": "Identifier of pipette to use for liquid handling.", + "title": "Pipetteid", + "type": "string" + }, + "tipPickUpParams": { + "anyOf": [ + { + "$ref": "#/$defs/TipPickUpParams" + }, + { + "type": "null" + } + ], + "default": null, + "description": "Specific parameters for " + }, + "wellLocation": { + "$ref": "#/$defs/PickUpTipWellLocation", + "description": "Relative well location at which to pick up the tip." + }, + "wellName": { + "description": "Name of well to use in labware.", + "title": "Wellname", + "type": "string" + } + }, + "required": ["pipetteId", "labwareId", "wellName"], + "title": "EvotipSealPipetteParams", + "type": "object" + }, + "EvotipUnsealPipetteCreate": { + "description": "Evotip unseal command creation request model.", + "properties": { + "commandType": { + "const": "evotipUnsealPipette", + "default": "evotipUnsealPipette", + "enum": ["evotipUnsealPipette"], + "title": "Commandtype", + "type": "string" + }, + "intent": { + "$ref": "#/$defs/CommandIntent", + "description": "The reason the command was added. If not specified or `protocol`, the command will be treated as part of the protocol run itself, and added to the end of the existing command queue.\n\nIf `setup`, the command will be treated as part of run setup. A setup command may only be enqueued if the run has not started.\n\nUse setup commands for activities like pre-run calibration checks and module setup, like pre-heating.", + "title": "Intent" + }, + "key": { + "description": "A key value, unique in this run, that can be used to track the same logical command across multiple runs of the same protocol. If a value is not provided, one will be generated.", + "title": "Key", + "type": "string" + }, + "params": { + "$ref": "#/$defs/EvotipUnsealPipetteParams" + } + }, + "required": ["params"], + "title": "EvotipUnsealPipetteCreate", + "type": "object" + }, + "EvotipUnsealPipetteParams": { + "description": "Payload required to drop a tip in a specific well.", + "properties": { + "labwareId": { + "description": "Identifier of labware to use.", + "title": "Labwareid", + "type": "string" + }, + "pipetteId": { + "description": "Identifier of pipette to use for liquid handling.", + "title": "Pipetteid", + "type": "string" + }, + "wellLocation": { + "$ref": "#/$defs/DropTipWellLocation", + "description": "Relative well location at which to drop the tip." + }, + "wellName": { + "description": "Name of well to use in labware.", + "title": "Wellname", + "type": "string" + } + }, + "required": ["pipetteId", "labwareId", "wellName"], + "title": "EvotipUnsealPipetteParams", + "type": "object" + }, "GetNextTipCreate": { "description": "Get next tip command creation request model.", "properties": { @@ -4564,6 +4753,31 @@ "title": "Submerge", "type": "object" }, + "TipPickUpParams": { + "description": "Payload used to specify press-tip parameters for a seal command.", + "properties": { + "ejectorPushMm": { + "default": 0, + "description": "The distance to back off to ensure that the tip presence sensors are not triggered.", + "title": "Ejectorpushmm", + "type": "number" + }, + "prepDistance": { + "default": 0, + "description": "The distance to move down to fit the tips on.", + "title": "Prepdistance", + "type": "number" + }, + "pressDistance": { + "default": 0, + "description": "The distance to press on tips.", + "title": "Pressdistance", + "type": "number" + } + }, + "title": "TipPickUpParams", + "type": "object" + }, "TipPresenceStatus": { "description": "Tip presence status reported by a pipette.", "enum": ["present", "absent", "unknown"], @@ -5721,6 +5935,9 @@ "dispenseInPlace": "#/$defs/DispenseInPlaceCreate", "dropTip": "#/$defs/DropTipCreate", "dropTipInPlace": "#/$defs/DropTipInPlaceCreate", + "evotipDispense": "#/$defs/EvotipDispenseCreate", + "evotipSealPipette": "#/$defs/EvotipSealPipetteCreate", + "evotipUnsealPipette": "#/$defs/EvotipUnsealPipetteCreate", "getNextTip": "#/$defs/GetNextTipCreate", "getTipPresence": "#/$defs/GetTipPresenceCreate", "heaterShaker/closeLabwareLatch": "#/$defs/CloseLabwareLatchCreate", @@ -5914,6 +6131,15 @@ { "$ref": "#/$defs/TryLiquidProbeCreate" }, + { + "$ref": "#/$defs/EvotipSealPipetteCreate" + }, + { + "$ref": "#/$defs/EvotipDispenseCreate" + }, + { + "$ref": "#/$defs/EvotipUnsealPipetteCreate" + }, { "$ref": "#/$defs/opentrons__protocol_engine__commands__heater_shaker__wait_for_temperature__WaitForTemperatureCreate" }, diff --git a/shared-data/labware/definitions/3/evotips_opentrons_96_labware/1.json b/shared-data/labware/definitions/3/evotips_opentrons_96_labware/1.json index 43b2616c6f5..d2ecbdce217 100644 --- a/shared-data/labware/definitions/3/evotips_opentrons_96_labware/1.json +++ b/shared-data/labware/definitions/3/evotips_opentrons_96_labware/1.json @@ -1025,6 +1025,20 @@ }, "gripHeightFromLabwareBottom": 14.5, "gripForce": 15, + "gripperOffsets": { + "default": { + "pickUpOffset": { + "x": 0, + "y": 0, + "z": 15 + }, + "dropOffset": { + "x": 0, + "y": 0, + "z": 15 + } + } + }, "stackingOffsetWithLabware": { "nest_96_wellplate_2ml_deep": { "x": 0,