first pass: add fluid handling

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

@@ -25,6 +25,7 @@
 from ..errors.error_occurrence import ErrorOccurrence
 from ..errors.exceptions import PipetteNotReadyToAspirateError
 from ..state.update_types import StateUpdate
+from ..types import AspiratedFluid, FluidKind
 
 if TYPE_CHECKING:
     from ..execution import PipettingHandler, GantryMover
@@ -128,6 +129,7 @@ async def execute(self, params: AirGapInPlaceParams) -> _ExecuteReturn:
                 state_update=state_update,
             )
         else:
+            state_update.set_fluid_aspirated(pipette_id=pipette_id, fluid=AspiratedFluid(kind=FluidKind.AIR, volume=volume))
             return SuccessData(
                 public=AirGapInPlaceResult(volume=volume),
                 private=None,

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

@@ -25,7 +25,7 @@
 from opentrons.hardware_control import HardwareControlAPI
 
 from ..state.update_types import StateUpdate
-from ..types import WellLocation, WellOrigin, CurrentWell, DeckPoint
+from ..types import WellLocation, WellOrigin, CurrentWell, DeckPoint, AspiratedFluid, FluidKind
 
 if TYPE_CHECKING:
     from ..execution import MovementHandler, PipettingHandler
@@ -163,6 +163,7 @@ async def execute(self, params: AspirateParams) -> _ExecuteReturn:
                 well_name=well_name,
                 volume=-volume_aspirated,
             )
+            state_update.set_fluid_aspirated(pipette_id=pipette_id, fluid=AspiratedFluid(kind=FluidKind.LIQUID, volume=volume_aspirated))
             return SuccessData(
                 public=AspirateResult(
                     volume=volume_aspirated,

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

@@ -25,7 +25,7 @@
 from ..errors.error_occurrence import ErrorOccurrence
 from ..errors.exceptions import PipetteNotReadyToAspirateError
 from ..state.update_types import StateUpdate
-from ..types import CurrentWell
+from ..types import CurrentWell, AspiratedFluid, FluidKind
 
 if TYPE_CHECKING:
     from ..execution import PipettingHandler, GantryMover
@@ -140,6 +140,7 @@ async def execute(self, params: AspirateInPlaceParams) -> _ExecuteReturn:
                 state_update=state_update,
             )
         else:
+            state_update.set_fluid_aspirated(pipette_id=pipette_id, fluid=AspiratedFluid(kind=FluidKind.LIQUID, volume=volume_aspirated))
             if (
                 isinstance(current_location, CurrentWell)
                 and current_location.pipette_id == params.pipetteId
@@ -149,6 +150,8 @@ async def execute(self, params: AspirateInPlaceParams) -> _ExecuteReturn:
                     well_name=current_location.well_name,
                     volume=-volume,
                 )
+
+
             return SuccessData(
                 public=AspirateInPlaceResult(volume=volume),
                 private=None,

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

@@ -5,7 +5,7 @@
 from typing_extensions import Literal
 
 
-from ..state.update_types import StateUpdate
+from ..state.update_types import StateUpdate, CLEAR
 from ..types import DeckPoint
 from .pipetting_common import (
     OverpressureError,
@@ -114,6 +114,7 @@ async def execute(self, params: BlowOutParams) -> _ExecuteReturn:
                 ),
             )
         else:
+            state_update.set_fluid_ejected(pipette_id=pipette_id, volume=CLEAR)
             return SuccessData(
                 public=BlowOutResult(position=deck_point),
                 private=None,

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

@@ -99,6 +99,7 @@ async def execute(self, params: BlowOutInPlaceParams) -> _ExecuteReturn:
                 ),
             )
         else:
+            state_update.set_fluid_ejected(pipette_id=pipette_id, volume=CLEAR)
             return SuccessData(public=BlowOutInPlaceResult(), private=None)
 
 

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

@@ -134,6 +134,7 @@ async def execute(self, params: DispenseParams) -> _ExecuteReturn:
                 well_name=well_name,
                 volume=volume,
             )
+            state_update.set_fluid_ejected(pipette_id=pipette_id, volume=volume)
             return SuccessData(
                 public=DispenseResult(volume=volume, position=deck_point),
                 private=None,

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

@@ -119,6 +119,7 @@ async def execute(self, params: DispenseInPlaceParams) -> _ExecuteReturn:
                 state_update=state_update,
             )
         else:
+            state_update.set_fluid_ejected(pipette_id=pipette_id, volume=volume)
             if (
                 isinstance(current_location, CurrentWell)
                 and current_location.pipette_id == params.pipetteId

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

@@ -148,6 +148,7 @@ async def execute(self, params: DropTipParams) -> _ExecuteReturn:
             )
             return DefinedErrorData(public=error, state_update=state_update)
         else:
+            state_update.set_fluid_ejected(pipette_id=pipette_id, volume=CLEAR)
             state_update.update_pipette_tip_state(
                 pipette_id=params.pipetteId, tip_geometry=None
             )

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

@@ -85,6 +85,7 @@ async def execute(self, params: DropTipInPlaceParams) -> _ExecuteReturn:
             )
             return DefinedErrorData(public=error, state_update=state_update)
         else:
+            state_update.set_fluid_ejected(pipette_id=pipette_id, volume=CLEAR)
             state_update.update_pipette_tip_state(
                 pipette_id=params.pipetteId, tip_geometry=None
             )

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

@@ -166,6 +166,7 @@ async def execute(
             state_update.mark_tips_as_used(
                 pipette_id=pipette_id, labware_id=labware_id, well_name=well_name
             )
+            state_update.set_fluid_ejected(pipette_id=pipette_id, volume=CLEAR)
             return SuccessData(
                 public=PickUpTipResult(
                     tipVolume=tip_geometry.volume,

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

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import dataclasses
+from logging import getLogger
 from typing import (
     Dict,
     List,
@@ -11,6 +12,8 @@
     Union,
 )
 
+from numpy import isclose
+
 from opentrons_shared_data.pipette import pipette_definition
 from opentrons.config.defaults_ot2 import Z_RETRACT_DISTANCE
 from opentrons.hardware_control.dev_types import PipetteDict
@@ -33,6 +36,7 @@
     CurrentAddressableArea,
     CurrentPipetteLocation,
     TipGeometry,
+    AspiratedFluid
 )
 from ..actions import (
     Action,
@@ -43,6 +47,7 @@
 )
 from ._abstract_store import HasState, HandlesActions
 
+LOG = getLogger(__name__)
 
 @dataclasses.dataclass(frozen=True)
 class HardwarePipette:
@@ -99,6 +104,8 @@ class StaticPipetteConfig:
     default_nozzle_map: NozzleMap  # todo(mm, 2024-10-14): unused, remove?
     lld_settings: Optional[Dict[str, Dict[str, float]]]
 
+FluidStack = List[AspiratedFluid]
+
 
 @dataclasses.dataclass
 class PipetteState:
@@ -108,8 +115,7 @@ class PipetteState:
     # attributes are populated at the appropriate times. Refactor to a
     # single dict-of-many-things instead of many dicts-of-single-things.
     pipettes_by_id: Dict[str, LoadedPipette]
-    aspirated_volume_by_id: Dict[str, Optional[float]]
-    aspirated_liquid_by_id: Dict[str, Optional[float]]
+    pipette_contents_by_id: Dict[str, List[AspiratedFluid]]
     current_location: Optional[CurrentPipetteLocation]
     current_deck_point: CurrentDeckPoint
     attached_tip_by_id: Dict[str, Optional[TipGeometry]]
@@ -129,8 +135,7 @@ def __init__(self) -> None:
         """Initialize a PipetteStore and its state."""
         self._state = PipetteState(
             pipettes_by_id={},
-            aspirated_volume_by_id={},
-            aspirated_liquid_by_id={},
+            pipette_contents_by_id={}
             attached_tip_by_id={},
             current_location=None,
             current_deck_point=CurrentDeckPoint(mount=None, deck_point=None),
@@ -150,9 +155,7 @@ def handle_action(self, action: Action) -> None:
             self._update_pipette_config(state_update)
             self._update_pipette_nozzle_map(state_update)
             self._update_tip_state(state_update)
-
-        if isinstance(action, (SucceedCommandAction, FailCommandAction)):
-            self._update_volumes(action)
+            self._update_volumes(state_update)
 
         elif isinstance(action, SetPipetteMovementSpeedAction):
             self._state.movement_speed_by_id[action.pipette_id] = action.speed
@@ -169,8 +172,7 @@ def _set_load_pipette(self, state_update: update_types.StateUpdate) -> None:
             self._state.liquid_presence_detection_by_id[pipette_id] = (
                 state_update.loaded_pipette.liquid_presence_detection or False
             )
-            self._state.aspirated_volume_by_id[pipette_id] = None
-            self._state.aspirated_liquid_by_id[pipette_id] = None
+            self._state.pipette_contents_by_id[pipette_id] = []
             self._state.movement_speed_by_id[pipette_id] = None
             self._state.attached_tip_by_id[pipette_id] = None
 
@@ -181,8 +183,7 @@ def _update_tip_state(self, state_update: update_types.StateUpdate) -> None:
                 attached_tip = state_update.pipette_tip_state.tip_geometry
 
                 self._state.attached_tip_by_id[pipette_id] = attached_tip
-                self._state.aspirated_volume_by_id[pipette_id] = 0
-                self._state.aspirated_liquid_by_id[pipette_id] = 0
+                self._state.pipette_contents_by_id[pipette_id] = []
 
                 static_config = self._state.static_config_by_id.get(pipette_id)
                 if static_config:
@@ -209,8 +210,7 @@ def _update_tip_state(self, state_update: update_types.StateUpdate) -> None:
 
             else:
                 pipette_id = state_update.pipette_tip_state.pipette_id
-                self._state.aspirated_volume_by_id[pipette_id] = None
-                self._state.aspirated_liquid_by_id[pipette_id] = None
+                self._state.pipette_contents_by_id[pipette_id] = []
                 self._state.attached_tip_by_id[pipette_id] = None
 
                 static_config = self._state.static_config_by_id.get(pipette_id)
@@ -315,69 +315,58 @@ def _update_pipette_nozzle_map(
             ] = state_update.pipette_nozzle_map.nozzle_map
 
     def _update_volumes(
-        self, action: Union[SucceedCommandAction, FailCommandAction]
+        self, state_update: update_types.PipetteAspiratedFluidUpdate | update_types.PipetteEjectedFluidUpdate
     ) -> None:
-        # todo(mm, 2024-10-10): Port these isinstance checks to StateUpdate.
-        # https://opentrons.atlassian.net/browse/EXEC-754
-
-        if isinstance(action, SucceedCommandAction) and isinstance(
-            action.command.result,
-            (commands.AspirateResult, commands.AspirateInPlaceResult),
-        ):
-            pipette_id = action.command.params.pipetteId
-            previous_volume = self._state.aspirated_volume_by_id[pipette_id] or 0
-            previous_liquid = self._state.aspirated_liquid_by_id[pipette_id] or 0
-            # PipetteHandler will have clamped action.command.result.volume for us, so
-            # next_volume should always be in bounds.
-            next_volume = previous_volume + action.command.result.volume
-            next_liquid = previous_liquid + action.command.result.volume
-
-            self._state.aspirated_volume_by_id[pipette_id] = next_volume
-            self._state.aspirated_liquid_by_id[pipette_id] = next_liquid
-
-        elif isinstance(action, SucceedCommandAction) and isinstance(
-            action.command.result,
-            (commands.DispenseResult, commands.DispenseInPlaceResult),
-        ):
-            pipette_id = action.command.params.pipetteId
-            previous_volume = self._state.aspirated_volume_by_id[pipette_id] or 0
-            previous_liquid = self._state.aspirated_liquid_by_id[pipette_id] or 0
-            # PipetteHandler will have clamped action.command.result.volume for us, so
-            # next_volume should always be in bounds.
-            next_volume = previous_volume - action.command.result.volume
-            next_liquid = previous_liquid - action.command.result.volume
-            self._state.aspirated_volume_by_id[pipette_id] = next_volume
-            self._state.aspirated_liquid_by_id[pipette_id] = next_liquid
-
-        elif isinstance(action, SucceedCommandAction) and isinstance(
-            action.command.result,
-            (
-                commands.BlowOutResult,
-                commands.BlowOutInPlaceResult,
-                commands.unsafe.UnsafeBlowOutInPlaceResult,
-            ),
-        ):
-            pipette_id = action.command.params.pipetteId
-            self._state.aspirated_volume_by_id[pipette_id] = None
-            self._state.aspirated_liquid_by_id[pipette_id] = None
+        if state_update.pipette_aspirated_fluid != update_types.NO_CHANGE:
+            self._state.pipette_contents_by_id[state_update.pipette_id] = self._add_fluid(fluid_stack, state_update.fluid)
+        if state_update.pipette_ejected_fluid != update_types.NO_CHANGE:
+            if state_update.pipette_ejected_fluid.volume == update_types.CLEAR:
+                self._state.pipette_contents_by_id[state_update.pipette_id] = []
+            else:
+                self._state.pipette_contents_by_id[state_update.pipette_id] = self._remove_fluid(fluid_stack, state_update.volume)
 
-        elif isinstance(action, SucceedCommandAction) and isinstance(
-            action.command.result, commands.PrepareToAspirateResult
-        ):
-            pipette_id = action.command.params.pipetteId
-            self._state.aspirated_volume_by_id[pipette_id] = 0
-            self._state.aspirated_liquid_by_id[pipette_id] = 0
-        elif isinstance(action, SucceedCommandAction) and isinstance(
-            action.command.result, commands.AirGapInPlaceResult
-        ):
-            # The point of the air_gap command as a separate thing from the aspirate command
-            # is that air gap updates the aspirated volume but _not_ the aspirated liquid, since
-            # it is an air gap.
-            pipette_id = action.command.params.pipetteId
-            previous_volume = self._state.aspirated_volume_by_id[pipette_id] or 0
-            self._state.aspirated_volume_by_id[pipette_id] = (
-                previous_volume + action.command.result.volume
-            )
+        @staticmethod
+        def _add_fluid(fluid_stack: FluidStack, new: AspiratedFluid) -> FluidStack:
+            if len(fluid_stack) == 0 or fluid_stack[-1].kind != new.kind:
+                # this is a new kind of fluid, append the record
+                fluid_stack.append(new)
+            else:
+                # this is more of the same kind of fluid, add the volumes
+                old_fluid = fluid_stack.pop(-1)
+                fluid_stack.append(AspiratedFluid(kind=new.kind, volume=old_fluid.volume + new.volume))
+            return fluid_stack
+
+        @staticmethod
+        def _alter_fluid_records(fluid_stack: FluidStack, remove: int, new_last: AspiratedFluid | None) -> FluidStack:
+            if remove == len(fluid_stack) or len(fluid_stack) == 0:
+                return []
+            if remove != 0:
+                removed = fluid_stack[:-remove]
+            else:
+                removed = fluid_stack
+            if new_last:
+                fluid_stack.pop(-1)
+                removed.append(new_last)
+            return removed
+
+
+        @staticmethod
+        def _remove_fluid(fluid_stack: FluidStack, volume: float) -> FluidStack:
+            fluid_stack_iterator = reversed(fluid_stack)
+            removed_elements: List[AspiratedFluid] = []
+            while volume > 0:
+                try:
+                    last_stack_element = next(fluid_stack_iterator)
+                except StopIteration:
+                    LOG.error(f'Attempting to remove more fluid than present, {volume}uL left over')
+                    return _alter_fluid_records(fluid_stack, len(removed_elements), None)
+                if last_stack_element.volume < volume:
+                    return _alter_fluid_records(fluid_stack, len(removed_elements), AspiratedFluid(kind=last_stack_element.kind, volume=last_stack_element.volume - volume))
+                elif isclose(last_stack_element.volume, volume):
+                    return _alter_fluid_records(fluid_stack, len(removed_elements) + 1, None)
+                else:
+                    removed_elements.append(last_stack_element)
+                    volume -= last_stack_element.volume
 
 
 class PipetteView(HasState[PipetteState]):