feat(api): add load-empty-liquid (#16676)

We have a loadLiquid command and associated python protocol API method,
which allow users to specify that a well contains a given amount of a
given (predefined) liquid. This was all that we needed for a while
because we didn't really do anything with that information except
display the starting deck state based on it.

Now that we track liquid, however, this isn't enough, because we also
need a way for users to specify that a well is known to be empty. To
that end, introduce a special EMPTY sentinel value for liquid ID and
restrict loadLiquid commands that use it to setting a 0 volume.

I think we probably should have a better API for setting multiple wells
but that's not this PR.

Closes EXEC-801

## reviews
- seem like a valid thing to do? Didn't make sense to have a whole
different command

## testing
- make sure empty liquids get loaded

## notes
- [x] will rebase this PR to edge once the current target is merged

---------

Co-authored-by: Ed Cormany <[email protected]>

Author: sfoster1

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

@@ -130,7 +130,10 @@ def load_liquid(
         liquid: Liquid,
         volume: float,
     ) -> None:
-        """Load liquid into a well."""
+        """Load liquid into a well.
+
+        If the well is known to be empty, use ``load_empty()`` instead of calling this with a 0.0 volume.
+        """
         self._engine_client.execute_command(
             cmd.LoadLiquidParams(
                 labwareId=self._labware_id,
@@ -139,6 +142,22 @@ def load_liquid(
             )
         )
 
+    def load_empty(
+        self,
+    ) -> None:
+        """Inform the system that a well is known to be empty.
+
+        This should be done early in the protocol, at the same time as a load_liquid command might
+        be used.
+        """
+        self._engine_client.execute_command(
+            cmd.LoadLiquidParams(
+                labwareId=self._labware_id,
+                liquidId="EMPTY",
+                volumeByWell={self._name: 0.0},
+            )
+        )
+
     def from_center_cartesian(self, x: float, y: float, z: float) -> Point:
         """Gets point in deck coordinates based on percentage of the radius of each axis."""
         well_size = self._engine_client.state.labware.get_well_size(

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

@@ -114,6 +114,10 @@ def load_liquid(
         """Load liquid into a well."""
         raise APIVersionError(api_element="Loading a liquid")
 
+    def load_empty(self) -> None:
+        """Mark a well as empty."""
+        assert False, "load_empty only supported on engine core"
+
     def from_center_cartesian(self, x: float, y: float, z: float) -> Point:
         """Gets point in deck coordinates based on percentage of the radius of each axis."""
         return self._geometry.from_center_cartesian(x, y, z)

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

@@ -79,6 +79,10 @@ def load_liquid(
     ) -> None:
         """Load liquid into a well."""
 
+    @abstractmethod
+    def load_empty(self) -> None:
+        """Mark a well as containing no liquid."""
+
     @abstractmethod
     def from_center_cartesian(self, x: float, y: float, z: float) -> Point:
         """Gets point in deck coordinates based on percentage of the radius of each axis."""

api/src/opentrons/protocol_api/labware.py

@@ -280,12 +280,20 @@ def load_liquid(self, liquid: Liquid, volume: float) -> None:
 
         :param Liquid liquid: The liquid to load into the well.
         :param float volume: The volume of liquid to load, in µL.
+
+        .. note::
+            In API version 2.22 and later, use :py:meth:`~.Well.load_empty()` to mark a well as empty at the beginning of a protocol, rather than using this method with ``volume=0``.
         """
         self._core.load_liquid(
             liquid=liquid,
             volume=volume,
         )
 
+    @requires_version(2, 22)
+    def load_empty(self) -> None:
+        """Mark a well as empty."""
+        self._core.load_empty()
+
     def _from_center_cartesian(self, x: float, y: float, z: float) -> Point:
         """
         Private version of from_center_cartesian. Present only for backward

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

@@ -5,6 +5,8 @@
 from typing_extensions import Literal
 
 from opentrons.protocol_engine.state.update_types import StateUpdate
+from opentrons.protocol_engine.types import LiquidId
+from opentrons.protocol_engine.errors import InvalidLiquidError
 
 from .command import AbstractCommandImpl, BaseCommand, BaseCommandCreate, SuccessData
 from ..errors.error_occurrence import ErrorOccurrence
@@ -19,17 +21,17 @@
 class LoadLiquidParams(BaseModel):
     """Payload required to load a liquid into a well."""
 
-    liquidId: str = Field(
+    liquidId: LiquidId = Field(
         ...,
-        description="Unique identifier of the liquid to load.",
+        description="Unique identifier of the liquid to load. If this is the sentinel value EMPTY, all values of volumeByWell must be 0.",
     )
     labwareId: str = Field(
         ...,
         description="Unique identifier of labware to load liquid into.",
     )
     volumeByWell: Dict[str, float] = Field(
         ...,
-        description="Volume of liquid, in µL, loaded into each well by name, in this labware.",
+        description="Volume of liquid, in µL, loaded into each well by name, in this labware. If the liquid id is the sentinel value EMPTY, all volumes must be 0.",
     )
 
 
@@ -57,6 +59,12 @@ async def execute(self, params: LoadLiquidParams) -> SuccessData[LoadLiquidResul
         self._state_view.labware.validate_liquid_allowed_in_labware(
             labware_id=params.labwareId, wells=params.volumeByWell
         )
+        if params.liquidId == "EMPTY":
+            for well_name, volume in params.volumeByWell.items():
+                if volume != 0.0:
+                    raise InvalidLiquidError(
+                        'loadLiquid commands that specify the special liquid "EMPTY" must set volume to be 0.0, but the volume for {well_name} is {volume}'
+                    )
 
         state_update = StateUpdate()
         state_update.set_liquid_loaded(

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

@@ -77,6 +77,7 @@
     OperationLocationNotInWellError,
     InvalidDispenseVolumeError,
     StorageLimitReachedError,
+    InvalidLiquidError,
 )
 
 from .error_occurrence import ErrorOccurrence, ProtocolCommandFailedError
@@ -137,6 +138,7 @@
     "InvalidTargetSpeedError",
     "InvalidBlockVolumeError",
     "InvalidHoldTimeError",
+    "InvalidLiquidError",
     "CannotPerformModuleAction",
     "ResumeFromRecoveryNotAllowedError",
     "PauseNotAllowedError",

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

@@ -244,6 +244,19 @@ def __init__(
         super().__init__(ErrorCodes.GENERAL_ERROR, message, details, wrapping)
 
 
+class InvalidLiquidError(ProtocolEngineError):
+    """Raised when attempting to add a liquid with an invalid property."""
+
+    def __init__(
+        self,
+        message: Optional[str] = None,
+        details: Optional[Dict[str, Any]] = None,
+        wrapping: Optional[Sequence[EnumeratedError]] = None,
+    ) -> None:
+        """Build an InvalidLiquidError."""
+        super().__init__(ErrorCodes.INVALID_PROTOCOL_DATA, message, details, wrapping)
+
+
 class LabwareDefinitionDoesNotExistError(ProtocolEngineError):
     """Raised when referencing a labware definition that does not exist."""
 

api/src/opentrons/protocol_engine/protocol_engine.py

@@ -566,9 +566,12 @@ def add_liquid(
             description=(description or ""),
             displayColor=color,
         )
+        validated_liquid = self._state_store.liquid.validate_liquid_allowed(
+            liquid=liquid
+        )
 
-        self._action_dispatcher.dispatch(AddLiquidAction(liquid=liquid))
-        return liquid
+        self._action_dispatcher.dispatch(AddLiquidAction(liquid=validated_liquid))
+        return validated_liquid
 
     def add_addressable_area(self, addressable_area_name: str) -> None:
         """Add an addressable area to state."""

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

@@ -1,11 +1,11 @@
 """Basic liquid data state and store."""
 from dataclasses import dataclass
 from typing import Dict, List
-from opentrons.protocol_engine.types import Liquid
+from opentrons.protocol_engine.types import Liquid, LiquidId
 
 from ._abstract_store import HasState, HandlesActions
 from ..actions import Action, AddLiquidAction
-from ..errors import LiquidDoesNotExistError
+from ..errors import LiquidDoesNotExistError, InvalidLiquidError
 
 
 @dataclass
@@ -51,11 +51,23 @@ def get_all(self) -> List[Liquid]:
         """Get all protocol liquids."""
         return list(self._state.liquids_by_id.values())
 
-    def validate_liquid_id(self, liquid_id: str) -> str:
+    def validate_liquid_id(self, liquid_id: LiquidId) -> LiquidId:
         """Check if liquid_id exists in liquids."""
+        is_empty = liquid_id == "EMPTY"
+        if is_empty:
+            return liquid_id
         has_liquid = liquid_id in self._state.liquids_by_id
         if not has_liquid:
             raise LiquidDoesNotExistError(
                 f"Supplied liquidId: {liquid_id} does not exist in the loaded liquids."
             )
         return liquid_id
+
+    def validate_liquid_allowed(self, liquid: Liquid) -> Liquid:
+        """Validate that a liquid is legal to load."""
+        is_empty = liquid.id == "EMPTY"
+        if is_empty:
+            raise InvalidLiquidError(
+                message='Protocols may not define a liquid with the special id "EMPTY".'
+            )
+        return liquid

api/src/opentrons/protocol_engine/types.py

@@ -828,6 +828,10 @@ def _color_is_a_valid_hex(cls, v: str) -> str:
         return v
 
 
+EmptyLiquidId = Literal["EMPTY"]
+LiquidId = str | EmptyLiquidId
+
+
 class Liquid(BaseModel):
     """Payload required to create a liquid."""