add load_liquid_class() to engine core, add transfer_liquid placehold…

…ers in cores, add tests

api/src/opentrons/protocol_api/core/engine/instrument.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 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
@@ -28,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
@@ -36,9 +36,8 @@
 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:
@@ -855,6 +854,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)
@@ -941,14 +979,3 @@ def nozzle_configuration_valid_for_lld(self) -> bool:
         return self._engine_client.state.pipettes.get_nozzle_configuration_supports_lld(
             self.pipette_id
         )
-
-    def transfer_liquid(
-        self,
-        liquid_class: LiquidClass,
-        volume: float,
-        source: List[WellCore],
-        dest: List[WellCore],
-        new_tip: TransferTipPolicyV2,
-        trash_location: Union[Location, TrashBin, WasteChute],
-    ) -> None:
-        """Execute transfer using liquid class properties."""

api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py

@@ -556,17 +556,27 @@ def configure_nozzle_layout(
         """This will never be called because it was added in API 2.16."""
         pass
 
-    def transfer_liquid(
+    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[types.Location, TrashBin, WasteChute],
+        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 error to include API version
+        """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:

api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py

@@ -474,17 +474,27 @@ def configure_nozzle_layout(
         """This will never be called because it was added in API 2.15."""
         pass
 
-    def transfer_liquid(
+    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[types.Location, TrashBin, WasteChute],
+        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 error to include API version
+        # 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:

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:

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

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,13 +39,15 @@
     SingleNozzleLayoutConfiguration,
     ColumnNozzleLayoutConfiguration,
     AddressableOffsetVector,
+    LiquidClassRecord,
 )
 from opentrons.protocol_api.disposal_locations import (
     TrashBin,
     WasteChute,
     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"