feat(api): add getNexTip protocol engine command

api/src/opentrons/protocol_engine/commands/__init__.py

@@ -332,6 +332,14 @@
     VerifyTipPresenceCommandType,
 )
 
+from .get_next_tip import (
+    GetNextTip,
+    GetNextTipCreate,
+    GetNextTipParams,
+    GetNextTipResult,
+    GetNextTipCommandType,
+)
+
 from .liquid_probe import (
     LiquidProbe,
     LiquidProbeParams,
@@ -595,6 +603,12 @@
     "VerifyTipPresenceParams",
     "VerifyTipPresenceResult",
     "VerifyTipPresenceCommandType",
+    # get next tip command bundle
+    "GetNextTip",
+    "GetNextTipCreate",
+    "GetNextTipParams",
+    "GetNextTipResult",
+    "GetNextTipCommandType",
     # liquid probe command bundle
     "LiquidProbe",
     "LiquidProbeParams",

api/src/opentrons/protocol_engine/commands/command_unions.py

@@ -323,6 +323,14 @@
     GetTipPresenceCommandType,
 )
 
+from .get_next_tip import (
+    GetNextTip,
+    GetNextTipCreate,
+    GetNextTipParams,
+    GetNextTipResult,
+    GetNextTipCommandType,
+)
+
 from .liquid_probe import (
     LiquidProbe,
     LiquidProbeParams,
@@ -375,6 +383,7 @@
         SetStatusBar,
         VerifyTipPresence,
         GetTipPresence,
+        GetNextTip,
         LiquidProbe,
         TryLiquidProbe,
         heater_shaker.WaitForTemperature,
@@ -460,6 +469,7 @@
     SetStatusBarParams,
     VerifyTipPresenceParams,
     GetTipPresenceParams,
+    GetNextTipParams,
     LiquidProbeParams,
     TryLiquidProbeParams,
     heater_shaker.WaitForTemperatureParams,
@@ -543,6 +553,7 @@
     SetStatusBarCommandType,
     VerifyTipPresenceCommandType,
     GetTipPresenceCommandType,
+    GetNextTipCommandType,
     LiquidProbeCommandType,
     TryLiquidProbeCommandType,
     heater_shaker.WaitForTemperatureCommandType,
@@ -627,6 +638,7 @@
         SetStatusBarCreate,
         VerifyTipPresenceCreate,
         GetTipPresenceCreate,
+        GetNextTipCreate,
         LiquidProbeCreate,
         TryLiquidProbeCreate,
         heater_shaker.WaitForTemperatureCreate,
@@ -712,6 +724,7 @@
     SetStatusBarResult,
     VerifyTipPresenceResult,
     GetTipPresenceResult,
+    GetNextTipResult,
     LiquidProbeResult,
     TryLiquidProbeResult,
     heater_shaker.WaitForTemperatureResult,

api/src/opentrons/protocol_engine/commands/get_next_tip.py

@@ -0,0 +1,123 @@
+"""Get next tip command request, result, and implementation models."""
+
+from __future__ import annotations
+from pydantic import BaseModel, Field
+from typing import TYPE_CHECKING, Optional, Type, List, Literal, Union
+
+from opentrons.types import NozzleConfigurationType
+
+from ..errors import ErrorOccurrence
+from ..types import NextTipInfo, NoTipAvailable, NoTipReason
+from .pipetting_common import PipetteIdMixin
+
+from .command import (
+    AbstractCommandImpl,
+    BaseCommand,
+    BaseCommandCreate,
+    SuccessData,
+)
+
+if TYPE_CHECKING:
+    from ..state.state import StateView
+
+
+GetNextTipCommandType = Literal["getNextTip"]
+
+
+class GetNextTipParams(PipetteIdMixin):
+    """Payload needed to resolve the next available tip."""
+
+    labwareIds: List[str] = Field(
+        ...,
+        description="Labware ID(s) of tip racks to resolve next available tip(s) from"
+        " Labware IDs will be resolved sequentially",
+    )
+    startingTipWell: Optional[str] = Field(
+        None,
+        description="Name of starting tip rack 'well'."
+        " This only applies to the first tip rack in the list provided in labwareIDs",
+    )
+
+
+class GetNextTipResult(BaseModel):
+    """Result data from the execution of a GetNextTip."""
+
+    nextTipInfo: Union[NextTipInfo, NoTipAvailable] = Field(
+        ...,
+        description="Labware ID and well name of next available tip for a pipette,"
+        " or information why no tip could be resolved.",
+    )
+
+
+class GetNextTipImplementation(
+    AbstractCommandImpl[GetNextTipParams, SuccessData[GetNextTipResult]]
+):
+    """Get next tip command implementation."""
+
+    def __init__(
+        self,
+        state_view: StateView,
+        **kwargs: object,
+    ) -> None:
+        self._state_view = state_view
+
+    async def execute(self, params: GetNextTipParams) -> SuccessData[GetNextTipResult]:
+        """Get the next available tip for the requested pipette."""
+        pipette_id = params.pipetteId
+        starting_tip_name = params.startingTipWell
+
+        num_tips = self._state_view.tips.get_pipette_active_channels(pipette_id)
+        nozzle_map = self._state_view.tips.get_pipette_nozzle_map(pipette_id)
+
+        if (
+            starting_tip_name is not None
+            and nozzle_map.configuration != NozzleConfigurationType.FULL
+        ):
+            return SuccessData(
+                public=GetNextTipResult(
+                    nextTipInfo=NoTipAvailable(
+                        noTipReason=NoTipReason.STARTING_TIP_WITH_PARTIAL,
+                        message="Cannot automatically resolve next tip with starting tip and partial tip configuration.",
+                    )
+                )
+            )
+
+        next_tip: Union[NextTipInfo, NoTipAvailable]
+        for labware_id in params.labwareIds:
+            well_name = self._state_view.tips.get_next_tip(
+                labware_id=labware_id,
+                num_tips=num_tips,
+                starting_tip_name=starting_tip_name,
+                nozzle_map=nozzle_map,
+            )
+            if well_name is not None:
+                next_tip = NextTipInfo(labwareId=labware_id, tipOriginWell=well_name)
+                break
+            # After the first tip rack is exhausted, starting tip no longer applies
+            starting_tip_name = None
+        else:
+            next_tip = NoTipAvailable(
+                noTipReason=NoTipReason.NO_AVAILABLE_TIPS,
+                message="No available tips for given pipette, nozzle configuration and provided tip racks.",
+            )
+
+        return SuccessData(public=GetNextTipResult(nextTipInfo=next_tip))
+
+
+class GetNextTip(BaseCommand[GetNextTipParams, GetNextTipResult, ErrorOccurrence]):
+    """Get next tip command model."""
+
+    commandType: GetNextTipCommandType = "getNextTip"
+    params: GetNextTipParams
+    result: Optional[GetNextTipResult]
+
+    _ImplementationCls: Type[GetNextTipImplementation] = GetNextTipImplementation
+
+
+class GetNextTipCreate(BaseCommandCreate[GetNextTipParams]):
+    """Get next tip command creation request model."""
+
+    commandType: GetNextTipCommandType = "getNextTip"
+    params: GetNextTipParams
+
+    _CommandCls: Type[GetNextTip] = GetNextTip

api/src/opentrons/protocol_engine/types.py

@@ -1115,6 +1115,37 @@ def from_hw_state(cls, state: HwTipStateType) -> "TipPresenceStatus":
         }[state]
 
 
+class NextTipInfo(BaseModel):
+    """Next available tip labware and well name data."""
+
+    labwareId: str = Field(
+        ...,
+        description="The labware ID of the tip rack where the next available tip(s) are located.",
+    )
+    tipOriginWell: str = Field(
+        ..., description="The (starting) well name of the next available tip(s)."
+    )
+
+
+class NoTipReason(Enum):
+    """The cause of no tip being available for a pipette and tip rack(s)."""
+
+    NO_AVAILABLE_TIPS = "noAvailableTips"
+    STARTING_TIP_WITH_PARTIAL = "startingTipWithPartial"
+    INCOMPATIBLE_CONFIGURATION = "incompatibleConfiguration"
+
+
+class NoTipAvailable(BaseModel):
+    """No available next tip data."""
+
+    noTipReason: NoTipReason = Field(
+        ..., description="The reason why no next available tip could be provided."
+    )
+    message: Optional[str] = Field(
+        None, description="Optional message explaining why a tip wasn't available."
+    )
+
+
 # TODO (spp, 2024-04-02): move all RTP types to runner
 class RTPBase(BaseModel):
     """Parameters defined in a protocol."""

api/tests/opentrons/protocol_engine/commands/test_get_next_tip.py

@@ -0,0 +1,142 @@
+"""Test get next tip in place commands."""
+from decoy import Decoy
+
+from opentrons.types import NozzleConfigurationType
+from opentrons.protocol_engine import StateView
+from opentrons.protocol_engine.types import NextTipInfo, NoTipAvailable, NoTipReason
+from opentrons.protocol_engine.commands.command import SuccessData
+from opentrons.protocol_engine.commands.get_next_tip import (
+    GetNextTipParams,
+    GetNextTipResult,
+    GetNextTipImplementation,
+)
+
+from opentrons.hardware_control.nozzle_manager import NozzleMap
+
+
+async def test_get_next_tip_implementation(
+    decoy: Decoy,
+    state_view: StateView,
+) -> None:
+    """A GetNextTip command should have an execution implementation."""
+    subject = GetNextTipImplementation(state_view=state_view)
+    params = GetNextTipParams(
+        pipetteId="abc", labwareIds=["123"], startingTipWell="xyz"
+    )
+    mock_nozzle_map = decoy.mock(cls=NozzleMap)
+
+    decoy.when(state_view.tips.get_pipette_active_channels("abc")).then_return(42)
+    decoy.when(state_view.tips.get_pipette_nozzle_map("abc")).then_return(
+        mock_nozzle_map
+    )
+    decoy.when(mock_nozzle_map.configuration).then_return(NozzleConfigurationType.FULL)
+
+    decoy.when(
+        state_view.tips.get_next_tip(
+            labware_id="123",
+            num_tips=42,
+            starting_tip_name="xyz",
+            nozzle_map=mock_nozzle_map,
+        )
+    ).then_return("foo")
+
+    result = await subject.execute(params)
+
+    assert result == SuccessData(
+        public=GetNextTipResult(
+            nextTipInfo=NextTipInfo(labwareId="123", tipOriginWell="foo")
+        ),
+    )
+
+
+async def test_get_next_tip_implementation_multiple_tip_racks(
+    decoy: Decoy,
+    state_view: StateView,
+) -> None:
+    """A GetNextTip command with multiple tip racks should not apply starting tip to the following ones."""
+    subject = GetNextTipImplementation(state_view=state_view)
+    params = GetNextTipParams(
+        pipetteId="abc", labwareIds=["123", "456"], startingTipWell="xyz"
+    )
+    mock_nozzle_map = decoy.mock(cls=NozzleMap)
+
+    decoy.when(state_view.tips.get_pipette_active_channels("abc")).then_return(42)
+    decoy.when(state_view.tips.get_pipette_nozzle_map("abc")).then_return(
+        mock_nozzle_map
+    )
+    decoy.when(mock_nozzle_map.configuration).then_return(NozzleConfigurationType.FULL)
+
+    decoy.when(
+        state_view.tips.get_next_tip(
+            labware_id="456",
+            num_tips=42,
+            starting_tip_name=None,
+            nozzle_map=mock_nozzle_map,
+        )
+    ).then_return("foo")
+
+    result = await subject.execute(params)
+
+    assert result == SuccessData(
+        public=GetNextTipResult(
+            nextTipInfo=NextTipInfo(labwareId="456", tipOriginWell="foo")
+        ),
+    )
+
+
+async def test_get_next_tip_implementation_no_tips(
+    decoy: Decoy,
+    state_view: StateView,
+) -> None:
+    """A GetNextTip command should return with NoTipAvailable if there are no available tips."""
+    subject = GetNextTipImplementation(state_view=state_view)
+    params = GetNextTipParams(
+        pipetteId="abc", labwareIds=["123", "456"], startingTipWell="xyz"
+    )
+    mock_nozzle_map = decoy.mock(cls=NozzleMap)
+
+    decoy.when(state_view.tips.get_pipette_active_channels("abc")).then_return(42)
+    decoy.when(state_view.tips.get_pipette_nozzle_map("abc")).then_return(
+        mock_nozzle_map
+    )
+    decoy.when(mock_nozzle_map.configuration).then_return(NozzleConfigurationType.FULL)
+
+    result = await subject.execute(params)
+
+    assert result == SuccessData(
+        public=GetNextTipResult(
+            nextTipInfo=NoTipAvailable(
+                noTipReason=NoTipReason.NO_AVAILABLE_TIPS,
+                message="No available tips for given pipette, nozzle configuration and provided tip racks.",
+            )
+        ),
+    )
+
+
+async def test_get_next_tip_implementation_partial_with_starting_tip(
+    decoy: Decoy,
+    state_view: StateView,
+) -> None:
+    """A GetNextTip command should return with NoTipAvailable if there's a starting tip and a partial config."""
+    subject = GetNextTipImplementation(state_view=state_view)
+    params = GetNextTipParams(
+        pipetteId="abc", labwareIds=["123", "456"], startingTipWell="xyz"
+    )
+    mock_nozzle_map = decoy.mock(cls=NozzleMap)
+
+    decoy.when(state_view.tips.get_pipette_active_channels("abc")).then_return(42)
+    decoy.when(state_view.tips.get_pipette_nozzle_map("abc")).then_return(
+        mock_nozzle_map
+    )
+    decoy.when(mock_nozzle_map.configuration).then_return(NozzleConfigurationType.ROW)
+
+    result = await subject.execute(params)
+
+    assert result == SuccessData(
+        public=GetNextTipResult(
+            nextTipInfo=NoTipAvailable(
+                noTipReason=NoTipReason.STARTING_TIP_WITH_PARTIAL,
+                message="Cannot automatically resolve next tip with starting tip and partial tip configuration.",
+            )
+        ),
+    )