feat(api): add new InstrumentContext.distribute_liquid() method (#17355)

# Overview

Adds a new `InstrumentContext.distribute_liquid()` method that takes in
a single source and multiple destinations for liquid distribution. This
method does argument and state verification before delegating the
implementation to the cores.

## Risk assessment

None. Adds a new feature that doesn't affect anything else.

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

@@ -1098,6 +1098,18 @@ def _pick_up_tip() -> None:
         if new_tip != TransferTipPolicyV2.NEVER:
             _drop_tip()
 
+    def distribute_liquid(
+        self,
+        liquid_class: LiquidClass,
+        volume: float,
+        source: Tuple[Location, WellCore],
+        dest: List[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

@@ -326,6 +326,23 @@ def transfer_liquid(
         """Transfer a liquid from source to dest according to liquid class properties."""
         ...
 
+    @abstractmethod
+    def distribute_liquid(
+        self,
+        liquid_class: LiquidClass,
+        volume: float,
+        source: Tuple[types.Location, WellCoreType],
+        dest: List[Tuple[types.Location, WellCoreType]],
+        new_tip: TransferTipPolicyV2,
+        tip_racks: List[Tuple[types.Location, LabwareCoreType]],
+        trash_location: Union[types.Location, TrashBin, WasteChute],
+    ) -> None:
+        """
+        Distribute a liquid from single source to multiple destinations
+        according to 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

@@ -570,10 +570,22 @@ def transfer_liquid(
         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 .."""
-        # TODO(spp, 2024-11-20): update the docstring and error to include API version
+        """This will never be called because it was added in API 2.23"""
         assert False, "transfer_liquid is not supported in legacy context"
 
+    def distribute_liquid(
+        self,
+        liquid_class: LiquidClass,
+        volume: float,
+        source: Tuple[types.Location, LegacyWellCore],
+        dest: List[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, "distribute_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

@@ -490,10 +490,22 @@ def transfer_liquid(
         tip_racks: List[Tuple[types.Location, LegacyLabwareCore]],
         trash_location: Union[types.Location, TrashBin, WasteChute],
     ) -> None:
-        """Transfer a liquid from source to dest according to liquid class properties."""
-        # TODO(spp, 2024-11-20): update the docstring and error to include API version
+        """This will never be called because it was added in API 2.23."""
         assert False, "transfer_liquid is not supported in legacy context"
 
+    def distribute_liquid(
+        self,
+        liquid_class: LiquidClass,
+        volume: float,
+        source: Tuple[types.Location, LegacyWellCore],
+        dest: List[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, "distribute_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

@@ -1604,6 +1604,91 @@ def transfer_liquid(
         )
         return self
 
+    def distribute_liquid(
+        self,
+        liquid_class: LiquidClass,
+        volume: float,
+        source: labware.Well,
+        dest: Union[
+            labware.Well, Sequence[labware.Well], Sequence[Sequence[labware.Well]]
+        ],
+        new_tip: TransferTipPolicyV2Type = "once",
+        trash_location: Optional[
+            Union[types.Location, labware.Well, TrashBin, WasteChute]
+        ] = None,
+    ) -> InstrumentContext:
+        """
+        Distribute liquid from a single source to multiple destinations
+        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(source, labware.Well):
+            raise ValueError(f"Source should be a single Well but received {source}.")
+        flat_dests_list = validation.ensure_valid_flat_wells_list_for_transfer_v2(dest)
+        for well in [source] + flat_dests_list:
+            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.distribute_liquid(
+            liquid_class=liquid_class,
+            volume=volume,
+            source=(types.Location(types.Point(), labware=source), source._core),
+            dest=[
+                (types.Location(types.Point(), labware=well), well._core)
+                for well in flat_dests_list
+            ],
+            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:
         """