diff --git a/api/src/opentrons/legacy_commands/commands.py b/api/src/opentrons/legacy_commands/commands.py index 8db3720eb07..bd07ab18d53 100755 --- a/api/src/opentrons/legacy_commands/commands.py +++ b/api/src/opentrons/legacy_commands/commands.py @@ -328,12 +328,18 @@ def transfer_with_liquid_class( liquid_class: LiquidClass, volume: float, source: Union[Well, Sequence[Well], Sequence[Sequence[Well]]], - destination: Union[Well, Sequence[Well], Sequence[Sequence[Well]]], + destination: Union[ + Well, Sequence[Well], Sequence[Sequence[Well]], TrashBin, WasteChute + ], ) -> command_types.TransferWithLiquidClassCommand: + if isinstance(destination, (TrashBin, WasteChute)): + destination_text = stringify_disposal_location(destination) + else: + destination_text = stringify_well_list(destination) text = ( "Transferring " + f"{volume} uL of {liquid_class.display_name} liquid class from " - + f"{stringify_well_list(source)} to {stringify_well_list(destination)}" + + f"{stringify_well_list(source)} to {destination_text}" ) return { "name": command_types.TRANSFER_WITH_LIQUID_CLASS, @@ -378,12 +384,18 @@ def consolidate_with_liquid_class( liquid_class: LiquidClass, volume: float, source: Union[Well, Sequence[Well], Sequence[Sequence[Well]]], - destination: Union[Well, Sequence[Well], Sequence[Sequence[Well]]], + destination: Union[ + Well, Sequence[Well], Sequence[Sequence[Well]], TrashBin, WasteChute + ], ) -> command_types.ConsolidateWithLiquidClassCommand: + if isinstance(destination, (TrashBin, WasteChute)): + destination_text = stringify_disposal_location(destination) + else: + destination_text = stringify_well_list(destination) text = ( "Consolidating " + f"{volume} uL of {liquid_class.display_name} liquid class from " - + f"{stringify_well_list(source)} to {stringify_well_list(destination)}" + + f"{stringify_well_list(source)} to {destination_text}" ) return { "name": command_types.CONSOLIDATE_WITH_LIQUID_CLASS, diff --git a/api/src/opentrons/legacy_commands/types.py b/api/src/opentrons/legacy_commands/types.py index 83e615f0ca5..f81f587a725 100755 --- a/api/src/opentrons/legacy_commands/types.py +++ b/api/src/opentrons/legacy_commands/types.py @@ -555,7 +555,9 @@ class LiquidClassCommandPayload(TextOnlyPayload, SingleInstrumentPayload): liquid_class: LiquidClass volume: float source: Union[Well, Sequence[Well], Sequence[Sequence[Well]]] - destination: Union[Well, Sequence[Well], Sequence[Sequence[Well]]] + destination: Union[ + Well, Sequence[Well], Sequence[Sequence[Well]], TrashBin, WasteChute + ] class TransferWithLiquidClassCommand(TypedDict): diff --git a/api/src/opentrons/protocol_api/_transfer_liquid_validation.py b/api/src/opentrons/protocol_api/_transfer_liquid_validation.py index d87c21a8272..4aea3b6f696 100644 --- a/api/src/opentrons/protocol_api/_transfer_liquid_validation.py +++ b/api/src/opentrons/protocol_api/_transfer_liquid_validation.py @@ -19,8 +19,8 @@ @dataclass class TransferInfo: - sources_list: List[Well] - destinations_list: List[Well] + source: List[Well] + dest: Union[List[Well], TrashBin, WasteChute] tip_policy: TransferTipPolicyV2 tip_racks: List[Labware] trash_location: Union[Location, TrashBin, WasteChute] @@ -28,7 +28,7 @@ class TransferInfo: def verify_and_normalize_transfer_args( source: Union[Well, Sequence[Well], Sequence[Sequence[Well]]], - dest: Union[Well, Sequence[Well], Sequence[Sequence[Well]]], + dest: Union[Well, Sequence[Well], Sequence[Sequence[Well]], TrashBin, WasteChute], tip_policy: TransferTipPolicyV2Type, last_tip_picked_up_from: Optional[Well], tip_racks: List[Labware], @@ -38,7 +38,11 @@ def verify_and_normalize_transfer_args( trash_location: Union[Location, Well, Labware, TrashBin, WasteChute], ) -> TransferInfo: flat_sources_list = validation.ensure_valid_flat_wells_list_for_transfer_v2(source) - flat_dests_list = validation.ensure_valid_flat_wells_list_for_transfer_v2(dest) + if not isinstance(dest, (TrashBin, WasteChute)): + flat_dests_list = validation.ensure_valid_flat_wells_list_for_transfer_v2(dest) + else: + # If trash bin or waste chute, set this to empty to have less isinstance checks after this + flat_dests_list = [] if not target_all_wells and nozzle_map.tip_count > 1: flat_sources_list = tx_liquid_utils.group_wells_for_multi_channel_transfer( flat_sources_list, nozzle_map @@ -83,8 +87,8 @@ def verify_and_normalize_transfer_args( ) return TransferInfo( - sources_list=flat_sources_list, - destinations_list=flat_dests_list, + source=flat_sources_list, + dest=flat_dests_list if not isinstance(dest, (TrashBin, WasteChute)) else dest, tip_policy=valid_new_tip, tip_racks=valid_tip_racks, trash_location=valid_trash_location, diff --git a/api/src/opentrons/protocol_api/core/engine/instrument.py b/api/src/opentrons/protocol_api/core/engine/instrument.py index d739154a3aa..77200f4a68e 100644 --- a/api/src/opentrons/protocol_api/core/engine/instrument.py +++ b/api/src/opentrons/protocol_api/core/engine/instrument.py @@ -10,6 +10,7 @@ cast, Union, List, + Sequence, Tuple, NamedTuple, Generator, @@ -1212,7 +1213,7 @@ def transfer_with_liquid_class( # noqa: C901 liquid_class: LiquidClass, volume: float, source: List[Tuple[Location, WellCore]], - dest: List[Tuple[Location, WellCore]], + dest: Union[List[Tuple[Location, WellCore]], TrashBin, WasteChute], new_tip: TransferTipPolicyV2, tip_racks: List[Tuple[Location, LabwareCore]], starting_tip: Optional[WellCore], @@ -1258,10 +1259,17 @@ def transfer_with_liquid_class( # noqa: C901 tiprack_uri=tiprack_uri_for_transfer_props, ) + target_destinations: Sequence[ + Union[Tuple[Location, WellCore], TrashBin, WasteChute] + ] + if isinstance(dest, (TrashBin, WasteChute)): + target_destinations = [dest] * len(source) + else: + target_destinations = dest source_dest_per_volume_step = ( tx_commons.expand_for_volume_constraints_for_liquid_classes( volumes=[volume for _ in range(len(source))], - targets=zip(source, dest), + targets=zip(source, target_destinations), max_volume=min( self.get_max_volume(), self._engine_client.state.geometry.get_nominal_tip_geometry( @@ -1728,7 +1736,7 @@ def consolidate_with_liquid_class( # noqa: C901 liquid_class: LiquidClass, volume: float, source: List[Tuple[Location, WellCore]], - dest: Tuple[Location, WellCore], + dest: Union[Tuple[Location, WellCore], TrashBin, WasteChute], new_tip: Literal[TransferTipPolicyV2.NEVER, TransferTipPolicyV2.ONCE], tip_racks: List[Tuple[Location, LabwareCore]], starting_tip: Optional[WellCore], @@ -2059,7 +2067,7 @@ def remove_air_gap_during_transfer_with_liquid_class( self, last_air_gap: float, dispense_props: SingleDispenseProperties, - location: Location, + location: Union[Location, TrashBin, WasteChute], ) -> None: """Remove an air gap that was previously added during a transfer.""" if last_air_gap == 0: @@ -2090,7 +2098,7 @@ def remove_air_gap_during_transfer_with_liquid_class( def dispense_liquid_class( self, volume: float, - dest: Tuple[Location, WellCore], + dest: Union[Tuple[Location, WellCore], TrashBin, WasteChute], source: Optional[Tuple[Location, WellCore]], transfer_properties: TransferProperties, transfer_type: tx_comps_executor.TransferType, @@ -2129,17 +2137,21 @@ def dispense_liquid_class( List of liquid and air gap pairs in tip. """ dispense_props = transfer_properties.dispense - dest_loc, dest_well = dest - dispense_point = ( - tx_comps_executor.absolute_point_from_position_reference_and_offset( + dispense_location: Union[Location, TrashBin, WasteChute] + if isinstance(dest, tuple): + dest_loc, dest_well = dest + dispense_point = tx_comps_executor.absolute_point_from_position_reference_and_offset( well=dest_well, well_volume_difference=volume, position_reference=dispense_props.dispense_position.position_reference, offset=dispense_props.dispense_position.offset, mount=self.get_mount(), ) - ) - dispense_location = Location(dispense_point, labware=dest_loc.labware) + dispense_location = Location(dispense_point, labware=dest_loc.labware) + else: + dispense_location = dest + dest_well = None + last_liquid_and_airgap_in_tip = ( tip_contents[-1] if tip_contents diff --git a/api/src/opentrons/protocol_api/core/engine/transfer_components_executor.py b/api/src/opentrons/protocol_api/core/engine/transfer_components_executor.py index 9af8b6a268a..cc7ca68df66 100644 --- a/api/src/opentrons/protocol_api/core/engine/transfer_components_executor.py +++ b/api/src/opentrons/protocol_api/core/engine/transfer_components_executor.py @@ -118,8 +118,8 @@ def __init__( self, instrument_core: InstrumentCore, transfer_properties: TransferProperties, - target_location: Location, - target_well: WellCore, + target_location: Union[Location, TrashBin, WasteChute], + target_well: Optional[WellCore], tip_state: TipState, transfer_type: TransferType, ) -> None: @@ -135,7 +135,8 @@ def __init__( will be used to initialize info about the dispense by assigning values to class attributes as follows- - target_location: the dispense location - - target_well: the well associated with dispense location + - target_well: the well associated with dispense location, will be None when the + target_location argument is a TrashBin or WasteChute - tip_state: the state of the tip before dispense component steps are executed - transfer_type: whether the dispense component is being called as a part of a 1-to-1 transfer or a consolidation or a distribution @@ -166,29 +167,38 @@ def submerge( Should raise an error if this point is inside the liquid? For liquid meniscus this is easy to tell. Can’t be below meniscus For reference pos of anything else, do not allow submerge position to be below aspirate position - 2. move to aspirate position at desired speed + 2. move to aspirate/dispense position at desired speed 3. delay + + If target location is a trash bin or waste chute, the pipette will move to the disposal location given, + remove air gap and delay """ - submerge_start_point = absolute_point_from_position_reference_and_offset( - well=self._target_well, - well_volume_difference=0, - position_reference=submerge_properties.start_position.position_reference, - offset=submerge_properties.start_position.offset, - mount=self._instrument.get_mount(), - ) - submerge_start_location = Location( - point=submerge_start_point, labware=self._target_location.labware - ) - tx_utils.raise_if_location_inside_liquid( - location=submerge_start_location, - well_location=self._target_location, - well_core=self._target_well, - location_check_descriptors=LocationCheckDescriptors( - location_type="submerge start", - pipetting_action=post_submerge_action, - ), - logger=log, - ) + submerge_start_location: Union[Location, TrashBin, WasteChute] + if isinstance(self._target_location, Location): + assert self._target_well is not None + submerge_start_point = absolute_point_from_position_reference_and_offset( + well=self._target_well, + well_volume_difference=0, + position_reference=submerge_properties.start_position.position_reference, + offset=submerge_properties.start_position.offset, + mount=self._instrument.get_mount(), + ) + submerge_start_location = Location( + point=submerge_start_point, labware=self._target_location.labware + ) + tx_utils.raise_if_location_inside_liquid( + location=submerge_start_location, + well_location=self._target_location, + well_core=self._target_well, + location_check_descriptors=LocationCheckDescriptors( + location_type="submerge start", + pipetting_action=post_submerge_action, + ), + logger=log, + ) + else: + submerge_start_location = self._target_location + self._instrument.move_to( location=submerge_start_location, well_core=self._target_well, @@ -197,19 +207,25 @@ def submerge( speed=None, ) self._remove_air_gap(location=submerge_start_location) - self._instrument.move_to( - location=self._target_location, - well_core=self._target_well, - force_direct=True, - minimum_z_height=None, - speed=submerge_properties.speed, - ) + if isinstance(self._target_location, Location): + self._instrument.move_to( + location=self._target_location, + well_core=self._target_well, + force_direct=True, + minimum_z_height=None, + speed=submerge_properties.speed, + ) + if submerge_properties.delay.enabled and submerge_properties.delay.duration: self._instrument.delay(submerge_properties.delay.duration) def aspirate_and_wait(self, volume: float) -> None: """Aspirate according to aspirate properties and wait if enabled.""" # TODO: handle volume correction + assert ( + isinstance(self._target_location, Location) + and self._target_well is not None + ) aspirate_props = self._transfer_properties.aspirate correction_volume = aspirate_props.correction_by_volume.get_for_volume(volume) self._instrument.aspirate( @@ -265,11 +281,15 @@ def mix(self, mix_properties: MixProperties, last_dispense_push_out: bool) -> No NOTE: For most of our built-in definitions, we will keep _mix_ off because it is a very application specific thing. We should mention in our docs that users should adjust this property according to their application. """ - if not mix_properties.enabled: + if not mix_properties.enabled or not isinstance( + self._target_location, Location + ): return # Assertion only for mypy purposes assert ( - mix_properties.repetitions is not None and mix_properties.volume is not None + mix_properties.repetitions is not None + and mix_properties.volume is not None + and self._target_well is not None ) push_out_vol = ( self._transfer_properties.dispense.push_out_by_volume.get_for_volume( @@ -327,6 +347,10 @@ def retract_after_aspiration( during a multi-dispense. """ # TODO: Raise error if retract is below the meniscus + assert ( + isinstance(self._target_location, Location) + and self._target_well is not None + ) retract_props = self._transfer_properties.aspirate.retract retract_point = absolute_point_from_position_reference_and_offset( well=self._target_well, @@ -423,37 +447,47 @@ def retract_after_dispensing( - Prepare-to-aspirate (top of well) - Do air-gap (top of well) 7. If drop tip, move to drop tip location, drop tip + + If target location is a trash bin or waste chute, the retract movement step is skipped along with touch tip, + even if it is enabled. """ # TODO: Raise error if retract is below the meniscus - retract_props = self._transfer_properties.dispense.retract - retract_point = absolute_point_from_position_reference_and_offset( - well=self._target_well, - well_volume_difference=0, - position_reference=retract_props.end_position.position_reference, - offset=retract_props.end_position.offset, - mount=self._instrument.get_mount(), - ) - retract_location = Location( - retract_point, labware=self._target_location.labware - ) - tx_utils.raise_if_location_inside_liquid( - location=retract_location, - well_location=self._target_location, - well_core=self._target_well, - location_check_descriptors=LocationCheckDescriptors( - location_type="retract end", - pipetting_action="dispense", - ), - logger=log, - ) - self._instrument.move_to( - location=retract_location, - well_core=self._target_well, - force_direct=True, - minimum_z_height=None, - speed=retract_props.speed, - ) + + retract_location: Union[Location, TrashBin, WasteChute] + if isinstance(self._target_location, Location): + assert self._target_well is not None + retract_point = absolute_point_from_position_reference_and_offset( + well=self._target_well, + well_volume_difference=0, + position_reference=retract_props.end_position.position_reference, + offset=retract_props.end_position.offset, + mount=self._instrument.get_mount(), + ) + retract_location = Location( + retract_point, labware=self._target_location.labware + ) + tx_utils.raise_if_location_inside_liquid( + location=retract_location, + well_location=self._target_location, + well_core=self._target_well, + location_check_descriptors=LocationCheckDescriptors( + location_type="retract end", + pipetting_action="dispense", + ), + logger=log, + ) + self._instrument.move_to( + location=retract_location, + well_core=self._target_well, + force_direct=True, + minimum_z_height=None, + speed=retract_props.speed, + ) + else: + retract_location = self._target_location + + # TODO should we delay here for a trash despite not having a "retract"? retract_delay = retract_props.delay if retract_delay.enabled and retract_delay.duration: self._instrument.delay(retract_delay.duration) @@ -481,7 +515,9 @@ def retract_after_dispensing( # then skip the final air gap if we have been told to do so. self._do_touch_tip_and_air_gap( touch_tip_properties=retract_props.touch_tip, - location=retract_location, + location=retract_location + if isinstance(retract_location, Location) + else None, well=self._target_well, add_air_gap=False if is_final_air_gap and not add_final_air_gap else True, ) @@ -567,7 +603,10 @@ def retract_during_multi_dispensing( and whether we are moving to another dispense or going back to the source. """ # TODO: Raise error if retract is below the meniscus - + assert ( + isinstance(self._target_location, Location) + and self._target_well is not None + ) assert self._transfer_properties.multi_dispense is not None retract_props = self._transfer_properties.multi_dispense.retract @@ -799,7 +838,7 @@ def _add_air_gap(self, air_gap_volume: float) -> None: self._instrument.delay(delay_props.duration) self._tip_state.append_air_gap(air_gap_volume) - def _remove_air_gap(self, location: Location) -> None: + def _remove_air_gap(self, location: Union[Location, TrashBin, WasteChute]) -> None: """Remove a previously added air gap.""" last_air_gap = self._tip_state.last_liquid_and_air_gap_in_tip.air_gap dispense_props = self._transfer_properties.dispense diff --git a/api/src/opentrons/protocol_api/core/instrument.py b/api/src/opentrons/protocol_api/core/instrument.py index 4790d6ee416..e7f11ae9388 100644 --- a/api/src/opentrons/protocol_api/core/instrument.py +++ b/api/src/opentrons/protocol_api/core/instrument.py @@ -365,7 +365,7 @@ def transfer_with_liquid_class( liquid_class: LiquidClass, volume: float, source: List[Tuple[types.Location, WellCoreType]], - dest: List[Tuple[types.Location, WellCoreType]], + dest: Union[List[Tuple[types.Location, WellCoreType]], TrashBin, WasteChute], new_tip: TransferTipPolicyV2, tip_racks: List[Tuple[types.Location, LabwareCoreType]], starting_tip: Optional[WellCoreType], @@ -400,7 +400,7 @@ def consolidate_with_liquid_class( liquid_class: LiquidClass, volume: float, source: List[Tuple[types.Location, WellCoreType]], - dest: Tuple[types.Location, WellCoreType], + dest: Union[Tuple[types.Location, WellCoreType], TrashBin, WasteChute], new_tip: Literal[TransferTipPolicyV2.NEVER, TransferTipPolicyV2.ONCE], tip_racks: List[Tuple[types.Location, LabwareCoreType]], starting_tip: Optional[WellCoreType], 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 5d67e9c46d0..b91ce585048 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 @@ -605,7 +605,7 @@ def transfer_with_liquid_class( liquid_class: LiquidClass, volume: float, source: List[Tuple[types.Location, LegacyWellCore]], - dest: List[Tuple[types.Location, LegacyWellCore]], + dest: Union[List[Tuple[types.Location, LegacyWellCore]], TrashBin, WasteChute], new_tip: TransferTipPolicyV2, tip_racks: List[Tuple[types.Location, LegacyLabwareCore]], starting_tip: Optional[LegacyWellCore], @@ -635,7 +635,7 @@ def consolidate_with_liquid_class( liquid_class: LiquidClass, volume: float, source: List[Tuple[types.Location, LegacyWellCore]], - dest: Tuple[types.Location, LegacyWellCore], + dest: Union[Tuple[types.Location, LegacyWellCore], TrashBin, WasteChute], new_tip: Literal[TransferTipPolicyV2.NEVER, TransferTipPolicyV2.ONCE], tip_racks: List[Tuple[types.Location, LegacyLabwareCore]], starting_tip: Optional[LegacyWellCore], 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 d029a23a08c..baaf429cc02 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 @@ -519,7 +519,7 @@ def transfer_with_liquid_class( liquid_class: LiquidClass, volume: float, source: List[Tuple[types.Location, LegacyWellCore]], - dest: List[Tuple[types.Location, LegacyWellCore]], + dest: Union[List[Tuple[types.Location, LegacyWellCore]], TrashBin, WasteChute], new_tip: TransferTipPolicyV2, tip_racks: List[Tuple[types.Location, LegacyLabwareCore]], starting_tip: Optional[LegacyWellCore], @@ -549,7 +549,7 @@ def consolidate_with_liquid_class( liquid_class: LiquidClass, volume: float, source: List[Tuple[types.Location, LegacyWellCore]], - dest: Tuple[types.Location, LegacyWellCore], + dest: Union[Tuple[types.Location, LegacyWellCore], TrashBin, WasteChute], new_tip: Literal[TransferTipPolicyV2.NEVER, TransferTipPolicyV2.ONCE], tip_racks: List[Tuple[types.Location, LegacyLabwareCore]], starting_tip: Optional[LegacyWellCore], diff --git a/api/src/opentrons/protocol_api/instrument_context.py b/api/src/opentrons/protocol_api/instrument_context.py index 7b119dad436..e827bc8c158 100644 --- a/api/src/opentrons/protocol_api/instrument_context.py +++ b/api/src/opentrons/protocol_api/instrument_context.py @@ -1,7 +1,7 @@ from __future__ import annotations import logging from contextlib import ExitStack -from typing import Any, List, Optional, Sequence, Union, cast +from typing import Any, List, Optional, Sequence, Union, cast, Tuple from opentrons_shared_data.errors.exceptions import ( CommandPreconditionViolated, CommandParameterLimitViolated, @@ -32,7 +32,7 @@ UnsupportedAPIError, ) -from .core.common import InstrumentCore, ProtocolCore +from .core.common import InstrumentCore, ProtocolCore, WellCore from .core.engine import ENGINE_CORE_API_VERSION from .core.legacy.legacy_instrument_core import LegacyInstrumentCore from .config import Clearances @@ -1619,7 +1619,11 @@ def transfer_with_liquid_class( labware.Well, Sequence[labware.Well], Sequence[Sequence[labware.Well]] ], dest: Union[ - labware.Well, Sequence[labware.Well], Sequence[Sequence[labware.Well]] + labware.Well, + Sequence[labware.Well], + Sequence[Sequence[labware.Well]], + TrashBin, + WasteChute, ], new_tip: TransferTipPolicyV2Type = "once", trash_location: Optional[ @@ -1641,7 +1645,7 @@ def transfer_with_liquid_class( :param volume: The amount, in µL, to aspirate from each source and dispense to each destination. :param source: A single well or a list of wells to aspirate liquid from. - :param dest: A single well or a list of wells to dispense liquid into. + :param dest: A single well, list of wells, trash bin, or waste chute to dispense liquid into. :param new_tip: When to pick up and drop tips during the command. Defaults to ``"once"``. @@ -1680,12 +1684,23 @@ def transfer_with_liquid_class( trash_location if trash_location is not None else self.trash_container ), ) - if len(transfer_args.sources_list) != len(transfer_args.destinations_list): - raise ValueError( - "Sources and destinations should be of the same length in order to perform a transfer." - " To transfer liquid from one source to many destinations, use 'distribute_liquid'," - " to transfer liquid onto one destinations from many sources, use 'consolidate_liquid'." - ) + + verified_dest: Union[ + List[Tuple[types.Location, WellCore]], TrashBin, WasteChute + ] + if isinstance(transfer_args.dest, (TrashBin, WasteChute)): + verified_dest = transfer_args.dest + else: + if len(transfer_args.source) != len(transfer_args.dest): + raise ValueError( + "Sources and destinations should be of the same length in order to perform a transfer." + " To transfer liquid from one source to many destinations, use 'distribute_liquid'," + " to transfer liquid to one destination from many sources, use 'consolidate_liquid'." + ) + verified_dest = [ + (types.Location(types.Point(), labware=well), well._core) + for well in transfer_args.dest + ] with publisher.publish_context( broker=self.broker, @@ -1702,12 +1717,9 @@ def transfer_with_liquid_class( volume=volume, source=[ (types.Location(types.Point(), labware=well), well._core) - for well in transfer_args.sources_list - ], - dest=[ - (types.Location(types.Point(), labware=well), well._core) - for well in transfer_args.destinations_list + for well in transfer_args.source ], + dest=verified_dest, new_tip=transfer_args.tip_policy, tip_racks=[ (types.Location(types.Point(), labware=rack), rack._core) @@ -1750,7 +1762,8 @@ def distribute_with_liquid_class( :param volume: The amount, in µL, to aspirate from the source and dispense to each destination. - :param source: A single well to aspirate liquid from. + :param source: A single well for the pipette to target, or a group of wells to + target in a single aspirate for a multi-channel pipette. :param dest: A list of wells to dispense liquid into. :param new_tip: When to pick up and drop tips during the command. Defaults to ``"once"``. @@ -1787,10 +1800,15 @@ def distribute_with_liquid_class( trash_location if trash_location is not None else self.trash_container ), ) - if len(transfer_args.sources_list) != 1: + if isinstance(transfer_args.dest, (TrashBin, WasteChute)): + raise ValueError( + "distribute_with_liquid_class() does not support trash bin or waste chute" + " as a destination." + ) + if len(transfer_args.source) != 1: raise ValueError( f"Source should be a single well (or resolve to a single transfer for multi-channel) " - f"but received {transfer_args.sources_list}." + f"but received {transfer_args.source}." ) if transfer_args.tip_policy not in [ TransferTipPolicyV2.ONCE, @@ -1802,7 +1820,7 @@ def distribute_with_liquid_class( f" 'once' and 'never'." ) - verified_source = transfer_args.sources_list[0] + verified_source = transfer_args.source[0] with publisher.publish_context( broker=self.broker, command=cmds.distribute_with_liquid_class( @@ -1822,7 +1840,7 @@ def distribute_with_liquid_class( ), dest=[ (types.Location(types.Point(), labware=well), well._core) - for well in transfer_args.destinations_list + for well in transfer_args.dest ], new_tip=transfer_args.tip_policy, # type: ignore[arg-type] tip_racks=[ @@ -1845,7 +1863,7 @@ def consolidate_with_liquid_class( source: Union[ labware.Well, Sequence[labware.Well], Sequence[Sequence[labware.Well]] ], - dest: Union[labware.Well, Sequence[labware.Well]], + dest: Union[labware.Well, Sequence[labware.Well], TrashBin, WasteChute], new_tip: TransferTipPolicyV2Type = "once", trash_location: Optional[ Union[types.Location, labware.Well, TrashBin, WasteChute] @@ -1867,7 +1885,9 @@ def consolidate_with_liquid_class( :param volume: The amount, in µL, to aspirate from the source and dispense to each destination. :param source: A list of wells to aspirate liquid from. - :param dest: A single well to dispense liquid into. + :param dest: A single well, list of wells, trash bin, or waste chute to dispense liquid into. + Multiple wells can only be given for multi-channel pipette configurations, and + must be able to be dispensed to in a single dispense. :param new_tip: When to pick up and drop tips during the command. Defaults to ``"once"``. @@ -1903,10 +1923,18 @@ def consolidate_with_liquid_class( trash_location if trash_location is not None else self.trash_container ), ) - if len(transfer_args.destinations_list) != 1: - raise ValueError( - f"Destination should be a single well (or resolve to a single transfer for multi-channel) " - f"but received {transfer_args.destinations_list}." + verified_dest: Union[Tuple[types.Location, WellCore], TrashBin, WasteChute] + if isinstance(transfer_args.dest, (TrashBin, WasteChute)): + verified_dest = transfer_args.dest + else: + if len(transfer_args.dest) != 1: + raise ValueError( + f"Destination should be a single well (or resolve to a single transfer for multi-channel) " + f"but received {transfer_args.dest}." + ) + verified_dest = ( + types.Location(types.Point(), labware=transfer_args.dest[0]), + transfer_args.dest[0]._core, ) if transfer_args.tip_policy not in [ TransferTipPolicyV2.ONCE, @@ -1918,7 +1946,6 @@ def consolidate_with_liquid_class( f" 'once' and 'never'." ) - verified_dest = transfer_args.destinations_list[0] with publisher.publish_context( broker=self.broker, command=cmds.consolidate_with_liquid_class( @@ -1934,12 +1961,9 @@ def consolidate_with_liquid_class( volume=volume, source=[ (types.Location(types.Point(), labware=well), well._core) - for well in transfer_args.sources_list + for well in transfer_args.source ], - dest=( - types.Location(types.Point(), labware=verified_dest), - verified_dest._core, - ), + dest=verified_dest, new_tip=transfer_args.tip_policy, # type: ignore[arg-type] tip_racks=[ (types.Location(types.Point(), labware=rack), rack._core) diff --git a/api/tests/opentrons/conftest.py b/api/tests/opentrons/conftest.py index 4ad7d0dba71..d3d7b8f9e06 100755 --- a/api/tests/opentrons/conftest.py +++ b/api/tests/opentrons/conftest.py @@ -953,7 +953,7 @@ def maximal_liquid_class_def() -> LiquidClassSchemaV1: ), speed=100, delay=DelayProperties( - enable=True, params=DelayParams(duration=0.0) + enable=True, params=DelayParams(duration=1.1) ), ), retract=RetractDispense( diff --git a/api/tests/opentrons/protocol_api/core/engine/test_transfer_components_executor.py b/api/tests/opentrons/protocol_api/core/engine/test_transfer_components_executor.py index edb1ad0c7cc..0db2a40a54f 100644 --- a/api/tests/opentrons/protocol_api/core/engine/test_transfer_components_executor.py +++ b/api/tests/opentrons/protocol_api/core/engine/test_transfer_components_executor.py @@ -8,7 +8,7 @@ BlowoutLocation, ) -from opentrons.protocol_api import TrashBin +from opentrons.protocol_api import TrashBin, WasteChute from opentrons.protocol_api._liquid import LiquidClass from opentrons.protocol_api._liquid_properties import TransferProperties from opentrons.protocol_api.core.engine.well import WellCore @@ -79,9 +79,8 @@ def patch_mock_raise_if_location_inside_liquid( argnames=[ "air_gap_volume", "air_gap_flow_rate_by_vol", - "expected_air_gap_flow_rate", ], - argvalues=[(0.123, 123, 123), (1.23, 0.123, 1.23)], + argvalues=[(0.123, 123), (1.23, 0.123)], ) def test_submerge( decoy: Decoy, @@ -89,7 +88,6 @@ def test_submerge( sample_transfer_props: TransferProperties, air_gap_volume: float, air_gap_flow_rate_by_vol: float, - expected_air_gap_flow_rate: float, ) -> None: """Should perform the expected submerge steps.""" source_well = decoy.mock(cls=WellCore) @@ -223,6 +221,65 @@ def test_submerge_without_starting_air_gap( ) +@pytest.mark.parametrize( + argnames=[ + "air_gap_volume", + "air_gap_flow_rate_by_vol", + ], + argvalues=[(0.123, 123), (1.23, 0.123)], +) +def test_submerge_with_trash_location( + decoy: Decoy, + mock_instrument_core: InstrumentCore, + sample_transfer_props: TransferProperties, + air_gap_volume: float, + air_gap_flow_rate_by_vol: float, +) -> None: + """Should perform the expected submerge steps.""" + air_gap_correction_by_vol = 0.321 + sample_transfer_props.dispense.flow_rate_by_volume.set_for_volume( + air_gap_volume, air_gap_flow_rate_by_vol + ) + sample_transfer_props.dispense.correction_by_volume.set_for_volume( + air_gap_volume, air_gap_correction_by_vol + ) + mock_trash_bin = decoy.mock(cls=TrashBin) + + subject = TransferComponentsExecutor( + instrument_core=mock_instrument_core, + transfer_properties=sample_transfer_props, + target_location=mock_trash_bin, + target_well=None, + tip_state=TipState( + ready_to_aspirate=True, + last_liquid_and_air_gap_in_tip=LiquidAndAirGapPair( + liquid=0, air_gap=air_gap_volume + ), + ), + transfer_type=TransferType.ONE_TO_ONE, + ) + subject.submerge( + submerge_properties=sample_transfer_props.dispense.submerge, + post_submerge_action="dispense", + ) + + decoy.verify( + mock_instrument_core.move_to( + location=mock_trash_bin, + well_core=None, + force_direct=False, + minimum_z_height=None, + speed=None, + ), + mock_instrument_core.remove_air_gap_during_transfer_with_liquid_class( + last_air_gap=air_gap_volume, + dispense_props=sample_transfer_props.dispense, + location=mock_trash_bin, + ), + mock_instrument_core.delay(1.1), + ) + + def test_submerge_raises_when_submerge_point_is_invalid( decoy: Decoy, mock_instrument_core: InstrumentCore, @@ -416,6 +473,47 @@ def test_dispense_and_wait_skips_delay( ) +def test_dispense_into_trash_and_wait( + decoy: Decoy, + mock_instrument_core: InstrumentCore, + sample_transfer_props: TransferProperties, +) -> None: + """It should execute a dispense and a delay according to properties.""" + mock_trash_bin = decoy.mock(cls=TrashBin) + dispense_flow_rate = ( + sample_transfer_props.dispense.flow_rate_by_volume.get_for_volume(10) + ) + correction_volume = ( + sample_transfer_props.dispense.correction_by_volume.get_for_volume(50) + ) + subject = TransferComponentsExecutor( + instrument_core=mock_instrument_core, + transfer_properties=sample_transfer_props, + target_location=mock_trash_bin, + target_well=None, + tip_state=TipState(), + transfer_type=TransferType.ONE_TO_ONE, + ) + subject.dispense_and_wait( + dispense_properties=sample_transfer_props.dispense, + volume=10, + push_out_override=123, + ) + decoy.verify( + mock_instrument_core.dispense( + location=mock_trash_bin, + well_core=None, + volume=10, + rate=1, + flow_rate=dispense_flow_rate, + in_place=True, + push_out=123, + correction_volume=correction_volume, + ), + mock_instrument_core.delay(0.5), + ) + + def test_mix( decoy: Decoy, mock_instrument_core: InstrumentCore, @@ -1319,6 +1417,247 @@ def test_retract_after_dispense_with_blowout_in_disposal_location( ) +@pytest.mark.parametrize( + "add_final_air_gap", + [True, False], +) +def test_retract_after_dispense_in_trash_with_blowout_in_source( + decoy: Decoy, + mock_instrument_core: InstrumentCore, + sample_transfer_props: TransferProperties, + add_final_air_gap: bool, +) -> None: + """It should execute steps to retract after a dispense into a trash.""" + source_location = Location(Point(1, 2, 3), labware=None) + source_well = decoy.mock(cls=WellCore) + target_chute = decoy.mock(cls=WasteChute) + + air_gap_volume = 0.123 + air_gap_flow_rate_by_vol = 123 + air_gap_correction_by_vol = 0.321 + + sample_transfer_props.aspirate.retract.air_gap_by_volume.set_for_volume( + 0, air_gap_volume + ) + sample_transfer_props.aspirate.flow_rate_by_volume.set_for_volume( + air_gap_volume, air_gap_flow_rate_by_vol + ) + sample_transfer_props.aspirate.correction_by_volume.set_for_volume( + air_gap_volume, air_gap_correction_by_vol + ) + + subject = TransferComponentsExecutor( + instrument_core=mock_instrument_core, + transfer_properties=sample_transfer_props, + target_location=target_chute, + target_well=None, + tip_state=TipState(), + transfer_type=TransferType.ONE_TO_ONE, + ) + + decoy.when(source_well.get_top(0)).then_return(Point(10, 20, 30)) + subject.retract_after_dispensing( + trash_location=Location(Point(), labware=None), + source_location=source_location, + source_well=source_well, + add_final_air_gap=add_final_air_gap, + ) + decoy.verify( + mock_instrument_core.delay(10), + mock_instrument_core.air_gap_in_place( + volume=air_gap_volume, + flow_rate=air_gap_flow_rate_by_vol, + correction_volume=air_gap_correction_by_vol, + ), + mock_instrument_core.delay(0.2), + mock_instrument_core.set_flow_rate(blow_out=100), + mock_instrument_core.blow_out( + location=Location(Point(10, 20, 30), labware=None), + well_core=source_well, + in_place=False, + ), + mock_instrument_core.touch_tip( + location=Location(Point(10, 20, 30), labware=None), + well_core=source_well, + radius=1, + mm_from_edge=0.75, + z_offset=-1, + speed=30, + ), + mock_instrument_core.move_to( + location=Location(Point(10, 20, 30), labware=None), + well_core=source_well, + force_direct=True, + minimum_z_height=None, + speed=None, + ), + mock_instrument_core.prepare_to_aspirate(), + *( + add_final_air_gap + and [ + mock_instrument_core.air_gap_in_place( # type: ignore[func-returns-value] + volume=air_gap_volume, + flow_rate=air_gap_flow_rate_by_vol, + correction_volume=air_gap_correction_by_vol, + ), + mock_instrument_core.delay(0.2), # type: ignore[func-returns-value] + ] + or [] + ), + ) + + +@pytest.mark.parametrize( + "add_final_air_gap", + [True, False], +) +def test_retract_after_dispense_in_trash_with_blowout_in_destination( + decoy: Decoy, + mock_instrument_core: InstrumentCore, + sample_transfer_props: TransferProperties, + add_final_air_gap: bool, +) -> None: + """It should execute steps to retract after a dispense into a trash.""" + source_well = decoy.mock(cls=WellCore) + target_trash = decoy.mock(cls=TrashBin) + + air_gap_volume = 0.123 + air_gap_flow_rate_by_vol = 123 + air_gap_correction_by_vol = 0.321 + + sample_transfer_props.aspirate.retract.air_gap_by_volume.set_for_volume( + 0, air_gap_volume + ) + sample_transfer_props.aspirate.flow_rate_by_volume.set_for_volume( + air_gap_volume, air_gap_flow_rate_by_vol + ) + sample_transfer_props.aspirate.correction_by_volume.set_for_volume( + air_gap_volume, air_gap_correction_by_vol + ) + + sample_transfer_props.dispense.retract.blowout.location = ( + BlowoutLocation.DESTINATION + ) + + subject = TransferComponentsExecutor( + instrument_core=mock_instrument_core, + transfer_properties=sample_transfer_props, + target_location=target_trash, + target_well=None, + tip_state=TipState( + ready_to_aspirate=True, + last_liquid_and_air_gap_in_tip=LiquidAndAirGapPair( + liquid=10, + air_gap=0, + ), + ), + transfer_type=TransferType.ONE_TO_ONE, + ) + + subject.retract_after_dispensing( + trash_location=Location(Point(), labware=None), + source_location=Location(Point(1, 2, 3), labware=None), + source_well=source_well, + add_final_air_gap=True, + ) + decoy.verify( + mock_instrument_core.delay(10), + mock_instrument_core.set_flow_rate(blow_out=100), + mock_instrument_core.blow_out( + location=target_trash, + well_core=None, + in_place=True, + ), + *( + add_final_air_gap + and [ + mock_instrument_core.air_gap_in_place( # type: ignore[func-returns-value] + volume=air_gap_volume, + flow_rate=air_gap_flow_rate_by_vol, + correction_volume=air_gap_correction_by_vol, + ), + mock_instrument_core.delay(0.2), # type: ignore[func-returns-value] + ] + or [] + ), + ) + + +@pytest.mark.parametrize( + "add_final_air_gap", + [True, False], +) +def test_retract_after_dispense_in_trash_with_blowout_in_disposal_location( + decoy: Decoy, + mock_instrument_core: InstrumentCore, + sample_transfer_props: TransferProperties, + add_final_air_gap: bool, +) -> None: + """It should execute steps to retract after a dispense into a trash.""" + source_location = Location(Point(1, 2, 3), labware=None) + source_well = decoy.mock(cls=WellCore) + target_trash = decoy.mock(cls=TrashBin) + trash_location = decoy.mock(cls=WasteChute) + + air_gap_volume = 0.123 + air_gap_flow_rate_by_vol = 123 + air_gap_correction_by_vol = 0.321 + + sample_transfer_props.aspirate.retract.air_gap_by_volume.set_for_volume( + 0, air_gap_volume + ) + sample_transfer_props.aspirate.flow_rate_by_volume.set_for_volume( + air_gap_volume, air_gap_flow_rate_by_vol + ) + sample_transfer_props.aspirate.correction_by_volume.set_for_volume( + air_gap_volume, air_gap_correction_by_vol + ) + + sample_transfer_props.dispense.retract.blowout.location = BlowoutLocation.TRASH + + subject = TransferComponentsExecutor( + instrument_core=mock_instrument_core, + transfer_properties=sample_transfer_props, + target_location=target_trash, + target_well=None, + tip_state=TipState(), + transfer_type=TransferType.ONE_TO_ONE, + ) + subject.retract_after_dispensing( + trash_location=trash_location, + source_location=source_location, + source_well=source_well, + add_final_air_gap=True, + ) + decoy.verify( + mock_instrument_core.delay(10), + mock_instrument_core.air_gap_in_place( + volume=air_gap_volume, + flow_rate=air_gap_flow_rate_by_vol, + correction_volume=air_gap_correction_by_vol, + ), + mock_instrument_core.delay(0.2), + mock_instrument_core.set_flow_rate(blow_out=100), + mock_instrument_core.blow_out( + location=trash_location, + well_core=None, + in_place=False, + ), + *( + add_final_air_gap + and [ + mock_instrument_core.air_gap_in_place( # type: ignore[func-returns-value] + volume=air_gap_volume, + flow_rate=air_gap_flow_rate_by_vol, + correction_volume=air_gap_correction_by_vol, + ), + mock_instrument_core.delay(0.2), # type: ignore[func-returns-value] + ] + or [] + ), + ) + + def test_retract_after_dispense_raises_for_invalid_retract_point( decoy: Decoy, mock_instrument_core: InstrumentCore, diff --git a/api/tests/opentrons/protocol_api/test_instrument_context.py b/api/tests/opentrons/protocol_api/test_instrument_context.py index 5b232e97d65..512ca5dab74 100644 --- a/api/tests/opentrons/protocol_api/test_instrument_context.py +++ b/api/tests/opentrons/protocol_api/test_instrument_context.py @@ -2119,6 +2119,56 @@ def test_transfer_liquid_multi_channel_delegates_to_engine_core( ) +@pytest.mark.parametrize("robot_type", ["OT-2 Standard", "OT-3 Standard"]) +def test_transfer_liquid_delegates_to_engine_core_with_trash_destination( + decoy: Decoy, + mock_protocol_core: ProtocolCore, + mock_instrument_core: InstrumentCore, + subject: InstrumentContext, + robot_type: RobotType, + minimal_liquid_class_def2: LiquidClassSchemaV1, +) -> None: + """It should delegate the transfer execution to core with a trash location as the destination.""" + test_liq_class = LiquidClass.create(minimal_liquid_class_def2) + mock_well = decoy.mock(cls=Well) + mock_starting_tip_well = decoy.mock(cls=Well) + mock_trash = decoy.mock(cls=TrashBin) + tip_racks = [decoy.mock(cls=Labware)] + trash_location = Location(point=Point(1, 2, 3), labware=mock_well) + next_tiprack = decoy.mock(cls=Labware) + subject.starting_tip = mock_starting_tip_well + subject._tip_racks = tip_racks + + decoy.when(mock_protocol_core.robot_type).then_return(robot_type) + decoy.when(mock_instrument_core.get_nozzle_map()).then_return(MOCK_MAP) + decoy.when(mock_instrument_core.get_active_channels()).then_return(2) + decoy.when(mock_instrument_core.get_current_volume()).then_return(0) + decoy.when(next_tiprack.uri).then_return("tiprack-uri") + decoy.when(mock_instrument_core.get_pipette_name()).then_return("pipette-name") + subject.transfer_with_liquid_class( + liquid_class=test_liq_class, + volume=10, + source=[mock_well], + dest=mock_trash, + new_tip="once", + trash_location=trash_location, + return_tip=True, + ) + decoy.verify( + mock_instrument_core.transfer_with_liquid_class( + liquid_class=test_liq_class, + volume=10, + source=[(Location(Point(), labware=mock_well), mock_well._core)], + dest=mock_trash, + new_tip=TransferTipPolicyV2.ONCE, + tip_racks=[(Location(Point(), labware=tip_racks[0]), tip_racks[0]._core)], + starting_tip=mock_starting_tip_well._core, + trash_location=trash_location, + return_tip=True, + ) + ) + + @pytest.mark.parametrize("robot_type", ["OT-2 Standard", "OT-3 Standard"]) def test_distribute_liquid_raises_if_more_than_one_source( decoy: Decoy, @@ -2691,3 +2741,54 @@ def test_consolidate_liquid_multi_channel_delegates_to_engine_core( return_tip=True, ) ) + + +@pytest.mark.parametrize("robot_type", ["OT-2 Standard", "OT-3 Standard"]) +def test_consolidate_liquid_delegates_to_engine_core_with_trash_destination( + decoy: Decoy, + mock_protocol_core: ProtocolCore, + mock_instrument_core: InstrumentCore, + subject: InstrumentContext, + robot_type: RobotType, + minimal_liquid_class_def2: LiquidClassSchemaV1, +) -> None: + """It should delegate the consolidate execution to core.""" + test_liq_class = LiquidClass.create(minimal_liquid_class_def2) + mock_well = decoy.mock(cls=Well) + mock_starting_tip_well = decoy.mock(cls=Well) + mock_waste_chute = decoy.mock(cls=WasteChute) + tip_racks = [decoy.mock(cls=Labware)] + trash_location = Location(point=Point(1, 2, 3), labware=mock_well) + next_tiprack = decoy.mock(cls=Labware) + subject.starting_tip = mock_starting_tip_well + subject._tip_racks = tip_racks + + decoy.when(mock_protocol_core.robot_type).then_return(robot_type) + decoy.when(mock_instrument_core.get_nozzle_map()).then_return(MOCK_MAP) + decoy.when(mock_instrument_core.get_active_channels()).then_return(2) + decoy.when(mock_instrument_core.get_current_volume()).then_return(0) + decoy.when(next_tiprack.uri).then_return("tiprack-uri") + decoy.when(mock_instrument_core.get_pipette_name()).then_return("pipette-name") + + subject.consolidate_with_liquid_class( + liquid_class=test_liq_class, + volume=10, + source=[mock_well], + dest=mock_waste_chute, + new_tip="once", + trash_location=trash_location, + return_tip=True, + ) + decoy.verify( + mock_instrument_core.consolidate_with_liquid_class( + liquid_class=test_liq_class, + volume=10, + source=[(Location(Point(), labware=mock_well), mock_well._core)], + dest=mock_waste_chute, + new_tip=TransferTipPolicyV2.ONCE, + tip_racks=[(Location(Point(), labware=tip_racks[0]), tip_racks[0]._core)], + starting_tip=mock_starting_tip_well._core, + trash_location=trash_location, + return_tip=True, + ) + )