feat(api): implement loadLiquidClass command in PE

api/src/opentrons/protocol_engine/commands/command_unions.py

@@ -136,6 +136,14 @@
     LoadLiquidCommandType,
 )
 
+from .load_liquid_class import (
+    LoadLiquidClass,
+    LoadLiquidClassParams,
+    LoadLiquidClassCreate,
+    LoadLiquidClassResult,
+    LoadLiquidClassCommandType,
+)
+
 from .load_module import (
     LoadModule,
     LoadModuleParams,
@@ -347,6 +355,7 @@
         LoadLabware,
         ReloadLabware,
         LoadLiquid,
+        LoadLiquidClass,
         LoadModule,
         LoadPipette,
         MoveLabware,
@@ -429,6 +438,7 @@
     LoadLabwareParams,
     ReloadLabwareParams,
     LoadLiquidParams,
+    LoadLiquidClassParams,
     LoadModuleParams,
     LoadPipetteParams,
     MoveLabwareParams,
@@ -509,6 +519,7 @@
     LoadLabwareCommandType,
     ReloadLabwareCommandType,
     LoadLiquidCommandType,
+    LoadLiquidClassCommandType,
     LoadModuleCommandType,
     LoadPipetteCommandType,
     MoveLabwareCommandType,
@@ -590,6 +601,7 @@
         LoadLabwareCreate,
         ReloadLabwareCreate,
         LoadLiquidCreate,
+        LoadLiquidClassCreate,
         LoadModuleCreate,
         LoadPipetteCreate,
         MoveLabwareCreate,
@@ -672,6 +684,7 @@
     LoadLabwareResult,
     ReloadLabwareResult,
     LoadLiquidResult,
+    LoadLiquidClassResult,
     LoadModuleResult,
     LoadPipetteResult,
     MoveLabwareResult,

api/src/opentrons/protocol_engine/commands/load_liquid_class.py

@@ -0,0 +1,137 @@
+"""LoadLiquidClass stores the liquid class settings used for a transfer into the Protocol Engine."""
+from __future__ import annotations
+
+from typing import Optional, Type, TYPE_CHECKING
+from typing_extensions import Literal
+from pydantic import BaseModel, Field
+
+from .command import AbstractCommandImpl, BaseCommand, BaseCommandCreate, SuccessData
+from ..errors import LiquidClassDoesNotExistError
+from ..errors.error_occurrence import ErrorOccurrence
+from ..errors.exceptions import LiquidClassRedefinitionError
+from ..state.update_types import LiquidClassLoadedUpdate, StateUpdate
+from ..types import LiquidClassRecord
+
+if TYPE_CHECKING:
+    from ..state.state import StateView
+    from ..resources import ModelUtils
+
+LoadLiquidClassCommandType = Literal["loadLiquidClass"]
+
+
+class LoadLiquidClassParams(BaseModel):
+    """The liquid class transfer properties to store."""
+
+    liquidClassId: Optional[str] = Field(
+        None,
+        description="Unique identifier for the liquid class to store. "
+        "If you do not supply a liquidClassId, we will generate one.",
+    )
+    liquidClassRecord: LiquidClassRecord = Field(
+        ...,
+        description="The liquid class to store.",
+    )
+
+
+class LoadLiquidClassResult(BaseModel):
+    """Result from execution of LoadLiquidClass command."""
+
+    liquidClassId: str = Field(
+        ...,
+        description="The ID for the liquid class that was loaded, either the one you "
+        "supplied or the one we generated.",
+    )
+
+
+class LoadLiquidClassImplementation(
+    AbstractCommandImpl[LoadLiquidClassParams, SuccessData[LoadLiquidClassResult]]
+):
+    """Load Liquid Class command implementation."""
+
+    def __init__(
+        self, state_view: StateView, model_utils: ModelUtils, **kwargs: object
+    ) -> None:
+        self._state_view = state_view
+        self._model_utils = model_utils
+
+    async def execute(
+        self, params: LoadLiquidClassParams
+    ) -> SuccessData[LoadLiquidClassResult]:
+        """Store the liquid class in the Protocol Engine."""
+        liquid_class_id: Optional[str]
+        already_loaded = False
+
+        if params.liquidClassId:
+            liquid_class_id = params.liquidClassId
+            if self._liquid_class_id_already_loaded(
+                liquid_class_id, params.liquidClassRecord
+            ):
+                already_loaded = True
+        else:
+            liquid_class_id = (
+                self._state_view.liquid_classes.get_id_for_liquid_class_record(
+                    params.liquidClassRecord
+                )  # if liquidClassRecord was already loaded, reuse the existing ID
+            )
+            if liquid_class_id:
+                already_loaded = True
+            else:
+                liquid_class_id = self._model_utils.generate_id()
+
+        if already_loaded:
+            state_update = StateUpdate()  # liquid class already loaded, do nothing
+        else:
+            state_update = StateUpdate(
+                liquid_class_loaded=LiquidClassLoadedUpdate(
+                    liquid_class_id=liquid_class_id,
+                    liquid_class_record=params.liquidClassRecord,
+                )
+            )
+
+        return SuccessData(
+            public=LoadLiquidClassResult(liquidClassId=liquid_class_id),
+            state_update=state_update,
+        )
+
+    def _liquid_class_id_already_loaded(
+        self, liquid_class_id: str, liquid_class_record: LiquidClassRecord
+    ) -> bool:
+        """Check if the liquid_class_id has already been loaded.
+
+        If it has, make sure that liquid_class_record matches the previously loaded definition.
+        """
+        try:
+            existing_liquid_class_record = self._state_view.liquid_classes.get(
+                liquid_class_id
+            )
+        except LiquidClassDoesNotExistError:
+            return False
+
+        if liquid_class_record != existing_liquid_class_record:
+            raise LiquidClassRedefinitionError(
+                f"Liquid class {liquid_class_id} conflicts with previously loaded definition."
+            )
+        return True
+
+
+class LoadLiquidClass(
+    BaseCommand[LoadLiquidClassParams, LoadLiquidClassResult, ErrorOccurrence]
+):
+    """Load Liquid Class command resource model."""
+
+    commandType: LoadLiquidClassCommandType = "loadLiquidClass"
+    params: LoadLiquidClassParams
+    result: Optional[LoadLiquidClassResult]
+
+    _ImplementationCls: Type[
+        LoadLiquidClassImplementation
+    ] = LoadLiquidClassImplementation
+
+
+class LoadLiquidClassCreate(BaseCommandCreate[LoadLiquidClassParams]):
+    """Load Liquid Class command creation request."""
+
+    commandType: LoadLiquidClassCommandType = "loadLiquidClass"
+    params: LoadLiquidClassParams
+
+    _CommandCls: Type[LoadLiquidClass] = LoadLiquidClass

api/src/opentrons/protocol_engine/errors/__init__.py

@@ -79,6 +79,7 @@
     StorageLimitReachedError,
     InvalidLiquidError,
     LiquidClassDoesNotExistError,
+    LiquidClassRedefinitionError,
 )
 
 from .error_occurrence import ErrorOccurrence, ProtocolCommandFailedError
@@ -166,4 +167,5 @@
     "InvalidDispenseVolumeError",
     "StorageLimitReachedError",
     "LiquidClassDoesNotExistError",
+    "LiquidClassRedefinitionError",
 ]

api/src/opentrons/protocol_engine/errors/exceptions.py

@@ -1167,3 +1167,15 @@ def __init__(
         wrapping: Optional[Sequence[EnumeratedError]] = None,
     ) -> None:
         super().__init__(ErrorCodes.GENERAL_ERROR, message, details, wrapping)
+
+
+class LiquidClassRedefinitionError(ProtocolEngineError):
+    """Raised when attempting to load a liquid class that conflicts with a liquid class already loaded."""
+
+    def __init__(
+        self,
+        message: Optional[str] = None,
+        details: Optional[Dict[str, Any]] = None,
+        wrapping: Optional[Sequence[EnumeratedError]] = None,
+    ) -> None:
+        super().__init__(ErrorCodes.GENERAL_ERROR, message, details, wrapping)

api/src/opentrons/protocol_engine/state/state.py

@@ -25,6 +25,7 @@
 from .pipettes import PipetteState, PipetteStore, PipetteView
 from .modules import ModuleState, ModuleStore, ModuleView
 from .liquids import LiquidState, LiquidView, LiquidStore
+from .liquid_classes import LiquidClassState, LiquidClassStore, LiquidClassView
 from .tips import TipState, TipView, TipStore
 from .wells import WellState, WellView, WellStore
 from .geometry import GeometryView
@@ -49,6 +50,7 @@ class State:
     pipettes: PipetteState
     modules: ModuleState
     liquids: LiquidState
+    liquid_classes: LiquidClassState
     tips: TipState
     wells: WellState
     files: FileState
@@ -64,6 +66,7 @@ class StateView(HasState[State]):
     _pipettes: PipetteView
     _modules: ModuleView
     _liquid: LiquidView
+    _liquid_classes: LiquidClassView
     _tips: TipView
     _wells: WellView
     _geometry: GeometryView
@@ -101,6 +104,11 @@ def liquid(self) -> LiquidView:
         """Get state view selectors for liquid state."""
         return self._liquid
 
+    @property
+    def liquid_classes(self) -> LiquidClassView:
+        """Get state view selectors for liquid class state."""
+        return self._liquid_classes
+
     @property
     def tips(self) -> TipView:
         """Get state view selectors for tip state."""
@@ -148,6 +156,7 @@ def get_summary(self) -> StateSummary:
             wells=self._wells.get_all(),
             hasEverEnteredErrorRecovery=self._commands.get_has_entered_recovery_mode(),
             files=self._state.files.file_ids,
+            # TODO(dc): Do we want to just dump all the liquid classes into the summary?
         )
 
 
@@ -213,6 +222,7 @@ def __init__(
             module_calibration_offsets=module_calibration_offsets,
         )
         self._liquid_store = LiquidStore()
+        self._liquid_class_store = LiquidClassStore()
         self._tip_store = TipStore()
         self._well_store = WellStore()
         self._file_store = FileStore()
@@ -224,6 +234,7 @@ def __init__(
             self._labware_store,
             self._module_store,
             self._liquid_store,
+            self._liquid_class_store,
             self._tip_store,
             self._well_store,
             self._file_store,
@@ -342,6 +353,7 @@ def _get_next_state(self) -> State:
             pipettes=self._pipette_store.state,
             modules=self._module_store.state,
             liquids=self._liquid_store.state,
+            liquid_classes=self._liquid_class_store.state,
             tips=self._tip_store.state,
             wells=self._well_store.state,
             files=self._file_store.state,
@@ -359,6 +371,7 @@ def _initialize_state(self) -> None:
         self._pipettes = PipetteView(state.pipettes)
         self._modules = ModuleView(state.modules)
         self._liquid = LiquidView(state.liquids)
+        self._liquid_classes = LiquidClassView(state.liquid_classes)
         self._tips = TipView(state.tips)
         self._wells = WellView(state.wells)
         self._files = FileView(state.files)
@@ -391,6 +404,7 @@ def _update_state_views(self) -> None:
         self._pipettes._state = next_state.pipettes
         self._modules._state = next_state.modules
         self._liquid._state = next_state.liquids
+        self._liquid_classes._state = next_state.liquid_classes
         self._tips._state = next_state.tips
         self._wells._state = next_state.wells
         self._change_notifier.notify()