From 0a140699863861752550ff17f44f9c8b5174115e Mon Sep 17 00:00:00 2001 From: jbleon95 Date: Wed, 23 Apr 2025 16:03:48 -0400 Subject: [PATCH 1/8] allow trash bin/chute as transfer component executor destination --- .../protocol_api/core/engine/instrument.py | 28 ++- .../engine/transfer_components_executor.py | 208 ++++++++++-------- .../opentrons/protocol_api/core/instrument.py | 2 +- .../core/legacy/legacy_instrument_core.py | 2 +- .../legacy_instrument_core.py | 2 +- 5 files changed, 143 insertions(+), 99 deletions(-) diff --git a/api/src/opentrons/protocol_api/core/engine/instrument.py b/api/src/opentrons/protocol_api/core/engine/instrument.py index f9fc06dc046..6c9cfb22eb8 100644 --- a/api/src/opentrons/protocol_api/core/engine/instrument.py +++ b/api/src/opentrons/protocol_api/core/engine/instrument.py @@ -9,6 +9,7 @@ cast, Union, List, + Sequence, Tuple, NamedTuple, Generator, @@ -1200,7 +1201,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], @@ -1246,10 +1247,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( @@ -2030,7 +2038,7 @@ def aspirate_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, @@ -2069,15 +2077,19 @@ 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, position_reference=dispense_props.dispense_position.position_reference, offset=dispense_props.dispense_position.offset, ) - ) - 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 92fe7e236fe..071adaf6fb1 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 @@ -111,8 +111,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: @@ -140,56 +140,64 @@ 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 """ - submerge_start_point = absolute_point_from_position_reference_and_offset( - well=self._target_well, - position_reference=submerge_properties.start_position.position_reference, - offset=submerge_properties.start_position.offset, - ) - submerge_start_location = Location( - point=submerge_start_point, labware=self._target_location.labware - ) - prep_before_moving_to_submerge = ( - post_submerge_action == "aspirate" - and volume_for_pipette_mode_configuration is not None - ) - if prep_before_moving_to_submerge: - # Move to the tip probe start position - self._instrument.move_to( - location=Location( - point=self._target_well.get_top( - LIQUID_PROBE_START_OFFSET_FROM_WELL_TOP.z + 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, + position_reference=submerge_properties.start_position.position_reference, + offset=submerge_properties.start_position.offset, + ) + submerge_start_location = Location( + point=submerge_start_point, labware=self._target_location.labware + ) + prep_before_moving_to_submerge = ( + post_submerge_action == "aspirate" + and volume_for_pipette_mode_configuration is not None + ) + + if prep_before_moving_to_submerge: + # Move to the tip probe start position + self._instrument.move_to( + location=Location( + point=self._target_well.get_top( + LIQUID_PROBE_START_OFFSET_FROM_WELL_TOP.z + ), + labware=self._target_location.labware, ), - labware=self._target_location.labware, - ), + well_core=self._target_well, + force_direct=False, + minimum_z_height=None, + speed=None, + ) + self._remove_air_gap(location=submerge_start_location) + if ( + self._transfer_type != TransferType.MANY_TO_ONE + and self._instrument.get_liquid_presence_detection() + ): + self._instrument.liquid_probe_with_recovery( + well_core=self._target_well, loc=submerge_start_location + ) + # TODO: do volume configuration + prepare for aspirate only if the mode needs to be changed + self._instrument.configure_for_volume(volume_for_pipette_mode_configuration) # type: ignore[arg-type] + self._instrument.prepare_to_aspirate() + tx_utils.raise_if_location_inside_liquid( + location=submerge_start_location, + well_location=self._target_location, well_core=self._target_well, - force_direct=False, - minimum_z_height=None, - speed=None, + location_check_descriptors=LocationCheckDescriptors( + location_type="submerge start", + pipetting_action=post_submerge_action, + ), + logger=log, ) - self._remove_air_gap(location=submerge_start_location) - if ( - self._transfer_type != TransferType.MANY_TO_ONE - and self._instrument.get_liquid_presence_detection() - ): - self._instrument.liquid_probe_with_recovery( - well_core=self._target_well, loc=submerge_start_location - ) - # TODO: do volume configuration + prepare for aspirate only if the mode needs to be changed - self._instrument.configure_for_volume(volume_for_pipette_mode_configuration) # type: ignore[arg-type] - self._instrument.prepare_to_aspirate() - 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 + prep_before_moving_to_submerge = False + self._instrument.move_to( location=submerge_start_location, well_core=self._target_well, @@ -199,19 +207,25 @@ def submerge( ) if not prep_before_moving_to_submerge: 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, + ) + # TODO do we want this for trash bins/waste chutes? 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( @@ -267,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 isinstance( + self._target_location, (TrashBin, WasteChute) + ): 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( @@ -329,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, @@ -425,36 +447,41 @@ def retract_after_dispensing( 7. If drop tip, move to drop tip location, drop tip """ # 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, - position_reference=retract_props.end_position.position_reference, - offset=retract_props.end_position.offset, - ) - 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_delay = retract_props.delay - if retract_delay.enabled and retract_delay.duration: - self._instrument.delay(retract_delay.duration) + + 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, + position_reference=retract_props.end_position.position_reference, + offset=retract_props.end_position.offset, + ) + 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_delay = retract_props.delay + if retract_delay.enabled and retract_delay.duration: + self._instrument.delay(retract_delay.duration) + else: + retract_location = self._target_location blowout_props = retract_props.blowout if ( @@ -479,7 +506,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, ) @@ -562,7 +591,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 @@ -787,7 +819,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 if last_air_gap == 0: diff --git a/api/src/opentrons/protocol_api/core/instrument.py b/api/src/opentrons/protocol_api/core/instrument.py index 00a48e67cd0..1138627f6c1 100644 --- a/api/src/opentrons/protocol_api/core/instrument.py +++ b/api/src/opentrons/protocol_api/core/instrument.py @@ -364,7 +364,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], 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 7f09deebd1c..fbda5582a5c 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 @@ -603,7 +603,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], 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 28640e16811..b95d367aac5 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 @@ -518,7 +518,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], From 3cb11c3e418da55363631d67927158b7d7ce573e Mon Sep 17 00:00:00 2001 From: jbleon95 Date: Thu, 24 Apr 2025 13:48:19 -0400 Subject: [PATCH 2/8] add trash bin and waste chute as arguments for transfer and consolidate --- api/src/opentrons/legacy_commands/commands.py | 20 ++++-- api/src/opentrons/legacy_commands/types.py | 4 +- .../_transfer_liquid_validation.py | 16 +++-- .../protocol_api/core/engine/instrument.py | 2 +- .../engine/transfer_components_executor.py | 4 +- .../opentrons/protocol_api/core/instrument.py | 2 +- .../core/legacy/legacy_instrument_core.py | 2 +- .../legacy_instrument_core.py | 2 +- .../protocol_api/instrument_context.py | 65 ++++++++++++------- 9 files changed, 79 insertions(+), 38 deletions(-) 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 e9901cd24d8..d9ed4438f95 100755 --- a/api/src/opentrons/legacy_commands/types.py +++ b/api/src/opentrons/legacy_commands/types.py @@ -548,7 +548,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..eb3341b7a62 100644 --- a/api/src/opentrons/protocol_api/_transfer_liquid_validation.py +++ b/api/src/opentrons/protocol_api/_transfer_liquid_validation.py @@ -20,7 +20,9 @@ class TransferInfo: sources_list: List[Well] - destinations_list: List[Well] + destinations_list: Union[ + List[Well], TrashBin, WasteChute + ] # TODO rename this maybe? tip_policy: TransferTipPolicyV2 tip_racks: List[Labware] trash_location: Union[Location, TrashBin, WasteChute] @@ -28,7 +30,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 +40,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 @@ -84,7 +90,9 @@ def verify_and_normalize_transfer_args( return TransferInfo( sources_list=flat_sources_list, - destinations_list=flat_dests_list, + destinations_list=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 6c9cfb22eb8..6a1d04fcb32 100644 --- a/api/src/opentrons/protocol_api/core/engine/instrument.py +++ b/api/src/opentrons/protocol_api/core/engine/instrument.py @@ -1735,7 +1735,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: TransferTipPolicyV2, tip_racks: List[Tuple[Location, LabwareCore]], starting_tip: Optional[WellCore], 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 071adaf6fb1..fcbbf10f658 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 @@ -281,8 +281,8 @@ 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 or isinstance( - self._target_location, (TrashBin, WasteChute) + if not mix_properties.enabled or not isinstance( + self._target_location, Location, ): return # Assertion only for mypy purposes diff --git a/api/src/opentrons/protocol_api/core/instrument.py b/api/src/opentrons/protocol_api/core/instrument.py index 1138627f6c1..138f4ec69e0 100644 --- a/api/src/opentrons/protocol_api/core/instrument.py +++ b/api/src/opentrons/protocol_api/core/instrument.py @@ -399,7 +399,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: TransferTipPolicyV2, 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 fbda5582a5c..b0592e1384d 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 @@ -633,7 +633,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: TransferTipPolicyV2, 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 b95d367aac5..126b8478c70 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 @@ -548,7 +548,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: TransferTipPolicyV2, 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 298b9dd20d5..b336ecb8116 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 @@ -1606,7 +1606,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[ @@ -1662,12 +1666,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.destinations_list, (TrashBin, WasteChute)): + verified_dest = transfer_args.destinations_list + else: + 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 = [ + (types.Location(types.Point(), labware=well), well._core) + for well in transfer_args.destinations_list + ] with publisher.publish_context( broker=self.broker, @@ -1686,10 +1701,7 @@ def transfer_with_liquid_class( (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 - ], + dest=verified_dest, new_tip=transfer_args.tip_policy, tip_racks=[ (types.Location(types.Point(), labware=rack), rack._core) @@ -1765,6 +1777,7 @@ def distribute_with_liquid_class( trash_location if trash_location is not None else self.trash_container ), ) + assert not isinstance(transfer_args.destinations_list, (TrashBin, WasteChute)) if len(transfer_args.sources_list) != 1: raise ValueError( f"Source should be a single well (or resolve to a single transfer for multi-channel) " @@ -1818,7 +1831,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] @@ -1873,17 +1886,26 @@ 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.destinations_list, (TrashBin, WasteChute)): + verified_dest = transfer_args.destinations_list + else: + 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 = ( + types.Location( + types.Point(), labware=transfer_args.destinations_list[0] + ), + transfer_args.destinations_list[0]._core, ) if transfer_args.tip_policy == TransferTipPolicyV2.PER_SOURCE: raise RuntimeError( 'Tip transfer policy "per source" incompatible with consolidate.' ) - verified_dest = transfer_args.destinations_list[0] with publisher.publish_context( broker=self.broker, command=cmds.consolidate_with_liquid_class( @@ -1901,10 +1923,7 @@ def consolidate_with_liquid_class( (types.Location(types.Point(), labware=well), well._core) for well in transfer_args.sources_list ], - dest=( - types.Location(types.Point(), labware=verified_dest), - verified_dest._core, - ), + dest=verified_dest, new_tip=transfer_args.tip_policy, tip_racks=[ (types.Location(types.Point(), labware=rack), rack._core) From b0db9c4a94c7606196f11b1902b1c4da491a7c2e Mon Sep 17 00:00:00 2001 From: jbleon95 Date: Thu, 24 Apr 2025 14:28:44 -0400 Subject: [PATCH 3/8] renaming and doc strings --- .../_transfer_liquid_validation.py | 12 ++--- .../engine/transfer_components_executor.py | 8 +++- .../protocol_api/instrument_context.py | 44 +++++++++---------- 3 files changed, 33 insertions(+), 31 deletions(-) diff --git a/api/src/opentrons/protocol_api/_transfer_liquid_validation.py b/api/src/opentrons/protocol_api/_transfer_liquid_validation.py index eb3341b7a62..4aea3b6f696 100644 --- a/api/src/opentrons/protocol_api/_transfer_liquid_validation.py +++ b/api/src/opentrons/protocol_api/_transfer_liquid_validation.py @@ -19,10 +19,8 @@ @dataclass class TransferInfo: - sources_list: List[Well] - destinations_list: Union[ - List[Well], TrashBin, WasteChute - ] # TODO rename this maybe? + source: List[Well] + dest: Union[List[Well], TrashBin, WasteChute] tip_policy: TransferTipPolicyV2 tip_racks: List[Labware] trash_location: Union[Location, TrashBin, WasteChute] @@ -89,10 +87,8 @@ def verify_and_normalize_transfer_args( ) return TransferInfo( - sources_list=flat_sources_list, - destinations_list=flat_dests_list - if not isinstance(dest, (TrashBin, WasteChute)) - else dest, + 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/transfer_components_executor.py b/api/src/opentrons/protocol_api/core/engine/transfer_components_executor.py index 216a05411e7..aa44c96be45 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 @@ -150,6 +150,9 @@ def submerge( For reference pos of anything else, do not allow submerge position to be below aspirate position 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_location: Union[Location, TrashBin, WasteChute] if isinstance(self._target_location, Location): @@ -290,7 +293,7 @@ def mix(self, mix_properties: MixProperties, last_dispense_push_out: bool) -> No We should mention in our docs that users should adjust this property according to their application. """ if not mix_properties.enabled or not isinstance( - self._target_location, Location, + self._target_location, Location ): return # Assertion only for mypy purposes @@ -453,6 +456,9 @@ 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 diff --git a/api/src/opentrons/protocol_api/instrument_context.py b/api/src/opentrons/protocol_api/instrument_context.py index e652c908128..9a97576ca42 100644 --- a/api/src/opentrons/protocol_api/instrument_context.py +++ b/api/src/opentrons/protocol_api/instrument_context.py @@ -1632,7 +1632,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, or 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"``. @@ -1675,10 +1675,10 @@ def transfer_with_liquid_class( verified_dest: Union[ List[Tuple[types.Location, WellCore]], TrashBin, WasteChute ] - if isinstance(transfer_args.destinations_list, (TrashBin, WasteChute)): - verified_dest = transfer_args.destinations_list + if isinstance(transfer_args.dest, (TrashBin, WasteChute)): + verified_dest = transfer_args.dest else: - if len(transfer_args.sources_list) != len(transfer_args.destinations_list): + 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'," @@ -1686,7 +1686,7 @@ def transfer_with_liquid_class( ) verified_dest = [ (types.Location(types.Point(), labware=well), well._core) - for well in transfer_args.destinations_list + for well in transfer_args.dest ] with publisher.publish_context( @@ -1704,7 +1704,7 @@ def transfer_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=verified_dest, new_tip=transfer_args.tip_policy, @@ -1749,7 +1749,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 or group of wells for a multi-channel pipette to + aspirate liquid from. :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"``. @@ -1786,11 +1787,11 @@ def distribute_with_liquid_class( trash_location if trash_location is not None else self.trash_container ), ) - assert not isinstance(transfer_args.destinations_list, (TrashBin, WasteChute)) - if len(transfer_args.sources_list) != 1: + assert not isinstance(transfer_args.dest, (TrashBin, WasteChute)) + 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 +1803,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 +1823,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=[ @@ -1867,7 +1868,8 @@ 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, group of wells for a multi-channel pipette, or + 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"``. @@ -1904,19 +1906,17 @@ def consolidate_with_liquid_class( ), ) verified_dest: Union[Tuple[types.Location, WellCore], TrashBin, WasteChute] - if isinstance(transfer_args.destinations_list, (TrashBin, WasteChute)): - verified_dest = transfer_args.destinations_list + if isinstance(transfer_args.dest, (TrashBin, WasteChute)): + verified_dest = transfer_args.dest else: - if len(transfer_args.destinations_list) != 1: + 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.destinations_list}." + f"but received {transfer_args.dest}." ) verified_dest = ( - types.Location( - types.Point(), labware=transfer_args.destinations_list[0] - ), - transfer_args.destinations_list[0]._core, + types.Location(types.Point(), labware=transfer_args.dest[0]), + transfer_args.dest[0]._core, ) if transfer_args.tip_policy not in [ TransferTipPolicyV2.ONCE, @@ -1943,7 +1943,7 @@ 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=verified_dest, new_tip=transfer_args.tip_policy, # type: ignore[arg-type] From 4acc0045bcf3d616553dacf305b2a7ea1f75ee1b Mon Sep 17 00:00:00 2001 From: jbleon95 Date: Mon, 28 Apr 2025 10:35:11 -0400 Subject: [PATCH 4/8] unit tests --- .../engine/transfer_components_executor.py | 8 +- api/tests/opentrons/conftest.py | 2 +- .../test_transfer_components_executor.py | 352 +++++++++++++++++- .../protocol_api/test_instrument_context.py | 101 +++++ 4 files changed, 458 insertions(+), 5 deletions(-) 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 aa44c96be45..e05ac868345 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 @@ -491,12 +491,14 @@ def retract_after_dispensing( minimum_z_height=None, speed=retract_props.speed, ) - retract_delay = retract_props.delay - if retract_delay.enabled and retract_delay.duration: - self._instrument.delay(retract_delay.duration) 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) + blowout_props = retract_props.blowout if ( blowout_props.enabled 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 0e3d91cadca..bf4e099106e 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 @@ -257,6 +257,74 @@ def test_submerge_with_lpd( ) +@pytest.mark.parametrize( + 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)], +) +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, + expected_air_gap_flow_rate: 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", + volume_for_pipette_mode_configuration=123, + ) + + 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.dispense( + location=mock_trash_bin, + well_core=None, + volume=air_gap_volume, + rate=1, + flow_rate=expected_air_gap_flow_rate, + in_place=True, + push_out=0, + correction_volume=air_gap_correction_by_vol, + ), + mock_instrument_core.delay(0.5), + mock_instrument_core.delay(1.1), + ) + + def test_submerge_raises_when_submerge_point_is_invalid( decoy: Decoy, mock_instrument_core: InstrumentCore, @@ -451,6 +519,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, @@ -1354,6 +1463,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 from well 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 from well 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 from well 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 4ecf267bda8..81ec362c096 100644 --- a/api/tests/opentrons/protocol_api/test_instrument_context.py +++ b/api/tests/opentrons/protocol_api/test_instrument_context.py @@ -2065,6 +2065,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, @@ -2637,3 +2687,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, + ) + ) From b268b8a9b138d966317ab7813b3796f3053efeb4 Mon Sep 17 00:00:00 2001 From: jbleon95 Date: Mon, 28 Apr 2025 14:36:50 -0400 Subject: [PATCH 5/8] fix merge issues --- .../protocol_api/core/engine/instrument.py | 2 +- .../test_transfer_components_executor.py | 21 +++++-------------- 2 files changed, 6 insertions(+), 17 deletions(-) diff --git a/api/src/opentrons/protocol_api/core/engine/instrument.py b/api/src/opentrons/protocol_api/core/engine/instrument.py index 973e515e934..77200f4a68e 100644 --- a/api/src/opentrons/protocol_api/core/engine/instrument.py +++ b/api/src/opentrons/protocol_api/core/engine/instrument.py @@ -2067,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: 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 e711be87af7..af6e7a27fc3 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 @@ -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) @@ -227,9 +225,8 @@ def test_submerge_without_starting_air_gap( 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_with_trash_location( decoy: Decoy, @@ -237,7 +234,6 @@ def test_submerge_with_trash_location( 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.""" air_gap_correction_by_vol = 0.321 @@ -265,7 +261,6 @@ def test_submerge_with_trash_location( subject.submerge( submerge_properties=sample_transfer_props.dispense.submerge, post_submerge_action="dispense", - volume_for_pipette_mode_configuration=123, ) decoy.verify( @@ -276,17 +271,11 @@ def test_submerge_with_trash_location( minimum_z_height=None, speed=None, ), - mock_instrument_core.dispense( + 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, - well_core=None, - volume=air_gap_volume, - rate=1, - flow_rate=expected_air_gap_flow_rate, - in_place=True, - push_out=0, - correction_volume=air_gap_correction_by_vol, ), - mock_instrument_core.delay(0.5), mock_instrument_core.delay(1.1), ) From 3c4d3f015e3b278c9c741a08a9205142219ea236 Mon Sep 17 00:00:00 2001 From: jbleon95 Date: Mon, 28 Apr 2025 14:55:31 -0400 Subject: [PATCH 6/8] small doc string update --- .../protocol_api/core/engine/transfer_components_executor.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) 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 720ab87149c..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 @@ -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 From 4e5e27e4727c2d7df28064c24895fba4d1281392 Mon Sep 17 00:00:00 2001 From: jbleon95 Date: Thu, 1 May 2025 10:02:37 -0400 Subject: [PATCH 7/8] doc string updates --- .../opentrons/protocol_api/instrument_context.py | 13 +++++++------ .../engine/test_transfer_components_executor.py | 6 +++--- 2 files changed, 10 insertions(+), 9 deletions(-) diff --git a/api/src/opentrons/protocol_api/instrument_context.py b/api/src/opentrons/protocol_api/instrument_context.py index 9e887b6df96..6ce1b40d84a 100644 --- a/api/src/opentrons/protocol_api/instrument_context.py +++ b/api/src/opentrons/protocol_api/instrument_context.py @@ -1645,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, list of wells, or trash bin or waste chute 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"``. @@ -1695,7 +1695,7 @@ def transfer_with_liquid_class( 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'." + " to transfer liquid to one destination from many sources, use 'consolidate_liquid'." ) verified_dest = [ (types.Location(types.Point(), labware=well), well._core) @@ -1762,8 +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 or group of wells for a multi-channel pipette 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"``. @@ -1881,8 +1881,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, group of wells for a multi-channel pipette, or - trash bin or waste chute 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"``. 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 af6e7a27fc3..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 @@ -1427,7 +1427,7 @@ def test_retract_after_dispense_in_trash_with_blowout_in_source( sample_transfer_props: TransferProperties, add_final_air_gap: bool, ) -> None: - """It should execute steps to retract from well after a dispense into a trash.""" + """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) @@ -1517,7 +1517,7 @@ def test_retract_after_dispense_in_trash_with_blowout_in_destination( sample_transfer_props: TransferProperties, add_final_air_gap: bool, ) -> None: - """It should execute steps to retract from well after a dispense into a trash.""" + """It should execute steps to retract after a dispense into a trash.""" source_well = decoy.mock(cls=WellCore) target_trash = decoy.mock(cls=TrashBin) @@ -1593,7 +1593,7 @@ def test_retract_after_dispense_in_trash_with_blowout_in_disposal_location( sample_transfer_props: TransferProperties, add_final_air_gap: bool, ) -> None: - """It should execute steps to retract from well after a dispense into a trash.""" + """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) From 7ddba771f51ff7dff6ce4ffb42f7b9b016acc55c Mon Sep 17 00:00:00 2001 From: jbleon95 Date: Thu, 1 May 2025 13:27:30 -0400 Subject: [PATCH 8/8] user friendly error message for trash as destination for distribute --- api/src/opentrons/protocol_api/instrument_context.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/api/src/opentrons/protocol_api/instrument_context.py b/api/src/opentrons/protocol_api/instrument_context.py index 6ce1b40d84a..e827bc8c158 100644 --- a/api/src/opentrons/protocol_api/instrument_context.py +++ b/api/src/opentrons/protocol_api/instrument_context.py @@ -1800,7 +1800,11 @@ def distribute_with_liquid_class( trash_location if trash_location is not None else self.trash_container ), ) - assert not isinstance(transfer_args.dest, (TrashBin, WasteChute)) + 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) "