revert merge

Author: caila-marashaj

api/docs/v2/conf.py

@@ -445,6 +445,7 @@
     ("py:class", r".*protocol_api\.config.*"),
     ("py:class", r".*opentrons_shared_data.*"),
     ("py:class", r".*protocol_api._parameters.Parameters.*"),
+    ("py:class", r".*protocol_api._liquid_properties.TransferProperties*"),
     ("py:class", r".*RobotContext"),  # shh it's a secret (for now)
     ("py:class", r".*FlexStackerContext"),  # ssh it's a secret (for now)
     (

api/docs/v2/new_protocol_api.rst

@@ -47,6 +47,9 @@ Wells and Liquids
 
 .. autoclass:: opentrons.protocol_api.Liquid
 
+.. autoclass:: opentrons.protocol_api.LiquidClass
+   :members:
+
 .. _protocol-api-modules:
 
 Modules

api/src/opentrons/legacy_commands/commands.py

@@ -1,12 +1,18 @@
 from __future__ import annotations
-from typing import TYPE_CHECKING, List, Union, overload
+from typing import TYPE_CHECKING, List, Sequence, Union, overload
 
 
-from .helpers import stringify_location, stringify_disposal_location, listify
+from .helpers import (
+    stringify_location,
+    stringify_disposal_location,
+    stringify_well_list,
+    listify,
+)
 from . import types as command_types
 
 from opentrons.types import Location
 from opentrons.protocol_api.disposal_locations import TrashBin, WasteChute
+from opentrons.protocol_api._liquid import LiquidClass
 
 if TYPE_CHECKING:
     from opentrons.protocol_api import InstrumentContext
@@ -317,6 +323,81 @@ def move_to_disposal_location(
     }
 
 
+def transfer_with_liquid_class(
+    instrument: InstrumentContext,
+    liquid_class: LiquidClass,
+    volume: float,
+    source: Union[Well, Sequence[Well], Sequence[Sequence[Well]]],
+    destination: Union[Well, Sequence[Well], Sequence[Sequence[Well]]],
+) -> command_types.TransferWithLiquidClassCommand:
+    text = (
+        "Transferring "
+        + f"{volume} uL of {liquid_class.display_name} liquid class from "
+        + f"{stringify_well_list(source)} to {stringify_well_list(destination)}"
+    )
+    return {
+        "name": command_types.TRANSFER_WITH_LIQUID_CLASS,
+        "payload": {
+            "instrument": instrument,
+            "liquid_class": liquid_class,
+            "volume": volume,
+            "source": source,
+            "destination": destination,
+            "text": text,
+        },
+    }
+
+
+def distribute_with_liquid_class(
+    instrument: InstrumentContext,
+    liquid_class: LiquidClass,
+    volume: float,
+    source: Union[Well, Sequence[Well], Sequence[Sequence[Well]]],
+    destination: Union[Well, Sequence[Well], Sequence[Sequence[Well]]],
+) -> command_types.DistributeWithLiquidClassCommand:
+    text = (
+        "Distributing "
+        + f"{volume} uL of {liquid_class.display_name} liquid class from "
+        + f"{stringify_well_list(source)} to {stringify_well_list(destination)}"
+    )
+    return {
+        "name": command_types.DISTRIBUTE_WITH_LIQUID_CLASS,
+        "payload": {
+            "instrument": instrument,
+            "liquid_class": liquid_class,
+            "volume": volume,
+            "source": source,
+            "destination": destination,
+            "text": text,
+        },
+    }
+
+
+def consolidate_with_liquid_class(
+    instrument: InstrumentContext,
+    liquid_class: LiquidClass,
+    volume: float,
+    source: Union[Well, Sequence[Well], Sequence[Sequence[Well]]],
+    destination: Union[Well, Sequence[Well], Sequence[Sequence[Well]]],
+) -> command_types.ConsolidateWithLiquidClassCommand:
+    text = (
+        "Consolidating "
+        + f"{volume} uL of {liquid_class.display_name} liquid class from "
+        + f"{stringify_well_list(source)} to {stringify_well_list(destination)}"
+    )
+    return {
+        "name": command_types.CONSOLIDATE_WITH_LIQUID_CLASS,
+        "payload": {
+            "instrument": instrument,
+            "liquid_class": liquid_class,
+            "volume": volume,
+            "source": source,
+            "destination": destination,
+            "text": text,
+        },
+    }
+
+
 def seal(
     instrument: InstrumentContext,
     location: Well,

api/src/opentrons/legacy_commands/helpers.py

@@ -1,4 +1,4 @@
-from typing import List, Union
+from typing import List, Union, Sequence
 
 from opentrons.protocol_api.labware import Well, Labware
 from opentrons.protocol_api.module_contexts import ModuleContext
@@ -48,6 +48,64 @@ def stringify_disposal_location(location: Union[TrashBin, WasteChute]) -> str:
         return "Waste Chute"
 
 
+def _group_wells_by_labware(wells: List[Well]) -> List[List[Well]]:
+    wells_by_labware: List[List[Well]] = []
+    sub_list = []
+    active_parent_labware = None
+    for well in wells:
+        if well.parent == active_parent_labware:
+            sub_list.append(well)
+        else:
+            active_parent_labware = well.parent
+            if sub_list:
+                wells_by_labware.append(sub_list)
+            sub_list = [well]
+    if sub_list:
+        wells_by_labware.append(sub_list)
+
+    return wells_by_labware
+
+
+def _stringify_multiple_wells_for_labware(wells: List[Well]) -> str:
+    if len(wells) == 0:
+        return ""
+    elif len(wells) == 1:
+        return str(wells[0])
+    # TODO(jbl 2025-04-10) this logic can be improved to more intelligently group wells
+    elif len(wells) < 9:  # At most we'll print out a full column's worth of well
+        return ", ".join([well.well_name for well in wells[:-1]]) + f", {wells[-1]}"
+    else:  # Otherwise print the first and last three
+        return (
+            ", ".join([well.well_name for well in wells[:3]])
+            + ", ... "
+            + ", ".join([well.well_name for well in wells[-3:-1]])
+            + f", {wells[-1]}"
+        )
+
+
+def stringify_well_list(
+    wells: Union[Well, Sequence[Well], Sequence[Sequence[Well]]]
+) -> str:
+    """Takes an arbitrary sequence of wells and returns a string representation of each well, associated by labware."""
+    if isinstance(wells, Well):
+        well_list = [wells]
+    elif len(wells) == 0:
+        well_list = []
+    elif isinstance(wells, list) and isinstance(wells[0], list):
+        well_list = [well for sub_well_list in wells for well in sub_well_list]
+    elif isinstance(wells, list):
+        well_list = wells
+    else:
+        return ""
+
+    return "; ".join(
+        [
+            _stringify_multiple_wells_for_labware(wells_by_labware)
+            for wells_by_labware in _group_wells_by_labware(well_list)
+        ]
+    )
+
+
 def _stringify_labware_movement_location(
     location: Union[
         DeckLocation, OffDeckType, Labware, ModuleContext, WasteChute, TrashBin

api/src/opentrons/legacy_commands/types.py

@@ -8,6 +8,7 @@
     from opentrons.protocol_api import InstrumentContext
     from opentrons.protocol_api.labware import Well
     from opentrons.protocol_api.disposal_locations import TrashBin, WasteChute
+    from opentrons.protocol_api._liquid import LiquidClass
 
 from opentrons.types import Location
 
@@ -43,6 +44,9 @@
 RETURN_TIP: Final = "command.RETURN_TIP"
 MOVE_TO: Final = "command.MOVE_TO"
 MOVE_TO_DISPOSAL_LOCATION: Final = "command.MOVE_TO_DISPOSAL_LOCATION"
+TRANSFER_WITH_LIQUID_CLASS: Final = "command.TRANSFER_WITH_LIQUID_CLASS"
+DISTRIBUTE_WITH_LIQUID_CLASS: Final = "command.DISTRIBUTE_WITH_LIQUID_CLASS"
+CONSOLIDATE_WITH_LIQUID_CLASS: Final = "command.CONSOLIDATE_WITH_LIQUID_CLASS"
 SEAL: Final = "command.SEAL"
 UNSEAL: Final = "command.UNSEAL"
 PRESSURIZE: Final = "command.PRESSURIZE"
@@ -540,6 +544,28 @@ class MoveLabwareCommandPayload(TextOnlyPayload):
     pass
 
 
+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]]]
+
+
+class TransferWithLiquidClassCommand(TypedDict):
+    name: Literal["command.TRANSFER_WITH_LIQUID_CLASS"]
+    payload: LiquidClassCommandPayload
+
+
+class DistributeWithLiquidClassCommand(TypedDict):
+    name: Literal["command.DISTRIBUTE_WITH_LIQUID_CLASS"]
+    payload: LiquidClassCommandPayload
+
+
+class ConsolidateWithLiquidClassCommand(TypedDict):
+    name: Literal["command.CONSOLIDATE_WITH_LIQUID_CLASS"]
+    payload: LiquidClassCommandPayload
+
+
 class SealCommandPayload(TextOnlyPayload):
     instrument: InstrumentContext
     location: Union[None, Location, Well]
@@ -622,6 +648,9 @@ class PressurizeCommand(TypedDict):
     MoveToCommand,
     MoveToDisposalLocationCommand,
     MoveLabwareCommand,
+    TransferWithLiquidClassCommand,
+    DistributeWithLiquidClassCommand,
+    ConsolidateWithLiquidClassCommand,
     SealCommand,
     UnsealCommand,
     PressurizeCommand,
@@ -674,6 +703,7 @@ class PressurizeCommand(TypedDict):
     MoveToCommandPayload,
     MoveToDisposalLocationCommandPayload,
     MoveLabwareCommandPayload,
+    LiquidClassCommandPayload,
     SealCommandPayload,
     UnsealCommandPayload,
     PressurizeCommandPayload,

api/src/opentrons/protocol_api/core/engine/instrument.py

@@ -1141,7 +1141,7 @@ def get_next_tip(
             result.nextTipInfo if isinstance(result.nextTipInfo, NextTipInfo) else None
         )
 
-    def transfer_liquid(  # noqa: C901
+    def transfer_with_liquid_class(  # noqa: C901
         self,
         liquid_class: LiquidClass,
         volume: float,
@@ -1335,7 +1335,7 @@ def _pick_up_tip() -> WellCore:
             _drop_tip()
 
     # TODO(spp, 2025-02-25): wire up return tip
-    def distribute_liquid(  # noqa: C901
+    def distribute_with_liquid_class(  # noqa: C901
         self,
         liquid_class: LiquidClass,
         volume: float,
@@ -1417,7 +1417,7 @@ def distribute_liquid(  # noqa: C901
                 tip_working_volume=working_volume,
             )
         ):
-            self.transfer_liquid(
+            self.transfer_with_liquid_class(
                 liquid_class=liquid_class,
                 volume=volume,
                 source=[source for _ in range(len(dest))],
@@ -1657,7 +1657,7 @@ def _tip_can_hold_volume_for_multi_dispensing(
             <= tip_working_volume
         )
 
-    def consolidate_liquid(  # noqa: C901
+    def consolidate_with_liquid_class(  # noqa: C901
         self,
         liquid_class: LiquidClass,
         volume: float,

api/src/opentrons/protocol_api/core/engine/transfer_components_executor.py

@@ -207,8 +207,7 @@ def submerge(
             minimum_z_height=None,
             speed=submerge_properties.speed,
         )
-        if submerge_properties.delay.enabled:
-            assert submerge_properties.delay.duration is not None
+        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:
@@ -227,9 +226,7 @@ def aspirate_and_wait(self, volume: float) -> None:
         )
         self._tip_state.append_liquid(volume)
         delay_props = aspirate_props.delay
-        if delay_props.enabled:
-            # Assertion only for mypy purposes
-            assert delay_props.duration is not None
+        if delay_props.enabled and delay_props.duration:
             self._instrument.delay(delay_props.duration)
 
     def dispense_and_wait(
@@ -257,8 +254,7 @@ def dispense_and_wait(
             self._tip_state.ready_to_aspirate = False
         self._tip_state.delete_liquid(volume)
         dispense_delay = dispense_properties.delay
-        if dispense_delay.enabled:
-            assert dispense_delay.duration is not None
+        if dispense_delay.enabled and dispense_delay.duration:
             self._instrument.delay(dispense_delay.duration)
 
     def mix(self, mix_properties: MixProperties, last_dispense_push_out: bool) -> None:
@@ -361,8 +357,7 @@ def retract_after_aspiration(
             speed=retract_props.speed,
         )
         retract_delay = retract_props.delay
-        if retract_delay.enabled:
-            assert retract_delay.duration is not None
+        if retract_delay.enabled and retract_delay.duration:
             self._instrument.delay(retract_delay.duration)
         touch_tip_props = retract_props.touch_tip
         if touch_tip_props.enabled:
@@ -459,8 +454,7 @@ def retract_after_dispensing(
             speed=retract_props.speed,
         )
         retract_delay = retract_props.delay
-        if retract_delay.enabled:
-            assert retract_delay.duration is not None
+        if retract_delay.enabled and retract_delay.duration:
             self._instrument.delay(retract_delay.duration)
 
         blowout_props = retract_props.blowout
@@ -599,8 +593,7 @@ def retract_during_multi_dispensing(
             speed=retract_props.speed,
         )
         retract_delay = retract_props.delay
-        if retract_delay.enabled:
-            assert retract_delay.duration is not None
+        if retract_delay.enabled and retract_delay.duration:
             self._instrument.delay(retract_delay.duration)
 
         blowout_props = retract_props.blowout
@@ -780,8 +773,8 @@ def _add_air_gap(self, air_gap_volume: float) -> None:
         correction_volume = aspirate_props.correction_by_volume.get_for_volume(
             air_gap_volume
         )
-        # The maximum flow rate should be air_gap_volume per second
-        flow_rate = min(
+        # The minimum flow rate should be air_gap_volume per second
+        flow_rate = max(
             aspirate_props.flow_rate_by_volume.get_for_volume(air_gap_volume),
             air_gap_volume,
         )
@@ -791,9 +784,7 @@ def _add_air_gap(self, air_gap_volume: float) -> None:
             correction_volume=correction_volume,
         )
         delay_props = aspirate_props.delay
-        if delay_props.enabled:
-            # Assertion only for mypy purposes
-            assert delay_props.duration is not None
+        if delay_props.enabled and delay_props.duration:
             self._instrument.delay(delay_props.duration)
         self._tip_state.append_air_gap(air_gap_volume)
 
@@ -807,8 +798,8 @@ def _remove_air_gap(self, location: Location) -> None:
         correction_volume = dispense_props.correction_by_volume.get_for_volume(
             last_air_gap
         )
-        # The maximum flow rate should be air_gap_volume per second
-        flow_rate = min(
+        # The minimum flow rate should be air_gap_volume per second
+        flow_rate = max(
             dispense_props.flow_rate_by_volume.get_for_volume(last_air_gap),
             last_air_gap,
         )
@@ -824,8 +815,7 @@ def _remove_air_gap(self, location: Location) -> None:
         )
         self._tip_state.delete_air_gap(last_air_gap)
         dispense_delay = dispense_props.delay
-        if dispense_delay.enabled:
-            assert dispense_delay.duration is not None
+        if dispense_delay.enabled and dispense_delay.duration:
             self._instrument.delay(dispense_delay.duration)