diff --git a/api/src/opentrons/protocol_api/core/engine/instrument.py b/api/src/opentrons/protocol_api/core/engine/instrument.py index 010f3110fdb..acd4d9f68ed 100644 --- a/api/src/opentrons/protocol_api/core/engine/instrument.py +++ b/api/src/opentrons/protocol_api/core/engine/instrument.py @@ -2,13 +2,17 @@ from __future__ import annotations -from typing import Optional, TYPE_CHECKING, cast, Union, List +from typing import Optional, TYPE_CHECKING, cast, Union, List, Tuple from opentrons.types import Location, Mount, NozzleConfigurationType, NozzleMapInterface from opentrons.hardware_control import SyncHardwareAPI from opentrons.hardware_control.dev_types import PipetteDict from opentrons.protocols.api_support.util import FlowRates, find_value_for_api_version from opentrons.protocols.api_support.types import APIVersion -from opentrons.protocols.advanced_control.transfers.common import TransferTipPolicyV2 +from opentrons.protocols.advanced_control.transfers.common import ( + TransferTipPolicyV2, + check_valid_volume_parameters, + expand_for_volume_constraints, +) from opentrons.protocol_engine import commands as cmd from opentrons.protocol_engine import ( DeckPoint, @@ -38,6 +42,7 @@ ) from opentrons.protocol_api._nozzle_layout import NozzleLayout from . import overlap_versions, pipette_movement_conflict +from . import transfer_components_executor as tx_comps_executor from .well import WellCore from ..instrument import AbstractInstrument @@ -46,6 +51,7 @@ if TYPE_CHECKING: from .protocol import ProtocolCore from opentrons.protocol_api._liquid import LiquidClass + from opentrons.protocol_api._liquid_properties import TransferProperties _DISPENSE_VOLUME_VALIDATION_ADDED_IN = APIVersion(2, 17) @@ -892,16 +898,131 @@ def load_liquid_class( ) return result.liquidClassId + # TODO: update with getNextTip implementation + def get_next_tip(self) -> None: + """Get the next tip to pick up.""" + def transfer_liquid( self, - liquid_class_id: str, + liquid_class: LiquidClass, volume: float, - source: List[WellCore], - dest: List[WellCore], + source: List[Tuple[Location, WellCore]], + dest: List[Tuple[Location, WellCore]], new_tip: TransferTipPolicyV2, - trash_location: Union[WellCore, Location, TrashBin, WasteChute], + tiprack_uri: str, + tip_drop_location: Union[WellCore, Location, TrashBin, WasteChute], + ) -> None: + """Execute transfer using liquid class properties. + + Args: + liquid_class: The liquid class to use for transfer properties. + volume: Volume to transfer per well. + source: List of source wells, with each well represented as a tuple of + types.Location and WellCore. + types.Location is only necessary for saving the last accessed location. + dest: List of destination wells, with each well represented as a tuple of + types.Location and WellCore. + types.Location is only necessary for saving the last accessed location. + new_tip: Whether the transfer should use a new tip 'once', 'never', 'always', + or 'per source'. + tiprack_uri: The URI of the tiprack that the transfer settings are for. + tip_drop_location: Location where the tip will be dropped (if appropriate). + """ + # This function is WIP + # TODO: use the ID returned by load_liquid_class in command annotations + self.load_liquid_class( + liquid_class=liquid_class, + pipette_load_name=self.get_pipette_name(), # TODO: update this to use load name instead + tiprack_uri=tiprack_uri, + ) + transfer_props = liquid_class.get_for( + # update this to fetch load name instead + pipette=self.get_pipette_name(), + tiprack=tiprack_uri, + ) + aspirate_props = transfer_props.aspirate + + check_valid_volume_parameters( + disposal_volume=0, # No disposal volume for 1-to-1 transfer + air_gap=aspirate_props.retract.air_gap_by_volume.get_for_volume(volume), + max_volume=self.get_max_volume(), + ) + source_dest_per_volume_step = expand_for_volume_constraints( + volumes=[volume for _ in range(len(source))], + targets=zip(source, dest), + max_volume=self.get_max_volume(), + ) + if new_tip == TransferTipPolicyV2.ONCE: + # TODO: update this once getNextTip is implemented + self.get_next_tip() + for step_volume, (src, dest) in source_dest_per_volume_step: # type: ignore[assignment] + if new_tip == TransferTipPolicyV2.ALWAYS: + # TODO: update this once getNextTip is implemented + self.get_next_tip() + + # TODO: add aspirate and dispense + + if new_tip == TransferTipPolicyV2.ALWAYS: + if isinstance(tip_drop_location, (TrashBin, WasteChute)): + self.drop_tip_in_disposal_location( + disposal_location=tip_drop_location, + home_after=False, + alternate_tip_drop=True, + ) + elif isinstance(tip_drop_location, Location): + self.drop_tip( + location=tip_drop_location, + well_core=tip_drop_location.labware.as_well()._core, # type: ignore[arg-type] + home_after=False, + alternate_drop_location=True, + ) + + def aspirate_liquid_class( + self, + volume: float, + source: Tuple[Location, WellCore], + transfer_properties: TransferProperties, ) -> None: - """Execute transfer using liquid class properties.""" + """Execute aspiration steps. + + 1. Submerge + 2. Mix + 3. pre-wet + 4. Aspirate + 5. Delay- wait inside the liquid + 6. Aspirate retract + """ + aspirate_props = transfer_properties.aspirate + source_loc, source_well = source + aspirate_point = ( + tx_comps_executor.absolute_point_from_position_reference_and_offset( + well=source_well, + position_reference=aspirate_props.position_reference, + offset=aspirate_props.offset, + ) + ) + aspirate_location = Location(aspirate_point, labware=source_loc.labware) + + components_executer = tx_comps_executor.get_transfer_components_executor( + instrument_core=self, + transfer_properties=transfer_properties, + target_location=aspirate_location, + target_well=source_well, + ) + components_executer.submerge( + submerge_properties=aspirate_props.submerge, + # Assuming aspirate is not called with *liquid* in the tip + # TODO: evaluate if using the current volume to find air gap is not a good idea. + air_gap_volume=self.get_current_volume(), + ) + # TODO: when aspirating for consolidation, do not perform mix + components_executer.mix(mix_properties=aspirate_props.mix) + # TODO: when aspirating for consolidation, do not preform pre-wet + components_executer.pre_wet( + volume=volume, + ) + components_executer.aspirate_and_wait(volume=volume) + components_executer.retract_after_aspiration(volume=volume) def retract(self) -> None: """Retract this instrument to the top of the gantry.""" @@ -994,3 +1115,7 @@ def nozzle_configuration_valid_for_lld(self) -> bool: return self._engine_client.state.pipettes.get_nozzle_configuration_supports_lld( self.pipette_id ) + + def delay(self, seconds: float) -> None: + """Call a protocol delay.""" + self._protocol_core.delay(seconds=seconds, msg=None) diff --git a/api/src/opentrons/protocol_api/core/engine/transfer_components_executor.py b/api/src/opentrons/protocol_api/core/engine/transfer_components_executor.py new file mode 100644 index 00000000000..2e4dae08fab --- /dev/null +++ b/api/src/opentrons/protocol_api/core/engine/transfer_components_executor.py @@ -0,0 +1,295 @@ +"""Executor for liquid class based complex commands.""" +from __future__ import annotations +from typing import TYPE_CHECKING, Optional + +from opentrons_shared_data.liquid_classes.liquid_class_definition import ( + PositionReference, + Coordinate, +) + +from opentrons.protocol_api._liquid_properties import ( + Submerge, + TransferProperties, + MixProperties, +) +from opentrons.types import Location, Point + +if TYPE_CHECKING: + from .well import WellCore + from .instrument import InstrumentCore + + +class TransferComponentsExecutor: + def __init__( + self, + instrument_core: InstrumentCore, + transfer_properties: TransferProperties, + target_location: Location, + target_well: WellCore, + ) -> None: + self._instrument = instrument_core + self._transfer_properties = transfer_properties + self._target_location = target_location + self._target_well = target_well + + def submerge( + self, + submerge_properties: Submerge, + air_gap_volume: float, + ) -> None: + """Execute submerge steps. + + 1. move to position shown by positionReference + offset (should practically be a point outside/above the liquid). + Should raise an error if this point is inside the liquid? + For liquid meniscus this is easy to tell. Can’t be below meniscus + For reference pos of anything else, do not allow submerge position to be below aspirate position + 2. move to aspirate position at desired speed + 3. delay + """ + # TODO: compare submerge start position and aspirate position and raise error if incompatible + submerge_start_point = absolute_point_from_position_reference_and_offset( + well=self._target_well, + position_reference=submerge_properties.position_reference, + offset=submerge_properties.offset, + ) + submerge_start_location = Location( + point=submerge_start_point, labware=self._target_location.labware + ) + self._instrument.move_to( + location=submerge_start_location, + well_core=self._target_well, + force_direct=False, + minimum_z_height=None, + speed=None, + ) + self._remove_air_gap( + location=submerge_start_location, + volume=air_gap_volume, + ) + self._instrument.move_to( + location=self._target_location, + well_core=self._target_well, + force_direct=True, + minimum_z_height=None, + speed=submerge_properties.speed, + ) + if submerge_properties.delay.enabled: + assert submerge_properties.delay.duration is not None + self._instrument.delay(submerge_properties.delay.duration) + + def aspirate_and_wait(self, volume: float) -> None: + """Aspirate according to aspirate properties and wait if enabled.""" + # TODO: handle volume correction + aspirate_props = self._transfer_properties.aspirate + self._instrument.aspirate( + location=self._target_location, + well_core=None, + volume=volume, + rate=1, + flow_rate=aspirate_props.flow_rate_by_volume.get_for_volume(volume), + in_place=True, + is_meniscus=None, # TODO: update this once meniscus is implemented + ) + delay_props = aspirate_props.delay + if delay_props.enabled: + # Assertion only for mypy purposes + assert delay_props.duration is not None + self._instrument.delay(delay_props.duration) + + def dispense_and_wait(self, volume: float, push_out: Optional[float]) -> None: + """Dispense according to dispense properties and wait if enabled.""" + # TODO: handle volume correction + dispense_props = self._transfer_properties.dispense + self._instrument.dispense( + location=self._target_location, + well_core=None, + volume=volume, + rate=1, + flow_rate=dispense_props.flow_rate_by_volume.get_for_volume(volume), + in_place=True, + push_out=push_out, + is_meniscus=None, + ) + dispense_delay = dispense_props.delay + if dispense_delay.enabled: + assert dispense_delay.duration is not None + self._instrument.delay(dispense_delay.duration) + + def mix(self, mix_properties: MixProperties) -> None: + """Execute mix steps. + + 1. Use same flow rates and delays as aspirate and dispense + 2. Do [(aspirate + dispense) x repetitions] at the same position + 3. Do NOT push out at the end of dispense + 4. USE the delay property from aspirate & dispense during mix as well (flow rate and delay are coordinated with each other) + 5. Do not mix during consolidation + NOTE: For most of our built-in definitions, we will keep _mix_ off because it is a very application specific thing. + We should mention in our docs that users should adjust this property according to their application. + """ + if not mix_properties.enabled: + return + # Assertion only for mypy purposes + assert ( + mix_properties.repetitions is not None and mix_properties.volume is not None + ) + for n in range(mix_properties.repetitions): + self.aspirate_and_wait(volume=mix_properties.volume) + # TODO: Update to doing a push out at the end of mix for a post-dispense mix + self.dispense_and_wait(volume=mix_properties.volume, push_out=0) + + def pre_wet( + self, + volume: float, + ) -> None: + """Do a pre-wet. + + - 1 combo of aspirate + dispense at the same flow rate as specified in asp & disp and the delays in asp & disp + - Use the target volume/ volume we will be aspirating + - No push out + - No pre-wet for consolidation + """ + if not self._transfer_properties.aspirate.pre_wet: + return + mix_props = MixProperties(_enabled=True, _repetitions=1, _volume=volume) + self.mix(mix_properties=mix_props) + + def retract_after_aspiration(self, volume: float) -> None: + """Execute post-aspiration retraction steps. + + 1. Move TO the position reference+offset AT the specified speed + Raise error if retract is below aspirate position or below the meniscus + 2. Delay + 3. Touch tip + - Move to the Z offset position + - Touch tip to the sides at the specified speed (tip moves back to the center as part of touch tip) + - Return back to the retract position + 4. Air gap + - Air gap volume depends on the amount of liquid in the pipette + So if total aspirated volume is 20, use the value for airGapByVolume[20] + Flow rate = min(aspirateFlowRate, (airGapByVolume)/sec) + - Use post-aspirate delay + """ + # TODO: Raise error if retract is below the meniscus + retract_props = self._transfer_properties.aspirate.retract + retract_point = absolute_point_from_position_reference_and_offset( + well=self._target_well, + position_reference=retract_props.position_reference, + offset=retract_props.offset, + ) + retract_location = Location( + retract_point, labware=self._target_location.labware + ) + self._instrument.move_to( + location=retract_location, + well_core=self._target_well, + force_direct=True, + minimum_z_height=None, + speed=retract_props.speed, + ) + retract_delay = retract_props.delay + if retract_delay.enabled: + assert retract_delay.duration is not None + self._instrument.delay(retract_delay.duration) + touch_tip_props = retract_props.touch_tip + if touch_tip_props.enabled: + assert ( + touch_tip_props.speed is not None + and touch_tip_props.z_offset is not None + and touch_tip_props.mm_to_edge is not None + ) + # TODO: update this once touch tip has mmToEdge + self._instrument.touch_tip( + location=retract_location, + well_core=self._target_well, + radius=0, + z_offset=touch_tip_props.z_offset, + speed=touch_tip_props.speed, + ) + self._instrument.move_to( + location=retract_location, + well_core=self._target_well, + force_direct=True, + minimum_z_height=None, + # Full speed because the tip will already be out of the liquid + speed=None, + ) + self._add_air_gap( + air_gap_volume=self._transfer_properties.aspirate.retract.air_gap_by_volume.get_for_volume( + volume + ) + ) + + def _add_air_gap(self, air_gap_volume: float) -> None: + """Add an air gap.""" + aspirate_props = self._transfer_properties.aspirate + # The maximum flow rate should be air_gap_volume per second + flow_rate = min( + aspirate_props.flow_rate_by_volume.get_for_volume(air_gap_volume), + air_gap_volume, + ) + self._instrument.air_gap_in_place(volume=air_gap_volume, flow_rate=flow_rate) + delay_props = aspirate_props.delay + if delay_props.enabled: + # Assertion only for mypy purposes + assert delay_props.duration is not None + self._instrument.delay(delay_props.duration) + + def _remove_air_gap(self, location: Location, volume: float) -> None: + """Remove a previously added air gap.""" + dispense_props = self._transfer_properties.dispense + # The maximum flow rate should be air_gap_volume per second + flow_rate = min( + dispense_props.flow_rate_by_volume.get_for_volume(volume), volume + ) + self._instrument.dispense( + location=location, + well_core=None, + volume=volume, + rate=1, + flow_rate=flow_rate, + in_place=True, + is_meniscus=None, + push_out=0, + ) + dispense_delay = dispense_props.delay + if dispense_delay.enabled: + assert dispense_delay.duration is not None + self._instrument.delay(dispense_delay.duration) + + +def get_transfer_components_executor( + instrument_core: InstrumentCore, + transfer_properties: TransferProperties, + target_location: Location, + target_well: WellCore, +) -> TransferComponentsExecutor: + """Get a TransferComponentsExecutor.""" + return TransferComponentsExecutor( + instrument_core=instrument_core, + transfer_properties=transfer_properties, + target_location=target_location, + target_well=target_well, + ) + + +def absolute_point_from_position_reference_and_offset( + well: WellCore, + position_reference: PositionReference, + offset: Coordinate, +) -> Point: + """Return the absolute point, given the well, the position reference and offset.""" + match position_reference: + case PositionReference.WELL_TOP: + reference_point = well.get_top(0) + case PositionReference.WELL_BOTTOM: + reference_point = well.get_bottom(0) + case PositionReference.WELL_CENTER: + reference_point = well.get_center() + case PositionReference.LIQUID_MENISCUS: + raise NotImplementedError( + "Liquid transfer using liquid-meniscus relative positioning" + " is not yet implemented." + ) + case _: + raise ValueError(f"Unknown position reference {position_reference}") + return reference_point + Point(offset.x, offset.y, offset.z) diff --git a/api/src/opentrons/protocol_api/core/instrument.py b/api/src/opentrons/protocol_api/core/instrument.py index bc1ec3669df..08a0611e306 100644 --- a/api/src/opentrons/protocol_api/core/instrument.py +++ b/api/src/opentrons/protocol_api/core/instrument.py @@ -3,7 +3,7 @@ from __future__ import annotations from abc import abstractmethod, ABC -from typing import Any, Generic, Optional, TypeVar, Union, List +from typing import Any, Generic, Optional, TypeVar, Union, List, Tuple from opentrons import types from opentrons.hardware_control.dev_types import PipetteDict @@ -310,27 +310,15 @@ def configure_nozzle_layout( """ ... - @abstractmethod - def load_liquid_class( - self, - liquid_class: LiquidClass, - pipette_load_name: str, - tiprack_uri: str, - ) -> str: - """Load the liquid class properties of given pipette and tiprack into the engine. - - Returns: ID of the liquid class record - """ - ... - @abstractmethod def transfer_liquid( self, - liquid_class_id: str, + liquid_class: LiquidClass, volume: float, - source: List[WellCoreType], - dest: List[WellCoreType], + source: List[Tuple[types.Location, WellCoreType]], + dest: List[Tuple[types.Location, WellCoreType]], new_tip: TransferTipPolicyV2, + tiprack_uri: str, trash_location: Union[WellCoreType, types.Location, TrashBin, WasteChute], ) -> None: """Transfer a liquid from source to dest according to liquid class properties.""" diff --git a/api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py b/api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py index d2d25051d49..ec725934d90 100644 --- a/api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py +++ b/api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py @@ -1,7 +1,7 @@ from __future__ import annotations import logging -from typing import TYPE_CHECKING, Optional, Union, List +from typing import TYPE_CHECKING, Optional, Union, List, Tuple from opentrons import types from opentrons.hardware_control import CriticalPoint @@ -556,23 +556,14 @@ def configure_nozzle_layout( """This will never be called because it was added in API 2.16.""" pass - def load_liquid_class( - self, - liquid_class: LiquidClass, - pipette_load_name: str, - tiprack_uri: str, - ) -> str: - """This will never be called because it was added in ..""" - # TODO(spp, 2024-11-20): update the docstring and error to include API version - assert False, "load_liquid_class is not supported in legacy context" - def transfer_liquid( self, - liquid_class_id: str, + liquid_class: LiquidClass, volume: float, - source: List[LegacyWellCore], - dest: List[LegacyWellCore], + source: List[Tuple[types.Location, LegacyWellCore]], + dest: List[Tuple[types.Location, LegacyWellCore]], new_tip: TransferTipPolicyV2, + tiprack_uri: str, trash_location: Union[LegacyWellCore, types.Location, TrashBin, WasteChute], ) -> None: """This will never be called because it was added in ..""" diff --git a/api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py b/api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py index ec194874528..aacac9392a7 100644 --- a/api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py +++ b/api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py @@ -1,7 +1,7 @@ from __future__ import annotations import logging -from typing import TYPE_CHECKING, Optional, Union, List +from typing import TYPE_CHECKING, Optional, Union, List, Tuple from opentrons import types from opentrons.hardware_control.dev_types import PipetteDict @@ -474,23 +474,14 @@ def configure_nozzle_layout( """This will never be called because it was added in API 2.15.""" pass - def load_liquid_class( - self, - liquid_class: LiquidClass, - pipette_load_name: str, - tiprack_uri: str, - ) -> str: - """This will never be called because it was added in ..""" - # TODO(spp, 2024-11-20): update the docstring and error to include API version - assert False, "load_liquid_class is not supported in legacy context" - def transfer_liquid( self, - liquid_class_id: str, + liquid_class: LiquidClass, volume: float, - source: List[LegacyWellCore], - dest: List[LegacyWellCore], + source: List[Tuple[types.Location, LegacyWellCore]], + dest: List[Tuple[types.Location, LegacyWellCore]], new_tip: TransferTipPolicyV2, + tiprack_uri: str, trash_location: Union[LegacyWellCore, types.Location, TrashBin, WasteChute], ) -> None: """Transfer a liquid from source to dest according to liquid class properties.""" diff --git a/api/src/opentrons/protocol_api/instrument_context.py b/api/src/opentrons/protocol_api/instrument_context.py index 9c6338270c7..e2be0905774 100644 --- a/api/src/opentrons/protocol_api/instrument_context.py +++ b/api/src/opentrons/protocol_api/instrument_context.py @@ -1522,7 +1522,7 @@ def transfer_liquid( new_tip: TransferTipPolicyV2Type = "once", tip_drop_location: Optional[ Union[types.Location, labware.Well, TrashBin, WasteChute] - ] = None, # Maybe call this 'tip_drop_location' which is similar to PD + ] = None, ) -> InstrumentContext: """Transfer liquid from source to dest using the specified liquid class properties. @@ -1563,6 +1563,7 @@ def transfer_liquid( else: tiprack = self._last_tip_picked_up_from.parent else: + # TODO: update this with getNextTip result from engine tiprack, well = labware.next_available_tip( starting_tip=self.starting_tip, tip_racks=self.tip_racks, @@ -1591,25 +1592,25 @@ def transfer_liquid( tip_drop_location=_trash_location ) ) - liquid_class_id = self._core.load_liquid_class( - liquid_class=liquid_class, - pipette_load_name=self.name, - tiprack_uri=tiprack.uri, - ) - self._core.transfer_liquid( - liquid_class_id=liquid_class_id, + liquid_class=liquid_class, volume=volume, - source=[well._core for well in flat_sources_list], - dest=[well._core for well in flat_dests_list], + source=[ + (types.Location(types.Point(), labware=well), well._core) + for well in flat_sources_list + ], + dest=[ + (types.Location(types.Point(), labware=well), well._core) + for well in flat_dests_list + ], new_tip=valid_new_tip, + tiprack_uri=tiprack.uri, trash_location=( checked_trash_location._core if isinstance(checked_trash_location, labware.Well) else checked_trash_location ), ) - return self @requires_version(2, 0) diff --git a/api/tests/opentrons/conftest.py b/api/tests/opentrons/conftest.py index 7be480cfe0b..13c1c88b47d 100755 --- a/api/tests/opentrons/conftest.py +++ b/api/tests/opentrons/conftest.py @@ -54,6 +54,11 @@ MixProperties, TouchTipProperties, BlowoutProperties, + MixParams, + LiquidClassTouchTipParams, + MultiDispenseProperties, + BlowoutParams, + BlowoutLocation, ) from opentrons_shared_data.deck.types import ( RobotModel, @@ -810,7 +815,7 @@ def minimal_liquid_class_def2() -> LiquidClassSchemaV1: tiprack="opentrons_flex_96_tiprack_50ul", aspirate=AspirateProperties( submerge=Submerge( - positionReference=PositionReference.LIQUID_MENISCUS, + positionReference=PositionReference.WELL_TOP, offset=Coordinate(x=0, y=0, z=-5), speed=100, delay=DelayProperties( @@ -865,3 +870,144 @@ def minimal_liquid_class_def2() -> LiquidClassSchemaV1: ) ], ) + + +@pytest.fixture +def maximal_liquid_class_def() -> LiquidClassSchemaV1: + """Return a liquid class def with all properties enabled.""" + return LiquidClassSchemaV1( + liquidClassName="test_water", + displayName="Test Water", + schemaVersion=1, + namespace="opentrons", + byPipette=[ + ByPipetteSetting( + pipetteModel="flex_1channel_50", + byTipType=[ + ByTipTypeSetting( + tiprack="opentrons_flex_96_tiprack_50ul", + aspirate=AspirateProperties( + submerge=Submerge( + positionReference=PositionReference.WELL_TOP, + offset=Coordinate(x=1, y=2, z=3), + speed=100, + delay=DelayProperties( + enable=True, params=DelayParams(duration=10.0) + ), + ), + retract=RetractAspirate( + positionReference=PositionReference.WELL_TOP, + offset=Coordinate(x=3, y=2, z=1), + speed=50, + airGapByVolume=[(1.0, 0.1), (49.9, 0.1), (50.0, 0.0)], + touchTip=TouchTipProperties( + enable=True, + params=LiquidClassTouchTipParams( + zOffset=-1, mmToEdge=0.5, speed=30 + ), + ), + delay=DelayProperties( + enable=True, params=DelayParams(duration=20) + ), + ), + positionReference=PositionReference.WELL_BOTTOM, + offset=Coordinate(x=10, y=20, z=30), + flowRateByVolume=[(1.0, 35.0), (10.0, 24.0), (50.0, 35.0)], + correctionByVolume=[(0.0, 0.0)], + preWet=True, + mix=MixProperties( + enable=True, params=MixParams(repetitions=1, volume=50) + ), + delay=DelayProperties( + enable=True, params=DelayParams(duration=0.2) + ), + ), + singleDispense=SingleDispenseProperties( + submerge=Submerge( + positionReference=PositionReference.WELL_TOP, + offset=Coordinate(x=30, y=20, z=10), + speed=100, + delay=DelayProperties( + enable=True, params=DelayParams(duration=0.0) + ), + ), + retract=RetractDispense( + positionReference=PositionReference.WELL_TOP, + offset=Coordinate(x=11, y=22, z=33), + speed=50, + airGapByVolume=[(1.0, 0.1), (49.9, 0.1), (50.0, 0.0)], + blowout=BlowoutProperties( + enable=True, + params=BlowoutParams( + location=BlowoutLocation.SOURCE, + flowRate=100, + ), + ), + touchTip=TouchTipProperties( + enable=True, + params=LiquidClassTouchTipParams( + zOffset=-1, mmToEdge=0.5, speed=30 + ), + ), + delay=DelayProperties( + enable=False, params=DelayParams(duration=0) + ), + ), + positionReference=PositionReference.WELL_BOTTOM, + offset=Coordinate(x=33, y=22, z=11), + flowRateByVolume=[(1.0, 50.0)], + correctionByVolume=[(0.0, 0.0)], + mix=MixProperties( + enable=True, params=MixParams(repetitions=1, volume=50) + ), + pushOutByVolume=[ + (1.0, 7.0), + (4.999, 7.0), + (5.0, 2.0), + (10.0, 2.0), + (50.0, 2.0), + ], + delay=DelayProperties( + enable=True, params=DelayParams(duration=0.5) + ), + ), + multiDispense=MultiDispenseProperties( + submerge=Submerge( + positionReference=PositionReference.WELL_TOP, + offset=Coordinate(x=0, y=0, z=2), + speed=100, + delay=DelayProperties( + enable=False, params=DelayParams(duration=0.0) + ), + ), + retract=RetractDispense( + positionReference=PositionReference.WELL_TOP, + offset=Coordinate(x=2, y=3, z=1), + speed=50, + airGapByVolume=[(1.0, 0.1), (49.9, 0.1), (50.0, 0.0)], + blowout=BlowoutProperties(enable=False, params=None), + touchTip=TouchTipProperties( + enable=False, + params=LiquidClassTouchTipParams( + zOffset=-1, mmToEdge=0.5, speed=30 + ), + ), + delay=DelayProperties( + enable=False, params=DelayParams(duration=0) + ), + ), + positionReference=PositionReference.WELL_BOTTOM, + offset=Coordinate(x=1, y=3, z=2), + flowRateByVolume=[(50.0, 50.0)], + correctionByVolume=[(0.0, 0.0)], + conditioningByVolume=[(1.0, 5.0), (45.0, 5.0), (50.0, 0.0)], + disposalByVolume=[(1.0, 5.0), (45.0, 5.0), (50.0, 0.0)], + delay=DelayProperties( + enable=True, params=DelayParams(duration=0.2) + ), + ), + ) + ], + ) + ], + ) diff --git a/api/tests/opentrons/protocol_api/core/engine/test_instrument_core.py b/api/tests/opentrons/protocol_api/core/engine/test_instrument_core.py index 352dcb35c58..b8600416de3 100644 --- a/api/tests/opentrons/protocol_api/core/engine/test_instrument_core.py +++ b/api/tests/opentrons/protocol_api/core/engine/test_instrument_core.py @@ -7,6 +7,8 @@ from decoy import errors from opentrons_shared_data.liquid_classes.liquid_class_definition import ( LiquidClassSchemaV1, + PositionReference, + Coordinate, ) from opentrons_shared_data.pipette.types import PipetteNameType @@ -14,6 +16,10 @@ from opentrons.hardware_control import SyncHardwareAPI from opentrons.hardware_control.dev_types import PipetteDict from opentrons.protocol_api._liquid_properties import TransferProperties +from opentrons.protocol_api.core.engine import transfer_components_executor +from opentrons.protocol_api.core.engine.transfer_components_executor import ( + TransferComponentsExecutor, +) from opentrons.protocol_engine import ( DeckPoint, LoadedPipette, @@ -90,6 +96,34 @@ def patch_mock_pipette_movement_safety_check( ) +@pytest.fixture +def mock_transfer_components_executor( + decoy: Decoy, +) -> TransferComponentsExecutor: + """Get a mocked out TransferComponentsExecutor.""" + return decoy.mock(cls=TransferComponentsExecutor) + + +@pytest.fixture(autouse=True) +def patch_mock_transfer_components_executor( + decoy: Decoy, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Replace transfer_components_executor functions with mocks.""" + monkeypatch.setattr( + transfer_components_executor, + "get_transfer_components_executor", + decoy.mock(func=transfer_components_executor.get_transfer_components_executor), + ) + monkeypatch.setattr( + transfer_components_executor, + "absolute_point_from_position_reference_and_offset", + decoy.mock( + func=transfer_components_executor.absolute_point_from_position_reference_and_offset + ), + ) + + @pytest.fixture def subject( decoy: Decoy, @@ -1556,3 +1590,54 @@ def test_load_liquid_class( tiprack_uri="opentrons_flex_96_tiprack_50ul", ) assert result == "liquid-class-id" + + +def test_aspirate_liquid_class( + decoy: Decoy, + mock_engine_client: EngineClient, + subject: InstrumentCore, + minimal_liquid_class_def2: LiquidClassSchemaV1, + mock_transfer_components_executor: TransferComponentsExecutor, +) -> None: + """It should call aspirate sub-steps execution based on liquid class.""" + source_well = decoy.mock(cls=WellCore) + source_location = Location(Point(1, 2, 3), labware=None) + test_liquid_class = LiquidClass.create(minimal_liquid_class_def2) + test_transfer_properties = test_liquid_class.get_for( + "flex_1channel_50", "opentrons_flex_96_tiprack_50ul" + ) + decoy.when( + transfer_components_executor.absolute_point_from_position_reference_and_offset( + well=source_well, + position_reference=PositionReference.WELL_BOTTOM, + offset=Coordinate(x=0, y=0, z=-5), + ) + ).then_return(Point(1, 2, 3)) + decoy.when( + transfer_components_executor.get_transfer_components_executor( + instrument_core=subject, + transfer_properties=test_transfer_properties, + target_location=Location(Point(1, 2, 3), labware=None), + target_well=source_well, + ) + ).then_return(mock_transfer_components_executor) + decoy.when( + mock_engine_client.state.pipettes.get_aspirated_volume("abc123") + ).then_return(111) + subject.aspirate_liquid_class( + volume=123, + source=(source_location, source_well), + transfer_properties=test_transfer_properties, + ) + decoy.verify( + mock_transfer_components_executor.submerge( + submerge_properties=test_transfer_properties.aspirate.submerge, + air_gap_volume=111, + ), + mock_transfer_components_executor.mix( + mix_properties=test_transfer_properties.aspirate.mix + ), + mock_transfer_components_executor.pre_wet(volume=123), + mock_transfer_components_executor.aspirate_and_wait(volume=123), + mock_transfer_components_executor.retract_after_aspiration(volume=123), + ) diff --git a/api/tests/opentrons/protocol_api/core/engine/test_transfer_components_executor.py b/api/tests/opentrons/protocol_api/core/engine/test_transfer_components_executor.py new file mode 100644 index 00000000000..7be520b173b --- /dev/null +++ b/api/tests/opentrons/protocol_api/core/engine/test_transfer_components_executor.py @@ -0,0 +1,540 @@ +"""Tests for complex commands executor.""" +import pytest +from decoy import Decoy +from opentrons_shared_data.liquid_classes.liquid_class_definition import ( + LiquidClassSchemaV1, + PositionReference, + Coordinate, +) + +from opentrons.protocol_api._liquid import LiquidClass +from opentrons.protocol_api._liquid_properties import TransferProperties +from opentrons.protocol_api.core.engine.well import WellCore +from opentrons.protocol_api.core.engine.instrument import InstrumentCore +from opentrons.protocol_api.core.engine.transfer_components_executor import ( + TransferComponentsExecutor, + absolute_point_from_position_reference_and_offset, +) +from opentrons.types import Location, Point + + +@pytest.fixture +def mock_instrument_core(decoy: Decoy) -> InstrumentCore: + """Return a mocked out instrument core.""" + return decoy.mock(cls=InstrumentCore) + + +@pytest.fixture +def sample_transfer_props( + maximal_liquid_class_def: LiquidClassSchemaV1, +) -> TransferProperties: + """Return a mocked out liquid class fixture.""" + return LiquidClass.create(maximal_liquid_class_def).get_for( + pipette="flex_1channel_50", tiprack="opentrons_flex_96_tiprack_50ul" + ) + + +""" Test aspirate properties: +"submerge": { + "positionReference": "well-top", + "offset": {"x": 1, "y": 2, "z": 3}, + "speed": 100, + "delay": {"enable": true, "params": {"duration": 10.0}}}, +"retract": { + "positionReference": "well-top", + "offset": {"x": 3, "y": 2, "z": 1}, + "speed": 50, + "airGapByVolume": [[1.0, 0.1], [49.9, 0.1], [50.0, 0.0]], + "touchTip": {"enable": true, "params": {"zOffset": -1, "mmToEdge": 0.5, "speed": 30}}, + "delay": {"enable": true, "params": {"duration": 20}}}, +"positionReference": "well-bottom", +"offset": {"x": 10, "y": 20, "z": 30}, +"flowRateByVolume": [[1.0, 35.0], [10.0, 24.0], [50.0, 35.0]], +"correctionByVolume": [[0.0, 0.0]], +"preWet": true, +"mix": {"enable": true, "params": {"repetitions": 1, "volume": 50}}, +"delay": {"enable": true, "params": {"duration": 0.2}} +""" + + +def test_submerge( + decoy: Decoy, + mock_instrument_core: InstrumentCore, + sample_transfer_props: TransferProperties, +) -> None: + """Should perform the expected submerge steps.""" + source_well = decoy.mock(cls=WellCore) + well_top_point = Point(1, 2, 3) + well_bottom_point = Point(4, 5, 6) + air_gap_removal_flow_rate = ( + sample_transfer_props.dispense.flow_rate_by_volume.get_for_volume(123) + ) + + subject = TransferComponentsExecutor( + instrument_core=mock_instrument_core, + transfer_properties=sample_transfer_props, + target_location=Location(Point(), labware=None), + target_well=source_well, + ) + decoy.when(source_well.get_bottom(0)).then_return(well_bottom_point) + decoy.when(source_well.get_top(0)).then_return(well_top_point) + + subject.submerge( + submerge_properties=sample_transfer_props.aspirate.submerge, + air_gap_volume=123, + ) + + decoy.verify( + mock_instrument_core.move_to( + location=Location(Point(x=2, y=4, z=6), labware=None), + well_core=source_well, + force_direct=False, + minimum_z_height=None, + speed=None, + ), + mock_instrument_core.dispense( + location=Location(Point(x=2, y=4, z=6), labware=None), + well_core=None, + volume=123, + rate=1, + flow_rate=air_gap_removal_flow_rate, + in_place=True, + is_meniscus=None, + push_out=0, + ), + mock_instrument_core.delay(0.5), + mock_instrument_core.move_to( + location=Location(Point(), labware=None), + well_core=source_well, + force_direct=True, + minimum_z_height=None, + speed=100, + ), + mock_instrument_core.delay(10), + ) + + +def test_aspirate_and_wait( + decoy: Decoy, + mock_instrument_core: InstrumentCore, + sample_transfer_props: TransferProperties, +) -> None: + """It should execute an aspirate and a delay according to properties.""" + source_well = decoy.mock(cls=WellCore) + aspirate_flow_rate = ( + sample_transfer_props.aspirate.flow_rate_by_volume.get_for_volume(10) + ) + + subject = TransferComponentsExecutor( + instrument_core=mock_instrument_core, + transfer_properties=sample_transfer_props, + target_location=Location(Point(1, 2, 3), labware=None), + target_well=source_well, + ) + subject.aspirate_and_wait(volume=10) + decoy.verify( + mock_instrument_core.aspirate( + location=Location(Point(1, 2, 3), labware=None), + well_core=None, + volume=10, + rate=1, + flow_rate=aspirate_flow_rate, + in_place=True, + is_meniscus=None, + ), + mock_instrument_core.delay(0.2), + ) + + +def test_aspirate_and_wait_skips_delay( + decoy: Decoy, + mock_instrument_core: InstrumentCore, + sample_transfer_props: TransferProperties, +) -> None: + """It should skip the wait after aspirate.""" + sample_transfer_props.aspirate.delay.enabled = False + source_well = decoy.mock(cls=WellCore) + + subject = TransferComponentsExecutor( + instrument_core=mock_instrument_core, + transfer_properties=sample_transfer_props, + target_location=Location(Point(1, 2, 3), labware=None), + target_well=source_well, + ) + subject.aspirate_and_wait(volume=10) + decoy.verify( + mock_instrument_core.delay(0.2), + times=0, + ) + + +def test_dispense_and_wait( + decoy: Decoy, + mock_instrument_core: InstrumentCore, + sample_transfer_props: TransferProperties, +) -> None: + """It should execute a dispense and a delay according to properties.""" + source_well = decoy.mock(cls=WellCore) + dispense_flow_rate = ( + sample_transfer_props.dispense.flow_rate_by_volume.get_for_volume(10) + ) + + subject = TransferComponentsExecutor( + instrument_core=mock_instrument_core, + transfer_properties=sample_transfer_props, + target_location=Location(Point(1, 2, 3), labware=None), + target_well=source_well, + ) + subject.dispense_and_wait(volume=10, push_out=123) + decoy.verify( + mock_instrument_core.dispense( + location=Location(Point(1, 2, 3), labware=None), + well_core=None, + volume=10, + rate=1, + flow_rate=dispense_flow_rate, + in_place=True, + push_out=123, + is_meniscus=None, + ), + mock_instrument_core.delay(0.5), + ) + + +def test_dispense_and_wait_skips_delay( + decoy: Decoy, + mock_instrument_core: InstrumentCore, + sample_transfer_props: TransferProperties, +) -> None: + """It should skip the wait after dispense.""" + sample_transfer_props.dispense.delay.enabled = False + source_well = decoy.mock(cls=WellCore) + + subject = TransferComponentsExecutor( + instrument_core=mock_instrument_core, + transfer_properties=sample_transfer_props, + target_location=Location(Point(1, 2, 3), labware=None), + target_well=source_well, + ) + subject.dispense_and_wait(volume=10, push_out=123) + decoy.verify( + mock_instrument_core.delay(0.2), + times=0, + ) + + +def test_mix( + decoy: Decoy, + mock_instrument_core: InstrumentCore, + sample_transfer_props: TransferProperties, +) -> None: + """It should execute mix steps.""" + source_well = decoy.mock(cls=WellCore) + aspirate_flow_rate = ( + sample_transfer_props.aspirate.flow_rate_by_volume.get_for_volume(50) + ) + dispense_flow_rate = ( + sample_transfer_props.dispense.flow_rate_by_volume.get_for_volume(50) + ) + subject = TransferComponentsExecutor( + instrument_core=mock_instrument_core, + transfer_properties=sample_transfer_props, + target_location=Location(Point(1, 2, 3), labware=None), + target_well=source_well, + ) + subject.mix(mix_properties=sample_transfer_props.aspirate.mix) + + decoy.verify( + mock_instrument_core.aspirate( + location=Location(Point(1, 2, 3), labware=None), + well_core=None, + volume=50, + rate=1, + flow_rate=aspirate_flow_rate, + in_place=True, + is_meniscus=None, + ), + mock_instrument_core.delay(0.2), + mock_instrument_core.dispense( + location=Location(Point(1, 2, 3), labware=None), + well_core=None, + volume=50, + rate=1, + flow_rate=dispense_flow_rate, + in_place=True, + push_out=0, + is_meniscus=None, + ), + mock_instrument_core.delay(0.5), + ) + + +def test_mix_disabled( + decoy: Decoy, + mock_instrument_core: InstrumentCore, + sample_transfer_props: TransferProperties, +) -> None: + """It should not perform a mix when it is disabled.""" + sample_transfer_props.aspirate.mix.enabled = False + source_well = decoy.mock(cls=WellCore) + aspirate_flow_rate = ( + sample_transfer_props.aspirate.flow_rate_by_volume.get_for_volume(50) + ) + subject = TransferComponentsExecutor( + instrument_core=mock_instrument_core, + transfer_properties=sample_transfer_props, + target_location=Location(Point(1, 2, 3), labware=None), + target_well=source_well, + ) + subject.mix(mix_properties=sample_transfer_props.aspirate.mix) + decoy.verify( + mock_instrument_core.aspirate( + location=Location(Point(1, 2, 3), labware=None), + well_core=None, + volume=50, + rate=1, + flow_rate=aspirate_flow_rate, + in_place=True, + is_meniscus=None, + ), + times=0, + ) + + +def test_pre_wet( + decoy: Decoy, + mock_instrument_core: InstrumentCore, + sample_transfer_props: TransferProperties, +) -> None: + """It should execute pre-wet steps.""" + source_well = decoy.mock(cls=WellCore) + aspirate_flow_rate = ( + sample_transfer_props.aspirate.flow_rate_by_volume.get_for_volume(40) + ) + dispense_flow_rate = ( + sample_transfer_props.dispense.flow_rate_by_volume.get_for_volume(40) + ) + subject = TransferComponentsExecutor( + instrument_core=mock_instrument_core, + transfer_properties=sample_transfer_props, + target_location=Location(Point(1, 2, 3), labware=None), + target_well=source_well, + ) + subject.pre_wet(volume=40) + + decoy.verify( + mock_instrument_core.aspirate( + location=Location(Point(1, 2, 3), labware=None), + well_core=None, + volume=40, + rate=1, + flow_rate=aspirate_flow_rate, + in_place=True, + is_meniscus=None, + ), + mock_instrument_core.delay(0.2), + mock_instrument_core.dispense( + location=Location(Point(1, 2, 3), labware=None), + well_core=None, + volume=40, + rate=1, + flow_rate=dispense_flow_rate, + in_place=True, + push_out=0, + is_meniscus=None, + ), + mock_instrument_core.delay(0.5), + ) + + +def test_pre_wet_disabled( + decoy: Decoy, + mock_instrument_core: InstrumentCore, + sample_transfer_props: TransferProperties, +) -> None: + """It should NOT execute pre-wet steps.""" + source_well = decoy.mock(cls=WellCore) + sample_transfer_props.aspirate.pre_wet = False + aspirate_flow_rate = ( + sample_transfer_props.aspirate.flow_rate_by_volume.get_for_volume(40) + ) + subject = TransferComponentsExecutor( + instrument_core=mock_instrument_core, + transfer_properties=sample_transfer_props, + target_location=Location(Point(1, 2, 3), labware=None), + target_well=source_well, + ) + subject.pre_wet(volume=40) + + decoy.verify( + mock_instrument_core.aspirate( + location=Location(Point(1, 2, 3), labware=None), + well_core=None, + volume=40, + rate=1, + flow_rate=aspirate_flow_rate, + in_place=True, + is_meniscus=None, + ), + times=0, + ) + + +def test_retract_after_aspiration( + decoy: Decoy, + mock_instrument_core: InstrumentCore, + sample_transfer_props: TransferProperties, +) -> None: + """It should execute steps to retract from well after an aspiration.""" + source_well = decoy.mock(cls=WellCore) + well_top_point = Point(1, 2, 3) + well_bottom_point = Point(4, 5, 6) + + air_gap_volume = ( + sample_transfer_props.aspirate.retract.air_gap_by_volume.get_for_volume(40) + ) + + subject = TransferComponentsExecutor( + instrument_core=mock_instrument_core, + transfer_properties=sample_transfer_props, + target_location=Location(Point(1, 1, 1), labware=None), + target_well=source_well, + ) + decoy.when(source_well.get_bottom(0)).then_return(well_bottom_point) + decoy.when(source_well.get_top(0)).then_return(well_top_point) + + subject.retract_after_aspiration(volume=40) + + decoy.verify( + mock_instrument_core.move_to( + location=Location(Point(x=4, y=4, z=4), labware=None), + well_core=source_well, + force_direct=True, + minimum_z_height=None, + speed=50, + ), + mock_instrument_core.delay(20), + mock_instrument_core.touch_tip( + location=Location(Point(x=4, y=4, z=4), labware=None), + well_core=source_well, + radius=0, # Update this to use mmToEdge once implemented + z_offset=-1, + speed=30, + ), + mock_instrument_core.move_to( + location=Location(Point(x=4, y=4, z=4), labware=None), + well_core=source_well, + force_direct=True, + minimum_z_height=None, + speed=None, + ), + mock_instrument_core.air_gap_in_place( + volume=air_gap_volume, + flow_rate=air_gap_volume, + ), + mock_instrument_core.delay(0.2), + ) + + +def test_retract_after_aspiration_without_touch_tip_and_delay( + decoy: Decoy, + mock_instrument_core: InstrumentCore, + sample_transfer_props: TransferProperties, +) -> None: + """It should execute steps to retract from well after an aspiration without a touch tip or delay.""" + source_well = decoy.mock(cls=WellCore) + well_top_point = Point(1, 2, 3) + well_bottom_point = Point(4, 5, 6) + + sample_transfer_props.aspirate.retract.touch_tip.enabled = False + sample_transfer_props.aspirate.retract.delay.enabled = False + + air_gap_volume = ( + sample_transfer_props.aspirate.retract.air_gap_by_volume.get_for_volume(40) + ) + + subject = TransferComponentsExecutor( + instrument_core=mock_instrument_core, + transfer_properties=sample_transfer_props, + target_location=Location(Point(1, 1, 1), labware=None), + target_well=source_well, + ) + decoy.when(source_well.get_bottom(0)).then_return(well_bottom_point) + decoy.when(source_well.get_top(0)).then_return(well_top_point) + + subject.retract_after_aspiration(volume=40) + + decoy.verify( + mock_instrument_core.move_to( + location=Location(Point(x=4, y=4, z=4), labware=None), + well_core=source_well, + force_direct=True, + minimum_z_height=None, + speed=50, + ), + mock_instrument_core.air_gap_in_place( + volume=air_gap_volume, + flow_rate=air_gap_volume, + ), + mock_instrument_core.delay(0.2), + ) + + +@pytest.mark.parametrize( + argnames=["position_reference", "offset", "expected_result"], + argvalues=[ + (PositionReference.WELL_TOP, Coordinate(x=11, y=12, z=13), Point(12, 14, 16)), + ( + PositionReference.WELL_BOTTOM, + Coordinate(x=21, y=22, z=23), + Point(25, 27, 29), + ), + ( + PositionReference.WELL_CENTER, + Coordinate(x=31, y=32, z=33), + Point(38, 40, 42), + ), + ], +) +def test_absolute_point_from_position_reference_and_offset( + decoy: Decoy, + position_reference: PositionReference, + offset: Coordinate, + expected_result: Point, +) -> None: + """It should return the correct absolute point based on well, position reference and offset.""" + well = decoy.mock(cls=WellCore) + + well_top_point = Point(1, 2, 3) + well_bottom_point = Point(4, 5, 6) + well_center_point = Point(7, 8, 9) + decoy.when(well.get_bottom(0)).then_return(well_bottom_point) + decoy.when(well.get_top(0)).then_return(well_top_point) + decoy.when(well.get_center()).then_return(well_center_point) + + assert ( + absolute_point_from_position_reference_and_offset( + well=well, position_reference=position_reference, offset=offset + ) + == expected_result + ) + + +def test_absolute_point_from_position_reference_and_offset_raises_errors( + decoy: Decoy, +) -> None: + """It should raise errors for invalid input.""" + well = decoy.mock(cls=WellCore) + with pytest.raises(NotImplementedError): + absolute_point_from_position_reference_and_offset( + well=well, + position_reference=PositionReference.LIQUID_MENISCUS, + offset=Coordinate(x=0, y=0, z=0), + ) + + with pytest.raises(ValueError, match="Unknown position reference"): + absolute_point_from_position_reference_and_offset( + well=well, + position_reference="PositionReference", # type: ignore[arg-type] + offset=Coordinate(x=0, y=0, z=0), + ) diff --git a/api/tests/opentrons/protocol_api/test_instrument_context.py b/api/tests/opentrons/protocol_api/test_instrument_context.py index 3f639aff922..965ad406739 100644 --- a/api/tests/opentrons/protocol_api/test_instrument_context.py +++ b/api/tests/opentrons/protocol_api/test_instrument_context.py @@ -1978,14 +1978,6 @@ def test_transfer_liquid_delegates_to_engine_core( ).then_return(trash_location.move(Point(1, 2, 3))) decoy.when(next_tiprack.uri).then_return("tiprack-uri") decoy.when(mock_instrument_core.get_pipette_name()).then_return("pipette-name") - decoy.when( - mock_instrument_core.load_liquid_class( - liquid_class=test_liq_class, - pipette_load_name="pipette-name", - tiprack_uri="tiprack-uri", - ) - ).then_return("liq-class-id") - subject.transfer_liquid( liquid_class=test_liq_class, volume=10, @@ -1996,11 +1988,12 @@ def test_transfer_liquid_delegates_to_engine_core( ) decoy.verify( mock_instrument_core.transfer_liquid( - liquid_class_id="liq-class-id", + liquid_class=test_liq_class, volume=10, - source=[mock_well._core], - dest=[mock_well._core], + source=[(Location(Point(), labware=mock_well), mock_well._core)], + dest=[(Location(Point(), labware=mock_well), mock_well._core)], new_tip=TransferTipPolicyV2.ONCE, + tiprack_uri="tiprack-uri", trash_location=trash_location.move(Point(1, 2, 3)), ) ) diff --git a/api/tests/opentrons/protocols/advanced_control/transfers/test_common_functions.py b/api/tests/opentrons/protocols/advanced_control/transfers/test_common_functions.py index 644c2b7094f..b80667a302b 100644 --- a/api/tests/opentrons/protocols/advanced_control/transfers/test_common_functions.py +++ b/api/tests/opentrons/protocols/advanced_control/transfers/test_common_functions.py @@ -44,20 +44,20 @@ def test_check_valid_volume_parameters( argvalues=[ ( [60, 70, 75], - [("a", "b"), ("c", "d"), ("e", "f")], + [(("a", "b"), (1, 2)), (("c", "d"), (3, 4)), (("e", "f"), (5, 6))], 20, [ - (20, ("a", "b")), - (20, ("a", "b")), - (20, ("a", "b")), - (20, ("c", "d")), - (20, ("c", "d")), - (15, ("c", "d")), - (15, ("c", "d")), - (20, ("e", "f")), - (20, ("e", "f")), - (17.5, ("e", "f")), - (17.5, ("e", "f")), + (20, (("a", "b"), (1, 2))), + (20, (("a", "b"), (1, 2))), + (20, (("a", "b"), (1, 2))), + (20, (("c", "d"), (3, 4))), + (20, (("c", "d"), (3, 4))), + (15, (("c", "d"), (3, 4))), + (15, (("c", "d"), (3, 4))), + (20, (("e", "f"), (5, 6))), + (20, (("e", "f"), (5, 6))), + (17.5, (("e", "f"), (5, 6))), + (17.5, (("e", "f"), (5, 6))), ], ), ],