feat(api): fully connected volume tracking

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

@@ -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,12 @@ async def execute(self, params: AspirateParams) -> _ExecuteReturn:
                 command_note_adder=self._command_note_adder,
             )
         except PipetteOverpressureError as e:
+            # TODO(pbm, 10-24-24): get new tip and LiquidProbe in error recovery to reestablish well liquid level
+            state_update.set_liquid_operated(
+                labware_id=labware_id,
+                well_name=well_name,
+                volume=None,
+            )
             return DefinedErrorData(
                 public=OverpressureError(
                     id=self._model_utils.generate_id(),
@@ -156,6 +158,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=-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
+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,16 @@ async def execute(self, params: AspirateInPlaceParams) -> _ExecuteReturn:
                 command_note_adder=self._command_note_adder,
             )
         except PipetteOverpressureError as e:
+            # TODO(pbm, 10-24-24): if location is a well, get new tip and LiquidProbe in error recovery to reestablish well liquid level
+            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=None,
+                )
             return DefinedErrorData(
                 public=OverpressureError(
                     id=self._model_utils.generate_id(),
@@ -121,10 +137,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=-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

@@ -107,6 +107,12 @@ async def execute(self, params: DispenseParams) -> _ExecuteReturn:
                 push_out=params.pushOut,
             )
         except PipetteOverpressureError as e:
+            # TODO(pbm, 10-24-24): get new tip and LiquidProbe in error recovery to reestablish well liquid level
+            state_update.set_liquid_operated(
+                labware_id=labware_id,
+                well_name=well_name,
+                volume=None,
+            )
             return DefinedErrorData(
                 public=OverpressureError(
                     id=self._model_utils.generate_id(),
@@ -123,6 +129,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=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
+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,16 @@ async def execute(self, params: DispenseInPlaceParams) -> _ExecuteReturn:
                 push_out=params.pushOut,
             )
         except PipetteOverpressureError as e:
+            # TODO(pbm, 10-24-24): if location is a well, get new tip and LiquidProbe in error recovery to reestablish well liquid level
+            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=None,
+                )
             return DefinedErrorData(
                 public=OverpressureError(
                     id=self._model_utils.generate_id(),
@@ -99,10 +116,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=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=None,
+                volume=None,
+                last_probed=self._model_utils.get_timestamp(),
+            )
             return DefinedErrorData(
                 public=LiquidNotFoundError(
                     id=self._model_utils.generate_id(),
@@ -220,6 +228,21 @@ async def execute(self, params: _CommonParams) -> _LiquidProbeExecuteReturn:
                 state_update=state_update,
             )
         else:
+            try:
+                well_volume = 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 = None
+            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 +262,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 +281,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 = None
+        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 = None
+
+        state_update.set_liquid_probed(
+            labware_id=params.labwareId,
+            well_name=params.wellName,
+            height=z_pos,
+            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]):

api/src/opentrons/protocol_engine/execution/queue_worker.py

@@ -69,7 +69,11 @@ async def join(self) -> None:
 
     async def _run_commands(self) -> None:
         async for command_id in self._command_generator():
-            await self._command_executor.execute(command_id=command_id)
+            try:
+                await self._command_executor.execute(command_id=command_id)
+            except BaseException:
+                log.exception("Unhandled failure in command executor")
+                raise
             # Yield to the event loop in case we're executing a long sequence of commands
             # that never yields internally. For example, a long sequence of comment commands.
             await asyncio.sleep(0)