Chore: incremental mergeback of 8.4.0 into edge

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/instrument_context.py

@@ -606,7 +606,20 @@ def blow_out(
                     "Blow_out being performed on a tiprack. "
                     "Please re-check your code"
                 )
-            move_to_location = target.location or target.well.top()
+            if target.location:
+                # because the lower levels of blowout don't handle LiquidHandlingWellLocation and
+                # there is no "operation_volume" for blowout we need to convert the relative location
+                # given with a .meniscus to an absolute point. To maintain the meniscus behavior
+                # we can just add the offset to the current liquid height.
+                if target.location.meniscus_tracking:
+                    move_to_location = target.well.bottom(
+                        target.well.current_liquid_height()  # type: ignore [arg-type]
+                        + target.location.point.z
+                    )
+                else:
+                    move_to_location = target.location
+            else:
+                move_to_location = target.well.top()
             well = target.well
         elif isinstance(target, validation.PointTarget):
             move_to_location = target.location
@@ -2698,7 +2711,8 @@ def measure_liquid_height(self, well: labware.Well) -> LiquidTrackingType:
         """
         self._raise_if_pressure_not_supported_by_pipette()
         loc = well.top()
-        return self._core.liquid_probe_without_recovery(well._core, loc)
+        self._core.liquid_probe_with_recovery(well._core, loc)
+        return well.current_liquid_height()
 
     def _raise_if_configuration_not_supported_by_pipette(
         self, style: NozzleLayout

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/src/opentrons/protocol_engine/types/__init__.py

@@ -135,6 +135,7 @@
     WellInfoSummary,
     WellLiquidInfo,
     LiquidTrackingType,
+    SimulatedProbeResult,
 )
 from .liquid_handling import FlowRates
 from .labware_movement import LabwareMovementStrategy, LabwareMovementOffsetData
@@ -280,6 +281,7 @@
     "WellInfoSummary",
     "WellLiquidInfo",
     "LiquidTrackingType",
+    "SimulatedProbeResult",
     # Liquid handling
     "FlowRates",
     # Labware movement

api/src/opentrons/protocol_engine/types/liquid_level_detection.py

@@ -1,9 +1,10 @@
 """Protocol Engine types to do with liquid level detection."""
+
 from __future__ import annotations
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Optional, List
-from pydantic import BaseModel, model_serializer, field_validator
+from typing import Optional, List, Any
+from pydantic import BaseModel, model_serializer, model_validator
 
 
 class SimulatedProbeResult(BaseModel):
@@ -17,6 +18,14 @@ def serialize_model(self) -> str:
         """Serialize instances of this class as a string."""
         return "SimulatedProbeResult"
 
+    @model_validator(mode="before")
+    @classmethod
+    def validate_model(cls, data: object) -> Any:
+        """Handle deserializing from a simulated probe result."""
+        if isinstance(data, str) and data == "SimulatedProbeResult":
+            return {}
+        return data
+
     def __add__(
         self, other: float | SimulatedProbeResult
     ) -> float | SimulatedProbeResult:
@@ -75,7 +84,9 @@ def simulate_probed_aspirate_dispense(self, volume: float) -> None:
         self.operations_after_probe.append(volume)
 
 
-LiquidTrackingType = SimulatedProbeResult | float
+# Work around https://github.com/pydantic/pydantic/issues/6830 - do not change the order of
+# this union
+LiquidTrackingType = float | SimulatedProbeResult
 
 
 class LoadedVolumeInfo(BaseModel):
@@ -104,23 +115,6 @@ class ProbedVolumeInfo(BaseModel):
 class WellInfoSummary(BaseModel):
     """Payload for a well's liquid info in StateSummary."""
 
-    # TODO(cm): 3/21/25: refactor SimulatedLiquidProbe in a way that
-    # doesn't require models like this one that are just using it to
-    # need a custom validator
-    @field_validator("probed_height", "probed_volume", mode="before")
-    @classmethod
-    def validate_simulated_probe_result(
-        cls, input_val: object
-    ) -> LiquidTrackingType | None:
-        """Return the appropriate input to WellInfoSummary from json data."""
-        if input_val is None:
-            return None
-        if isinstance(input_val, LiquidTrackingType):
-            return input_val
-        if isinstance(input_val, str) and input_val == "SimulatedProbeResult":
-            return SimulatedProbeResult()
-        raise ValueError(f"Invalid input value {input_val} to WellInfoSummary")
-
     labware_id: str
     well_name: str
     loaded_volume: Optional[float] = None

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