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

Closes AUTH-843

# Overview

Adds `InstrumentContext.transfer_liquid()` method that does the
following-
- validates parameters of `transfer_liquid()`
- loads the liquid class properties for the relevant pipette and tiprack
into protocol engine
- delegates to engine core to perform the actual transfer

This PR does not cover engine core's transfer method execution.

## Risk assessment

No risk so far since this is a code-only change.

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 list(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_shared_data_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_shared_data_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_shared_data_model(self) -> SharedDataTouchTipProperties:
+        return SharedDataTouchTipProperties(
+            enable=self._enabled,
+            params=self._get_shared_data_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_shared_data_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_shared_data_model(self) -> SharedDataMixProperties:
+        return SharedDataMixProperties(
+            enable=self._enabled,
+            params=self._get_shared_data_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_shared_data_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_shared_data_model(self) -> SharedDataBlowoutProperties:
+        return SharedDataBlowoutProperties(
+            enable=self._enabled,
+            params=self._get_shared_data_params(),
+        )
+
 
 @dataclass
 class SubmergeRetractCommon:
@@ -271,6 +340,14 @@ def delay(self) -> DelayProperties:
 class Submerge(SubmergeRetractCommon):
     ...
 
+    def as_shared_data_model(self) -> SharedDataSubmerge:
+        return SharedDataSubmerge(
+            positionReference=self._position_reference,
+            offset=self._offset,
+            speed=self._speed,
+            delay=self._delay.as_shared_data_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_shared_data_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_shared_data_model(),
+            delay=self._delay.as_shared_data_model(),
+        )
+
 
 @dataclass
 class RetractDispense(SubmergeRetractCommon):
@@ -306,6 +393,17 @@ def touch_tip(self) -> TouchTipProperties:
     def blowout(self) -> BlowoutProperties:
         return self._blowout
 
+    def as_shared_data_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_shared_data_model(),
+            touchTip=self._touch_tip.as_shared_data_model(),
+            delay=self._delay.as_shared_data_model(),
+        )
+
 
 @dataclass
 class BaseLiquidHandlingProperties:
@@ -375,6 +473,19 @@ def retract(self) -> RetractAspirate:
     def mix(self) -> MixProperties:
         return self._mix
 
+    def as_shared_data_model(self) -> SharedDataAspirateProperties:
+        return SharedDataAspirateProperties(
+            submerge=self._submerge.as_shared_data_model(),
+            retract=self._retract.as_shared_data_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_shared_data_model(),
+            delay=self._delay.as_shared_data_model(),
+            correctionByVolume=self._correction_by_volume.as_list_of_tuples(),
+        )
+
 
 @dataclass
 class SingleDispenseProperties(BaseLiquidHandlingProperties):
@@ -395,6 +506,19 @@ def retract(self) -> RetractDispense:
     def mix(self) -> MixProperties:
         return self._mix
 
+    def as_shared_data_model(self) -> SharedDataSingleDispenseProperties:
+        return SharedDataSingleDispenseProperties(
+            submerge=self._submerge.as_shared_data_model(),
+            retract=self._retract.as_shared_data_model(),
+            positionReference=self._position_reference,
+            offset=self._offset,
+            flowRateByVolume=self._flow_rate_by_volume.as_list_of_tuples(),
+            mix=self._mix.as_shared_data_model(),
+            pushOutByVolume=self._push_out_by_volume.as_list_of_tuples(),
+            delay=self._delay.as_shared_data_model(),
+            correctionByVolume=self._correction_by_volume.as_list_of_tuples(),
+        )
+
 
 @dataclass
 class MultiDispenseProperties(BaseLiquidHandlingProperties):
@@ -415,6 +539,19 @@ def conditioning_by_volume(self) -> LiquidHandlingPropertyByVolume:
     def disposal_by_volume(self) -> LiquidHandlingPropertyByVolume:
         return self._disposal_by_volume
 
+    def as_shared_data_model(self) -> SharedDataMultiDispenseProperties:
+        return SharedDataMultiDispenseProperties(
+            submerge=self._submerge.as_shared_data_model(),
+            retract=self._retract.as_shared_data_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_shared_data_model(),
+            correctionByVolume=self._correction_by_volume.as_list_of_tuples(),
+        )
+
 
 @dataclass
 class TransferProperties:

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_shared_data_model(),
+            singleDispense=transfer_props.dispense.as_shared_data_model(),
+            multiDispense=transfer_props.multi_dispense.as_shared_data_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)

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."""