feat(api): add new InstrumentContext.transfer_liquid() method

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

@@ -2,7 +2,8 @@
 
 from __future__ import annotations
 
-from typing import Optional, TYPE_CHECKING, cast, Union
+import contextlib
+from typing import Optional, TYPE_CHECKING, cast, Union, Iterator
 from opentrons.protocols.api_support.types import APIVersion
 
 from opentrons.types import Location, Mount, NozzleConfigurationType, NozzleMapInterface
@@ -42,6 +43,7 @@
 
 if TYPE_CHECKING:
     from .protocol import ProtocolCore
+    from opentrons.protocol_api._liquid import LiquidClass
 
 
 _DISPENSE_VOLUME_VALIDATION_ADDED_IN = APIVersion(2, 17)
@@ -85,6 +87,7 @@ def __init__(
         self._liquid_presence_detection = bool(
             self._engine_client.state.pipettes.get_liquid_presence_detection(pipette_id)
         )
+        self._current_liquid_class: Optional[LiquidClass] = None
 
     @property
     def pipette_id(self) -> str:
@@ -940,3 +943,13 @@ def nozzle_configuration_valid_for_lld(self) -> bool:
         return self._engine_client.state.pipettes.get_nozzle_configuration_supports_lld(
             self.pipette_id
         )
+
+    @contextlib.contextmanager
+    def load_liquid_class(self, liquid_class: LiquidClass) -> Iterator[None]:
+        """Load a liquid class into the engine."""
+        try:
+            # TODO: issue a loadLiquidClass command
+            self._current_liquid_class = liquid_class
+            yield
+        finally:
+            self._current_liquid_class = None

api/src/opentrons/protocol_api/instrument_context.py

@@ -1,12 +1,14 @@
 from __future__ import annotations
 import logging
 from contextlib import ExitStack
-from typing import Any, List, Optional, Sequence, Union, cast, Dict
+from typing import Any, List, Optional, Sequence, Union, cast, Dict, Literal
 from opentrons_shared_data.errors.exceptions import (
     CommandPreconditionViolated,
     CommandParameterLimitViolated,
     UnexpectedTipRemovalError,
 )
+from opentrons_shared_data.robot.types import RobotTypeEnum
+
 from opentrons.legacy_broker import LegacyBroker
 from opentrons.hardware_control.dev_types import PipetteDict
 from opentrons import types
@@ -15,7 +17,9 @@
 from opentrons.legacy_commands import publisher
 from opentrons.protocols.advanced_control.mix import mix_from_kwargs
 from opentrons.protocols.advanced_control.transfers import transfer as v1_transfer
-
+from opentrons.protocols.advanced_control.transfers import (
+    transfer_liquid as v2_transfer,
+)
 from opentrons.protocols.api_support.deck_type import NoTrashDefinedError
 from opentrons.protocols.api_support.types import APIVersion
 from opentrons.protocols.api_support import instrument
@@ -34,9 +38,10 @@
 from .config import Clearances
 from .disposal_locations import TrashBin, WasteChute
 from ._nozzle_layout import NozzleLayout
+from ._liquid import LiquidClass
 from . import labware, validation
-
-AdvancedLiquidHandling = v1_transfer.AdvancedLiquidHandling
+from ..config import feature_flags
+from ..protocols.advanced_control.transfers.common import TransferTipPolicyV2
 
 _DEFAULT_ASPIRATE_CLEARANCE = 1.0
 _DEFAULT_DISPENSE_CLEARANCE = 1.0
@@ -60,6 +65,8 @@
 _AIR_GAP_TRACKING_ADDED_IN = APIVersion(2, 22)
 """The version after which air gaps should be implemented with a separate call instead of an aspirate for better liquid volume tracking."""
 
+AdvancedLiquidHandling = v1_transfer.AdvancedLiquidHandling
+
 
 class InstrumentContext(publisher.CommandPublisher):
     """
@@ -1215,7 +1222,6 @@ def home_plunger(self) -> InstrumentContext:
         self._core.home_plunger()
         return self
 
-    # TODO (spp, 2024-03-08): verify if ok to & change source & dest types to AdvancedLiquidHandling
     @publisher.publish(command=cmds.distribute)
     @requires_version(2, 0)
     def distribute(
@@ -1255,7 +1261,6 @@ def distribute(
 
         return self.transfer(volume, source, dest, **kwargs)
 
-    # TODO (spp, 2024-03-08): verify if ok to & change source & dest types to AdvancedLiquidHandling
     @publisher.publish(command=cmds.consolidate)
     @requires_version(2, 0)
     def consolidate(
@@ -1501,6 +1506,80 @@ def _execute_transfer(self, plan: v1_transfer.TransferPlan) -> None:
         for cmd in plan:
             getattr(self, cmd["method"])(*cmd["args"], **cmd["kwargs"])
 
+    def transfer_liquid(
+        self,
+        liquid_class: LiquidClass,
+        volume: float,
+        source: AdvancedLiquidHandling,
+        dest: AdvancedLiquidHandling,
+        new_tip: Literal["once", "always", "never"] = "once",
+        trash_location: Optional[Union[types.Location, TrashBin, WasteChute]] = None,
+    ) -> InstrumentContext:
+        """Transfer liquid from source to dest using the specified liquid class properties."""
+        if not feature_flags.allow_liquid_classes(
+            robot_type=RobotTypeEnum.robot_literal_to_enum(
+                self._protocol_core.robot_type
+            )
+        ):
+            raise NotImplementedError("This method is not implemented.")
+
+        flat_sources_list = validation.ensure_valid_flat_wells_list(source)
+        flat_dest_list = validation.ensure_valid_flat_wells_list(dest)
+
+        if len(flat_sources_list) != len(flat_dest_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'."
+            )
+
+        valid_new_tip = validation.ensure_new_tip_policy(new_tip)
+        if valid_new_tip == TransferTipPolicyV2.NEVER:
+            if self._last_tip_picked_up_from is None:
+                raise RuntimeError(
+                    "Pipette has no tip attached to perform transfer."
+                    " Either do a pick_up_tip beforehand or specify a new_tip parameter"
+                    " of 'once' or 'always'."
+                )
+            else:
+                tiprack = self._last_tip_picked_up_from.parent
+        else:
+            tiprack, well = labware.next_available_tip(
+                starting_tip=self.starting_tip,
+                tip_racks=self.tip_racks,
+                channels=self.active_channels,
+                nozzle_map=self._core.get_nozzle_map(),
+            )
+        if self.current_volume != 0:
+            raise RuntimeError(
+                "A transfer on a liquid class cannot start with liquid already in the tip."
+                " Ensure that all previously aspirated liquid is dispensed before starting"
+                " a new transfer."
+            )
+        liquid_class_props = liquid_class.get_for(
+            pipette=self.name, tiprack=tiprack.name
+        )
+        checked_trash_location: Union[
+            types.Location, labware.Labware, TrashBin, WasteChute
+        ]
+        if trash_location is None:
+            checked_trash_location = (
+                self.trash_container
+            )  # Could be a labware or a trash fixture
+        else:
+            checked_trash_location = trash_location
+
+        v2_transfer.get_transfer_steps(
+            aspirate_properties=liquid_class_props.aspirate,
+            single_dispense_properties=liquid_class_props.dispense,
+            volume=volume,
+            source=flat_sources_list,
+            dest=flat_dest_list,
+            trash_location=checked_trash_location,
+            new_tip=valid_new_tip,
+        )
+        return self
+
     @requires_version(2, 0)
     def delay(self, *args: Any, **kwargs: Any) -> None:
         """

api/src/opentrons/protocol_api/validation.py

@@ -21,6 +21,7 @@
 from opentrons.protocols.api_support.types import APIVersion, ThermocyclerStep
 from opentrons.protocols.api_support.util import APIVersionError
 from opentrons.protocols.models import LabwareDefinition
+from opentrons.protocols.advanced_control.transfers.common import TransferTipPolicyV2
 from opentrons.types import (
     Mount,
     DeckSlotName,
@@ -42,6 +43,7 @@
 
 from .disposal_locations import TrashBin, WasteChute
 
+
 if TYPE_CHECKING:
     from .labware import Well
 
@@ -634,3 +636,55 @@ def validate_coordinates(value: Sequence[float]) -> Tuple[float, float, float]:
     if not all(isinstance(v, (float, int)) for v in value):
         raise ValueError("All values in coordinates must be floats.")
     return float(value[0]), float(value[1]), float(value[2])
+
+
+def ensure_new_tip_policy(value: str) -> TransferTipPolicyV2:
+    """Ensure that new_tip value is a valid TransferTipPolicy value."""
+    try:
+        return TransferTipPolicyV2(value.lower())
+    except ValueError:
+        raise ValueError(
+            f"'{value}' is invalid value for 'new_tip'."
+            f" Acceptable value is either 'never', 'once' or 'always'."
+        )
+
+
+def _verify_each_list_element_is_valid_location(
+    locations: Sequence[Union[Well, Location]]
+) -> None:
+    from .labware import Well
+
+    for loc in locations:
+        if not (isinstance(loc, Well) or isinstance(loc, Location)):
+            raise ValueError(
+                f"'{loc}' is not a valid location for transfer. Should be of type 'Well' or 'Location'"
+            )
+
+
+def ensure_valid_flat_wells_list(
+    target: Union[
+        Well,
+        Location,
+        Sequence[Union[Well, Location]],
+        Sequence[Sequence[Well]],
+    ],
+) -> Sequence[Union[Well, Location]]:
+    """Ensure that the given target(s) for a liquid transfer are valid and in a flat list."""
+    from .labware import Well
+
+    if isinstance(target, Well) or isinstance(target, Location):
+        return [target]
+    elif isinstance(target, List):
+        if isinstance(target[0], List):
+            for sub_list in target:
+                _verify_each_list_element_is_valid_location(sub_list)
+            return [loc for sub_list in target for loc in sub_list]
+
+        _verify_each_list_element_is_valid_location(target)
+        return target
+    else:
+        raise ValueError(
+            f"'{target}' is not a valid location for transfer."
+            f" Expected locations are of type 'Well', 'Location'"
+            f" or a one- or two-dimensional list of 'Well' or 'Location'."
+        )

api/src/opentrons/protocols/advanced_control/transfers/common.py

@@ -1,6 +1,14 @@
 """Common functions between v1 transfer and liquid-class-based transfer."""
+import enum
 from typing import Iterable, Generator, Tuple, TypeVar
 
+
+class TransferTipPolicyV2(enum.Enum):
+    ONCE = "once"
+    NEVER = "never"
+    ALWAYS = "always"
+
+
 Target = TypeVar("Target")
 
 

api/src/opentrons/protocols/advanced_control/transfers/transfer.py

@@ -22,7 +22,6 @@
 from . import common as tx_commons
 from ..common import Mix, MixOpts, MixStrategy
 
-
 AdvancedLiquidHandling = Union[
     Well,
     types.Location,

api/src/opentrons/protocols/advanced_control/transfers/transfer_liquid.py

@@ -0,0 +1,48 @@
+"""Steps builder for transfer, consolidate and distribute using liquid class."""
+import dataclasses
+from typing import (
+    Optional,
+    Dict,
+    Any,
+    Sequence,
+    Union,
+    TYPE_CHECKING,
+)
+
+from opentrons.protocol_api._liquid_properties import (
+    AspirateProperties,
+    SingleDispenseProperties,
+)
+from opentrons import types
+from .common import TransferTipPolicyV2
+
+# from opentrons.protocol_api.labware import Labware, Well
+#
+if TYPE_CHECKING:
+    from opentrons.protocol_api import TrashBin, WasteChute, Well, Labware
+#
+# AdvancedLiquidHandling = Union[
+#     Well,
+#     types.Location,
+#     Sequence[Union[Well, types.Location]],
+#     Sequence[Sequence[Well]],
+# ]
+
+
+@dataclasses.dataclass
+class TransferStep:
+    method: str
+    kwargs: Optional[Dict[str, Any]]
+
+
+def get_transfer_steps(
+    aspirate_properties: AspirateProperties,
+    single_dispense_properties: SingleDispenseProperties,
+    volume: float,
+    source: Sequence[Union[Well, types.Location]],
+    dest: Sequence[Union[Well, types.Location]],
+    trash_location: Union[Labware, types.Location, TrashBin, WasteChute],
+    new_tip: TransferTipPolicyV2,
+) -> None:
+    """Return the PAPI function steps to perform for this transfer."""
+    # TODO: check for valid volume params of disposal vol, air gap and max volume