refactor(api): move last tip picked up tracking from context to engine (#19061)

# Overview

This PR refactors the way the Python API tracks where a tip has been
picked up from. Previously we were relying on the
`InstrumentContext._last_tip_picked_up_from` to keep track of which tip
rack well the current tip was picked up from, mostly for the purposes of
storing it for `return_tip`. The issue with this approach is that
anything that did not go through the top level instrument context,
whether it is via the cores or directly in protocol engine, would lead
to this not being updated and lead to either incorrect behavior for tip
returns or an error.

This became more acute with the introduction of liquid classes and
allowing them to return tips and keep them at the end, as this variable
wasn't getting updated and we had to manually code around that.

Now the tracking of the last tip rack well has been moved entirely to
the engine, where it will store it as a combination of `labware_id` and
`well_name`. This is then reconstructed into a `LabwareCore` and
`WellCore` by the `InstrumentCore`, which is further reconstructed into
a `Well` object by the top level `InstrumentContext`. This allowed a lot
of the liquid class scaffolding around this issue to be removed and
those methods to be simplified.

This change also required adding this functionality to the legacy core.
With no engine, in those older cores we are now simply storing the
legacy labware and well cores when a pick up tip or drop tip is called,
allowing this new method to work via both intefaces.

Because it was discovered a lot of older protocols rely on
`_last_tip_picked_up_from`, a property of the same name was added that
uses the new interface. In addition, a version gated
`current_tip_source_well()` function was added to provide a
version-protected method that should now be used instead.

## Test Plan and Hands on Testing

Snapshots and integration tests should cover a lot of the regression
testing, but the following protocols will be tested:

```
requirements = {
	"robotType": "Flex",
	"apiLevel": "2.25"
}


metadata = {
    "protocolName":'Return tip new',
    'author':'Jeremy'
}

def run(protocol_context):
	tiprack = protocol_context.load_labware("opentrons_flex_96_tiprack_200ul", "C2")
	trash = protocol_context.load_trash_bin('A3')
	pipette_1k = protocol_context.load_instrument("flex_1channel_1000", "right", tip_racks=[tiprack])
	nest_plate = protocol_context.load_labware("nest_96_wellplate_2ml_deep", "D1")
	arma_plate = protocol_context.load_labware("armadillo_96_wellplate_200ul_pcr_full_skirt", "D3")

	pipette_1k.pick_up_tip()

	pipette_1k.return_tip()

	water_class = protocol_context.get_liquid_class("water")

	pipette_1k.transfer_with_liquid_class(
		liquid_class=water_class,
		volume=200,
		source=nest_plate.wells()[:4],
		dest=arma_plate.wells()[:4],
		new_tip="always",
		trash_location=trash,
		return_tip=True,
	)

	pipette_1k.transfer_with_liquid_class(
		liquid_class=water_class,
		volume=200,
		source=nest_plate.wells()[0],
		dest=arma_plate.wells()[0],
		new_tip="once",
		trash_location=trash,
		keep_last_tip=True
	)


	pipette_1k.return_tip()
```
```
requirements = {
    "robotType": "OT-2",
    "apiLevel": "2.13"
}

metadata = {
    "protocolName":'Return tip OT-2 2.13',
    'author':'Jeremy'
}

def run(protocol_context):
    p300rack = protocol_context.load_labware('opentrons_96_tiprack_300ul', 1)
    p300 = protocol_context.load_instrument('p300_single_gen2', 'left', tip_racks=[p300rack])

    p300.pick_up_tip()

    p300.return_tip()
```

## Changelog

- Refactored `_last_tip_picked_up_from` to use engine and cores to keep
track of tip, not the instrument context

## Review requests

Does the way we reconstruct the `Well` object make sense?

## Risk assessment

Medium/high, this refactor touches a lot of long existing methods and is
not gated by version. There's no new logic added but this affects not
only Flex run code but OT-2 older API versions as well.

---------

Co-authored-by: David Chau <[email protected]>

Author: jbleon95

api/src/opentrons/protocol_api/_transfer_liquid_validation.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from typing import List, Union, Sequence, Optional, Tuple
+from typing import List, Union, Sequence, Optional
 
 from opentrons.types import Location, NozzleMapInterface
 from opentrons.protocols.api_support import instrument
@@ -13,7 +13,6 @@
 
 from .disposal_locations import TrashBin, WasteChute
 from .labware import Labware, Well
-from .core.common import WellCore
 from . import validation
 
 
@@ -25,7 +24,6 @@ class TransferInfo:
     tip_policy: TransferTipPolicyV2
     tip_racks: List[Labware]
     trash_location: Union[Location, TrashBin, WasteChute]
-    last_tip_location: Optional[Tuple[Location, WellCore]]
 
 
 def verify_and_normalize_transfer_args(
@@ -88,22 +86,12 @@ def verify_and_normalize_transfer_args(
         trash_location=_trash_location
     )
 
-    if last_tip_well is not None:
-        parent_tip_rack = last_tip_well.parent
-        last_tip_location = (
-            Location(last_tip_well.top().point, parent_tip_rack),
-            last_tip_well._core,
-        )
-    else:
-        last_tip_location = None
-
     return TransferInfo(
         source=flat_sources_list,
         dest=flat_dests_list if not isinstance(dest, (TrashBin, WasteChute)) else dest,
         tip_policy=valid_new_tip,
         tip_racks=valid_tip_racks,
         trash_location=valid_trash_location,
-        last_tip_location=last_tip_location,
     )
 
 

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

@@ -1047,6 +1047,25 @@ def get_nozzle_configuration(self) -> NozzleConfigurationType:
     def get_liquid_presence_detection(self) -> bool:
         return self._liquid_presence_detection
 
+    def get_tip_origin(
+        self,
+    ) -> Optional[Tuple[LabwareCore, WellCore]]:
+        last_tip_pickup_info = (
+            self._engine_client.state.pipettes.get_tip_rack_well_picked_up_from(
+                self._pipette_id
+            )
+        )
+        if last_tip_pickup_info is None:
+            return None
+        else:
+            tip_rack_labware_core = self._protocol_core._labware_cores_by_id[
+                last_tip_pickup_info.labware_id
+            ]
+            tip_well_core = tip_rack_labware_core.get_well_core(
+                last_tip_pickup_info.well_name
+            )
+            return tip_rack_labware_core, tip_well_core
+
     def is_tip_tracking_available(self) -> bool:
         if self.get_nozzle_configuration() == NozzleConfigurationType.FULL:
             return True
@@ -1215,8 +1234,7 @@ def transfer_with_liquid_class(  # noqa: C901
         trash_location: Union[Location, TrashBin, WasteChute],
         return_tip: bool,
         keep_last_tip: bool,
-        last_tip_location: Optional[Tuple[Location, WellCore]],
-    ) -> Optional[Tuple[Location, WellCore]]:
+    ) -> None:
         """Execute transfer using liquid class properties.
 
         Args:
@@ -1238,10 +1256,6 @@ def transfer_with_liquid_class(  # noqa: C901
             return_tip: If `True`, return tips to the tip rack location they were picked up from,
                         otherwise drop in `trash_location`
             keep_last_tip: When set to `True`, do not drop the final tip used in the transfer.
-            last_tip_location: If a tip is already attached, this will be the tiprack and well it was
-                           picked up from, represented as a tuple of types.Location and WellCore.
-                           Used so a tip can be returned if it was picked up outside this function
-                           as could be the case for a new_tip of `never`.
         """
         if not tip_racks:
             raise RuntimeError(
@@ -1278,9 +1292,8 @@ def transfer_with_liquid_class(  # noqa: C901
             )
         )
 
-        last_tip = last_tip_location
         if new_tip == TransferTipPolicyV2.ONCE:
-            last_tip = self._pick_up_tip_for_liquid_class(
+            self._pick_up_tip_for_liquid_class(
                 tip_racks, starting_tip, tiprack_uri_for_transfer_props
             )
 
@@ -1319,10 +1332,8 @@ def transfer_with_liquid_class(  # noqa: C901
                 )
             ):
                 if prev_src is not None and prev_dest is not None:
-                    self._drop_tip_for_liquid_class(
-                        trash_location, last_tip, return_tip
-                    )
-                last_tip = self._pick_up_tip_for_liquid_class(
+                    self._drop_tip_for_liquid_class(trash_location, return_tip)
+                self._pick_up_tip_for_liquid_class(
                     tip_racks, starting_tip, tiprack_uri_for_transfer_props
                 )
                 post_disp_tip_contents = [
@@ -1354,10 +1365,7 @@ def transfer_with_liquid_class(  # noqa: C901
             prev_dest = step_destination
 
         if not keep_last_tip:
-            self._drop_tip_for_liquid_class(trash_location, last_tip, return_tip)
-            last_tip = None
-
-        return last_tip
+            self._drop_tip_for_liquid_class(trash_location, return_tip)
 
     def distribute_with_liquid_class(  # noqa: C901
         self,
@@ -1375,8 +1383,7 @@ def distribute_with_liquid_class(  # noqa: C901
         trash_location: Union[Location, TrashBin, WasteChute],
         return_tip: bool,
         keep_last_tip: bool,
-        last_tip_location: Optional[Tuple[Location, WellCore]],
-    ) -> Optional[Tuple[Location, WellCore]]:
+    ) -> None:
         """Execute a distribution using liquid class properties.
 
         Args:
@@ -1399,10 +1406,6 @@ def distribute_with_liquid_class(  # noqa: C901
             return_tip: If `True`, return tips to the tip rack location they were picked up from,
                         otherwise drop in `trash_location`
             keep_last_tip: When set to `True`, do not drop the final tip used in the distribute.
-            last_tip_location: If a tip is already attached, this will be the tiprack and well it was
-                           picked up from, represented as a tuple of types.Location and WellCore.
-                           Used so a tip can be returned if it was picked up outside this function
-                           as could be the case for a new_tip of `never`
 
         This method distributes the liquid in the source well into multiple destinations.
         It can accomplish this by either doing a multi-dispense (aspirate once and then
@@ -1462,7 +1465,6 @@ def distribute_with_liquid_class(  # noqa: C901
                 trash_location=trash_location,
                 return_tip=return_tip,
                 keep_last_tip=keep_last_tip,
-                last_tip_location=last_tip_location,
             )
 
         # TODO: use the ID returned by load_liquid_class in command annotations
@@ -1484,9 +1486,8 @@ def distribute_with_liquid_class(  # noqa: C901
             )
         )
 
-        last_tip = last_tip_location
         if new_tip != TransferTipPolicyV2.NEVER:
-            last_tip = self._pick_up_tip_for_liquid_class(
+            self._pick_up_tip_for_liquid_class(
                 tip_racks, starting_tip, tiprack_uri_for_transfer_props
             )
 
@@ -1553,8 +1554,8 @@ def distribute_with_liquid_class(  # noqa: C901
                 )
 
             if not is_first_step and new_tip == TransferTipPolicyV2.ALWAYS:
-                self._drop_tip_for_liquid_class(trash_location, last_tip, return_tip)
-                last_tip = self._pick_up_tip_for_liquid_class(
+                self._drop_tip_for_liquid_class(trash_location, return_tip)
+                self._pick_up_tip_for_liquid_class(
                     tip_racks, starting_tip, tiprack_uri_for_transfer_props
                 )
                 tip_contents = [
@@ -1614,10 +1615,7 @@ def distribute_with_liquid_class(  # noqa: C901
                 is_first_step = False
 
         if not keep_last_tip:
-            self._drop_tip_for_liquid_class(trash_location, last_tip, return_tip)
-            last_tip = None
-
-        return last_tip
+            self._drop_tip_for_liquid_class(trash_location, return_tip)
 
     def _tip_can_hold_volume_for_multi_dispensing(
         self,
@@ -1656,8 +1654,7 @@ def consolidate_with_liquid_class(  # noqa: C901
         trash_location: Union[Location, TrashBin, WasteChute],
         return_tip: bool,
         keep_last_tip: bool,
-        last_tip_location: Optional[Tuple[Location, WellCore]],
-    ) -> Optional[Tuple[Location, WellCore]]:
+    ) -> None:
         """Execute consolidate using liquid class properties.
 
         Args:
@@ -1681,10 +1678,6 @@ def consolidate_with_liquid_class(  # noqa: C901
             return_tip: If `True`, return tips to the tip rack location they were picked up from,
                         otherwise drop in `trash_location`
             keep_last_tip: When set to `True`, do not drop the final tip used in the consolidate.
-            last_tip_location: If a tip is already attached, this will be the tiprack and well it was
-                           picked up from, represented as a tuple of types.Location and WellCore.
-                           Used so a tip can be returned if it was picked up outside this function
-                           as could be the case for a new_tip of `never`.
         """
         if not tip_racks:
             raise RuntimeError(
@@ -1730,9 +1723,8 @@ def consolidate_with_liquid_class(  # noqa: C901
             )
         )
 
-        last_tip = last_tip_location
         if new_tip in [TransferTipPolicyV2.ONCE, TransferTipPolicyV2.ALWAYS]:
-            last_tip = self._pick_up_tip_for_liquid_class(
+            self._pick_up_tip_for_liquid_class(
                 tip_racks, starting_tip, tiprack_uri_for_transfer_props
             )
 
@@ -1764,8 +1756,8 @@ def consolidate_with_liquid_class(  # noqa: C901
                     break
 
             if not is_first_step and new_tip == TransferTipPolicyV2.ALWAYS:
-                self._drop_tip_for_liquid_class(trash_location, last_tip, return_tip)
-                last_tip = self._pick_up_tip_for_liquid_class(
+                self._drop_tip_for_liquid_class(trash_location, return_tip)
+                self._pick_up_tip_for_liquid_class(
                     tip_racks, starting_tip, tiprack_uri_for_transfer_props
                 )
                 tip_contents = [
@@ -1802,10 +1794,7 @@ def consolidate_with_liquid_class(  # noqa: C901
             )
 
         if not keep_last_tip:
-            self._drop_tip_for_liquid_class(trash_location, last_tip, return_tip)
-            last_tip = None
-
-        return last_tip
+            self._drop_tip_for_liquid_class(trash_location, return_tip)
 
     def _get_location_and_well_core_from_next_tip_info(
         self,
@@ -1857,7 +1846,7 @@ def _pick_up_tip_for_liquid_class(
         tip_racks: List[Tuple[Location, LabwareCore]],
         starting_tip: Optional[WellCore],
         tiprack_uri_for_transfer_props: str,
-    ) -> Tuple[Location, WellCore]:
+    ) -> None:
         """Resolve next tip and pick it up, for use in liquid class transfer code."""
         next_tip = self.get_next_tip(
             tip_racks=[core for loc, core in tip_racks],
@@ -1884,16 +1873,15 @@ def _pick_up_tip_for_liquid_class(
             presses=None,
             increment=None,
         )
-        return tiprack_loc, tip_well
 
     def _drop_tip_for_liquid_class(
         self,
         trash_location: Union[Location, TrashBin, WasteChute],
-        last_tip: Optional[Tuple[Location, WellCore]],
         return_tip: bool,
     ) -> None:
         """Drop or return tip for usage in liquid class transfers."""
         if return_tip:
+            last_tip = self.get_tip_origin()
             assert last_tip is not None
             _, tip_well = last_tip
             self.drop_tip(

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

@@ -310,6 +310,12 @@ def get_blow_out_flow_rate(self, rate: float = 1.0) -> float:
     def get_liquid_presence_detection(self) -> bool:
         ...
 
+    @abstractmethod
+    def get_tip_origin(
+        self,
+    ) -> Optional[Tuple[LabwareCoreType, WellCoreType]]:
+        ...
+
     @abstractmethod
     def _pressure_supported_by_pipette(self) -> bool:
         ...
@@ -372,8 +378,7 @@ def transfer_with_liquid_class(
         trash_location: Union[types.Location, TrashBin, WasteChute],
         return_tip: bool,
         keep_last_tip: bool,
-        last_tip_location: Optional[Tuple[types.Location, WellCoreType]],
-    ) -> Optional[Tuple[types.Location, WellCoreType]]:
+    ) -> None:
         """Transfer a liquid from source to dest according to liquid class properties."""
         ...
 
@@ -390,8 +395,7 @@ def distribute_with_liquid_class(
         trash_location: Union[types.Location, TrashBin, WasteChute],
         return_tip: bool,
         keep_last_tip: bool,
-        last_tip_location: Optional[Tuple[types.Location, WellCoreType]],
-    ) -> Optional[Tuple[types.Location, WellCoreType]]:
+    ) -> None:
         """
         Distribute a liquid from single source to multiple destinations
         according to liquid class properties.
@@ -411,8 +415,7 @@ def consolidate_with_liquid_class(
         trash_location: Union[types.Location, TrashBin, WasteChute],
         return_tip: bool,
         keep_last_tip: bool,
-        last_tip_location: Optional[Tuple[types.Location, WellCoreType]],
-    ) -> Optional[Tuple[types.Location, WellCoreType]]:
+    ) -> None:
         """
         Consolidate liquid from multiple sources to a single destination
         using the specified liquid class properties.

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

@@ -67,6 +67,9 @@ def __init__(
             api_level=self._api_version,
         )
         self._liquid_presence_detection = False
+        self._last_tip_rack_and_well: Optional[
+            Tuple[LegacyLabwareCore, LegacyWellCore]
+        ] = None
 
     def get_default_speed(self) -> float:
         """Gets the speed at which the robot's gantry moves."""
@@ -252,6 +255,7 @@ def pick_up_tip(
             num_channels=self.get_channels(),
             fail_if_full=self._api_version < APIVersion(2, 2),
         )
+        self._last_tip_rack_and_well = tip_rack_core, well_core
 
     def drop_tip(
         self,
@@ -298,6 +302,7 @@ def drop_tip(
         hw = self._protocol_interface.get_hardware()
         self.move_to(location=location)
         hw.drop_tip(self._mount, home_after=True if home_after is None else home_after)
+        self._last_tip_rack_and_well = None
 
         if self._api_version < APIVersion(2, 2) and labware_core.is_tip_rack():
             # If this is a tiprack we can try and add the dirty tip back to the tracker
@@ -532,6 +537,11 @@ def get_blow_out_flow_rate(self, rate: float = 1.0) -> float:
     def get_speed(self) -> PlungerSpeeds:
         return self._speeds
 
+    def get_tip_origin(
+        self,
+    ) -> Optional[Tuple[LegacyLabwareCore, LegacyWellCore]]:
+        return self._last_tip_rack_and_well
+
     def set_flow_rate(
         self,
         aspirate: Optional[float] = None,
@@ -612,8 +622,7 @@ def transfer_with_liquid_class(
         trash_location: Union[types.Location, TrashBin, WasteChute],
         return_tip: bool,
         keep_last_tip: bool,
-        last_tip_location: Optional[Tuple[types.Location, LegacyWellCore]],
-    ) -> Optional[Tuple[types.Location, LegacyWellCore]]:
+    ) -> None:
         """This will never be called because it was added in API 2.23"""
         assert False, "transfer_liquid is not supported in legacy context"
 
@@ -629,8 +638,7 @@ def distribute_with_liquid_class(
         trash_location: Union[types.Location, TrashBin, WasteChute],
         return_tip: bool,
         keep_last_tip: bool,
-        last_tip_location: Optional[Tuple[types.Location, LegacyWellCore]],
-    ) -> Optional[Tuple[types.Location, LegacyWellCore]]:
+    ) -> None:
         """This will never be called because it was added in API 2.23"""
         assert False, "distribute_liquid is not supported in legacy context"
 
@@ -646,8 +654,7 @@ def consolidate_with_liquid_class(
         trash_location: Union[types.Location, TrashBin, WasteChute],
         return_tip: bool,
         keep_last_tip: bool,
-        last_tip_location: Optional[Tuple[types.Location, LegacyWellCore]],
-    ) -> Optional[Tuple[types.Location, LegacyWellCore]]:
+    ) -> None:
         """This will never be called because it was added in API 2.23."""
         assert False, "consolidate_liquid is not supported in legacy context"