Skip to content

Commit 1b95421

Browse files
jbleon95jbleon95
authored
feat(api): add getNexTip protocol engine command (#17038)
Adds a new `getNextTip` for use in getting next tip information directly from the engine
1 parent 7671c31 commit 1b95421

File tree

6 files changed

+387
-0
lines changed

6 files changed

+387
-0
lines changed

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

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -341,6 +341,14 @@
341341
VerifyTipPresenceCommandType,
342342
)
343343

344+
from .get_next_tip import (
345+
GetNextTip,
346+
GetNextTipCreate,
347+
GetNextTipParams,
348+
GetNextTipResult,
349+
GetNextTipCommandType,
350+
)
351+
344352
from .liquid_probe import (
345353
LiquidProbe,
346354
LiquidProbeParams,
@@ -611,6 +619,12 @@
611619
"VerifyTipPresenceParams",
612620
"VerifyTipPresenceResult",
613621
"VerifyTipPresenceCommandType",
622+
# get next tip command bundle
623+
"GetNextTip",
624+
"GetNextTipCreate",
625+
"GetNextTipParams",
626+
"GetNextTipResult",
627+
"GetNextTipCommandType",
614628
# liquid probe command bundle
615629
"LiquidProbe",
616630
"LiquidProbeParams",

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

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -323,6 +323,14 @@
323323
GetTipPresenceCommandType,
324324
)
325325

326+
from .get_next_tip import (
327+
GetNextTip,
328+
GetNextTipCreate,
329+
GetNextTipParams,
330+
GetNextTipResult,
331+
GetNextTipCommandType,
332+
)
333+
326334
from .liquid_probe import (
327335
LiquidProbe,
328336
LiquidProbeParams,
@@ -375,6 +383,7 @@
375383
SetStatusBar,
376384
VerifyTipPresence,
377385
GetTipPresence,
386+
GetNextTip,
378387
LiquidProbe,
379388
TryLiquidProbe,
380389
heater_shaker.WaitForTemperature,
@@ -460,6 +469,7 @@
460469
SetStatusBarParams,
461470
VerifyTipPresenceParams,
462471
GetTipPresenceParams,
472+
GetNextTipParams,
463473
LiquidProbeParams,
464474
TryLiquidProbeParams,
465475
heater_shaker.WaitForTemperatureParams,
@@ -543,6 +553,7 @@
543553
SetStatusBarCommandType,
544554
VerifyTipPresenceCommandType,
545555
GetTipPresenceCommandType,
556+
GetNextTipCommandType,
546557
LiquidProbeCommandType,
547558
TryLiquidProbeCommandType,
548559
heater_shaker.WaitForTemperatureCommandType,
@@ -627,6 +638,7 @@
627638
SetStatusBarCreate,
628639
VerifyTipPresenceCreate,
629640
GetTipPresenceCreate,
641+
GetNextTipCreate,
630642
LiquidProbeCreate,
631643
TryLiquidProbeCreate,
632644
heater_shaker.WaitForTemperatureCreate,
@@ -712,6 +724,7 @@
712724
SetStatusBarResult,
713725
VerifyTipPresenceResult,
714726
GetTipPresenceResult,
727+
GetNextTipResult,
715728
LiquidProbeResult,
716729
TryLiquidProbeResult,
717730
heater_shaker.WaitForTemperatureResult,

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

Lines changed: 127 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,127 @@
1+
"""Get next tip command request, result, and implementation models."""
2+
3+
from __future__ import annotations
4+
from pydantic import BaseModel, Field
5+
from typing import TYPE_CHECKING, Optional, Type, List, Literal, Union
6+
7+
from opentrons.types import NozzleConfigurationType
8+
9+
from ..errors import ErrorOccurrence
10+
from ..types import NextTipInfo, NoTipAvailable, NoTipReason
11+
from .pipetting_common import PipetteIdMixin
12+
13+
from .command import (
14+
AbstractCommandImpl,
15+
BaseCommand,
16+
BaseCommandCreate,
17+
SuccessData,
18+
)
19+
20+
if TYPE_CHECKING:
21+
from ..state.state import StateView
22+
23+
24+
GetNextTipCommandType = Literal["getNextTip"]
25+
26+
27+
class GetNextTipParams(PipetteIdMixin):
28+
"""Payload needed to resolve the next available tip."""
29+
30+
labwareIds: List[str] = Field(
31+
...,
32+
description="Labware ID(s) of tip racks to resolve next available tip(s) from"
33+
" Labware IDs will be resolved sequentially",
34+
)
35+
startingTipWell: Optional[str] = Field(
36+
None,
37+
description="Name of starting tip rack 'well'."
38+
" This only applies to the first tip rack in the list provided in labwareIDs",
39+
)
40+
41+
42+
class GetNextTipResult(BaseModel):
43+
"""Result data from the execution of a GetNextTip."""
44+
45+
nextTipInfo: Union[NextTipInfo, NoTipAvailable] = Field(
46+
...,
47+
description="Labware ID and well name of next available tip for a pipette,"
48+
" or information why no tip could be resolved.",
49+
)
50+
51+
52+
class GetNextTipImplementation(
53+
AbstractCommandImpl[GetNextTipParams, SuccessData[GetNextTipResult]]
54+
):
55+
"""Get next tip command implementation."""
56+
57+
def __init__(
58+
self,
59+
state_view: StateView,
60+
**kwargs: object,
61+
) -> None:
62+
self._state_view = state_view
63+
64+
async def execute(self, params: GetNextTipParams) -> SuccessData[GetNextTipResult]:
65+
"""Get the next available tip for the requested pipette."""
66+
pipette_id = params.pipetteId
67+
starting_tip_name = params.startingTipWell
68+
69+
num_tips = self._state_view.tips.get_pipette_active_channels(pipette_id)
70+
nozzle_map = self._state_view.tips.get_pipette_nozzle_map(pipette_id)
71+
72+
if (
73+
starting_tip_name is not None
74+
and nozzle_map.configuration != NozzleConfigurationType.FULL
75+
):
76+
# This is to match the behavior found in PAPI, but also because we don't have logic to automatically find
77+
# the next tip with partial configuration and a starting tip. This will never work for a 96-channel due to
78+
# x-axis overlap, but could eventually work with 8-channel if we better define starting tip USED or CLEAN
79+
# state when starting a protocol to prevent accidental tip pick-up with starting non-full tip racks.
80+
return SuccessData(
81+
public=GetNextTipResult(
82+
nextTipInfo=NoTipAvailable(
83+
noTipReason=NoTipReason.STARTING_TIP_WITH_PARTIAL,
84+
message="Cannot automatically resolve next tip with starting tip and partial tip configuration.",
85+
)
86+
)
87+
)
88+
89+
next_tip: Union[NextTipInfo, NoTipAvailable]
90+
for labware_id in params.labwareIds:
91+
well_name = self._state_view.tips.get_next_tip(
92+
labware_id=labware_id,
93+
num_tips=num_tips,
94+
starting_tip_name=starting_tip_name,
95+
nozzle_map=nozzle_map,
96+
)
97+
if well_name is not None:
98+
next_tip = NextTipInfo(labwareId=labware_id, tipStartingWell=well_name)
99+
break
100+
# After the first tip rack is exhausted, starting tip no longer applies
101+
starting_tip_name = None
102+
else:
103+
next_tip = NoTipAvailable(
104+
noTipReason=NoTipReason.NO_AVAILABLE_TIPS,
105+
message="No available tips for given pipette, nozzle configuration and provided tip racks.",
106+
)
107+
108+
return SuccessData(public=GetNextTipResult(nextTipInfo=next_tip))
109+
110+
111+
class GetNextTip(BaseCommand[GetNextTipParams, GetNextTipResult, ErrorOccurrence]):
112+
"""Get next tip command model."""
113+
114+
commandType: GetNextTipCommandType = "getNextTip"
115+
params: GetNextTipParams
116+
result: Optional[GetNextTipResult]
117+
118+
_ImplementationCls: Type[GetNextTipImplementation] = GetNextTipImplementation
119+
120+
121+
class GetNextTipCreate(BaseCommandCreate[GetNextTipParams]):
122+
"""Get next tip command creation request model."""
123+
124+
commandType: GetNextTipCommandType = "getNextTip"
125+
params: GetNextTipParams
126+
127+
_CommandCls: Type[GetNextTip] = GetNextTip

‎api/src/opentrons/protocol_engine/types.py

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1115,6 +1115,37 @@ def from_hw_state(cls, state: HwTipStateType) -> "TipPresenceStatus":
11151115
}[state]
11161116

11171117

1118+
class NextTipInfo(BaseModel):
1119+
"""Next available tip labware and well name data."""
1120+
1121+
labwareId: str = Field(
1122+
...,
1123+
description="The labware ID of the tip rack where the next available tip(s) are located.",
1124+
)
1125+
tipStartingWell: str = Field(
1126+
..., description="The (starting) well name of the next available tip(s)."
1127+
)
1128+
1129+
1130+
class NoTipReason(Enum):
1131+
"""The cause of no tip being available for a pipette and tip rack(s)."""
1132+
1133+
NO_AVAILABLE_TIPS = "noAvailableTips"
1134+
STARTING_TIP_WITH_PARTIAL = "startingTipWithPartial"
1135+
INCOMPATIBLE_CONFIGURATION = "incompatibleConfiguration"
1136+
1137+
1138+
class NoTipAvailable(BaseModel):
1139+
"""No available next tip data."""
1140+
1141+
noTipReason: NoTipReason = Field(
1142+
..., description="The reason why no next available tip could be provided."
1143+
)
1144+
message: Optional[str] = Field(
1145+
None, description="Optional message explaining why a tip wasn't available."
1146+
)
1147+
1148+
11181149
# TODO (spp, 2024-04-02): move all RTP types to runner
11191150
class RTPBase(BaseModel):
11201151
"""Parameters defined in a protocol."""

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

Lines changed: 142 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,142 @@
1+
"""Test get next tip in place commands."""
2+
from decoy import Decoy
3+
4+
from opentrons.types import NozzleConfigurationType
5+
from opentrons.protocol_engine import StateView
6+
from opentrons.protocol_engine.types import NextTipInfo, NoTipAvailable, NoTipReason
7+
from opentrons.protocol_engine.commands.command import SuccessData
8+
from opentrons.protocol_engine.commands.get_next_tip import (
9+
GetNextTipParams,
10+
GetNextTipResult,
11+
GetNextTipImplementation,
12+
)
13+
14+
from opentrons.hardware_control.nozzle_manager import NozzleMap
15+
16+
17+
async def test_get_next_tip_implementation(
18+
decoy: Decoy,
19+
state_view: StateView,
20+
) -> None:
21+
"""A GetNextTip command should have an execution implementation."""
22+
subject = GetNextTipImplementation(state_view=state_view)
23+
params = GetNextTipParams(
24+
pipetteId="abc", labwareIds=["123"], startingTipWell="xyz"
25+
)
26+
mock_nozzle_map = decoy.mock(cls=NozzleMap)
27+
28+
decoy.when(state_view.tips.get_pipette_active_channels("abc")).then_return(42)
29+
decoy.when(state_view.tips.get_pipette_nozzle_map("abc")).then_return(
30+
mock_nozzle_map
31+
)
32+
decoy.when(mock_nozzle_map.configuration).then_return(NozzleConfigurationType.FULL)
33+
34+
decoy.when(
35+
state_view.tips.get_next_tip(
36+
labware_id="123",
37+
num_tips=42,
38+
starting_tip_name="xyz",
39+
nozzle_map=mock_nozzle_map,
40+
)
41+
).then_return("foo")
42+
43+
result = await subject.execute(params)
44+
45+
assert result == SuccessData(
46+
public=GetNextTipResult(
47+
nextTipInfo=NextTipInfo(labwareId="123", tipStartingWell="foo")
48+
),
49+
)
50+
51+
52+
async def test_get_next_tip_implementation_multiple_tip_racks(
53+
decoy: Decoy,
54+
state_view: StateView,
55+
) -> None:
56+
"""A GetNextTip command with multiple tip racks should not apply starting tip to the following ones."""
57+
subject = GetNextTipImplementation(state_view=state_view)
58+
params = GetNextTipParams(
59+
pipetteId="abc", labwareIds=["123", "456"], startingTipWell="xyz"
60+
)
61+
mock_nozzle_map = decoy.mock(cls=NozzleMap)
62+
63+
decoy.when(state_view.tips.get_pipette_active_channels("abc")).then_return(42)
64+
decoy.when(state_view.tips.get_pipette_nozzle_map("abc")).then_return(
65+
mock_nozzle_map
66+
)
67+
decoy.when(mock_nozzle_map.configuration).then_return(NozzleConfigurationType.FULL)
68+
69+
decoy.when(
70+
state_view.tips.get_next_tip(
71+
labware_id="456",
72+
num_tips=42,
73+
starting_tip_name=None,
74+
nozzle_map=mock_nozzle_map,
75+
)
76+
).then_return("foo")
77+
78+
result = await subject.execute(params)
79+
80+
assert result == SuccessData(
81+
public=GetNextTipResult(
82+
nextTipInfo=NextTipInfo(labwareId="456", tipStartingWell="foo")
83+
),
84+
)
85+
86+
87+
async def test_get_next_tip_implementation_no_tips(
88+
decoy: Decoy,
89+
state_view: StateView,
90+
) -> None:
91+
"""A GetNextTip command should return with NoTipAvailable if there are no available tips."""
92+
subject = GetNextTipImplementation(state_view=state_view)
93+
params = GetNextTipParams(
94+
pipetteId="abc", labwareIds=["123", "456"], startingTipWell="xyz"
95+
)
96+
mock_nozzle_map = decoy.mock(cls=NozzleMap)
97+
98+
decoy.when(state_view.tips.get_pipette_active_channels("abc")).then_return(42)
99+
decoy.when(state_view.tips.get_pipette_nozzle_map("abc")).then_return(
100+
mock_nozzle_map
101+
)
102+
decoy.when(mock_nozzle_map.configuration).then_return(NozzleConfigurationType.FULL)
103+
104+
result = await subject.execute(params)
105+
106+
assert result == SuccessData(
107+
public=GetNextTipResult(
108+
nextTipInfo=NoTipAvailable(
109+
noTipReason=NoTipReason.NO_AVAILABLE_TIPS,
110+
message="No available tips for given pipette, nozzle configuration and provided tip racks.",
111+
)
112+
),
113+
)
114+
115+
116+
async def test_get_next_tip_implementation_partial_with_starting_tip(
117+
decoy: Decoy,
118+
state_view: StateView,
119+
) -> None:
120+
"""A GetNextTip command should return with NoTipAvailable if there's a starting tip and a partial config."""
121+
subject = GetNextTipImplementation(state_view=state_view)
122+
params = GetNextTipParams(
123+
pipetteId="abc", labwareIds=["123", "456"], startingTipWell="xyz"
124+
)
125+
mock_nozzle_map = decoy.mock(cls=NozzleMap)
126+
127+
decoy.when(state_view.tips.get_pipette_active_channels("abc")).then_return(42)
128+
decoy.when(state_view.tips.get_pipette_nozzle_map("abc")).then_return(
129+
mock_nozzle_map
130+
)
131+
decoy.when(mock_nozzle_map.configuration).then_return(NozzleConfigurationType.ROW)
132+
133+
result = await subject.execute(params)
134+
135+
assert result == SuccessData(
136+
public=GetNextTipResult(
137+
nextTipInfo=NoTipAvailable(
138+
noTipReason=NoTipReason.STARTING_TIP_WITH_PARTIAL,
139+
message="Cannot automatically resolve next tip with starting tip and partial tip configuration.",
140+
)
141+
),
142+
)

0 commit comments

Comments
 (0)