code and type fixes

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

@@ -19,7 +19,7 @@
 
 WellCore = AbstractWellCore
 LabwareCore = AbstractLabware[WellCore]
-InstrumentCore = AbstractInstrument[WellCore]
+InstrumentCore = AbstractInstrument[WellCore, LabwareCore]
 ModuleCore = AbstractModuleCore
 TemperatureModuleCore = AbstractTemperatureModuleCore
 MagneticModuleCore = AbstractMagneticModuleCore

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

@@ -2,17 +2,22 @@
 
 from __future__ import annotations
 
-from typing import Optional, TYPE_CHECKING, cast, Union, List, Tuple
+from typing import (
+    Optional,
+    TYPE_CHECKING,
+    cast,
+    Union,
+    List,
+    Tuple,
+    NamedTuple,
+)
 from opentrons.types import Location, Mount, NozzleConfigurationType, NozzleMapInterface
 from opentrons.hardware_control import SyncHardwareAPI
 from opentrons.hardware_control.dev_types import PipetteDict
 from opentrons.protocols.api_support.util import FlowRates, find_value_for_api_version
 from opentrons.protocols.api_support.types import APIVersion
-from opentrons.protocols.advanced_control.transfers.common import (
-    TransferTipPolicyV2,
-    check_valid_volume_parameters,
-    expand_for_volume_constraints,
-)
+from opentrons.protocols.advanced_control.transfers.common import TransferTipPolicyV2
+from opentrons.protocols.advanced_control.transfers import common as tx_commons
 from opentrons.protocol_engine import commands as cmd
 from opentrons.protocol_engine import (
     DeckPoint,
@@ -33,7 +38,6 @@
     AddressableOffsetVector,
     LiquidClassRecord,
     NextTipInfo,
-    NoTipAvailable,
 )
 from opentrons.protocol_engine.errors.exceptions import TipNotAttachedError
 from opentrons.protocol_engine.clients import SyncClient as EngineClient
@@ -59,7 +63,7 @@
 _DISPENSE_VOLUME_VALIDATION_ADDED_IN = APIVersion(2, 17)
 
 
-class InstrumentCore(AbstractInstrument[WellCore]):
+class InstrumentCore(AbstractInstrument[WellCore, LabwareCore]):
     """Instrument API core using a ProtocolEngine.
 
     Args:
@@ -716,9 +720,11 @@ def get_pipette_load_name(self) -> str:
         """
         pipette = self._engine_client.state.pipettes.get(self._pipette_id)
         assert pipette.pipetteName in PIPETTE_API_NAMES_MAP.values()
-        for pip_api_name, pip_name in PIPETTE_API_NAMES_MAP.items():
-            if pip_name == pipette.pipetteName:
-                return pip_api_name
+        return [
+            pip_api_name
+            for pip_api_name, pip_name in PIPETTE_API_NAMES_MAP.items()
+            if pip_name == pipette.pipetteName
+        ][0]
 
     def get_model(self) -> str:
         return self._engine_client.state.pipettes.get_model_name(self._pipette_id)
@@ -926,7 +932,7 @@ def get_next_tip(
         )
         return next_tip_info if isinstance(next_tip_info, NextTipInfo) else None
 
-    def transfer_liquid(
+    def transfer_liquid(  # noqa: C901
         self,
         liquid_class: LiquidClass,
         volume: float,
@@ -952,27 +958,24 @@ def transfer_liquid(
             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).
         """
-        # This function is WIP
-        # TODO: Validate that the tipracks we need to use from the tipracks list
-        #  (based on number of tips we'll need) have the same uri.
         if not tip_racks:
             raise RuntimeError(
                 "No tipracks found for pipette in order to perform transfer"
             )
-        tiprack_uri = tip_racks[0][1].get_uri()
+        tiprack_uri_for_transfer_props = tip_racks[0][1].get_uri()
 
         # TODO: use the ID returned by load_liquid_class in command annotations
         self.load_liquid_class(
             liquid_class=liquid_class,
             pipette_load_name=self.get_pipette_load_name(),
-            tiprack_uri=tiprack_uri,
+            tiprack_uri=tiprack_uri_for_transfer_props,
         )
         transfer_props = liquid_class.get_for(
             pipette=self.get_pipette_load_name(),
-            tiprack=tiprack_uri,
+            tiprack=tiprack_uri_for_transfer_props,
         )
         # TODO: add multi-channel pipette handling here
-        source_dest_per_volume_step = expand_for_volume_constraints(
+        source_dest_per_volume_step = tx_commons.expand_for_volume_constraints(
             volumes=[volume for _ in range(len(source))],
             targets=zip(source, dest),
             max_volume=self.get_max_volume(),
@@ -988,45 +991,59 @@ def _drop_tip() -> None:
             elif isinstance(trash_location, Location):
                 self.drop_tip(
                     location=trash_location,
-                    well_core=trash_location.labware.as_well()._core,
-                    # type: ignore[arg-type]
+                    well_core=trash_location.labware.as_well()._core,  # type: ignore[arg-type]
                     home_after=False,
                     alternate_drop_location=True,
                 )
 
         def _pick_up_tip() -> None:
             next_tip = self.get_next_tip(
-                tip_racks=tip_racks,
+                tip_racks=[core for loc, core in tip_racks],
                 starting_well=None,
             )
             if not next_tip:
                 raise RuntimeError(
                     f"No tip available among {tip_racks} for this transfer."
                 )
-            tiprack_loc, tip_well = self._get_location_and_well_core_from_next_tip_info(
-                next_tip
-            )
+            (
+                tiprack_loc,
+                tiprack_uri,
+                tip_well,
+            ) = self._get_location_and_well_core_from_next_tip_info(next_tip, tip_racks)
+            if tiprack_uri != tiprack_uri_for_transfer_props:
+                raise RuntimeError(
+                    f"Tiprack {tiprack_uri} does not match the tiprack designated "
+                    f"for this transfer- {tiprack_uri_for_transfer_props}."
+                )
             self.pick_up_tip(
-                location=tiprack_loc[0],
+                location=tiprack_loc,
                 well_core=tip_well,
+                presses=None,
+                increment=None,
             )
 
         if new_tip == TransferTipPolicyV2.ONCE:
             _pick_up_tip()
 
-        prev_src: Optional[Tuple(Location, WellCore)] = None
+        prev_src: Optional[Tuple[Location, WellCore]] = None
         post_disp_tip_contents = [
             tx_comps_executor.LiquidAndAirGapPair(
                 liquid=0,
                 air_gap=0,
             )
         ]
+        src_dest_combo: Tuple[Tuple[Location, WellCore], Tuple[Location, WellCore]]
         for step_volume, src_dest_combo in source_dest_per_volume_step:
-            source, destination = src_dest_combo
+            step_source: Tuple[Location, WellCore]
+            step_destination: Tuple[Location, WellCore]
+            step_source, step_destination = src_dest_combo
             if new_tip == TransferTipPolicyV2.ALWAYS or (
-                new_tip == TransferTipPolicyV2.PER_SOURCE and source != prev_src
+                new_tip == TransferTipPolicyV2.PER_SOURCE and step_source != prev_src
             ):
-                if new_tip == TransferTipPolicyV2.PER_SOURCE and source != prev_src:
+                if (
+                    new_tip == TransferTipPolicyV2.PER_SOURCE
+                    and step_source != prev_src
+                ):
                     _drop_tip()
                     post_disp_tip_contents = [
                         tx_comps_executor.LiquidAndAirGapPair(
@@ -1038,17 +1055,19 @@ def _pick_up_tip() -> None:
 
             post_asp_tip_contents = self.aspirate_liquid_class(
                 volume=step_volume,
-                source=source,
+                source=step_source,
                 transfer_properties=transfer_props,
                 transfer_type=tx_comps_executor.TransferType.ONE_TO_ONE,
                 tip_contents=post_disp_tip_contents,
             )
             post_disp_tip_contents = self.dispense_liquid_class(
                 volume=step_volume,
-                dest=destination,
+                dest=step_destination,
+                source=step_source,
                 transfer_properties=transfer_props,
                 transfer_type=tx_comps_executor.TransferType.ONE_TO_ONE,
                 tip_contents=post_asp_tip_contents,
+                trash_location=trash_location,
             )
 
             if new_tip == TransferTipPolicyV2.ALWAYS:
@@ -1059,20 +1078,22 @@ def _pick_up_tip() -> None:
                         air_gap=0,
                     )
                 ]
-            prev_src = source
+            prev_src = step_source
 
     def _get_location_and_well_core_from_next_tip_info(
         self,
         tip_info: NextTipInfo,
-    ) -> Tuple[Location, WellCore]:
-        tiprack_id, tip_well_id = tip_info
-        tiprack_labware_core = self._protocol_core._labware_cores_by_id[tiprack_id]
-        tip_well = tiprack_labware_core.get_well_core(tip_well_id)
+        tip_racks: List[Tuple[Location, LabwareCore]],
+    ) -> _TipInfo:
+        tiprack_labware_core = self._protocol_core._labware_cores_by_id[
+            tip_info.labwareId
+        ]
+        tip_well = tiprack_labware_core.get_well_core(tip_info.tipStartingWell)
 
         tiprack_loc = [
             loc for loc, lw_core in tip_racks if lw_core == tiprack_labware_core
         ]
-        return (tiprack_loc, tip_well)
+        return _TipInfo(tiprack_loc[0], tiprack_labware_core.get_uri(), tip_well)
 
     def aspirate_liquid_class(
         self,
@@ -1094,7 +1115,7 @@ def aspirate_liquid_class(
         Return: List of liquid and air gap pairs in tip.
         """
         aspirate_props = transfer_properties.aspirate
-        check_valid_volume_parameters(
+        tx_commons.check_valid_volume_parameters(
             disposal_volume=0,  # No disposal volume for 1-to-1 transfer
             air_gap=aspirate_props.retract.air_gap_by_volume.get_for_volume(volume),
             max_volume=self.get_max_volume(),
@@ -1115,6 +1136,7 @@ def aspirate_liquid_class(
                 liquid=0,
                 air_gap=0,
             )
+            tip_contents = [last_liquid_and_airgap_in_tip]
         components_executor = tx_comps_executor.TransferComponentsExecutor(
             instrument_core=self,
             transfer_properties=transfer_properties,
@@ -1137,8 +1159,8 @@ def aspirate_liquid_class(
         components_executor.aspirate_and_wait(volume=volume)
         components_executor.retract_after_aspiration(volume=volume)
         last_contents = components_executor.tip_state.last_liquid_and_air_gap_in_tip
-        tip_contents.pop()
-        return tip_contents.append(last_contents)
+        tip_contents[-1] = last_contents
+        return tip_contents
 
     def dispense_liquid_class(
         self,
@@ -1196,6 +1218,7 @@ def dispense_liquid_class(
                 liquid=0,
                 air_gap=0,
             )
+            tip_contents = [last_liquid_and_airgap_in_tip]
         components_executor = tx_comps_executor.TransferComponentsExecutor(
             instrument_core=self,
             transfer_properties=transfer_properties,
@@ -1226,8 +1249,8 @@ def dispense_liquid_class(
             source_well=source[1] if source else None,
         )
         last_contents = components_executor.tip_state.last_liquid_and_air_gap_in_tip
-        tip_contents.pop()
-        return tip_contents.append(last_contents)
+        tip_contents[-1] = last_contents
+        return tip_contents
 
     def retract(self) -> None:
         """Retract this instrument to the top of the gantry."""
@@ -1324,3 +1347,9 @@ def nozzle_configuration_valid_for_lld(self) -> bool:
     def delay(self, seconds: float) -> None:
         """Call a protocol delay."""
         self._protocol_core.delay(seconds=seconds, msg=None)
+
+
+class _TipInfo(NamedTuple):
+    tiprack_location: Location
+    tiprack_uri: str
+    tip_well: WellCore

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

@@ -3,7 +3,7 @@
 
 from enum import Enum
 from typing import TYPE_CHECKING, Optional, Union
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 
 from opentrons_shared_data.liquid_classes.liquid_class_definition import (
     PositionReference,
@@ -28,8 +28,8 @@
 class LiquidAndAirGapPair:
     """Pairing of a liquid and air gap in a tip, in that order."""
 
-    liquid: float
-    air_gap: float
+    liquid: float = 0
+    air_gap: float = 0
 
 
 @dataclass
@@ -50,9 +50,8 @@ class TipState:
 
     ready_to_aspirate: bool = True
     # TODO: maybe use the tip contents from engine state instead.
-    last_liquid_and_air_gap_in_tip: LiquidAndAirGapPair = LiquidAndAirGapPair(
-        liquid=0,
-        air_gap=0,
+    last_liquid_and_air_gap_in_tip: LiquidAndAirGapPair = field(
+        default_factory=LiquidAndAirGapPair
     )
 
     def add_liquid(self, volume: float) -> None:

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

@@ -16,7 +16,7 @@
 from .labware import LabwareCoreType
 
 
-class AbstractInstrument(ABC, Generic[WellCoreType]):
+class AbstractInstrument(ABC, Generic[WellCoreType, LabwareCoreType]):
     @abstractmethod
     def get_default_speed(self) -> float:
         ...
@@ -359,4 +359,4 @@ def nozzle_configuration_valid_for_lld(self) -> bool:
         """Check if the nozzle configuration currently supports LLD."""
 
 
-InstrumentCoreType = TypeVar("InstrumentCoreType", bound=AbstractInstrument[Any])
+InstrumentCoreType = TypeVar("InstrumentCoreType", bound=AbstractInstrument[Any, Any])

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

@@ -38,7 +38,7 @@
 """In PAPIv2.1 and below, tips are always dropped 10 mm from the bottom of the well."""
 
 
-class LegacyInstrumentCore(AbstractInstrument[LegacyWellCore]):
+class LegacyInstrumentCore(AbstractInstrument[LegacyWellCore, LegacyLabwareCore]):
     """Implementation of the InstrumentContext interface."""
 
     def __init__(

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

@@ -43,7 +43,9 @@
 """In PAPIv2.1 and below, tips are always dropped 10 mm from the bottom of the well."""
 
 
-class LegacyInstrumentCoreSimulator(AbstractInstrument[LegacyWellCore]):
+class LegacyInstrumentCoreSimulator(
+    AbstractInstrument[LegacyWellCore, LegacyLabwareCore]
+):
     """A simulation of an instrument context."""
 
     def __init__(

api/src/opentrons/protocol_api/instrument_context.py

@@ -1561,15 +1561,9 @@ def transfer_liquid(
                     " of 'once' or 'always'."
                 )
             else:
-                tiprack = self._last_tip_picked_up_from.parent
+                tip_racks = [self._last_tip_picked_up_from.parent]
         else:
-            # TODO: update this with getNextTip result from engine
-            tiprack, well = labware.next_available_tip(
-                starting_tip=self.starting_tip,
-                tip_racks=self.tip_racks,
-                channels=self.active_channels,
-                nozzle_map=self._core.get_nozzle_map(),
-            )
+            tip_racks = self._tip_racks
         if self.current_volume != 0:
             raise RuntimeError(
                 "A transfer on a liquid class cannot start with liquid already in the tip."
@@ -1602,9 +1596,9 @@ def transfer_liquid(
                 for well in flat_dests_list
             ],
             new_tip=valid_new_tip,
-            tipracks=[
-                (types.Location(point=Point(), labware=rack), rack._core)
-                for rack in self._tip_racks
+            tip_racks=[
+                (types.Location(types.Point(), labware=rack), rack._core)
+                for rack in tip_racks
             ],
             trash_location=checked_trash_location,
         )