Skip to content

feat(api): allow disposal locations as destinations for transfer and consolidate with liquid class #18196

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 10 commits into from
May 1, 2025
20 changes: 16 additions & 4 deletions api/src/opentrons/legacy_commands/commands.py
Original file line number Diff line number Diff line change
@@ -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,
4 changes: 3 additions & 1 deletion api/src/opentrons/legacy_commands/types.py
Original file line number Diff line number Diff line change
@@ -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):
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
@@ -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],
@@ -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,
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
@@ -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
Loading
Oops, something went wrong.