refactor(api): make estimating liquid height after pipetting account for nozzles/well

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

@@ -3,7 +3,7 @@
 
 from opentrons_shared_data.labware.constants import WELL_NAME_PATTERN
 
-from opentrons.types import Point
+from opentrons.types import Point, Mount, MountType
 
 from opentrons.protocol_engine import WellLocation, WellOrigin, WellOffset
 from opentrons.protocol_engine import commands as cmd
@@ -13,6 +13,7 @@
     SimulatedProbeResult,
     LiquidTrackingType,
 )
+from opentrons.protocol_engine.errors import PipetteNotAttachedError
 
 from . import point_calculations
 from . import stringify
@@ -181,15 +182,25 @@ def from_center_cartesian(self, x: float, y: float, z: float) -> Point:
 
     def estimate_liquid_height_after_pipetting(
         self,
+        mount: Mount | str,
         operation_volume: float,
     ) -> LiquidTrackingType:
         """Return an estimate of liquid height after pipetting without raising an error."""
         labware_id = self.labware_id
         well_name = self._name
+        if isinstance(mount, Mount):
+            mount_type = MountType.from_hw_mount(mount)
+        else:
+            mount_type = MountType(mount)
+        pipette_from_mount = self._engine_client.state.pipettes.get_by_mount(mount_type)
+        if pipette_from_mount is None:
+            raise PipetteNotAttachedError(f"No pipette present on mount {mount}")
+        pipette_id = pipette_from_mount.id
         starting_liquid_height = self.current_liquid_height()
         projected_final_height = self._engine_client.state.geometry.get_well_height_after_liquid_handling_no_error(
             labware_id=labware_id,
             well_name=well_name,
+            pipette_id=pipette_id,
             initial_height=starting_liquid_height,
             volume=operation_volume,
         )

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

@@ -5,7 +5,7 @@
 
 from opentrons.protocols.api_support.util import APIVersionError
 
-from opentrons.types import Point
+from opentrons.types import Point, Mount
 
 from opentrons.protocol_engine.types.liquid_level_detection import (
     SimulatedProbeResult,
@@ -129,6 +129,7 @@ def from_center_cartesian(self, x: float, y: float, z: float) -> Point:
 
     def estimate_liquid_height_after_pipetting(
         self,
+        mount: Mount | str,
         operation_volume: float,
     ) -> LiquidTrackingType:
         """Estimate what the liquid height will be after pipetting, without raising an error."""

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

@@ -3,7 +3,7 @@
 from abc import ABC, abstractmethod
 from typing import TypeVar, Optional, Union
 
-from opentrons.types import Point
+from opentrons.types import Point, Mount
 from opentrons.protocol_engine.types import LiquidTrackingType
 
 from .._liquid import Liquid
@@ -91,6 +91,7 @@ def from_center_cartesian(self, x: float, y: float, z: float) -> Point:
     @abstractmethod
     def estimate_liquid_height_after_pipetting(
         self,
+        mount: Mount | str,
         operation_volume: float,
     ) -> LiquidTrackingType:
         """Estimate what the liquid height will be after pipetting, without raising an error."""

api/src/opentrons/protocol_api/labware.py

@@ -39,6 +39,7 @@
     Point,
     NozzleMapInterface,
     MeniscusTrackingTarget,
+    Mount,
 )
 from opentrons.protocols.api_support.types import APIVersion
 from opentrons.protocols.api_support.util import (
@@ -348,6 +349,7 @@ def current_liquid_volume(self) -> LiquidTrackingType:
     @requires_version(2, 21)
     def estimate_liquid_height_after_pipetting(
         self,
+        mount: Mount | str,
         operation_volume: float,
     ) -> LiquidTrackingType:
         """Check the height of the liquid within a well.
@@ -360,7 +362,7 @@ def estimate_liquid_height_after_pipetting(
         """
 
         projected_final_height = self._core.estimate_liquid_height_after_pipetting(
-            operation_volume=operation_volume,
+            operation_volume=operation_volume, mount=mount
         )
         return projected_final_height
 

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

@@ -199,6 +199,7 @@ async def aspirate_while_tracking(
             labware_id=labware_id,
             well_name=well_name,
             operation_volume=volume * -1,
+            pipette_id=pipette_id,
         )
         if isinstance(aspirate_z_distance, SimulatedProbeResult):
             raise InvalidLiquidHeightFound(
@@ -234,6 +235,7 @@ async def dispense_while_tracking(
             labware_id=labware_id,
             well_name=well_name,
             operation_volume=volume,
+            pipette_id=pipette_id,
         )
         if isinstance(dispense_z_distance, SimulatedProbeResult):
             raise InvalidLiquidHeightFound(

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

@@ -520,6 +520,7 @@ def get_well_position(
                 well_location=well_location,
                 well_depth=well_depth,
                 operation_volume=operation_volume,
+                pipette_id=pipette_id,
             )
             if not isinstance(offset_adjustment, SimulatedProbeResult):
                 offset = offset.model_copy(update={"z": offset.z + offset_adjustment})
@@ -1844,6 +1845,7 @@ def get_liquid_handling_z_change(
         self,
         labware_id: str,
         well_name: str,
+        pipette_id: str,
         operation_volume: float,
     ) -> float:
         """Get the change in height from a liquid handling operation."""
@@ -1853,6 +1855,7 @@ def get_liquid_handling_z_change(
         final_height = self.get_well_height_after_liquid_handling(
             labware_id=labware_id,
             well_name=well_name,
+            pipette_id=pipette_id,
             initial_height=initial_handling_height,
             volume=operation_volume,
         )
@@ -1873,6 +1876,7 @@ def get_well_offset_adjustment(
         well_name: str,
         well_location: WellLocationType,
         well_depth: float,
+        pipette_id: Optional[str] = None,
         operation_volume: Optional[float] = None,
     ) -> LiquidTrackingType:
         """Return a z-axis distance that accounts for well handling height and operation volume.
@@ -1906,9 +1910,14 @@ def get_well_offset_adjustment(
                 volume = well_location.volumeOffset
 
         if volume:
+            if pipette_id is None:
+                raise ValueError(
+                    "cannot get liquid handling offset without pipette id."
+                )
             liquid_height_after = self.get_well_height_after_liquid_handling(
                 labware_id=labware_id,
                 well_name=well_name,
+                pipette_id=pipette_id,
                 initial_height=initial_handling_height,
                 volume=volume,
             )
@@ -2030,6 +2039,7 @@ def get_well_height_after_liquid_handling(
         self,
         labware_id: str,
         well_name: str,
+        pipette_id: str,
         initial_height: LiquidTrackingType,
         volume: float,
     ) -> LiquidTrackingType:
@@ -2044,7 +2054,14 @@ def get_well_height_after_liquid_handling(
             initial_volume = find_volume_at_well_height(
                 target_height=initial_height, well_geometry=well_geometry
             )
-            final_volume = initial_volume + volume
+            final_volume = initial_volume + (
+                volume
+                * self.get_nozzles_per_well(
+                    labware_id=labware_id,
+                    target_well_name=well_name,
+                    pipette_id=pipette_id,
+                )
+            )
             return find_height_at_well_volume(
                 target_volume=final_volume, well_geometry=well_geometry
             )
@@ -2058,6 +2075,7 @@ def get_well_height_after_liquid_handling_no_error(
         self,
         labware_id: str,
         well_name: str,
+        pipette_id: str,
         initial_height: LiquidTrackingType,
         volume: float,
     ) -> LiquidTrackingType:
@@ -2073,7 +2091,14 @@ def get_well_height_after_liquid_handling_no_error(
             initial_volume = find_volume_at_well_height(
                 target_height=initial_height, well_geometry=well_geometry
             )
-            final_volume = initial_volume + volume
+            final_volume = initial_volume + (
+                volume
+                * self.get_nozzles_per_well(
+                    labware_id=labware_id,
+                    target_well_name=well_name,
+                    pipette_id=pipette_id,
+                )
+            )
             well_volume = find_height_at_well_volume(
                 target_volume=final_volume,
                 well_geometry=well_geometry,

api/tests/opentrons/protocol_api/core/engine/test_well_core.py

@@ -10,9 +10,15 @@
     RectangularWellDefinition2,
     CircularWellDefinition2,
 )
+from opentrons_shared_data.pipette.types import PipetteNameType
 
 from opentrons.protocol_api import MAX_SUPPORTED_VERSION
-from opentrons.protocol_engine import WellLocation, WellOrigin, WellOffset
+from opentrons.protocol_engine import (
+    WellLocation,
+    WellOrigin,
+    WellOffset,
+    LoadedPipette,
+)
 from opentrons.protocol_engine import commands as cmd
 from opentrons.protocol_engine.clients import SyncClient as EngineClient
 from opentrons.protocol_engine.errors.exceptions import (
@@ -21,7 +27,7 @@
 )
 from opentrons.protocols.api_support.types import APIVersion
 from opentrons.protocols.api_support.util import UnsupportedAPIError
-from opentrons.types import Point
+from opentrons.types import Point, Mount, MountType
 from opentrons_shared_data.labware.labware_definition import (
     InnerWellGeometry,
     ConicalFrustum,
@@ -312,11 +318,13 @@ def test_current_liquid_volume(
 
 
 @pytest.mark.parametrize("operation_volume", [0.0, 100, -100, 2, -4, 5])
+@pytest.mark.parametrize("mount", [Mount.LEFT, "left"])
 def test_estimate_liquid_height_after_pipetting(
     decoy: Decoy,
     subject: WellCore,
     mock_engine_client: EngineClient,
     operation_volume: float,
+    mount: Mount | str,
 ) -> None:
     """Make sure estimate_liquid_height_after_pipetting returns the correct value and does not raise an error."""
     fake_well_geometry = InnerWellGeometry(
@@ -355,14 +363,24 @@ def test_estimate_liquid_height_after_pipetting(
         mock_engine_client.state.geometry.get_well_height_after_liquid_handling_no_error(
             labware_id="labware-id",
             well_name="well-name",
+            pipette_id="pipette-id",
             initial_height=initial_liquid_height,
             volume=operation_volume,
         )
     ).then_return(fake_final_height)
+    decoy.when(
+        mock_engine_client.state.pipettes.get_by_mount(MountType.LEFT)
+    ).then_return(
+        LoadedPipette(
+            id="pipette-id",
+            pipetteName=PipetteNameType.P300_SINGLE,
+            mount=MountType.LEFT,
+        )
+    )
 
     # make sure that no error was raised
     final_height = subject.estimate_liquid_height_after_pipetting(
-        operation_volume=operation_volume,
+        operation_volume=operation_volume, mount=mount
     )
     assert final_height == fake_final_height
 

api/tests/opentrons/protocol_engine/execution/test_pipetting_handler.py

@@ -332,7 +332,10 @@ async def test_hw_aspirate_while_tracking(
 
     decoy.when(
         mock_state_view.geometry.get_liquid_handling_z_change(
-            labware_id="labware-id", well_name="A1", operation_volume=-25.0
+            labware_id="labware-id",
+            well_name="A1",
+            pipette_id="pipette_id",
+            operation_volume=-25.0,
         )
     ).then_return(4.544)
 

api/tests/opentrons/protocol_engine/state/test_geometry_view.py

@@ -1773,6 +1773,10 @@ def test_get_well_position_with_meniscus_and_literal_volume_offset(
     slot_pos = Point(4, 5, 6)
     well_def = well_plate_def.wells["B2"]
 
+    pip_type = PipetteNameType.P300_SINGLE
+    decoy.when(mock_pipette_view.get_nozzle_configuration("pipette-id")).then_return(
+        get_default_nozzle_map(pip_type)
+    )
     decoy.when(mock_labware_view.get("labware-id")).then_return(labware_data)
     decoy.when(mock_labware_view.get_definition("labware-id")).then_return(
         well_plate_def
@@ -1843,7 +1847,10 @@ def test_get_well_position_with_meniscus_and_float_volume_offset(
     calibration_offset = LabwareOffsetVector(x=1, y=-2, z=3)
     slot_pos = Point(4, 5, 6)
     well_def = well_plate_def.wells["B2"]
-
+    pip_type = PipetteNameType.P300_SINGLE
+    decoy.when(mock_pipette_view.get_nozzle_configuration("pipette-id")).then_return(
+        get_default_nozzle_map(pip_type)
+    )
     decoy.when(mock_labware_view.get("labware-id")).then_return(labware_data)
     decoy.when(mock_labware_view.get_definition("labware-id")).then_return(
         well_plate_def
@@ -1941,6 +1948,10 @@ def test_get_well_position_raises_validation_error(
     decoy.when(mock_labware_view.get_well_geometry("labware-id", "B2")).then_return(
         _TEST_INNER_WELL_GEOMETRY
     )
+    pip_type = PipetteNameType.P300_SINGLE
+    decoy.when(mock_pipette_view.get_nozzle_configuration("pipette-id")).then_return(
+        get_default_nozzle_map(pip_type)
+    )
     decoy.when(
         mock_pipette_view.get_current_tip_lld_settings(pipette_id="pipette-id")
     ).then_return(0.5)
@@ -4184,17 +4195,27 @@ def test_virtual_get_well_height_after_liquid_handling_no_error(
     decoy: Decoy,
     subject: GeometryView,
     mock_labware_view: LabwareView,
+    mock_pipette_view: PipetteView,
+    well_plate_def: LabwareDefinition,
     initial_liquid_height: LiquidTrackingType,
 ) -> None:
     """Make sure SimulatedLiquidProbe doesn't change geometry behavior."""
+    pip_type = PipetteNameType.P300_SINGLE
+    decoy.when(mock_pipette_view.get_nozzle_configuration("pipette-id")).then_return(
+        get_default_nozzle_map(pip_type)
+    )
+    decoy.when(mock_labware_view.get_definition("labware-id")).then_return(
+        well_plate_def
+    )
+
     decoy.when(mock_labware_view.get_well_geometry("labware-id", "B2")).then_return(
         _TEST_INNER_WELL_GEOMETRY
     )
     operation_volume = 1000.0
-
     result_estimate = subject.get_well_height_after_liquid_handling_no_error(
         labware_id="labware-id",
         well_name="B2",
+        pipette_id="pipette-id",
         initial_height=initial_liquid_height,
         volume=operation_volume,
     )
@@ -4228,13 +4249,23 @@ def test_virtual_find_height_and_volume(
 def test_get_liquid_handling_z_change(
     decoy: Decoy,
     subject: GeometryView,
+    well_plate_def: LabwareDefinition,
     mock_labware_view: LabwareView,
+    mock_pipette_view: PipetteView,
     mock_well_view: WellView,
 ) -> None:
     """Test for get_liquid_handling_z_change math."""
+    pip_type = PipetteNameType.P300_SINGLE
+    decoy.when(mock_pipette_view.get_nozzle_configuration("pipette-id")).then_return(
+        get_default_nozzle_map(pip_type)
+    )
+
     decoy.when(mock_labware_view.get_well_definition("labware-id", "A1")).then_return(
         RectangularWellDefinition3.model_construct(totalLiquidVolume=1100000)  # type: ignore[call-arg]
     )
+    decoy.when(mock_labware_view.get_definition("labware-id")).then_return(
+        well_plate_def
+    )
     decoy.when(mock_labware_view.get_well_geometry("labware-id", "A1")).then_return(
         _TEST_INNER_WELL_GEOMETRY
     )
@@ -4251,7 +4282,10 @@ def test_get_liquid_handling_z_change(
     )
     # make sure that liquid handling z change math stays the same
     change = subject.get_liquid_handling_z_change(
-        labware_id="labware-id", well_name="A1", operation_volume=199.0
+        labware_id="labware-id",
+        well_name="A1",
+        pipette_id="pipette-id",
+        operation_volume=199.0,
     )
     expected_change = 3.2968
     assert isclose(change, expected_change)