diff --git a/api/src/opentrons/protocol_api/_liquid_properties.py b/api/src/opentrons/protocol_api/_liquid_properties.py index 5aaed51edbe..8b99a5f00cf 100644 --- a/api/src/opentrons/protocol_api/_liquid_properties.py +++ b/api/src/opentrons/protocol_api/_liquid_properties.py @@ -1,15 +1,19 @@ from dataclasses import dataclass from numpy import interp -from typing import Optional, Dict, Sequence, Tuple +from typing import Optional, Dict, Sequence, Tuple, List from opentrons_shared_data.liquid_classes.liquid_class_definition import ( AspirateProperties as SharedDataAspirateProperties, SingleDispenseProperties as SharedDataSingleDispenseProperties, MultiDispenseProperties as SharedDataMultiDispenseProperties, DelayProperties as SharedDataDelayProperties, + DelayParams as SharedDataDelayParams, TouchTipProperties as SharedDataTouchTipProperties, + LiquidClassTouchTipParams as SharedDataTouchTipParams, MixProperties as SharedDataMixProperties, + MixParams as SharedDataMixParams, BlowoutProperties as SharedDataBlowoutProperties, + BlowoutParams as SharedDataBlowoutParams, ByTipTypeSetting as SharedByTipTypeSetting, Submerge as SharedDataSubmerge, RetractAspirate as SharedDataRetractAspirate, @@ -37,6 +41,10 @@ def as_dict(self) -> Dict[float, float]: """Get a dictionary representation of all set volumes and values along with the default.""" return self._properties_by_volume + def as_list_of_tuples(self) -> List[Tuple[float, float]]: + """Get as list of tuples.""" + return [(k, v) for k, v in self._properties_by_volume.items()] + def get_for_volume(self, volume: float) -> float: """Get a value by volume for this property. Volumes not defined will be interpolated between set volumes.""" validated_volume = validation.ensure_positive_float(volume) @@ -101,6 +109,14 @@ def duration(self, new_duration: float) -> None: validated_duration = validation.ensure_positive_float(new_duration) self._duration = validated_duration + def as_schema_v1_model(self) -> SharedDataDelayProperties: + return SharedDataDelayProperties( + enable=self._enabled, + params=SharedDataDelayParams(duration=self.duration) + if self.duration is not None + else None, + ) + @dataclass class TouchTipProperties: @@ -152,6 +168,27 @@ def speed(self, new_speed: float) -> None: validated_speed = validation.ensure_positive_float(new_speed) self._speed = validated_speed + def _get_schema_v1_params(self) -> Optional[SharedDataTouchTipParams]: + """Get the touch tip params in schema v1 shape.""" + if ( + self._z_offset is not None + and self._mm_to_edge is not None + and self._speed is not None + ): + return SharedDataTouchTipParams( + zOffset=self._z_offset, + mmToEdge=self._mm_to_edge, + speed=self._speed, + ) + else: + return None + + def as_schema_v1_model(self) -> SharedDataTouchTipProperties: + return SharedDataTouchTipProperties( + enable=self._enabled, + params=self._get_schema_v1_params(), + ) + @dataclass class MixProperties: @@ -189,6 +226,22 @@ def volume(self, new_volume: float) -> None: validated_volume = validation.ensure_positive_float(new_volume) self._volume = validated_volume + def _get_schema_v1_params(self) -> Optional[SharedDataMixParams]: + """Get the mix params in schema v1 shape.""" + if self._repetitions is not None and self._volume is not None: + return SharedDataMixParams( + repetitions=self._repetitions, + volume=self._volume, + ) + else: + return None + + def as_schema_v1_model(self) -> SharedDataMixProperties: + return SharedDataMixProperties( + enable=self._enabled, + params=self._get_schema_v1_params(), + ) + @dataclass class BlowoutProperties: @@ -227,6 +280,22 @@ def flow_rate(self, new_flow_rate: float) -> None: validated_flow_rate = validation.ensure_positive_float(new_flow_rate) self._flow_rate = validated_flow_rate + def _get_schema_v1_params(self) -> Optional[SharedDataBlowoutParams]: + """Get the mix params in schema v1 shape.""" + if self._location is not None and self._flow_rate is not None: + return SharedDataBlowoutParams( + location=self._location, + flowRate=self._flow_rate, + ) + else: + return None + + def as_schema_v1_model(self) -> SharedDataBlowoutProperties: + return SharedDataBlowoutProperties( + enable=self._enabled, + params=self._get_schema_v1_params(), + ) + @dataclass class SubmergeRetractCommon: @@ -271,6 +340,14 @@ def delay(self) -> DelayProperties: class Submerge(SubmergeRetractCommon): ... + def as_schema_v1_model(self) -> SharedDataSubmerge: + return SharedDataSubmerge( + positionReference=self._position_reference, + offset=self._offset, + speed=self._speed, + delay=self._delay.as_schema_v1_model(), + ) + @dataclass class RetractAspirate(SubmergeRetractCommon): @@ -286,6 +363,16 @@ def air_gap_by_volume(self) -> LiquidHandlingPropertyByVolume: def touch_tip(self) -> TouchTipProperties: return self._touch_tip + def as_schema_v1_model(self) -> SharedDataRetractAspirate: + return SharedDataRetractAspirate( + positionReference=self._position_reference, + offset=self._offset, + speed=self._speed, + airGapByVolume=self._air_gap_by_volume.as_list_of_tuples(), + touchTip=self._touch_tip.as_schema_v1_model(), + delay=self._delay.as_schema_v1_model(), + ) + @dataclass class RetractDispense(SubmergeRetractCommon): @@ -306,6 +393,17 @@ def touch_tip(self) -> TouchTipProperties: def blowout(self) -> BlowoutProperties: return self._blowout + def as_schema_v1_model(self) -> SharedDataRetractDispense: + return SharedDataRetractDispense( + positionReference=self._position_reference, + offset=self._offset, + speed=self._speed, + airGapByVolume=self._air_gap_by_volume.as_list_of_tuples(), + blowout=self._blowout.as_schema_v1_model(), + touchTip=self._touch_tip.as_schema_v1_model(), + delay=self._delay.as_schema_v1_model(), + ) + @dataclass class BaseLiquidHandlingProperties: @@ -370,6 +468,18 @@ def retract(self) -> RetractAspirate: def mix(self) -> MixProperties: return self._mix + def as_schema_v1_model(self) -> SharedDataAspirateProperties: + return SharedDataAspirateProperties( + submerge=self._submerge.as_schema_v1_model(), + retract=self._retract.as_schema_v1_model(), + positionReference=self._position_reference, + offset=self._offset, + flowRateByVolume=self._flow_rate_by_volume.as_list_of_tuples(), + preWet=self._pre_wet, + mix=self._mix.as_schema_v1_model(), + delay=self._delay.as_schema_v1_model(), + ) + @dataclass class SingleDispenseProperties(BaseLiquidHandlingProperties): @@ -390,6 +500,18 @@ def retract(self) -> RetractDispense: def mix(self) -> MixProperties: return self._mix + def as_schema_v1_model(self) -> SharedDataSingleDispenseProperties: + return SharedDataSingleDispenseProperties( + submerge=self._submerge.as_schema_v1_model(), + retract=self._retract.as_schema_v1_model(), + positionReference=self._position_reference, + offset=self._offset, + flowRateByVolume=self._flow_rate_by_volume.as_list_of_tuples(), + mix=self._mix.as_schema_v1_model(), + pushOutByVolume=self._push_out_by_volume.as_list_of_tuples(), + delay=self._delay.as_schema_v1_model(), + ) + @dataclass class MultiDispenseProperties(BaseLiquidHandlingProperties): @@ -410,6 +532,18 @@ def conditioning_by_volume(self) -> LiquidHandlingPropertyByVolume: def disposal_by_volume(self) -> LiquidHandlingPropertyByVolume: return self._disposal_by_volume + def as_schema_v1_model(self) -> SharedDataMultiDispenseProperties: + return SharedDataMultiDispenseProperties( + submerge=self._submerge.as_schema_v1_model(), + retract=self._retract.as_schema_v1_model(), + positionReference=self._position_reference, + offset=self._offset, + flowRateByVolume=self._flow_rate_by_volume.as_list_of_tuples(), + conditioningByVolume=self._conditioning_by_volume.as_list_of_tuples(), + disposalByVolume=self._disposal_by_volume.as_list_of_tuples(), + delay=self._delay.as_schema_v1_model(), + ) + @dataclass class TransferProperties: diff --git a/api/src/opentrons/protocol_api/core/engine/instrument.py b/api/src/opentrons/protocol_api/core/engine/instrument.py index 8fc707541f0..d7682848327 100644 --- a/api/src/opentrons/protocol_api/core/engine/instrument.py +++ b/api/src/opentrons/protocol_api/core/engine/instrument.py @@ -2,13 +2,13 @@ from __future__ import annotations -from typing import Optional, TYPE_CHECKING, cast, Union -from opentrons.protocols.api_support.types import APIVersion - +from typing import Optional, TYPE_CHECKING, cast, Union, List 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.protocol_engine import commands as cmd from opentrons.protocol_engine import ( DeckPoint, @@ -27,6 +27,7 @@ PRIMARY_NOZZLE_LITERAL, NozzleLayoutConfigurationType, AddressableOffsetVector, + LiquidClassRecord, ) from opentrons.protocol_engine.errors.exceptions import TipNotAttachedError from opentrons.protocol_engine.clients import SyncClient as EngineClient @@ -38,14 +39,13 @@ from opentrons.protocol_api._nozzle_layout import NozzleLayout from . import overlap_versions, pipette_movement_conflict -from ..instrument import AbstractInstrument from .well import WellCore - +from ..instrument import AbstractInstrument from ...disposal_locations import TrashBin, WasteChute if TYPE_CHECKING: from .protocol import ProtocolCore - + from opentrons.protocol_api._liquid import LiquidClass _DISPENSE_VOLUME_VALIDATION_ADDED_IN = APIVersion(2, 17) @@ -864,6 +864,45 @@ def configure_nozzle_layout( ) ) + def load_liquid_class( + self, + liquid_class: LiquidClass, + pipette_load_name: str, + tiprack_uri: str, + ) -> str: + """Load a liquid class into the engine and return its ID.""" + transfer_props = liquid_class.get_for( + pipette=pipette_load_name, tiprack=tiprack_uri + ) + + liquid_class_record = LiquidClassRecord( + liquidClassName=liquid_class.name, + pipetteModel=self.get_model(), # TODO: verify this is the correct 'model' to use + tiprack=tiprack_uri, + aspirate=transfer_props.aspirate.as_schema_v1_model(), + singleDispense=transfer_props.dispense.as_schema_v1_model(), + multiDispense=transfer_props.multi_dispense.as_schema_v1_model() + if transfer_props.multi_dispense + else None, + ) + result = self._engine_client.execute_command_without_recovery( + cmd.LoadLiquidClassParams( + liquidClassRecord=liquid_class_record, + ) + ) + return result.liquidClassId + + def transfer_liquid( + self, + liquid_class_id: str, + volume: float, + source: List[WellCore], + dest: List[WellCore], + new_tip: TransferTipPolicyV2, + trash_location: Union[WellCore, Location, TrashBin, WasteChute], + ) -> None: + """Execute transfer using liquid class properties.""" + def retract(self) -> None: """Retract this instrument to the top of the gantry.""" z_axis = self._engine_client.state.pipettes.get_z_axis(self._pipette_id) diff --git a/api/src/opentrons/protocol_api/core/instrument.py b/api/src/opentrons/protocol_api/core/instrument.py index f110bde928d..bc1ec3669df 100644 --- a/api/src/opentrons/protocol_api/core/instrument.py +++ b/api/src/opentrons/protocol_api/core/instrument.py @@ -3,13 +3,14 @@ from __future__ import annotations from abc import abstractmethod, ABC -from typing import Any, Generic, Optional, TypeVar, Union +from typing import Any, Generic, Optional, TypeVar, Union, List from opentrons import types from opentrons.hardware_control.dev_types import PipetteDict from opentrons.protocols.api_support.util import FlowRates +from opentrons.protocols.advanced_control.transfers.common import TransferTipPolicyV2 from opentrons.protocol_api._nozzle_layout import NozzleLayout - +from opentrons.protocol_api._liquid import LiquidClass from ..disposal_locations import TrashBin, WasteChute from .well import WellCoreType @@ -309,6 +310,32 @@ 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, + volume: float, + source: List[WellCoreType], + dest: List[WellCoreType], + new_tip: TransferTipPolicyV2, + trash_location: Union[WellCoreType, types.Location, TrashBin, WasteChute], + ) -> None: + """Transfer a liquid from source to dest according to liquid class properties.""" + ... + @abstractmethod def is_tip_tracking_available(self) -> bool: """Return whether auto tip tracking is available for the pipette's current nozzle configuration.""" 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 76d49b40557..d2d25051d49 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,12 +1,13 @@ from __future__ import annotations import logging -from typing import TYPE_CHECKING, Optional, Union +from typing import TYPE_CHECKING, Optional, Union, List from opentrons import types from opentrons.hardware_control import CriticalPoint from opentrons.hardware_control.dev_types import PipetteDict from opentrons.protocol_api.core.common import WellCore +from opentrons.protocols.advanced_control.transfers.common import TransferTipPolicyV2 from opentrons.protocols.api_support import instrument as instrument_support from opentrons.protocols.api_support.definitions import MAX_SUPPORTED_VERSION from opentrons.protocols.api_support.labware_like import LabwareLike @@ -19,6 +20,7 @@ ) from opentrons.protocols.geometry import planning from opentrons.protocol_api._nozzle_layout import NozzleLayout +from opentrons.protocol_api._liquid import LiquidClass from ...disposal_locations import TrashBin, WasteChute from ..instrument import AbstractInstrument @@ -554,6 +556,29 @@ 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, + volume: float, + source: List[LegacyWellCore], + dest: List[LegacyWellCore], + new_tip: TransferTipPolicyV2, + trash_location: Union[LegacyWellCore, types.Location, TrashBin, WasteChute], + ) -> None: + """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, "transfer_liquid is not supported in legacy context" + def get_active_channels(self) -> int: """This will never be called because it was added in API 2.16.""" assert False, "get_active_channels only supported in API 2.16 & later" 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 f55bf05c447..ec194874528 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,12 +1,13 @@ from __future__ import annotations import logging -from typing import TYPE_CHECKING, Optional, Union +from typing import TYPE_CHECKING, Optional, Union, List from opentrons import types from opentrons.hardware_control.dev_types import PipetteDict from opentrons.hardware_control.types import HardwareAction from opentrons.protocol_api.core.common import WellCore +from opentrons.protocols.advanced_control.transfers.common import TransferTipPolicyV2 from opentrons.protocols.api_support import instrument as instrument_support from opentrons.protocols.api_support.labware_like import LabwareLike from opentrons.protocols.api_support.types import APIVersion @@ -24,6 +25,7 @@ from ...disposal_locations import TrashBin, WasteChute from opentrons.protocol_api._nozzle_layout import NozzleLayout +from opentrons.protocol_api._liquid import LiquidClass from ..instrument import AbstractInstrument @@ -472,6 +474,29 @@ 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, + volume: float, + source: List[LegacyWellCore], + dest: List[LegacyWellCore], + new_tip: TransferTipPolicyV2, + trash_location: Union[LegacyWellCore, types.Location, TrashBin, WasteChute], + ) -> None: + """Transfer a liquid from source to dest according to liquid class properties.""" + # TODO(spp, 2024-11-20): update the docstring and error to include API version + assert False, "transfer_liquid is not supported in legacy context" + def get_active_channels(self) -> int: """This will never be called because it was added in API 2.16.""" assert False, "get_active_channels only supported in API 2.16 & later" diff --git a/api/src/opentrons/protocol_api/instrument_context.py b/api/src/opentrons/protocol_api/instrument_context.py index 7cc2d43bac2..d792c9c8f5c 100644 --- a/api/src/opentrons/protocol_api/instrument_context.py +++ b/api/src/opentrons/protocol_api/instrument_context.py @@ -1,13 +1,15 @@ 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, UnsupportedHardwareCommand, ) +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 @@ -16,7 +18,6 @@ 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.api_support.deck_type import NoTrashDefinedError from opentrons.protocols.api_support.types import APIVersion from opentrons.protocols.api_support import instrument @@ -35,9 +36,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 @@ -61,6 +63,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): """ @@ -1219,7 +1223,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( @@ -1259,7 +1262,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( @@ -1505,6 +1507,105 @@ 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: 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, labware.Well, TrashBin, WasteChute] + ] = None, + ) -> InstrumentContext: + """Transfer liquid from source to dest using the specified liquid class properties. + + TODO: Add args description. + """ + 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_for_transfer_v2( + source + ) + flat_dest_list = validation.ensure_valid_flat_wells_list_for_transfer_v2(dest) + for well in flat_sources_list + flat_dest_list: + instrument.validate_takes_liquid( + location=well.top(), + reject_module=True, + reject_adapter=True, + ) + 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." + ) + + _trash_location: Union[types.Location, labware.Well, TrashBin, WasteChute] + if trash_location is None: + saved_trash = self.trash_container + if isinstance(saved_trash, labware.Labware): + _trash_location = saved_trash.wells()[0] + else: + _trash_location = saved_trash + else: + _trash_location = trash_location + + checked_trash_location = validation.ensure_valid_trash_location_for_transfer_v2( + trash_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, + volume=volume, + source=[well._core for well in flat_sources_list], + dest=[well._core for well in flat_dest_list], + new_tip=valid_new_tip, + trash_location=checked_trash_location._core + if isinstance(checked_trash_location, labware.Well) + else checked_trash_location, + ) + + return self + @requires_version(2, 0) def delay(self, *args: Any, **kwargs: Any) -> None: """ diff --git a/api/src/opentrons/protocol_api/validation.py b/api/src/opentrons/protocol_api/validation.py index 44123571081..ce401697d02 100644 --- a/api/src/opentrons/protocol_api/validation.py +++ b/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, @@ -634,3 +635,81 @@ 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[Well]) -> None: + from .labware import Well + + for loc in locations: + if not isinstance(loc, Well): + raise ValueError( + f"'{loc}' is not a valid location for transfer." + f" Location should be a well instance." + ) + + +def ensure_valid_flat_wells_list_for_transfer_v2( + target: Union[Well, Sequence[Well], Sequence[Sequence[Well]]], +) -> List[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): + return [target] + + if isinstance(target, list) or isinstance(target, tuple): + if isinstance(target[0], list) or isinstance(target[0], tuple): + for sub_sequence in target: + _verify_each_list_element_is_valid_location(sub_sequence) + return [loc for sub_sequence in target for loc in sub_sequence] + else: + _verify_each_list_element_is_valid_location(target) + return list(target) + else: + raise ValueError( + f"'{target}' is not a valid location for transfer." + f" Location should be a well instance, or a 1-dimensional or" + f" 2-dimensional sequence of well instances." + ) + + +def ensure_valid_trash_location_for_transfer_v2( + trash_location: Union[Location, Well, TrashBin, WasteChute] +) -> Union[Location, Well, TrashBin, WasteChute]: + """Ensure that the trash location is valid for v2 transfer.""" + if ( + isinstance(trash_location, Well) + or isinstance(trash_location, TrashBin) + or isinstance(trash_location, WasteChute) + ): + return trash_location + elif isinstance(trash_location, Location): + _, maybe_well = trash_location.labware.get_parent_labware_and_well() + + if maybe_well is None: + raise TypeError( + "If a location is specified as a `types.Location`" + " (for instance, as the result of a call to `Well.top()`)," + " it must be a location relative to a well," + " since that is where a tip is dropped." + f" However, the given location refers to {trash_location.labware}" + ) + return trash_location + else: + raise TypeError( + f"If specified, location should be an instance of" + f" `types.Location` (e.g. the return value from `Well.top()`)" + f" or `Well` (e.g. `tiprack.wells()[0]`) or an instance of `TrashBin` or `WasteChute`." + f" However, it is {trash_location}." + ) diff --git a/api/src/opentrons/protocol_engine/clients/sync_client.py b/api/src/opentrons/protocol_engine/clients/sync_client.py index 3460c13d463..71837a7a2ca 100644 --- a/api/src/opentrons/protocol_engine/clients/sync_client.py +++ b/api/src/opentrons/protocol_engine/clients/sync_client.py @@ -89,6 +89,12 @@ def execute_command_without_recovery( ) -> commands.TryLiquidProbeResult: pass + @overload + def execute_command_without_recovery( + self, params: commands.LoadLiquidClassParams + ) -> commands.LoadLiquidClassResult: + pass + def execute_command_without_recovery( self, params: commands.CommandParams ) -> commands.CommandResult: diff --git a/api/src/opentrons/protocol_engine/commands/__init__.py b/api/src/opentrons/protocol_engine/commands/__init__.py index 8e1e91bec50..f25293f85fb 100644 --- a/api/src/opentrons/protocol_engine/commands/__init__.py +++ b/api/src/opentrons/protocol_engine/commands/__init__.py @@ -147,6 +147,15 @@ LoadLiquidImplementation, ) +from .load_liquid_class import ( + LoadLiquidClass, + LoadLiquidClassParams, + LoadLiquidClassCreate, + LoadLiquidClassResult, + LoadLiquidClassCommandType, + LoadLiquidClassImplementation, +) + from .load_module import ( LoadModule, LoadModuleParams, @@ -553,6 +562,13 @@ "LoadLiquidParams", "LoadLiquidResult", "LoadLiquidCommandType", + # load liquid class command models + "LoadLiquidClass", + "LoadLiquidClassParams", + "LoadLiquidClassCreate", + "LoadLiquidClassResult", + "LoadLiquidClassImplementation", + "LoadLiquidClassCommandType", # hardware control command models # hardware module command bundles "absorbance_reader", diff --git a/api/src/opentrons/protocols/advanced_control/transfers/common.py b/api/src/opentrons/protocols/advanced_control/transfers/common.py index e7f41f2e8e9..e0c25e3c3cf 100644 --- a/api/src/opentrons/protocols/advanced_control/transfers/common.py +++ b/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") diff --git a/api/src/opentrons/protocols/advanced_control/transfers/transfer.py b/api/src/opentrons/protocols/advanced_control/transfers/transfer.py index 1c6c9a78288..3f5f90ab550 100644 --- a/api/src/opentrons/protocols/advanced_control/transfers/transfer.py +++ b/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, 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 b2dff4a7254..cdad23fab10 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 @@ -5,11 +5,15 @@ import pytest from decoy import Decoy from decoy import errors +from opentrons_shared_data.liquid_classes.liquid_class_definition import ( + LiquidClassSchemaV1, +) from opentrons_shared_data.pipette.types import PipetteNameType 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_engine import ( DeckPoint, LoadedPipette, @@ -35,6 +39,7 @@ SingleNozzleLayoutConfiguration, ColumnNozzleLayoutConfiguration, AddressableOffsetVector, + LiquidClassRecord, ) from opentrons.protocol_api.disposal_locations import ( TrashBin, @@ -42,6 +47,7 @@ DisposalOffset, ) from opentrons.protocol_api._nozzle_layout import NozzleLayout +from opentrons.protocol_api._liquid import LiquidClass from opentrons.protocol_api.core.engine import ( InstrumentCore, WellCore, @@ -1494,3 +1500,59 @@ def test_liquid_probe_with_recovery( ) ) ) + + +def test_load_liquid_class( + decoy: Decoy, + mock_engine_client: EngineClient, + subject: InstrumentCore, + minimal_liquid_class_def2: LiquidClassSchemaV1, +) -> None: + """It should send the load liquid class command to the engine.""" + sample_aspirate_data = minimal_liquid_class_def2.byPipette[0].byTipType[0].aspirate + sample_single_dispense_data = ( + minimal_liquid_class_def2.byPipette[0].byTipType[0].singleDispense + ) + sample_multi_dispense_data = ( + minimal_liquid_class_def2.byPipette[0].byTipType[0].multiDispense + ) + + test_liq_class = decoy.mock(cls=LiquidClass) + test_transfer_props = decoy.mock(cls=TransferProperties) + + decoy.when( + test_liq_class.get_for("flex_1channel_50", "opentrons_flex_96_tiprack_50ul") + ).then_return(test_transfer_props) + decoy.when(test_liq_class.name).then_return("water") + decoy.when( + mock_engine_client.state.pipettes.get_model_name(subject.pipette_id) + ).then_return("flex_1channel_50") + decoy.when(test_transfer_props.aspirate.as_schema_v1_model()).then_return( + sample_aspirate_data + ) + decoy.when(test_transfer_props.dispense.as_schema_v1_model()).then_return( + sample_single_dispense_data + ) + decoy.when(test_transfer_props.multi_dispense.as_schema_v1_model()).then_return( # type: ignore[union-attr] + sample_multi_dispense_data + ) + decoy.when( + mock_engine_client.execute_command_without_recovery( + cmd.LoadLiquidClassParams( + liquidClassRecord=LiquidClassRecord( + liquidClassName="water", + pipetteModel="flex_1channel_50", + tiprack="opentrons_flex_96_tiprack_50ul", + aspirate=sample_aspirate_data, + singleDispense=sample_single_dispense_data, + multiDispense=sample_multi_dispense_data, + ) + ) + ) + ).then_return(cmd.LoadLiquidClassResult(liquidClassId="liquid-class-id")) + result = subject.load_liquid_class( + liquid_class=test_liq_class, + pipette_load_name="flex_1channel_50", + tiprack_uri="opentrons_flex_96_tiprack_50ul", + ) + assert result == "liquid-class-id" diff --git a/api/tests/opentrons/protocol_api/test_instrument_context.py b/api/tests/opentrons/protocol_api/test_instrument_context.py index 1caae624377..99705a147b0 100644 --- a/api/tests/opentrons/protocol_api/test_instrument_context.py +++ b/api/tests/opentrons/protocol_api/test_instrument_context.py @@ -10,12 +10,14 @@ from decoy import Decoy from pytest_lazyfixture import lazy_fixture # type: ignore[import-untyped] +from opentrons.config import feature_flags as ff from opentrons.protocol_engine.commands.pipetting_common import LiquidNotFoundError from opentrons.protocol_engine.errors.error_occurrence import ( ProtocolCommandFailedError, ) from opentrons.legacy_broker import LegacyBroker +from opentrons.protocols.advanced_control.transfers.common import TransferTipPolicyV2 from tests.opentrons.protocol_api.partial_tip_configurations import ( PipetteReliantNozzleConfigSpec, @@ -42,6 +44,7 @@ Well, labware, validation as mock_validation, + LiquidClass, ) from opentrons.protocol_api.validation import WellTarget, PointTarget from opentrons.protocol_api.core.common import InstrumentCore, ProtocolCore @@ -51,12 +54,16 @@ from opentrons.hardware_control.nozzle_manager import NozzleMap from opentrons.protocol_api.disposal_locations import TrashBin, WasteChute -from opentrons_shared_data.pipette.pipette_definition import ValidNozzleMaps from opentrons.types import Location, Mount, Point +from opentrons_shared_data.pipette.pipette_definition import ValidNozzleMaps from opentrons_shared_data.errors.exceptions import ( CommandPreconditionViolated, ) +from opentrons_shared_data.liquid_classes.liquid_class_definition import ( + LiquidClassSchemaV1, +) +from opentrons_shared_data.robot.types import RobotTypeEnum, RobotType from . import versions_at_or_above, versions_between @@ -1723,3 +1730,283 @@ def test_air_gap_uses_air_gap( decoy.verify(mock_move_to(top_location, publish=False)) decoy.verify(mock_instrument_core.air_gap_in_place(10, 11)) + + +@pytest.mark.parametrize("robot_type", ["OT-2 Standard", "OT-3 Standard"]) +def test_transfer_liquid_raises_for_invalid_locations( + decoy: Decoy, + mock_protocol_core: ProtocolCore, + subject: InstrumentContext, + mock_feature_flags: None, + robot_type: RobotType, + minimal_liquid_class_def2: LiquidClassSchemaV1, +) -> None: + """It should raise errors if source or destination is invalid.""" + test_liq_class = LiquidClass.create(minimal_liquid_class_def2) + mock_well = decoy.mock(cls=Well) + decoy.when(mock_protocol_core.robot_type).then_return(robot_type) + decoy.when( + ff.allow_liquid_classes(RobotTypeEnum.robot_literal_to_enum(robot_type)) + ).then_return(True) + decoy.when( + mock_validation.ensure_valid_flat_wells_list_for_transfer_v2([mock_well]) + ).then_raise(ValueError("Oh no")) + with pytest.raises(ValueError): + subject.transfer_liquid( + liquid_class=test_liq_class, + volume=10, + source=[mock_well], + dest=[[mock_well]], + ) + + +@pytest.mark.parametrize("robot_type", ["OT-2 Standard", "OT-3 Standard"]) +def test_transfer_liquid_raises_for_unequal_source_and_dest( + decoy: Decoy, + mock_protocol_core: ProtocolCore, + subject: InstrumentContext, + mock_feature_flags: None, + robot_type: RobotType, + minimal_liquid_class_def2: LiquidClassSchemaV1, +) -> None: + """It should raise errors if source and destination are not of same length.""" + test_liq_class = LiquidClass.create(minimal_liquid_class_def2) + mock_well = decoy.mock(cls=Well) + decoy.when(mock_protocol_core.robot_type).then_return(robot_type) + decoy.when( + ff.allow_liquid_classes(RobotTypeEnum.robot_literal_to_enum(robot_type)) + ).then_return(True) + decoy.when( + mock_validation.ensure_valid_flat_wells_list_for_transfer_v2(mock_well) + ).then_return([mock_well, mock_well]) + decoy.when( + mock_validation.ensure_valid_flat_wells_list_for_transfer_v2([mock_well]) + ).then_return([mock_well]) + with pytest.raises( + ValueError, match="Sources and destinations should be of the same length" + ): + subject.transfer_liquid( + liquid_class=test_liq_class, + volume=10, + source=mock_well, + dest=[mock_well], + ) + + +@pytest.mark.parametrize("robot_type", ["OT-2 Standard", "OT-3 Standard"]) +def test_transfer_liquid_raises_for_non_liquid_handling_locations( + decoy: Decoy, + mock_protocol_core: ProtocolCore, + subject: InstrumentContext, + mock_feature_flags: None, + robot_type: RobotType, + minimal_liquid_class_def2: LiquidClassSchemaV1, +) -> None: + """It should raise errors if source and destination are not of same length.""" + test_liq_class = LiquidClass.create(minimal_liquid_class_def2) + mock_well = decoy.mock(cls=Well) + decoy.when(mock_protocol_core.robot_type).then_return(robot_type) + decoy.when( + ff.allow_liquid_classes(RobotTypeEnum.robot_literal_to_enum(robot_type)) + ).then_return(True) + decoy.when( + mock_validation.ensure_valid_flat_wells_list_for_transfer_v2([mock_well]) + ).then_return([mock_well]) + decoy.when( + mock_instrument_support.validate_takes_liquid( + mock_well.top(), reject_module=True, reject_adapter=True + ) + ).then_raise(ValueError("Uh oh")) + with pytest.raises(ValueError, match="Uh oh"): + subject.transfer_liquid( + liquid_class=test_liq_class, + volume=10, + source=[mock_well], + dest=[mock_well], + ) + + +@pytest.mark.parametrize("robot_type", ["OT-2 Standard", "OT-3 Standard"]) +def test_transfer_liquid_raises_for_bad_tip_policy( + decoy: Decoy, + mock_protocol_core: ProtocolCore, + subject: InstrumentContext, + mock_feature_flags: None, + robot_type: RobotType, + minimal_liquid_class_def2: LiquidClassSchemaV1, +) -> None: + """It should raise errors if new_tip is invalid.""" + test_liq_class = LiquidClass.create(minimal_liquid_class_def2) + mock_well = decoy.mock(cls=Well) + decoy.when(mock_protocol_core.robot_type).then_return(robot_type) + decoy.when( + ff.allow_liquid_classes(RobotTypeEnum.robot_literal_to_enum(robot_type)) + ).then_return(True) + decoy.when( + mock_validation.ensure_valid_flat_wells_list_for_transfer_v2([mock_well]) + ).then_return([mock_well]) + decoy.when(mock_validation.ensure_new_tip_policy("once")).then_raise( + ValueError("Uh oh") + ) + with pytest.raises(ValueError, match="Uh oh"): + subject.transfer_liquid( + liquid_class=test_liq_class, + volume=10, + source=[mock_well], + dest=[mock_well], + new_tip="once", + ) + + +@pytest.mark.parametrize("robot_type", ["OT-2 Standard", "OT-3 Standard"]) +def test_transfer_liquid_raises_for_no_tip( + decoy: Decoy, + mock_protocol_core: ProtocolCore, + subject: InstrumentContext, + mock_feature_flags: None, + robot_type: RobotType, + minimal_liquid_class_def2: LiquidClassSchemaV1, +) -> None: + """It should raise errors if there is no tip attached.""" + test_liq_class = LiquidClass.create(minimal_liquid_class_def2) + mock_well = decoy.mock(cls=Well) + decoy.when(mock_protocol_core.robot_type).then_return(robot_type) + decoy.when( + ff.allow_liquid_classes(RobotTypeEnum.robot_literal_to_enum(robot_type)) + ).then_return(True) + decoy.when( + mock_validation.ensure_valid_flat_wells_list_for_transfer_v2([mock_well]) + ).then_return([mock_well]) + decoy.when(mock_validation.ensure_new_tip_policy("never")).then_return( + TransferTipPolicyV2.NEVER + ) + with pytest.raises(RuntimeError, match="Pipette has no tip"): + subject.transfer_liquid( + liquid_class=test_liq_class, + volume=10, + source=[mock_well], + dest=[mock_well], + new_tip="never", + ) + + +@pytest.mark.parametrize("robot_type", ["OT-2 Standard", "OT-3 Standard"]) +def test_transfer_liquid_raises_if_tip_has_liquid( + decoy: Decoy, + mock_protocol_core: ProtocolCore, + mock_instrument_core: InstrumentCore, + subject: InstrumentContext, + mock_feature_flags: None, + robot_type: RobotType, + minimal_liquid_class_def2: LiquidClassSchemaV1, +) -> None: + """It should raise errors if there is no tip attached.""" + test_liq_class = LiquidClass.create(minimal_liquid_class_def2) + mock_well = decoy.mock(cls=Well) + tip_racks = [decoy.mock(cls=Labware)] + + subject.starting_tip = None + subject.tip_racks = tip_racks + + decoy.when(mock_protocol_core.robot_type).then_return(robot_type) + decoy.when( + ff.allow_liquid_classes(RobotTypeEnum.robot_literal_to_enum(robot_type)) + ).then_return(True) + decoy.when( + mock_validation.ensure_valid_flat_wells_list_for_transfer_v2([mock_well]) + ).then_return([mock_well]) + decoy.when(mock_validation.ensure_new_tip_policy("never")).then_return( + TransferTipPolicyV2.ONCE + ) + decoy.when(mock_instrument_core.get_nozzle_map()).then_return(MOCK_MAP) + decoy.when(mock_instrument_core.get_active_channels()).then_return(2) + decoy.when( + labware.next_available_tip( + starting_tip=None, + tip_racks=tip_racks, + channels=2, + nozzle_map=MOCK_MAP, + ) + ).then_return((decoy.mock(cls=Labware), decoy.mock(cls=Well))) + decoy.when(mock_instrument_core.get_current_volume()).then_return(1000) + with pytest.raises(RuntimeError, match="liquid already in the tip"): + subject.transfer_liquid( + liquid_class=test_liq_class, + volume=10, + source=[mock_well], + dest=[mock_well], + new_tip="never", + ) + + +@pytest.mark.parametrize("robot_type", ["OT-2 Standard", "OT-3 Standard"]) +def test_transfer_liquid_delegates_to_engine_core( + decoy: Decoy, + mock_protocol_core: ProtocolCore, + mock_instrument_core: InstrumentCore, + subject: InstrumentContext, + mock_feature_flags: None, + robot_type: RobotType, + minimal_liquid_class_def2: LiquidClassSchemaV1, +) -> None: + """It should load liquid class into engine and delegate the transfer execution to core.""" + test_liq_class = LiquidClass.create(minimal_liquid_class_def2) + mock_well = decoy.mock(cls=Well) + tip_racks = [decoy.mock(cls=Labware)] + trash_location = Location(point=Point(1, 2, 3), labware=mock_well) + next_tiprack = decoy.mock(cls=Labware) + subject.starting_tip = None + subject.tip_racks = tip_racks + + decoy.when(mock_protocol_core.robot_type).then_return(robot_type) + decoy.when( + ff.allow_liquid_classes(RobotTypeEnum.robot_literal_to_enum(robot_type)) + ).then_return(True) + decoy.when( + mock_validation.ensure_valid_flat_wells_list_for_transfer_v2([mock_well]) + ).then_return([mock_well]) + decoy.when(mock_validation.ensure_new_tip_policy("never")).then_return( + TransferTipPolicyV2.ONCE + ) + decoy.when(mock_instrument_core.get_nozzle_map()).then_return(MOCK_MAP) + decoy.when(mock_instrument_core.get_active_channels()).then_return(2) + decoy.when( + labware.next_available_tip( + starting_tip=None, + tip_racks=tip_racks, + channels=2, + nozzle_map=MOCK_MAP, + ) + ).then_return((next_tiprack, decoy.mock(cls=Well))) + decoy.when(mock_instrument_core.get_current_volume()).then_return(0) + decoy.when( + mock_validation.ensure_valid_trash_location_for_transfer_v2(trash_location) + ).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, + source=[mock_well], + dest=[mock_well], + new_tip="never", + trash_location=trash_location, + ) + decoy.verify( + mock_instrument_core.transfer_liquid( + liquid_class_id="liq-class-id", + volume=10, + source=[mock_well._core], + dest=[mock_well._core], + new_tip=TransferTipPolicyV2.ONCE, + trash_location=trash_location.move(Point(1, 2, 3)), + ) + ) diff --git a/api/tests/opentrons/protocol_api/test_liquid_class_properties.py b/api/tests/opentrons/protocol_api/test_liquid_class_properties.py index f7033afb5be..6f1ccab8179 100644 --- a/api/tests/opentrons/protocol_api/test_liquid_class_properties.py +++ b/api/tests/opentrons/protocol_api/test_liquid_class_properties.py @@ -51,6 +51,7 @@ def test_build_aspirate_settings() -> None: assert aspirate_properties.mix.volume == 15 assert aspirate_properties.delay.enabled is True assert aspirate_properties.delay.duration == 2 + assert aspirate_properties.as_schema_v1_model() == aspirate_data def test_build_single_dispense_settings() -> None: @@ -103,6 +104,7 @@ def test_build_single_dispense_settings() -> None: } assert single_dispense_properties.delay.enabled is True assert single_dispense_properties.delay.duration == 2.5 + assert single_dispense_properties.as_schema_v1_model() == single_dispense_data def test_build_multi_dispense_settings() -> None: @@ -154,6 +156,7 @@ def test_build_multi_dispense_settings() -> None: } assert multi_dispense_properties.delay.enabled is True assert multi_dispense_properties.delay.duration == 1 + assert multi_dispense_properties.as_schema_v1_model() == multi_dispense_data def test_build_multi_dispense_settings_none( @@ -169,6 +172,7 @@ def test_liquid_handling_property_by_volume() -> None: subject = LiquidHandlingPropertyByVolume([(5.0, 50.0), (10.0, 250.0)]) assert subject.as_dict() == {5.0: 50, 10.0: 250} assert subject.get_for_volume(7) == 130.0 + assert subject.as_list_of_tuples() == [(5.0, 50.0), (10.0, 250.0)] subject.set_for_volume(volume=7, value=175.5) assert subject.as_dict() == { diff --git a/api/tests/opentrons/protocol_api/test_validation.py b/api/tests/opentrons/protocol_api/test_validation.py index 9a111e6f81f..835988b68ac 100644 --- a/api/tests/opentrons/protocol_api/test_validation.py +++ b/api/tests/opentrons/protocol_api/test_validation.py @@ -6,6 +6,7 @@ import pytest import re +from opentrons.protocols.advanced_control.transfers.common import TransferTipPolicyV2 from opentrons_shared_data.labware.labware_definition import ( LabwareRole, Parameters as LabwareDefinitionParameters, @@ -722,3 +723,95 @@ def test_ensure_only_gantry_axis_map_type( """Check that gantry axis_map validation occurs for the given scenarios.""" with pytest.raises(subject.IncorrectAxisError, match=error_message): subject.ensure_only_gantry_axis_map_type(axis_map, robot_type) + + +@pytest.mark.parametrize( + ["value", "expected_result"], + [ + ("once", TransferTipPolicyV2.ONCE), + ("NEVER", TransferTipPolicyV2.NEVER), + ("alWaYs", TransferTipPolicyV2.ALWAYS), + ], +) +def test_ensure_new_tip_policy( + value: str, expected_result: TransferTipPolicyV2 +) -> None: + """It should return the expected tip policy.""" + assert subject.ensure_new_tip_policy(value) == expected_result + + +def test_ensure_new_tip_policy_raises() -> None: + """It should raise ValueError for invalid new_tip value.""" + with pytest.raises(ValueError, match="is invalid value for 'new_tip'"): + subject.ensure_new_tip_policy("blah") + + +@pytest.mark.parametrize( + ["target", "expected_raise"], + [ + ( + "a", + pytest.raises( + ValueError, match="'a' is not a valid location for transfer." + ), + ), + ( + ["a"], + pytest.raises( + ValueError, match="'a' is not a valid location for transfer." + ), + ), + ( + [("a",)], + pytest.raises( + ValueError, match="'a' is not a valid location for transfer." + ), + ), + ], +) +def test_ensure_valid_flat_wells_list_raises_for_invalid_targets( + target: Any, + expected_raise: ContextManager[Any], +) -> None: + """It should raise an error if target location is invalid.""" + with expected_raise: + subject.ensure_valid_flat_wells_list_for_transfer_v2(target) + + +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."): + subject.ensure_valid_flat_wells_list_for_transfer_v2(target1) # type: ignore[arg-type] + + target2 = [[decoy.mock(cls=Well)], ["a"]] + with pytest.raises(ValueError, match="'a' is not a valid location for transfer."): + subject.ensure_valid_flat_wells_list_for_transfer_v2(target2) # type: ignore[arg-type] + + +def test_ensure_valid_flat_wells_list(decoy: Decoy) -> None: + """It should convert the locations to flat lists correctly.""" + target1 = decoy.mock(cls=Well) + target2 = decoy.mock(cls=Well) + + assert subject.ensure_valid_flat_wells_list_for_transfer_v2(target1) == [target1] + assert subject.ensure_valid_flat_wells_list_for_transfer_v2([target1, target2]) == [ + target1, + target2, + ] + assert subject.ensure_valid_flat_wells_list_for_transfer_v2( + [ + [target1, target1], + [target2, target2], + ] + ) == [target1, target1, target2, target2] + assert subject.ensure_valid_flat_wells_list_for_transfer_v2((target1, target2)) == [ + target1, + target2, + ] + assert subject.ensure_valid_flat_wells_list_for_transfer_v2( + ( + [target1, target1], + [target2, target2], + ) + ) == [target1, target1, target2, target2]