feat(api): add new InstrumentContext.consolidate_liquid() method (#17359)

Adds consolidate_liquid method to instrument context

Author: jbleon95

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

@@ -1110,6 +1110,18 @@ def distribute_liquid(
     ) -> None:
         pass
 
+    def consolidate_liquid(
+        self,
+        liquid_class: LiquidClass,
+        volume: float,
+        source: List[Tuple[Location, WellCore]],
+        dest: Tuple[Location, WellCore],
+        new_tip: TransferTipPolicyV2,
+        tip_racks: List[Tuple[Location, LabwareCore]],
+        trash_location: Union[Location, TrashBin, WasteChute],
+    ) -> None:
+        pass
+
     def _get_location_and_well_core_from_next_tip_info(
         self,
         tip_info: NextTipInfo,

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

@@ -343,6 +343,23 @@ def distribute_liquid(
         """
         ...
 
+    @abstractmethod
+    def consolidate_liquid(
+        self,
+        liquid_class: LiquidClass,
+        volume: float,
+        source: List[Tuple[types.Location, WellCoreType]],
+        dest: Tuple[types.Location, WellCoreType],
+        new_tip: TransferTipPolicyV2,
+        tip_racks: List[Tuple[types.Location, LabwareCoreType]],
+        trash_location: Union[types.Location, TrashBin, WasteChute],
+    ) -> None:
+        """
+        Consolidate liquid from multiple sources to a single destination
+        using the specified liquid class properties.
+        """
+        ...
+
     @abstractmethod
     def is_tip_tracking_available(self) -> bool:
         """Return whether auto tip tracking is available for the pipette's current nozzle configuration."""

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

@@ -586,6 +586,19 @@ def distribute_liquid(
         """This will never be called because it was added in API 2.23"""
         assert False, "distribute_liquid is not supported in legacy context"
 
+    def consolidate_liquid(
+        self,
+        liquid_class: LiquidClass,
+        volume: float,
+        source: List[Tuple[types.Location, LegacyWellCore]],
+        dest: Tuple[types.Location, LegacyWellCore],
+        new_tip: TransferTipPolicyV2,
+        tip_racks: List[Tuple[types.Location, LegacyLabwareCore]],
+        trash_location: Union[types.Location, TrashBin, WasteChute],
+    ) -> None:
+        """This will never be called because it was added in API 2.23."""
+        assert False, "consolidate_liquid is not supported in legacy context"
+
     def get_active_channels(self) -> int:
         """This will never be called because it was added in API 2.16."""
         assert False, "get_active_channels only supported in API 2.16 & later"

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

@@ -506,6 +506,19 @@ def distribute_liquid(
         """This will never be called because it was added in API 2.23."""
         assert False, "distribute_liquid is not supported in legacy context"
 
+    def consolidate_liquid(
+        self,
+        liquid_class: LiquidClass,
+        volume: float,
+        source: List[Tuple[types.Location, LegacyWellCore]],
+        dest: Tuple[types.Location, LegacyWellCore],
+        new_tip: TransferTipPolicyV2,
+        tip_racks: List[Tuple[types.Location, LegacyLabwareCore]],
+        trash_location: Union[types.Location, TrashBin, WasteChute],
+    ) -> None:
+        """This will never be called because it was added in API 2.23."""
+        assert False, "consolidate_liquid is not supported in legacy context"
+
     def get_active_channels(self) -> int:
         """This will never be called because it was added in API 2.16."""
         assert False, "get_active_channels only supported in API 2.16 & later"

api/src/opentrons/protocol_api/instrument_context.py

@@ -1689,6 +1689,94 @@ def distribute_liquid(
         )
         return self
 
+    def consolidate_liquid(
+        self,
+        liquid_class: LiquidClass,
+        volume: float,
+        source: Union[
+            labware.Well, Sequence[labware.Well], Sequence[Sequence[labware.Well]]
+        ],
+        dest: labware.Well,
+        new_tip: TransferTipPolicyV2Type = "once",
+        trash_location: Optional[
+            Union[types.Location, labware.Well, TrashBin, WasteChute]
+        ] = None,
+    ) -> InstrumentContext:
+        """
+        Consolidate liquid from multiple sources to a single destination
+        using the specified liquid class properties.
+
+        TODO: Add args description.
+        """
+        if not feature_flags.allow_liquid_classes(
+            robot_type=RobotTypeEnum.robot_literal_to_enum(
+                self._protocol_core.robot_type
+            )
+        ):
+            raise NotImplementedError("This method is not implemented.")
+        if not isinstance(dest, labware.Well):
+            raise ValueError(
+                f"Destination should be a single Well but received {dest}."
+            )
+        flat_sources_list = validation.ensure_valid_flat_wells_list_for_transfer_v2(
+            source
+        )
+        for well in flat_sources_list + [dest]:
+            instrument.validate_takes_liquid(
+                location=well.top(),
+                reject_module=True,
+                reject_adapter=True,
+            )
+
+        valid_new_tip = validation.ensure_new_tip_policy(new_tip)
+        if valid_new_tip == TransferTipPolicyV2.NEVER:
+            if self._last_tip_picked_up_from is None:
+                raise RuntimeError(
+                    "Pipette has no tip attached to perform transfer."
+                    " Either do a pick_up_tip beforehand or specify a new_tip parameter"
+                    " of 'once' or 'always'."
+                )
+            else:
+                tip_racks = [self._last_tip_picked_up_from.parent]
+        else:
+            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."
+                " Ensure that all previously aspirated liquid is dispensed before starting"
+                " a new transfer."
+            )
+
+        _trash_location: Union[types.Location, labware.Well, TrashBin, WasteChute]
+        if trash_location is None:
+            saved_trash = self.trash_container
+            if isinstance(saved_trash, labware.Labware):
+                _trash_location = saved_trash.wells()[0]
+            else:
+                _trash_location = saved_trash
+        else:
+            _trash_location = trash_location
+
+        checked_trash_location = validation.ensure_valid_trash_location_for_transfer_v2(
+            trash_location=_trash_location
+        )
+        self._core.consolidate_liquid(
+            liquid_class=liquid_class,
+            volume=volume,
+            source=[
+                (types.Location(types.Point(), labware=well), well._core)
+                for well in flat_sources_list
+            ],
+            dest=(types.Location(types.Point(), labware=dest), dest._core),
+            new_tip=valid_new_tip,
+            tip_racks=[
+                (types.Location(types.Point(), labware=rack), rack._core)
+                for rack in tip_racks
+            ],
+            trash_location=checked_trash_location,
+        )
+        return self
+
     @requires_version(2, 0)
     def delay(self, *args: Any, **kwargs: Any) -> None:
         """