Skip to content
Merged
20 changes: 16 additions & 4 deletions api/src/opentrons/legacy_commands/commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down
4 changes: 3 additions & 1 deletion api/src/opentrons/legacy_commands/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down
16 changes: 10 additions & 6 deletions api/src/opentrons/protocol_api/_transfer_liquid_validation.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,16 +19,16 @@
@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]


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],
Expand All @@ -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
Expand Down Expand Up @@ -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,
Expand Down
32 changes: 22 additions & 10 deletions api/src/opentrons/protocol_api/core/engine/instrument.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
cast,
Union,
List,
Sequence,
Tuple,
NamedTuple,
Generator,
Expand Down Expand Up @@ -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],
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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],
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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
Expand Down
Loading