Merge branch 'chore_release-8.5.0' into 850-snapshot-testing

Author: y3rsh

api/src/opentrons/protocol_api/_transfer_liquid_validation.py

@@ -93,3 +93,16 @@ def verify_and_normalize_transfer_args(
         tip_racks=valid_tip_racks,
         trash_location=valid_trash_location,
     )
+
+
+def resolve_keep_last_tip(
+    keep_last_tip: Optional[bool], tip_strategy: TransferTipPolicyV2
+) -> bool:
+    """Resolve the liquid class transfer argument `keep_last_tip`
+
+    If set to a boolean value, maintains that setting. Otherwise, default to
+    `True` if tip policy is `NEVER`, otherwise default to `False`
+    """
+    if keep_last_tip is not None:
+        return keep_last_tip
+    return tip_strategy == TransferTipPolicyV2.NEVER

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

@@ -1205,6 +1205,7 @@ def transfer_with_liquid_class(  # noqa: C901
         starting_tip: Optional[WellCore],
         trash_location: Union[Location, TrashBin, WasteChute],
         return_tip: bool,
+        keep_last_tip: bool,
     ) -> None:
         """Execute transfer using liquid class properties.
 
@@ -1404,15 +1405,14 @@ def _pick_up_tip() -> WellCore:
                     transfer_type=tx_comps_executor.TransferType.ONE_TO_ONE,
                     tip_contents=post_asp_tip_contents,
                     add_final_air_gap=(
-                        False
-                        if is_last_step and new_tip == TransferTipPolicyV2.NEVER
-                        else True
+                        False if is_last_step and keep_last_tip else True
                     ),
                     trash_location=trash_location,
                 )
             prev_src = step_source
             prev_dest = step_destination
-        if new_tip != TransferTipPolicyV2.NEVER:
+
+        if not keep_last_tip:
             _drop_tip()
 
     # TODO(spp, 2025-02-25): wire up return tip
@@ -1422,11 +1422,16 @@ def distribute_with_liquid_class(  # noqa: C901
         volume: float,
         source: Tuple[Location, WellCore],
         dest: List[Tuple[Location, WellCore]],
-        new_tip: Literal[TransferTipPolicyV2.NEVER, TransferTipPolicyV2.ONCE],
+        new_tip: Literal[
+            TransferTipPolicyV2.NEVER,
+            TransferTipPolicyV2.ONCE,
+            TransferTipPolicyV2.ALWAYS,
+        ],
         tip_racks: List[Tuple[Location, LabwareCore]],
         starting_tip: Optional[WellCore],
         trash_location: Union[Location, TrashBin, WasteChute],
         return_tip: bool,
+        keep_last_tip: bool,
     ) -> None:
         """Execute a distribution using liquid class properties.
 
@@ -1438,7 +1443,10 @@ def distribute_with_liquid_class(  # noqa: C901
             dest: List of destination wells, with each well represented as a tuple of
                     types.Location and WellCore.
                     types.Location is only necessary for saving the last accessed location.
-            new_tip: Whether the transfer should use a new tip 'once' or 'never'.
+            new_tip: Whether the transfer should use a new tip 'once', 'always' or 'never'.
+                     'never': the transfer will never pick up a new tip
+                     'once': the transfer will pick up a new tip once at the start of transfer
+                     'always': the transfer will pick up a new tip before every aspirate
             tiprack_uri: The URI of the tiprack that the transfer settings are for.
             tip_drop_location: Location where the tip will be dropped (if appropriate).
 
@@ -1454,7 +1462,11 @@ def distribute_with_liquid_class(  # noqa: C901
             raise RuntimeError(
                 "No tipracks found for pipette in order to perform transfer"
             )
-        assert new_tip in [TransferTipPolicyV2.NEVER, TransferTipPolicyV2.ONCE]
+        assert new_tip in [
+            TransferTipPolicyV2.NEVER,
+            TransferTipPolicyV2.ONCE,
+            TransferTipPolicyV2.ALWAYS,
+        ]
 
         tiprack_uri_for_transfer_props = tip_racks[0][1].get_uri()
         working_volume = min(
@@ -1510,6 +1522,7 @@ def distribute_with_liquid_class(  # noqa: C901
                 starting_tip=starting_tip,
                 trash_location=trash_location,
                 return_tip=return_tip,
+                keep_last_tip=keep_last_tip,
             )
             return
 
@@ -1661,9 +1674,21 @@ def _pick_up_tip() -> WellCore:
                 and new_tip != TransferTipPolicyV2.NEVER
                 and is_first_step
             ):
+                # Do probing only once, regardless of whether you are coming back to refill
+                # with a fresh tip since there's only one source well
                 enable_lpd = True
             else:
                 enable_lpd = False
+
+            if not is_first_step and new_tip == TransferTipPolicyV2.ALWAYS:
+                _drop_tip()
+                last_tip_picked_up_from = _pick_up_tip()
+                tip_contents = [
+                    tx_comps_executor.LiquidAndAirGapPair(
+                        liquid=0,
+                        air_gap=0,
+                    )
+                ]
             with self.lpd_for_transfer(enable=enable_lpd):
                 # Aspirate the total volume determined by the loop above
                 tip_contents = self.aspirate_liquid_class(
@@ -1693,9 +1718,7 @@ def _pick_up_tip() -> WellCore:
                         transfer_type=tx_comps_executor.TransferType.ONE_TO_MANY,
                         tip_contents=tip_contents,
                         add_final_air_gap=(
-                            False
-                            if is_last_step and new_tip == TransferTipPolicyV2.NEVER
-                            else True
+                            False if is_last_step and keep_last_tip else True
                         ),
                         trash_location=trash_location,
                     )
@@ -1708,17 +1731,15 @@ def _pick_up_tip() -> WellCore:
                         transfer_type=tx_comps_executor.TransferType.ONE_TO_MANY,
                         tip_contents=tip_contents,
                         add_final_air_gap=(
-                            False
-                            if is_last_step and new_tip == TransferTipPolicyV2.NEVER
-                            else True
+                            False if is_last_step and keep_last_tip else True
                         ),
                         trash_location=trash_location,
                         conditioning_volume=conditioning_vol,
                         disposal_volume=disposal_vol,
                     )
                 is_first_step = False
 
-        if new_tip != TransferTipPolicyV2.NEVER:
+        if not keep_last_tip:
             _drop_tip()
 
     def _tip_can_hold_volume_for_multi_dispensing(
@@ -1748,17 +1769,28 @@ def consolidate_with_liquid_class(  # noqa: C901
         volume: float,
         source: List[Tuple[Location, WellCore]],
         dest: Union[Tuple[Location, WellCore], TrashBin, WasteChute],
-        new_tip: Literal[TransferTipPolicyV2.NEVER, TransferTipPolicyV2.ONCE],
+        new_tip: Literal[
+            TransferTipPolicyV2.NEVER,
+            TransferTipPolicyV2.ONCE,
+            TransferTipPolicyV2.ALWAYS,
+        ],
         tip_racks: List[Tuple[Location, LabwareCore]],
         starting_tip: Optional[WellCore],
         trash_location: Union[Location, TrashBin, WasteChute],
         return_tip: bool,
+        keep_last_tip: bool,
     ) -> None:
         if not tip_racks:
             raise RuntimeError(
                 "No tipracks found for pipette in order to perform transfer"
             )
-        assert new_tip in [TransferTipPolicyV2.NEVER, TransferTipPolicyV2.ONCE]
+        # NOTE: Tip option of "always" in consolidate is equivalent to "after every dispense",
+        #       or more specifically, "before the next chunk of aspirates".
+        assert new_tip in [
+            TransferTipPolicyV2.NEVER,
+            TransferTipPolicyV2.ONCE,
+            TransferTipPolicyV2.ALWAYS,
+        ]
         tiprack_uri_for_transfer_props = tip_racks[0][1].get_uri()
         try:
             transfer_props = liquid_class.get_for(
@@ -1860,7 +1892,7 @@ def _pick_up_tip() -> WellCore:
             )
             return tip_well
 
-        if new_tip == TransferTipPolicyV2.ONCE:
+        if new_tip in [TransferTipPolicyV2.ONCE, TransferTipPolicyV2.ALWAYS]:
             last_tip_picked_up_from = _pick_up_tip()
 
         tip_contents = [
@@ -1872,6 +1904,7 @@ def _pick_up_tip() -> WellCore:
         next_step_volume, next_source = next(source_per_volume_step)
         is_first_step = True
         is_last_step = False
+        prev_src: Optional[tuple[Location, WellCore]] = None
         while not is_last_step:
             total_dispense_volume = 0.0
             vol_aspirate_combo = []
@@ -1895,6 +1928,21 @@ def _pick_up_tip() -> WellCore:
                 and is_first_step
             ):
                 enable_lpd = True
+            elif not is_first_step and new_tip == TransferTipPolicyV2.ALWAYS:
+                _drop_tip()
+                last_tip_picked_up_from = _pick_up_tip()
+                tip_contents = [
+                    tx_comps_executor.LiquidAndAirGapPair(
+                        liquid=0,
+                        air_gap=0,
+                    )
+                ]
+                # Enable LPD if it's globally enabled, as long as
+                # the next source is different from previous source
+                enable_lpd = (
+                    self.get_liquid_presence_detection()
+                    and prev_src != vol_aspirate_combo[0][1]
+                )
             else:
                 enable_lpd = False
 
@@ -1912,6 +1960,7 @@ def _pick_up_tip() -> WellCore:
                         ),
                         current_volume=total_aspirated_volume,
                     )
+                prev_src = step_source
                 total_aspirated_volume += step_volume
                 is_first_step = False
                 enable_lpd = False
@@ -1922,14 +1971,11 @@ def _pick_up_tip() -> WellCore:
                 transfer_properties=transfer_props,
                 transfer_type=tx_comps_executor.TransferType.MANY_TO_ONE,
                 tip_contents=tip_contents,
-                add_final_air_gap=(
-                    False
-                    if is_last_step and new_tip == TransferTipPolicyV2.NEVER
-                    else True
-                ),
+                add_final_air_gap=(False if is_last_step and keep_last_tip else True),
                 trash_location=trash_location,
             )
-        if new_tip != TransferTipPolicyV2.NEVER:
+
+        if not keep_last_tip:
             _drop_tip()
 
     def _get_location_and_well_core_from_next_tip_info(
@@ -2011,10 +2057,7 @@ def aspirate_liquid_class(
                 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()
-            ):
+            if self.get_liquid_presence_detection():
                 self.liquid_probe_with_recovery(
                     well_core=source_well, loc=prep_location
                 )