feat(api): add liquid class in PAPI

api/src/opentrons/config/advanced_settings.py

@@ -222,6 +222,17 @@ class Setting(NamedTuple):
         robot_type=[RobotTypeEnum.OT2, RobotTypeEnum.FLEX],
         internal_only=True,
     ),
+    SettingDefinition(
+        _id="allowLiquidClasses",
+        title="Allow the use of liquid classes",
+        description=(
+            "Do not enable."
+            " This is an Opentrons internal setting to allow using in-development"
+            " liquid classes."
+        ),
+        robot_type=[RobotTypeEnum.OT2, RobotTypeEnum.FLEX],
+        internal_only=True,
+    ),
 ]
 
 

api/src/opentrons/config/feature_flags.py

@@ -78,3 +78,7 @@ def enable_performance_metrics(robot_type: RobotTypeEnum) -> bool:
 
 def oem_mode_enabled() -> bool:
     return advs.get_setting_with_env_overload("enableOEMMode", RobotTypeEnum.FLEX)
+
+
+def allow_liquid_classes(robot_type: RobotTypeEnum) -> bool:
+    return advs.get_setting_with_env_overload("allowLiquidClasses", robot_type)

api/src/opentrons/protocol_api/__init__.py

@@ -29,7 +29,7 @@
     AbsorbanceReaderContext,
 )
 from .disposal_locations import TrashBin, WasteChute
-from ._liquid import Liquid
+from ._liquid import Liquid, LiquidClass
 from ._types import OFF_DECK
 from ._nozzle_layout import (
     COLUMN,
@@ -67,6 +67,7 @@
     "WasteChute",
     "Well",
     "Liquid",
+    "LiquidClass",
     "Parameters",
     "COLUMN",
     "PARTIAL_COLUMN",

api/src/opentrons/protocol_api/_liquid.py

@@ -1,5 +1,14 @@
 from dataclasses import dataclass
-from typing import Optional
+from typing import Optional, Sequence
+
+from opentrons_shared_data.liquid_classes.liquid_class_definition import (
+    LiquidClassSchemaV1,
+    AspirateProperties,
+    SingleDispenseProperties,
+    MultiDispenseProperties,
+    ByPipetteSetting,
+    ByTipTypeSetting,
+)
 
 
 @dataclass(frozen=True)
@@ -18,3 +27,82 @@ class Liquid:
     name: str
     description: Optional[str]
     display_color: Optional[str]
+
+
+# TODO (spp, 2024-10-17): create PAPI-equivalent types for all the properties
+#  and have validation on value updates with user-facing error messages
+@dataclass
+class TransferProperties:
+    _aspirate: AspirateProperties
+    _dispense: SingleDispenseProperties
+    _multi_dispense: Optional[MultiDispenseProperties]
+
+    @property
+    def aspirate(self) -> AspirateProperties:
+        """Aspirate properties."""
+        return self._aspirate
+
+    @property
+    def dispense(self) -> SingleDispenseProperties:
+        """Single dispense properties."""
+        return self._dispense
+
+    @property
+    def multi_dispense(self) -> Optional[MultiDispenseProperties]:
+        """Multi dispense properties."""
+        return self._multi_dispense
+
+
+@dataclass
+class LiquidClass:
+    """A data class that contains properties of a specific class of liquids."""
+
+    _name: str
+    _display_name: str
+    _by_pipette_setting: Sequence[ByPipetteSetting]
+
+    @classmethod
+    def create(cls, liquid_class_definition: LiquidClassSchemaV1) -> "LiquidClass":
+        """Liquid class factory method."""
+
+        return LiquidClass(
+            _name=liquid_class_definition.liquidClassName,
+            _display_name=liquid_class_definition.displayName,
+            _by_pipette_setting=liquid_class_definition.byPipette,
+        )
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    @property
+    def display_name(self) -> str:
+        return self._display_name
+
+    def get_for(self, pipette: str, tiprack: str) -> TransferProperties:
+        """Get liquid class transfer properties for the specified pipette and tip."""
+        settings_for_pipette: Sequence[ByPipetteSetting] = list(
+            filter(
+                lambda pip_setting: pip_setting.pipetteModel == pipette,
+                self._by_pipette_setting,
+            )
+        )
+        if len(settings_for_pipette) == 0:
+            raise ValueError(
+                f"No properties found for {pipette} in {self._name} liquid class"
+            )
+        settings_for_tip: Sequence[ByTipTypeSetting] = list(
+            filter(
+                lambda tip_setting: tip_setting.tiprack == tiprack,
+                settings_for_pipette[0].byTipType,
+            )
+        )
+        if len(settings_for_tip) == 0:
+            raise ValueError(
+                f"No properties found for {tiprack} in {self._name} liquid class"
+            )
+        return TransferProperties(
+            _aspirate=settings_for_tip[0].aspirate,
+            _dispense=settings_for_tip[0].singleDispense,
+            _multi_dispense=settings_for_tip[0].multiDispense,
+        )

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

@@ -7,6 +7,10 @@
 from opentrons_shared_data.deck.types import DeckDefinitionV5, SlotDefV3
 from opentrons_shared_data.labware.labware_definition import LabwareDefinition
 from opentrons_shared_data.labware.types import LabwareDefinition as LabwareDefDict
+from opentrons_shared_data import liquid_classes
+from opentrons_shared_data.liquid_classes.liquid_class_definition import (
+    LiquidClassSchemaV1,
+)
 from opentrons_shared_data.pipette.types import PipetteNameType
 from opentrons_shared_data.robot.types import RobotType
 
@@ -51,7 +55,7 @@
 
 from ... import validation
 from ..._types import OffDeckType
-from ..._liquid import Liquid
+from ..._liquid import Liquid, LiquidClass
 from ...disposal_locations import TrashBin, WasteChute
 from ..protocol import AbstractProtocol
 from ..labware import LabwareLoadParams
@@ -103,6 +107,7 @@ def __init__(
             str, Union[ModuleCore, NonConnectedModuleCore]
         ] = {}
         self._disposal_locations: List[Union[Labware, TrashBin, WasteChute]] = []
+        self._defined_liquid_class_defs_by_name: Dict[str, LiquidClassSchemaV1] = {}
         self._load_fixed_trash()
 
     @property
@@ -747,6 +752,23 @@ def define_liquid(
             ),
         )
 
+    def define_liquid_class(self, name: str) -> LiquidClass:
+        """Define a liquid class for use in transfer functions."""
+        try:
+            # Check if we have already loaded this liquid class' definition
+            liquid_class_def = self._defined_liquid_class_defs_by_name[name]
+        except KeyError:
+            try:
+                # Fetching the liquid class data from file and parsing it
+                # is an expensive operation and should be avoided.
+                # Calling this often will degrade protocol execution performance.
+                liquid_class_def = liquid_classes.load_definition(name)
+                self._defined_liquid_class_defs_by_name[name] = liquid_class_def
+            except KeyError:
+                raise ValueError("Liquid class definition not found")
+
+        return LiquidClass.create(liquid_class_def)
+
     def get_labware_location(
         self, labware_core: LabwareCore
     ) -> Union[str, LabwareCore, ModuleCore, NonConnectedModuleCore, OffDeckType]:

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

@@ -17,7 +17,7 @@
 
 from ...labware import Labware
 from ...disposal_locations import TrashBin, WasteChute
-from ..._liquid import Liquid
+from ..._liquid import Liquid, LiquidClass
 from ..._types import OffDeckType
 from ..protocol import AbstractProtocol
 from ..labware import LabwareLoadParams
@@ -531,6 +531,10 @@ def define_liquid(
         """Define a liquid to load into a well."""
         assert False, "define_liquid only supported on engine core"
 
+    def define_liquid_class(self, name: str) -> LiquidClass:
+        """Define a liquid class."""
+        assert False, "define_liquid_class is only supported on engine core"
+
     def get_labware_location(
         self, labware_core: LegacyLabwareCore
     ) -> Union[

api/src/opentrons/protocol_api/core/protocol.py

@@ -18,7 +18,7 @@
 from .instrument import InstrumentCoreType
 from .labware import LabwareCoreType, LabwareLoadParams
 from .module import ModuleCoreType
-from .._liquid import Liquid
+from .._liquid import Liquid, LiquidClass
 from .._types import OffDeckType
 from ..disposal_locations import TrashBin, WasteChute
 
@@ -247,6 +247,10 @@ def define_liquid(
     ) -> Liquid:
         """Define a liquid to load into a well."""
 
+    @abstractmethod
+    def define_liquid_class(self, name: str) -> LiquidClass:
+        """Define a liquid class for use in transfer functions."""
+
     @abstractmethod
     def get_labware_location(
         self, labware_core: LabwareCoreType

api/src/opentrons/protocol_api/protocol_context.py

@@ -14,8 +14,10 @@
 
 from opentrons_shared_data.labware.types import LabwareDefinition
 from opentrons_shared_data.pipette.types import PipetteNameType
+from opentrons_shared_data.robot.types import RobotTypeEnum
 
 from opentrons.types import Mount, Location, DeckLocation, DeckSlotName, StagingSlotName
+from opentrons.config import feature_flags
 from opentrons.legacy_broker import LegacyBroker
 from opentrons.hardware_control.modules.types import (
     MagneticBlockModel,
@@ -61,7 +63,7 @@
 from .core.legacy.legacy_protocol_core import LegacyProtocolCore
 
 from . import validation
-from ._liquid import Liquid
+from ._liquid import Liquid, LiquidClass
 from .disposal_locations import TrashBin, WasteChute
 from .deck import Deck
 from .instrument_context import InstrumentContext
@@ -1284,6 +1286,18 @@ def define_liquid(
             display_color=display_color,
         )
 
+    def define_liquid_class(
+        self,
+        name: str,
+    ) -> LiquidClass:
+        """Define a liquid class for use in the protocol."""
+        if feature_flags.allow_liquid_classes(
+            robot_type=RobotTypeEnum.robot_literal_to_enum(self._core.robot_type)
+        ):
+            return self._core.define_liquid_class(name=name)
+        else:
+            raise NotImplementedError("This method is not implemented.")
+
     @property
     @requires_version(2, 5)
     def rail_lights_on(self) -> bool:

api/tests/opentrons/conftest.py

@@ -20,6 +20,7 @@
     Union,
     cast,
 )
+
 from typing_extensions import TypedDict
 
 import pytest
@@ -37,6 +38,23 @@
 from opentrons_shared_data.protocol.types import JsonProtocol
 from opentrons_shared_data.labware.types import LabwareDefinition
 from opentrons_shared_data.module.types import ModuleDefinitionV3
+from opentrons_shared_data.liquid_classes.liquid_class_definition import (
+    LiquidClassSchemaV1,
+    ByPipetteSetting,
+    ByTipTypeSetting,
+    AspirateProperties,
+    Submerge,
+    PositionReference,
+    DelayProperties,
+    DelayParams,
+    RetractAspirate,
+    SingleDispenseProperties,
+    RetractDispense,
+    Coordinate,
+    MixProperties,
+    TouchTipProperties,
+    BlowoutProperties,
+)
 from opentrons_shared_data.deck.types import (
     RobotModel,
     DeckDefinitionV3,
@@ -763,3 +781,84 @@ def minimal_module_def() -> ModuleDefinitionV3:
         "cornerOffsetFromSlot": {"x": 0.1, "y": 0.1, "z": 0.0},
         "twoDimensionalRendering": {},
     }
+
+
+@pytest.fixture
+def minimal_liquid_class_def1() -> LiquidClassSchemaV1:
+    return LiquidClassSchemaV1(
+        liquidClassName="water1",
+        displayName="water 1",
+        schemaVersion=1,
+        namespace="test-fixture-1",
+        byPipette=[],
+    )
+
+
+@pytest.fixture
+def minimal_liquid_class_def2() -> LiquidClassSchemaV1:
+    return LiquidClassSchemaV1(
+        liquidClassName="water2",
+        displayName="water 2",
+        schemaVersion=1,
+        namespace="test-fixture-2",
+        byPipette=[
+            ByPipetteSetting(
+                pipetteModel="p20_single_gen2",
+                byTipType=[
+                    ByTipTypeSetting(
+                        tiprack="opentrons_96_tiprack_20ul",
+                        aspirate=AspirateProperties(
+                            submerge=Submerge(
+                                positionReference=PositionReference.LIQUID_MENISCUS,
+                                offset=Coordinate(x=0, y=0, z=-5),
+                                speed=100,
+                                delay=DelayProperties(
+                                    enable=True, params=DelayParams(duration=1.5)
+                                ),
+                            ),
+                            retract=RetractAspirate(
+                                positionReference=PositionReference.WELL_TOP,
+                                offset=Coordinate(x=0, y=0, z=5),
+                                speed=100,
+                                airGapByVolume={"default": 2, "5": 3, "10": 4},
+                                touchTip=TouchTipProperties(enable=False),
+                                delay=DelayProperties(enable=False),
+                            ),
+                            positionReference=PositionReference.WELL_BOTTOM,
+                            offset=Coordinate(x=0, y=0, z=-5),
+                            flowRateByVolume={"default": 50, "10": 40, "20": 30},
+                            preWet=True,
+                            mix=MixProperties(enable=False),
+                            delay=DelayProperties(
+                                enable=True, params=DelayParams(duration=2)
+                            ),
+                        ),
+                        singleDispense=SingleDispenseProperties(
+                            submerge=Submerge(
+                                positionReference=PositionReference.LIQUID_MENISCUS,
+                                offset=Coordinate(x=0, y=0, z=-5),
+                                speed=100,
+                                delay=DelayProperties(enable=False),
+                            ),
+                            retract=RetractDispense(
+                                positionReference=PositionReference.WELL_TOP,
+                                offset=Coordinate(x=0, y=0, z=5),
+                                speed=100,
+                                airGapByVolume={"default": 2, "5": 3, "10": 4},
+                                blowout=BlowoutProperties(enable=False),
+                                touchTip=TouchTipProperties(enable=False),
+                                delay=DelayProperties(enable=False),
+                            ),
+                            positionReference=PositionReference.WELL_BOTTOM,
+                            offset=Coordinate(x=0, y=0, z=-5),
+                            flowRateByVolume={"default": 50, "10": 40, "20": 30},
+                            mix=MixProperties(enable=False),
+                            pushOutByVolume={"default": 5, "10": 7, "20": 10},
+                            delay=DelayProperties(enable=False),
+                        ),
+                        multiDispense=None,
+                    )
+                ],
+            )
+        ],
+    )