fix(api): meniscus relative positioning in LC based transfers

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

@@ -3,6 +3,7 @@
 from __future__ import annotations
 from contextlib import contextmanager
 from itertools import dropwhile
+from copy import deepcopy
 from typing import (
     Optional,
     TYPE_CHECKING,
@@ -88,6 +89,7 @@
     from opentrons.protocol_api._liquid_properties import (
         TransferProperties,
         MultiDispenseProperties,
+        SingleDispenseProperties,
     )
 
 _DISPENSE_VOLUME_VALIDATION_ADDED_IN = APIVersion(2, 17)
@@ -1955,22 +1957,54 @@ def aspirate_liquid_class(
             max_volume=self.get_working_volume(),
         )
         source_loc, source_well = source
+        last_liquid_and_airgap_in_tip = (
+            deepcopy(tip_contents[-1])  # don't modify caller's object
+            if tip_contents
+            else tx_comps_executor.LiquidAndAirGapPair(
+                liquid=0,
+                air_gap=0,
+            )
+        )
+        if volume_for_pipette_mode_configuration is not None:
+            prep_location = Location(
+                point=source_well.get_top(LIQUID_PROBE_START_OFFSET_FROM_WELL_TOP.z),
+                labware=source_loc.labware,
+            )
+            self.move_to(
+                location=prep_location,
+                well_core=source_well,
+                force_direct=False,
+                minimum_z_height=None,
+                speed=None,
+            )
+            self.remove_air_gap_during_transfer_with_liquid_class(
+                last_air_gap=last_liquid_and_airgap_in_tip.air_gap,
+                dispense_props=transfer_properties.dispense,
+                location=prep_location,
+            )
+            last_liquid_and_airgap_in_tip.air_gap = 0
+            if (
+                transfer_type != tx_comps_executor.TransferType.MANY_TO_ONE
+                and self.get_liquid_presence_detection()
+            ):
+                self.liquid_probe_with_recovery(
+                    well_core=source_well, loc=prep_location
+                )
+            # TODO: do volume configuration + prepare for aspirate only if the mode needs to be changed
+            self.configure_for_volume(volume_for_pipette_mode_configuration)
+            self.prepare_to_aspirate()
+
         aspirate_point = (
             tx_comps_executor.absolute_point_from_position_reference_and_offset(
                 well=source_well,
+                well_volume_difference=-volume,
                 position_reference=aspirate_props.aspirate_position.position_reference,
                 offset=aspirate_props.aspirate_position.offset,
+                mount=self.get_mount(),
             )
         )
         aspirate_location = Location(aspirate_point, labware=source_loc.labware)
-        last_liquid_and_airgap_in_tip = (
-            tip_contents[-1]
-            if tip_contents
-            else tx_comps_executor.LiquidAndAirGapPair(
-                liquid=0,
-                air_gap=0,
-            )
-        )
+
         components_executor = tx_comps_executor.TransferComponentsExecutor(
             instrument_core=self,
             transfer_properties=transfer_properties,
@@ -1982,9 +2016,7 @@ def aspirate_liquid_class(
             ),
         )
         components_executor.submerge(
-            submerge_properties=aspirate_props.submerge,
-            post_submerge_action="aspirate",
-            volume_for_pipette_mode_configuration=volume_for_pipette_mode_configuration,
+            submerge_properties=aspirate_props.submerge, post_submerge_action="aspirate"
         )
         # Do not do a pre-aspirate mix or pre-wet if consolidating
         if transfer_type != tx_comps_executor.TransferType.MANY_TO_ONE:
@@ -2023,6 +2055,38 @@ def aspirate_liquid_class(
         new_tip_contents = tip_contents[0:-1] + [last_contents]
         return new_tip_contents
 
+    def remove_air_gap_during_transfer_with_liquid_class(
+        self,
+        last_air_gap: float,
+        dispense_props: SingleDispenseProperties,
+        location: Location,
+    ) -> None:
+        """Remove an air gap that was previously added during a transfer."""
+        if last_air_gap == 0:
+            return
+
+        correction_volume = dispense_props.correction_by_volume.get_for_volume(
+            last_air_gap
+        )
+        # The minimum flow rate should be air_gap_volume per second
+        flow_rate = max(
+            dispense_props.flow_rate_by_volume.get_for_volume(last_air_gap),
+            last_air_gap,
+        )
+        self.dispense(
+            location=location,
+            well_core=None,
+            volume=last_air_gap,
+            rate=1,
+            flow_rate=flow_rate,
+            in_place=True,
+            push_out=0,
+            correction_volume=correction_volume,
+        )
+        dispense_delay = dispense_props.delay
+        if dispense_delay.enabled and dispense_delay.duration:
+            self.delay(dispense_delay.duration)
+
     def dispense_liquid_class(
         self,
         volume: float,
@@ -2069,8 +2133,10 @@ def dispense_liquid_class(
         dispense_point = (
             tx_comps_executor.absolute_point_from_position_reference_and_offset(
                 well=dest_well,
+                well_volume_difference=volume,
                 position_reference=dispense_props.dispense_position.position_reference,
                 offset=dispense_props.dispense_position.offset,
+                mount=self.get_mount(),
             )
         )
         dispense_location = Location(dispense_point, labware=dest_loc.labware)
@@ -2093,9 +2159,7 @@ def dispense_liquid_class(
             ),
         )
         components_executor.submerge(
-            submerge_properties=dispense_props.submerge,
-            post_submerge_action="dispense",
-            volume_for_pipette_mode_configuration=None,
+            submerge_properties=dispense_props.submerge, post_submerge_action="dispense"
         )
         push_out_vol = (
             0.0
@@ -2151,8 +2215,10 @@ def dispense_liquid_class_during_multi_dispense(
         dispense_point = (
             tx_comps_executor.absolute_point_from_position_reference_and_offset(
                 well=dest_well,
+                well_volume_difference=volume,
                 position_reference=dispense_props.dispense_position.position_reference,
                 offset=dispense_props.dispense_position.offset,
+                mount=self.get_mount(),
             )
         )
         dispense_location = Location(dispense_point, labware=dest_loc.labware)
@@ -2175,9 +2241,7 @@ def dispense_liquid_class_during_multi_dispense(
             ),
         )
         components_executor.submerge(
-            submerge_properties=dispense_props.submerge,
-            post_submerge_action="dispense",
-            volume_for_pipette_mode_configuration=None,
+            submerge_properties=dispense_props.submerge, post_submerge_action="dispense"
         )
         tip_starting_volume = self.get_current_volume()
         is_last_dispense_without_disposal_vol = (

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

@@ -12,7 +12,6 @@
     Coordinate,
     BlowoutLocation,
 )
-from opentrons_shared_data.pipette.types import LIQUID_PROBE_START_OFFSET_FROM_WELL_TOP
 
 from opentrons.protocol_api._liquid_properties import (
     Submerge,
@@ -23,7 +22,7 @@
     TouchTipProperties,
 )
 from opentrons.protocol_engine.errors import TouchTipDisabledError
-from opentrons.types import Location, Point
+from opentrons.types import Location, Point, Mount
 from opentrons.protocols.advanced_control.transfers.transfer_liquid_utils import (
     LocationCheckDescriptors,
 )
@@ -124,6 +123,26 @@ def __init__(
         tip_state: TipState,
         transfer_type: TransferType,
     ) -> None:
+        """Create a TransferComponentsExecutor instance.
+
+        One instance should be created to execute all the steps inside each of the
+        liquid class' transfer components- aspirate, dispense and multi-dispense.
+        The state of the TransferComponentsExecutor instance is expected to be valid
+        only for the component it was created.
+
+        For example, if we want to execute all the steps (submerge, dispense, retract, etc)
+        related to the 'dispense' component of a liquid-class based transfer, the class
+        will be used to initialize info about the dispense by assigning values
+        to class attributes as follows-
+        - target_location: the dispense location
+        - target_well: the well associated with dispense location
+        - tip_state: the state of the tip before dispense component steps are executed
+        - transfer_type: whether the dispense component is being called as a part of a
+                        1-to-1 transfer or a consolidation or a distribution
+
+        These attributes will remain the same throughout the component's execution,
+        except `tip_state`, which will keep updating as fluids are handled.
+        """
         self._instrument = instrument_core
         self._transfer_properties = transfer_properties
         self._target_location = target_location
@@ -140,7 +159,6 @@ def submerge(
         self,
         submerge_properties: Submerge,
         post_submerge_action: Literal["aspirate", "dispense"],
-        volume_for_pipette_mode_configuration: Optional[float],
     ) -> None:
         """Execute submerge steps.
 
@@ -153,41 +171,14 @@ def submerge(
         """
         submerge_start_point = absolute_point_from_position_reference_and_offset(
             well=self._target_well,
+            well_volume_difference=0,
             position_reference=submerge_properties.start_position.position_reference,
             offset=submerge_properties.start_position.offset,
+            mount=self._instrument.get_mount(),
         )
         submerge_start_location = Location(
             point=submerge_start_point, labware=self._target_location.labware
         )
-        prep_before_moving_to_submerge = (
-            post_submerge_action == "aspirate"
-            and volume_for_pipette_mode_configuration is not None
-        )
-        if prep_before_moving_to_submerge:
-            # Move to the tip probe start position
-            self._instrument.move_to(
-                location=Location(
-                    point=self._target_well.get_top(
-                        LIQUID_PROBE_START_OFFSET_FROM_WELL_TOP.z
-                    ),
-                    labware=self._target_location.labware,
-                ),
-                well_core=self._target_well,
-                force_direct=False,
-                minimum_z_height=None,
-                speed=None,
-            )
-            self._remove_air_gap(location=submerge_start_location)
-            if (
-                self._transfer_type != TransferType.MANY_TO_ONE
-                and self._instrument.get_liquid_presence_detection()
-            ):
-                self._instrument.liquid_probe_with_recovery(
-                    well_core=self._target_well, loc=submerge_start_location
-                )
-            # TODO: do volume configuration + prepare for aspirate only if the mode needs to be changed
-            self._instrument.configure_for_volume(volume_for_pipette_mode_configuration)  # type: ignore[arg-type]
-            self._instrument.prepare_to_aspirate()
         tx_utils.raise_if_location_inside_liquid(
             location=submerge_start_location,
             well_location=self._target_location,
@@ -205,8 +196,7 @@ def submerge(
             minimum_z_height=None,
             speed=None,
         )
-        if not prep_before_moving_to_submerge:
-            self._remove_air_gap(location=submerge_start_location)
+        self._remove_air_gap(location=submerge_start_location)
         self._instrument.move_to(
             location=self._target_location,
             well_core=self._target_well,
@@ -340,8 +330,10 @@ def retract_after_aspiration(
         retract_props = self._transfer_properties.aspirate.retract
         retract_point = absolute_point_from_position_reference_and_offset(
             well=self._target_well,
+            well_volume_difference=0,
             position_reference=retract_props.end_position.position_reference,
             offset=retract_props.end_position.offset,
+            mount=self._instrument.get_mount(),
         )
         retract_location = Location(
             retract_point, labware=self._target_location.labware
@@ -437,8 +429,10 @@ def retract_after_dispensing(
         retract_props = self._transfer_properties.dispense.retract
         retract_point = absolute_point_from_position_reference_and_offset(
             well=self._target_well,
+            well_volume_difference=0,
             position_reference=retract_props.end_position.position_reference,
             offset=retract_props.end_position.offset,
+            mount=self._instrument.get_mount(),
         )
         retract_location = Location(
             retract_point, labware=self._target_location.labware
@@ -579,8 +573,10 @@ def retract_during_multi_dispensing(
         retract_props = self._transfer_properties.multi_dispense.retract
         retract_point = absolute_point_from_position_reference_and_offset(
             well=self._target_well,
+            well_volume_difference=0,
             position_reference=retract_props.end_position.position_reference,
             offset=retract_props.end_position.offset,
+            mount=self._instrument.get_mount(),
         )
         retract_location = Location(
             retract_point, labware=self._target_location.labware
@@ -806,40 +802,30 @@ def _add_air_gap(self, air_gap_volume: float) -> None:
     def _remove_air_gap(self, location: Location) -> None:
         """Remove a previously added air gap."""
         last_air_gap = self._tip_state.last_liquid_and_air_gap_in_tip.air_gap
-        if last_air_gap == 0:
-            return
-
         dispense_props = self._transfer_properties.dispense
-        correction_volume = dispense_props.correction_by_volume.get_for_volume(
-            last_air_gap
-        )
-        # The minimum flow rate should be air_gap_volume per second
-        flow_rate = max(
-            dispense_props.flow_rate_by_volume.get_for_volume(last_air_gap),
-            last_air_gap,
-        )
-        self._instrument.dispense(
+        self._instrument.remove_air_gap_during_transfer_with_liquid_class(
+            last_air_gap=last_air_gap,
+            dispense_props=dispense_props,
             location=location,
-            well_core=None,
-            volume=last_air_gap,
-            rate=1,
-            flow_rate=flow_rate,
-            in_place=True,
-            push_out=0,
-            correction_volume=correction_volume,
         )
         self._tip_state.delete_air_gap(last_air_gap)
-        dispense_delay = dispense_props.delay
-        if dispense_delay.enabled and dispense_delay.duration:
-            self._instrument.delay(dispense_delay.duration)
 
 
 def absolute_point_from_position_reference_and_offset(
     well: WellCore,
+    well_volume_difference: float,
     position_reference: PositionReference,
     offset: Coordinate,
+    mount: Mount,
 ) -> Point:
-    """Return the absolute point, given the well, the position reference and offset."""
+    """Return the absolute point, given the well, the position reference and offset.
+
+    If using meniscus as the position reference, well_volume_difference should be specified.
+    `well_volume_difference` is the expected *difference* in well volume we want to consider
+    when estimating the height of the liquid meniscus after an aspirate/ dispense.
+    So, for liquid height estimation after an aspirate, well_volume_difference is
+    expected to be a -ve value while for a dispense, it will be a +ve value.
+    """
     match position_reference:
         case PositionReference.WELL_TOP:
             reference_point = well.get_top(0)
@@ -848,11 +834,17 @@ def absolute_point_from_position_reference_and_offset(
         case PositionReference.WELL_CENTER:
             reference_point = well.get_center()
         case PositionReference.LIQUID_MENISCUS:
-            meniscus_point = well.get_meniscus()
-            if not isinstance(meniscus_point, Point):
-                reference_point = well.get_center()
+            estimated_liquid_height = well.estimate_liquid_height_after_pipetting(
+                mount=mount,
+                operation_volume=well_volume_difference,
+            )
+            if isinstance(estimated_liquid_height, (float, int)):
+                reference_point = well.get_bottom(z_offset=estimated_liquid_height)
             else:
-                reference_point = meniscus_point
+                # If estimated liquid height gives a SimulatedProbeResult then
+                # assume meniscus is at well center.
+                # Will this cause more harm than good? Is there a better alternative to this?
+                reference_point = well.get_center()
         case _:
             raise ValueError(f"Unknown position reference {position_reference}")
     return reference_point + Point(offset.x, offset.y, offset.z)

api/src/opentrons/protocols/advanced_control/transfers/transfer_liquid_utils.py

@@ -48,7 +48,7 @@ def raise_if_location_inside_liquid(
         liquid_height_from_bottom = well_core.current_liquid_height()
     except LiquidHeightUnknownError:
         liquid_height_from_bottom = None
-    if liquid_height_from_bottom is not None:
+    if isinstance(liquid_height_from_bottom, (int, float)):
         if liquid_height_from_bottom + well_core.get_bottom(0).z > location.point.z:
             raise RuntimeError(
                 f"{location_check_descriptors.location_type.capitalize()} location {location} is"