transfer builder partial

api/src/opentrons/protocol_api/instrument_context.py

@@ -1024,6 +1024,8 @@ def pick_up_tip(  # noqa: C901
         instrument.validate_tiprack(self.name, tip_rack, _log)
 
         move_to_location = move_to_location or well.top()
+
+        # prep_after should be true for transfer
         prep_after = (
             prep_after
             if prep_after is not None
@@ -1510,8 +1512,12 @@ def transfer_liquid(
         self,
         liquid_class: LiquidClass,
         volume: float,
-        source: AdvancedLiquidHandling,
-        dest: AdvancedLiquidHandling,
+        source: Union[
+            labware.Well, Sequence[labware.Well], Sequence[Sequence[labware.Well]]
+        ],
+        dest: Union[
+            labware.Well, Sequence[labware.Well], Sequence[Sequence[labware.Well]]
+        ],
         new_tip: Literal["once", "always", "never"] = "once",
         trash_location: Optional[Union[types.Location, TrashBin, WasteChute]] = None,
     ) -> InstrumentContext:
@@ -1523,6 +1529,7 @@ def transfer_liquid(
         ):
             raise NotImplementedError("This method is not implemented.")
 
+        # TODO: verify source/dest is not trash bin/waste chute
         flat_sources_list = validation.ensure_valid_flat_wells_list(source)
         flat_dest_list = validation.ensure_valid_flat_wells_list(dest)
 
@@ -1556,6 +1563,7 @@ def transfer_liquid(
                 " Ensure that all previously aspirated liquid is dispensed before starting"
                 " a new transfer."
             )
+        # TODO (spp, 2024-11-18): verify that all tipracks being used will be the same
         liquid_class_props = liquid_class.get_for(
             pipette=self.name, tiprack=tiprack.name
         )
@@ -1569,17 +1577,20 @@ def transfer_liquid(
         else:
             checked_trash_location = trash_location
 
-        v2_transfer.get_transfer_steps(
+        command_builder = v2_transfer.ComplexCommandBuilder()
+        command_builder.build_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,
-        )
+            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,
+            instrument_info=)
+        # self._execute_transfer(plan_steps)
         return self
 
+    def _execute_transfer_liquid(self, plan: v1_transfer.TransferPlan) -> None:
+        for cmd in plan:
+            getattr(self, cmd["method"])(**cmd["kwargs"])
+
     @requires_version(2, 0)
     def delay(self, *args: Any, **kwargs: Any) -> None:
         """
@@ -1604,6 +1615,13 @@ def delay(self, *args: Any, **kwargs: Any) -> None:
             # that a protocol out in the wild does it, for some reason.
             pass
 
+    def _delay(self, seconds: float) -> None:
+        """Call a protocol core delay.
+
+        *** For internal use only.***
+        """
+        self._protocol_core.delay(seconds=seconds, msg=None)
+
     @requires_version(2, 0)
     def move_to(
         self,

api/src/opentrons/protocol_api/validation.py

@@ -649,30 +649,27 @@ def ensure_new_tip_policy(value: str) -> TransferTipPolicyV2:
         )
 
 
-def _verify_each_list_element_is_valid_location(
-    locations: Sequence[Union[Well, Location]]
-) -> None:
+def _verify_each_list_element_is_valid_location(locations: Sequence[Well]) -> None:
     from .labware import Well
 
     for loc in locations:
-        if not (isinstance(loc, Well) or isinstance(loc, Location)):
+        if not isinstance(loc, Well):
             raise ValueError(
-                f"'{loc}' is not a valid location for transfer. Should be of type 'Well' or 'Location'"
+                f"'{loc}' is not a valid location for transfer. Location should be of type 'Well'."
             )
 
 
 def ensure_valid_flat_wells_list(
     target: Union[
         Well,
-        Location,
-        Sequence[Union[Well, Location]],
+        Sequence[Well],
         Sequence[Sequence[Well]],
     ],
-) -> Sequence[Union[Well, Location]]:
+) -> Sequence[Well]:
     """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):
+    if isinstance(target, Well):
         return [target]
     elif isinstance(target, List):
         if isinstance(target[0], List):

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

@@ -1,26 +1,41 @@
 """Steps builder for transfer, consolidate and distribute using liquid class."""
-import dataclasses
+from __future__ import annotations
+from dataclasses import (
+    dataclass,
+)
 from typing import (
     Optional,
     Dict,
     Any,
     Sequence,
     Union,
     TYPE_CHECKING,
+    List,
+    Iterator,
+)
+
+from opentrons_shared_data.liquid_classes.liquid_class_definition import (
+    PositionReference, Coordinate
 )
 
 from opentrons.protocol_api._liquid_properties import (
     AspirateProperties,
     SingleDispenseProperties,
+    DelayProperties,
+    TouchTipProperties,
+    BlowoutProperties,
 )
 from opentrons import types
-from .common import TransferTipPolicyV2
+from opentrons.types import NozzleMapInterface
+from .common import (
+    TransferTipPolicyV2,
+    expand_for_volume_constraints,
+    check_valid_volume_parameters,
+)
 
-# 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,
@@ -29,20 +44,215 @@
 # ]
 
 
-@dataclasses.dataclass
+@dataclass
+class PipetteAndTipStateInfo:
+    max_volume: float
+    pipette_channels: int
+    nozzle_configuration: NozzleMapInterface
+
+
+@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
+@dataclass
+class BaseKwargs:
+    ...
+
+    def dict(self) -> Dict[str, Any]:
+        """As dictionary"""
+        return self.__dict__
+
+
+@dataclass
+class MoveToArgs(BaseKwargs):
+    location: types.Location
+    speed: Optional[float]
+
+
+@dataclass
+class DelayArgs(BaseKwargs):
+    seconds: float
+
+
+@dataclass
+class PickUpTipArgs(BaseKwargs):
+    location: Union[types.Location, Well, Labware, None]
+    presses: None = None
+    increment: None = None
+    prep_after: bool = True
+
+
+@dataclass
+class DropTipArgs(BaseKwargs):
+    location: Optional[
+            Union[
+                types.Location,
+                Well,
+                TrashBin,
+                WasteChute,
+            ]
+        ] = None,
+    home_after: bool = True
+
+
+@dataclass
+class MixArgs(BaseKwargs):
+    repetitions: int = 1
+    volume: None = None
+    location: None = None
+    rate: float = 1
+
+
+class ComplexCommandBuilder:
+    """Builder for transfer/ distribute/ consolidate steps."""
+
+    def __init__(self) -> None:
+        """Initialize complex command builder."""
+        self._aspirate_steps_builder = AspirateStepsBuilder()
+
+    def build_transfer_steps(
+        self,
+        aspirate_properties: AspirateProperties,
+        single_dispense_properties: SingleDispenseProperties,
+        volume: float,
+        source: Sequence[Well],
+        dest: Sequence[Well],
+        trash_location: Union[types.Location, TrashBin, WasteChute],
+        new_tip: TransferTipPolicyV2,
+        instrument_info: PipetteAndTipStateInfo,
+    ) -> Iterator[TransferStep]:
+        """Build steps for the transfer and return an iterator for them."""
+        check_valid_volume_parameters(
+            disposal_volume=0,  # No disposal volume for 1-to-1 transfer
+            air_gap=aspirate_properties.retract.air_gap_by_volume.get_for_volume(volume),
+            max_volume=instrument_info.max_volume,
+        )
+        source_dest_per_volume_step = expand_for_volume_constraints(
+            volumes=[volume for _ in range(len(source))],
+            targets=zip(source, dest),
+            max_volume=instrument_info.max_volume
+        )
+        if new_tip == TransferTipPolicyV2.ONCE:
+            yield TransferStep(
+                method="pick_up_tip",
+                kwargs=PickUpTipArgs(location=None).dict()
+            )
+        for step_volume, (src, dest) in source_dest_per_volume_step:
+            if new_tip == TransferTipPolicyV2.ALWAYS:
+                yield TransferStep(
+                    method="pick_up_tip",
+                    kwargs=PickUpTipArgs(location=None).dict()
+                )
+            yield from self._aspirate_steps_builder.build_aspirate_steps(
+                source=src,
+                aspirate_properties=aspirate_properties
+            )
+            # TODO: add dispense step builder
+            if new_tip == TransferTipPolicyV2.ALWAYS:
+                yield TransferStep(
+                    method="drop_tip",
+                    kwargs=DropTipArgs(location=trash_location).dict()
+                )
+
+
+class AspirateStepsBuilder:
+    """Builder for all steps associated with aspiration."""
+
+    def __init__(self) -> None:
+        """Initialize AspirateStepsBuilder."""
+        self._submerge_steps_builder = SubmergeStepsBuilder()
+        self._retract_steps_builder = RetractStepsBuilder()
+
+    def build_aspirate_steps(
+        self,
+        source: Well,
+        aspirate_properties: AspirateProperties,
+    ) -> Iterator[TransferStep]:
+        """Build steps associated with aspiration."""
+        yield self._submerge_steps_builder.build_submerge_steps(
+            target_well=source,
+            position_reference=aspirate_properties.submerge.position_reference,
+            offset=aspirate_properties.submerge.offset,
+            speed=aspirate_properties.submerge.speed,
+            delay=aspirate_properties.submerge.delay,
+        )
+        if aspirate_properties.mix.enabled:
+            yield TransferStep(
+                method="mix",
+                kwargs=MixArgs(
+                    repetitions=aspirate_properties.mix.repetitions,
+                    volume=aspirate_properties.mix.volume,
+                ).dict()
+            )
+        if aspirate_properties.pre_wet
+
+
+
+class SubmergeStepsBuilder:
+    """Class for building submerge steps."""
+
+    def __init__(self) -> None:
+        """Initialize SubmergeStepsBuilder."""
+        pass
+
+    def build_submerge_steps(
+        self,
+        target_well: Well,
+        position_reference: PositionReference,
+        offset: Coordinate,
+        speed: float,
+        delay: DelayProperties,
+    ) -> Iterator[TransferStep]:
+        """Build steps associated with submerging the pipette."""
+        # Move to top of well
+        yield TransferStep(
+            method="move_to",
+            kwargs=MoveToArgs(location=target_well.top(), speed=None).dict()
+        )
+
+        # Move to submerge position inside well at given speed
+        offset_point = types.Point(offset.x, offset.y, offset.z)
+        if position_reference == PositionReference.WELL_TOP:
+            well_position = target_well.top().move(offset_point)
+        elif position_reference == PositionReference.WELL_BOTTOM:
+            well_position = target_well.bottom().move(offset_point)
+        elif position_reference == PositionReference.WELL_CENTER:
+            well_position = target_well.center().move(offset_point)
+        else:
+            raise NotImplementedError(
+                "Only position reference of WELL_TOP, WELL_BOTTOM and WELL_CENTER is implemented."
+            )
+        yield TransferStep(
+            method="move_to",
+            kwargs=MoveToArgs(location=well_position, speed=speed).dict(),
+        )
+        # Delay
+        if delay.enabled:
+            yield TransferStep(
+                method="_delay",
+                kwargs=DelayArgs(seconds=delay.duration).dict()
+            )
+
+
+class RetractStepsBuilder:
+    """Class for building retraction steps."""
+
+    def __init__(self) -> None:
+        """Initialize RetractStepsBuilder."""
+        pass
+
+    def build_retract_steps(
+        self,
+        position_reference: PositionReference,
+        offset: Coordinate,
+        speed: float,
+        delay: DelayProperties,
+        air_gap: float,
+        touch_tip: TouchTipProperties,
+        blow_out: Optional[BlowoutProperties],
+    ) -> Iterator[TransferStep]:
+        """Build steps associated with retracting the pipette."""
+        pass