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,
+    ),
 ]
 
 
@@ -715,6 +726,16 @@ def _migrate34to35(previous: SettingsMap) -> SettingsMap:
     return newmap
 
 
+def _migrate35to36(previous: SettingsMap) -> SettingsMap:
+    """Migrate to version 36 of the feature flags file.
+
+    - Adds the allowLiquidClasses config element.
+    """
+    newmap = {k: v for k, v in previous.items()}
+    newmap["allowLiquidClasses"] = None
+    return newmap
+
+
 _MIGRATIONS = [
     _migrate0to1,
     _migrate1to2,
@@ -751,6 +772,7 @@ def _migrate34to35(previous: SettingsMap) -> SettingsMap:
     _migrate32to33,
     _migrate33to34,
     _migrate34to35,
+    _migrate35to36,
 ]
 """
 List of all migrations to apply, indexed by (version - 1). See _migrate below

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,80 @@ 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 cls(
+            _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] = [
+            pip_setting
+            for pip_setting in self._by_pipette_setting
+            if pip_setting.pipetteModel == pipette
+        ]
+        if len(settings_for_pipette) == 0:
+            raise ValueError(
+                f"No properties found for {pipette} in {self._name} liquid class"
+            )
+        settings_for_tip: Sequence[ByTipTypeSetting] = [
+            tip_setting
+            for tip_setting in settings_for_pipette[0].byTipType
+            if tip_setting.tiprack == tiprack
+        ]
+        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

@@ -2,11 +2,17 @@
 from __future__ import annotations
 from typing import Dict, Optional, Type, Union, List, Tuple, TYPE_CHECKING
 
+from opentrons_shared_data.liquid_classes import LiquidClassDefinitionDoesNotExist
+
 from opentrons.protocol_engine import commands as cmd
 from opentrons.protocol_engine.commands import LoadModuleResult
 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 +57,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 +109,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 +754,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 LiquidClassDefinitionDoesNotExist:
+                raise ValueError(f"Liquid class definition not found for '{name}'.")
+
+        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/config/test_advanced_settings_migration.py

@@ -8,7 +8,7 @@
 
 @pytest.fixture
 def migrated_file_version() -> int:
-    return 35
+    return 36
 
 
 # make sure to set a boolean value in default_file_settings only if
@@ -30,6 +30,7 @@ def default_file_settings() -> Dict[str, Any]:
         "enableErrorRecoveryExperiments": None,
         "enableOEMMode": None,
         "enablePerformanceMetrics": None,
+        "allowLiquidClasses": None,
     }
 
 
@@ -68,6 +69,7 @@ def v2_config(v1_config: Dict[str, Any]) -> Dict[str, Any]:
     r.update(
         {
             "_version": 2,
+            "disableLogAggregation": None,
         }
     )
     return r
@@ -410,6 +412,26 @@ def v34_config(v33_config: Dict[str, Any]) -> Dict[str, Any]:
     return r
 
 
+@pytest.fixture
+def v35_config(v34_config: Dict[str, Any]) -> Dict[str, Any]:
+    r = v34_config.copy()
+    r.pop("disableLogAggregation")
+    r["_version"] = 35
+    return r
+
+
+@pytest.fixture
+def v36_config(v35_config: Dict[str, Any]) -> Dict[str, Any]:
+    r = v35_config.copy()
+    r.update(
+        {
+            "_version": 36,
+            "allowLiquidClasses": None,
+        }
+    )
+    return r
+
+
 @pytest.fixture(
     scope="session",
     params=[
@@ -449,6 +471,8 @@ def v34_config(v33_config: Dict[str, Any]) -> Dict[str, Any]:
         lazy_fixture("v32_config"),
         lazy_fixture("v33_config"),
         lazy_fixture("v34_config"),
+        lazy_fixture("v35_config"),
+        lazy_fixture("v36_config"),
     ],
 )
 def old_settings(request: SubRequest) -> Dict[str, Any]:
@@ -539,4 +563,5 @@ def test_ensures_config() -> None:
         "enableErrorRecoveryExperiments": None,
         "enableOEMMode": None,
         "enablePerformanceMetrics": None,
+        "allowLiquidClasses": None,
     }