diff --git a/api/src/opentrons/protocol_api/instrument_context.py b/api/src/opentrons/protocol_api/instrument_context.py index 845c2756c6a..1798e043fc7 100644 --- a/api/src/opentrons/protocol_api/instrument_context.py +++ b/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, diff --git a/api/src/opentrons/protocol_api/validation.py b/api/src/opentrons/protocol_api/validation.py index 55048a30daa..fd5b4c64b92 100644 --- a/api/src/opentrons/protocol_api/validation.py +++ b/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): diff --git a/api/src/opentrons/protocols/advanced_control/transfers/transfer_liquid.py b/api/src/opentrons/protocols/advanced_control/transfers/transfer_liquid.py index 6cc9929e1f0..0b540776bc2 100644 --- a/api/src/opentrons/protocols/advanced_control/transfers/transfer_liquid.py +++ b/api/src/opentrons/protocols/advanced_control/transfers/transfer_liquid.py @@ -1,5 +1,8 @@ """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, @@ -7,20 +10,32 @@ 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 diff --git a/api/tests/opentrons/protocol_api/test_validation.py b/api/tests/opentrons/protocol_api/test_validation.py index e2970c7558e..8d80f2294cb 100644 --- a/api/tests/opentrons/protocol_api/test_validation.py +++ b/api/tests/opentrons/protocol_api/test_validation.py @@ -758,30 +758,17 @@ def test_ensure_new_tip_policy_raises() -> None: ( ["a"], pytest.raises( - ValueError, match="'a' is not a valid location for transfer. Should be" + ValueError, + match="'a' is not a valid location for transfer. Location should be of type 'Well'.", ), ), ( [["a"]], pytest.raises( - ValueError, match="'a' is not a valid location for transfer. Should be" + ValueError, + match="'a' is not a valid location for transfer. Location should be of type 'Well'.", ), ), - ( - [Location(point=Point(x=1, y=1, z=2), labware=None), "a"], - pytest.raises( - ValueError, match="'a' is not a valid location for transfer. Should be" - ), - ), - ( - [[Location(point=Point(x=1, y=1, z=2), labware=None)], ["a"]], - pytest.raises( - ValueError, match="'a' is not a valid location for transfer. Should be" - ), - ), - (Location(point=Point(x=1, y=1, z=2), labware=None), do_not_raise()), - ([Location(point=Point(x=1, y=1, z=2), labware=None)], do_not_raise()), - ([[Location(point=Point(x=1, y=1, z=2), labware=None)]], do_not_raise()), ], ) def test_ensure_valid_flat_wells_list_raises_for_invalid_targets( @@ -793,32 +780,36 @@ def test_ensure_valid_flat_wells_list_raises_for_invalid_targets( subject.ensure_valid_flat_wells_list(target) -sample_location1 = Location(point=Point(x=1, y=1, z=2), labware=None) -sample_location2 = Location(point=Point(x=2, y=1, z=2), labware=None) +def test_ensure_valid_flat_wells_list_raises_for_mixed_targets(decoy: Decoy) -> None: + """It should raise appropriate error if target has mixed valid and invalid wells.""" + target1 = [decoy.mock(cls=Well), "a"] + with pytest.raises( + ValueError, + match="'a' is not a valid location for transfer. Location should be of type 'Well'.", + ): + subject.ensure_valid_flat_wells_list(target1) # type: ignore[arg-type] + target2 = [[decoy.mock(cls=Well)], ["a"]] + with pytest.raises( + ValueError, + match="'a' is not a valid location for transfer. Location should be of type 'Well'.", + ): + subject.ensure_valid_flat_wells_list(target2) # type: ignore[arg-type] -@pytest.mark.parametrize( - ["target", "expected_result"], - [ - (sample_location1, [sample_location1]), - ([sample_location1, sample_location2], [sample_location1, sample_location2]), - ( - [ - [sample_location1, sample_location1], - [sample_location2, sample_location2], - ], - [sample_location1, sample_location1, sample_location2, sample_location2], - ), - ], -) -def test_ensure_valid_flat_wells_list( - target: Union[ - Well, - Location, - Sequence[Union[Well, Location]], - Sequence[Sequence[Well]], - ], - expected_result: Sequence[Union[Well, Location]], -) -> None: + +def test_ensure_valid_flat_wells_list(decoy: Decoy) -> None: """It should convert the locations to flat lists correctly.""" - assert subject.ensure_valid_flat_wells_list(target) == expected_result + target1 = decoy.mock(cls=Well) + target2 = decoy.mock(cls=Well) + + assert subject.ensure_valid_flat_wells_list(target1) == [target1] + assert subject.ensure_valid_flat_wells_list([target1, target2]) == [ + target1, + target2, + ] + assert subject.ensure_valid_flat_wells_list( + [ + [target1, target1], + [target2, target2], + ] + ) == [target1, target1, target2, target2] diff --git a/shared-data/liquid-class/fixtures/fixture_glycerol50.json b/shared-data/liquid-class/fixtures/fixture_glycerol50.json index 4b8a3480474..144c69c415b 100644 --- a/shared-data/liquid-class/fixtures/fixture_glycerol50.json +++ b/shared-data/liquid-class/fixtures/fixture_glycerol50.json @@ -11,7 +11,7 @@ "tiprack": "opentrons_96_tiprack_20ul", "aspirate": { "submerge": { - "positionReference": "liquid-meniscus", + "positionReference": "liquid-meniscus", # The starting point "offset": { "x": 0, "y": 0,