feat(api): fully connected volume tracking (#16532)

# Overview

This PR aims to consolidate well volume data from our existing relevant
commands into a single source of truth in WellStore. These commands are
LiquidProbe, LoadLiquid, Aspirate, and Dispense.

## volume estimation and height
Well volume and height are tracked in three parts:
- the probed height is a height gathered from a liquid probe since the
last volume-affecting operation occurred; it is nulled out if you carry
out a liquid operation on a well
- the tracked volume is a volume that started with a `LoadLiquid` and
was then modified by liquid volume tracking from aspirate and dispense
volumes
- the estimated volume is the same, but has been updated or started with
a liquid height measurement that was converted into an estimated volume

The reason for having all three is that they have different accuracy
characteristics, and we therefore use them for meniscus height
estimation in an order of precedence:
- The most accurate is a probed height
- If a probed height isn't available, use a tracked volume that we turn
into a height
- If that isn't available, use an estimated volume that we turn into a
height

The latter two values are only accessible if the labware the well is in
has internal geometry data. Even when we do, that geometry data is noisy
and can be inaccurate, making those values suitable for problems like "I
don't want to have to accurately measure how much is in this tube" and
not for problems like "I want to minimize external droplet formation on
my 1uL aspiration".

This PR implements all that, as well as setting the movement data for
actually going to a well meniscus.

## volume offsets

A pretty important thing to consider for _static_ meniscus-relative
movements is that when you're aspirating you don't want to move to where
the meniscus _is_ but rather where it will be when the operation ends,
so you don't leave your pipette high and dry. We signify this by
allowing a "volume offset" in a location. This is like a height offset
but is specified in uL or the special value "operationVolume". It will
add an extra Z-offset to the positioning to account for a volume change
in a well. This is calculated using the well's internal geometry and
thus only available if the well has internal geometry. If the value is
the string "operationVolume", which should be thought of as a sentinel
value, we even calculate the volume offset from the volume you passed to
`aspirate`; but specifying the value as a float is available for things
like a `moveToWell` preceding an `aspirateInPlace`.

There's some gross internals because of history. One particularly gross
one is the Location data type; this is an old friend indeed, and its
friendship continues to pay dividends since we want to add something to
it that is somewhat internal, and we have to do so in a really gross
way.

## testing
We really need to test this in actually used protocols and this
shouldn't be relied upon yet, but extensive test coverage should help.

---------

Co-authored-by: Seth Foster <[email protected]>

analyses-snapshot-testing/citools/generate_analyses.py

@@ -186,6 +186,7 @@ def analyze(protocol: TargetProtocol, container: docker.models.containers.Contai
     start_time = time.time()
     result = None
     exit_code = None
+    console.print(f"Beginning analysis of {protocol.host_protocol_file.name}")
     try:
         command_result = container.exec_run(cmd=command)
         exit_code = command_result.exit_code

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

@@ -153,6 +153,8 @@ def aspirate(
                 absolute_point=location.point,
                 is_meniscus=is_meniscus,
             )
+            if well_location.origin == WellOrigin.MENISCUS:
+                well_location.volumeOffset = "operationVolume"
             pipette_movement_conflict.check_safe_for_pipette_movement(
                 engine_state=self._engine_client.state,
                 pipette_id=self._pipette_id,

api/src/opentrons/protocol_engine/actions/get_state_update.py

@@ -1,5 +1,6 @@
 # noqa: D100
-
+from __future__ import annotations
+from typing import TYPE_CHECKING
 
 from .actions import (
     Action,
@@ -9,7 +10,9 @@
 )
 from ..commands.command import DefinedErrorData
 from ..error_recovery_policy import ErrorRecoveryType
-from ..state.update_types import StateUpdate
+
+if TYPE_CHECKING:
+    from ..state.update_types import StateUpdate
 
 
 def get_state_updates(action: Action) -> list[StateUpdate]:

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

@@ -24,7 +24,7 @@
 
 from opentrons.hardware_control import HardwareControlAPI
 
-from ..state.update_types import StateUpdate
+from ..state.update_types import StateUpdate, CLEAR
 from ..types import WellLocation, WellOrigin, CurrentWell, DeckPoint
 
 if TYPE_CHECKING:
@@ -112,15 +112,11 @@ async def execute(self, params: AspirateParams) -> _ExecuteReturn:
                 well_name=well_name,
             )
 
-        well_location = params.wellLocation
-        if well_location.origin == WellOrigin.MENISCUS:
-            well_location.volumeOffset = "operationVolume"
-
         position = await self._movement.move_to_well(
             pipette_id=pipette_id,
             labware_id=labware_id,
             well_name=well_name,
-            well_location=well_location,
+            well_location=params.wellLocation,
             current_well=current_well,
             operation_volume=-params.volume,
         )
@@ -140,6 +136,11 @@ async def execute(self, params: AspirateParams) -> _ExecuteReturn:
                 command_note_adder=self._command_note_adder,
             )
         except PipetteOverpressureError as e:
+            state_update.set_liquid_operated(
+                labware_id=labware_id,
+                well_name=well_name,
+                volume_added=CLEAR,
+            )
             return DefinedErrorData(
                 public=OverpressureError(
                     id=self._model_utils.generate_id(),
@@ -156,6 +157,11 @@ async def execute(self, params: AspirateParams) -> _ExecuteReturn:
                 state_update=state_update,
             )
         else:
+            state_update.set_liquid_operated(
+                labware_id=labware_id,
+                well_name=well_name,
+                volume_added=-volume_aspirated,
+            )
             return SuccessData(
                 public=AspirateResult(
                     volume=volume_aspirated,

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

@@ -24,6 +24,8 @@
 )
 from ..errors.error_occurrence import ErrorOccurrence
 from ..errors.exceptions import PipetteNotReadyToAspirateError
+from ..state.update_types import StateUpdate, CLEAR
+from ..types import CurrentWell
 
 if TYPE_CHECKING:
     from ..execution import PipettingHandler, GantryMover
@@ -91,6 +93,10 @@ async def execute(self, params: AspirateInPlaceParams) -> _ExecuteReturn:
                 " The first aspirate following a blow-out must be from a specific well"
                 " so the plunger can be reset in a known safe position."
             )
+
+        state_update = StateUpdate()
+        current_location = self._state_view.pipettes.get_current_location()
+
         try:
             current_position = await self._gantry_mover.get_position(params.pipetteId)
             volume = await self._pipetting.aspirate_in_place(
@@ -100,6 +106,15 @@ async def execute(self, params: AspirateInPlaceParams) -> _ExecuteReturn:
                 command_note_adder=self._command_note_adder,
             )
         except PipetteOverpressureError as e:
+            if (
+                isinstance(current_location, CurrentWell)
+                and current_location.pipette_id == params.pipetteId
+            ):
+                state_update.set_liquid_operated(
+                    labware_id=current_location.labware_id,
+                    well_name=current_location.well_name,
+                    volume_added=CLEAR,
+                )
             return DefinedErrorData(
                 public=OverpressureError(
                     id=self._model_utils.generate_id(),
@@ -121,10 +136,22 @@ async def execute(self, params: AspirateInPlaceParams) -> _ExecuteReturn:
                         }
                     ),
                 ),
+                state_update=state_update,
             )
         else:
+            if (
+                isinstance(current_location, CurrentWell)
+                and current_location.pipette_id == params.pipetteId
+            ):
+                state_update.set_liquid_operated(
+                    labware_id=current_location.labware_id,
+                    well_name=current_location.well_name,
+                    volume_added=-volume,
+                )
             return SuccessData(
-                public=AspirateInPlaceResult(volume=volume), private=None
+                public=AspirateInPlaceResult(volume=volume),
+                private=None,
+                state_update=state_update,
             )
 
 

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

@@ -8,7 +8,7 @@
 from pydantic import Field
 
 from ..types import DeckPoint
-from ..state.update_types import StateUpdate
+from ..state.update_types import StateUpdate, CLEAR
 from .pipetting_common import (
     PipetteIdMixin,
     DispenseVolumeMixin,
@@ -107,6 +107,11 @@ async def execute(self, params: DispenseParams) -> _ExecuteReturn:
                 push_out=params.pushOut,
             )
         except PipetteOverpressureError as e:
+            state_update.set_liquid_operated(
+                labware_id=labware_id,
+                well_name=well_name,
+                volume_added=CLEAR,
+            )
             return DefinedErrorData(
                 public=OverpressureError(
                     id=self._model_utils.generate_id(),
@@ -123,6 +128,11 @@ async def execute(self, params: DispenseParams) -> _ExecuteReturn:
                 state_update=state_update,
             )
         else:
+            state_update.set_liquid_operated(
+                labware_id=labware_id,
+                well_name=well_name,
+                volume_added=volume,
+            )
             return SuccessData(
                 public=DispenseResult(volume=volume, position=deck_point),
                 private=None,

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

@@ -21,10 +21,13 @@
     DefinedErrorData,
 )
 from ..errors.error_occurrence import ErrorOccurrence
+from ..state.update_types import StateUpdate, CLEAR
+from ..types import CurrentWell
 
 if TYPE_CHECKING:
     from ..execution import PipettingHandler, GantryMover
     from ..resources import ModelUtils
+    from ..state.state import StateView
 
 
 DispenseInPlaceCommandType = Literal["dispenseInPlace"]
@@ -59,16 +62,20 @@ class DispenseInPlaceImplementation(
     def __init__(
         self,
         pipetting: PipettingHandler,
+        state_view: StateView,
         gantry_mover: GantryMover,
         model_utils: ModelUtils,
         **kwargs: object,
     ) -> None:
         self._pipetting = pipetting
+        self._state_view = state_view
         self._gantry_mover = gantry_mover
         self._model_utils = model_utils
 
     async def execute(self, params: DispenseInPlaceParams) -> _ExecuteReturn:
         """Dispense without moving the pipette."""
+        state_update = StateUpdate()
+        current_location = self._state_view.pipettes.get_current_location()
         try:
             current_position = await self._gantry_mover.get_position(params.pipetteId)
             volume = await self._pipetting.dispense_in_place(
@@ -78,6 +85,15 @@ async def execute(self, params: DispenseInPlaceParams) -> _ExecuteReturn:
                 push_out=params.pushOut,
             )
         except PipetteOverpressureError as e:
+            if (
+                isinstance(current_location, CurrentWell)
+                and current_location.pipette_id == params.pipetteId
+            ):
+                state_update.set_liquid_operated(
+                    labware_id=current_location.labware_id,
+                    well_name=current_location.well_name,
+                    volume_added=CLEAR,
+                )
             return DefinedErrorData(
                 public=OverpressureError(
                     id=self._model_utils.generate_id(),
@@ -99,10 +115,22 @@ async def execute(self, params: DispenseInPlaceParams) -> _ExecuteReturn:
                         }
                     ),
                 ),
+                state_update=state_update,
             )
         else:
+            if (
+                isinstance(current_location, CurrentWell)
+                and current_location.pipette_id == params.pipetteId
+            ):
+                state_update.set_liquid_operated(
+                    labware_id=current_location.labware_id,
+                    well_name=current_location.well_name,
+                    volume_added=volume,
+                )
             return SuccessData(
-                public=DispenseInPlaceResult(volume=volume), private=None
+                public=DispenseInPlaceResult(volume=volume),
+                private=None,
+                state_update=state_update,
             )
 
 

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

@@ -11,6 +11,7 @@
     MustHomeError,
     PipetteNotReadyToAspirateError,
     TipNotEmptyError,
+    IncompleteLabwareDefinitionError,
 )
 from opentrons.types import MountType
 from opentrons_shared_data.errors.exceptions import (
@@ -205,6 +206,13 @@ async def execute(self, params: _CommonParams) -> _LiquidProbeExecuteReturn:
             self._state_view, self._movement, self._pipetting, params
         )
         if isinstance(z_pos_or_error, PipetteLiquidNotFoundError):
+            state_update.set_liquid_probed(
+                labware_id=params.labwareId,
+                well_name=params.wellName,
+                height=update_types.CLEAR,
+                volume=update_types.CLEAR,
+                last_probed=self._model_utils.get_timestamp(),
+            )
             return DefinedErrorData(
                 public=LiquidNotFoundError(
                     id=self._model_utils.generate_id(),
@@ -220,6 +228,23 @@ async def execute(self, params: _CommonParams) -> _LiquidProbeExecuteReturn:
                 state_update=state_update,
             )
         else:
+            try:
+                well_volume: float | update_types.ClearType = (
+                    self._state_view.geometry.get_well_volume_at_height(
+                        labware_id=params.labwareId,
+                        well_name=params.wellName,
+                        height=z_pos_or_error,
+                    )
+                )
+            except IncompleteLabwareDefinitionError:
+                well_volume = update_types.CLEAR
+            state_update.set_liquid_probed(
+                labware_id=params.labwareId,
+                well_name=params.wellName,
+                height=z_pos_or_error,
+                volume=well_volume,
+                last_probed=self._model_utils.get_timestamp(),
+            )
             return SuccessData(
                 public=LiquidProbeResult(
                     z_position=z_pos_or_error, position=deck_point
@@ -239,11 +264,13 @@ def __init__(
         state_view: StateView,
         movement: MovementHandler,
         pipetting: PipettingHandler,
+        model_utils: ModelUtils,
         **kwargs: object,
     ) -> None:
         self._state_view = state_view
         self._movement = movement
         self._pipetting = pipetting
+        self._model_utils = model_utils
 
     async def execute(self, params: _CommonParams) -> _TryLiquidProbeExecuteReturn:
         """Execute a `tryLiquidProbe` command.
@@ -256,11 +283,26 @@ async def execute(self, params: _CommonParams) -> _TryLiquidProbeExecuteReturn:
             self._state_view, self._movement, self._pipetting, params
         )
 
-        z_pos = (
-            None
-            if isinstance(z_pos_or_error, PipetteLiquidNotFoundError)
-            else z_pos_or_error
+        if isinstance(z_pos_or_error, PipetteLiquidNotFoundError):
+            z_pos = None
+            well_volume: float | update_types.ClearType = update_types.CLEAR
+        else:
+            z_pos = z_pos_or_error
+            try:
+                well_volume = self._state_view.geometry.get_well_volume_at_height(
+                    labware_id=params.labwareId, well_name=params.wellName, height=z_pos
+                )
+            except IncompleteLabwareDefinitionError:
+                well_volume = update_types.CLEAR
+
+        state_update.set_liquid_probed(
+            labware_id=params.labwareId,
+            well_name=params.wellName,
+            height=z_pos if z_pos is not None else update_types.CLEAR,
+            volume=well_volume,
+            last_probed=self._model_utils.get_timestamp(),
         )
+
         return SuccessData(
             public=TryLiquidProbeResult(
                 z_position=z_pos,

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

@@ -4,11 +4,14 @@
 from typing import Optional, Type, Dict, TYPE_CHECKING
 from typing_extensions import Literal
 
+from opentrons.protocol_engine.state.update_types import StateUpdate
+
 from .command import AbstractCommandImpl, BaseCommand, BaseCommandCreate, SuccessData
 from ..errors.error_occurrence import ErrorOccurrence
 
 if TYPE_CHECKING:
     from ..state.state import StateView
+    from ..resources import ModelUtils
 
 LoadLiquidCommandType = Literal["loadLiquid"]
 
@@ -41,8 +44,11 @@ class LoadLiquidImplementation(
 ):
     """Load liquid command implementation."""
 
-    def __init__(self, state_view: StateView, **kwargs: object) -> None:
+    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: LoadLiquidParams
@@ -54,7 +60,16 @@ async def execute(
             labware_id=params.labwareId, wells=params.volumeByWell
         )
 
-        return SuccessData(public=LoadLiquidResult(), private=None)
+        state_update = StateUpdate()
+        state_update.set_liquid_loaded(
+            labware_id=params.labwareId,
+            volumes=params.volumeByWell,
+            last_loaded=self._model_utils.get_timestamp(),
+        )
+
+        return SuccessData(
+            public=LoadLiquidResult(), private=None, state_update=state_update
+        )
 
 
 class LoadLiquid(BaseCommand[LoadLiquidParams, LoadLiquidResult, ErrorOccurrence]):