Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(engine, api): alternate tip drop location for addressable area trash bins #14196

Merged
26 changes: 18 additions & 8 deletions api/src/opentrons/protocol_api/core/engine/instrument.py
Original file line number Diff line number Diff line change
Expand Up @@ -487,6 +487,16 @@ def _move_to_disposal_location(

if isinstance(disposal_location, TrashBin):
addressable_area_name = disposal_location._addressable_area_name
self._engine_client.move_to_addressable_area_for_drop_tip(
pipette_id=self._pipette_id,
addressable_area_name=addressable_area_name,
offset=offset,
force_direct=force_direct,
speed=speed,
minimum_z_height=None,
alternate_drop_location=True,
jbleon95 marked this conversation as resolved.
Show resolved Hide resolved
)

if isinstance(disposal_location, WasteChute):
num_channels = self.get_channels()
addressable_area_name = {
Expand All @@ -495,14 +505,14 @@ def _move_to_disposal_location(
96: "96ChannelWasteChute",
}[num_channels]

self._engine_client.move_to_addressable_area(
pipette_id=self._pipette_id,
addressable_area_name=addressable_area_name,
offset=offset,
force_direct=force_direct,
speed=speed,
minimum_z_height=None,
)
self._engine_client.move_to_addressable_area(
pipette_id=self._pipette_id,
addressable_area_name=addressable_area_name,
offset=offset,
force_direct=force_direct,
speed=speed,
minimum_z_height=None,
)

def _drop_tip_in_place(self, home_after: Optional[bool]) -> None:
self._engine_client.drop_tip_in_place(
Expand Down
26 changes: 26 additions & 0 deletions api/src/opentrons/protocol_engine/clients/sync_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -205,6 +205,32 @@ def move_to_addressable_area(

return cast(commands.MoveToAddressableAreaResult, result)

def move_to_addressable_area_for_drop_tip(
self,
pipette_id: str,
addressable_area_name: str,
offset: AddressableOffsetVector,
minimum_z_height: Optional[float],
force_direct: bool,
speed: Optional[float],
alternate_drop_location: Optional[bool],
) -> commands.MoveToAddressableAreaForDropTipResult:
"""Execute a MoveToAddressableArea command and return the result."""
request = commands.MoveToAddressableAreaForDropTipCreate(
params=commands.MoveToAddressableAreaForDropTipParams(
pipetteId=pipette_id,
addressableAreaName=addressable_area_name,
offset=offset,
forceDirect=force_direct,
minimumZHeight=minimum_z_height,
speed=speed,
alternateDropLocation=alternate_drop_location,
)
)
result = self._transport.execute_command(request=request)

return cast(commands.MoveToAddressableAreaForDropTipResult, result)

def move_to_coordinates(
self,
pipette_id: str,
Expand Down
14 changes: 14 additions & 0 deletions api/src/opentrons/protocol_engine/commands/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -185,6 +185,14 @@
MoveToAddressableAreaCommandType,
)

from .move_to_addressable_area_for_drop_tip import (
MoveToAddressableAreaForDropTip,
MoveToAddressableAreaForDropTipParams,
MoveToAddressableAreaForDropTipCreate,
MoveToAddressableAreaForDropTipResult,
MoveToAddressableAreaForDropTipCommandType,
)

from .wait_for_resume import (
WaitForResume,
WaitForResumeParams,
Expand Down Expand Up @@ -435,6 +443,12 @@
"MoveToAddressableAreaCreate",
"MoveToAddressableAreaResult",
"MoveToAddressableAreaCommandType",
# move to addressable area for drop tip command models
"MoveToAddressableAreaForDropTip",
"MoveToAddressableAreaForDropTipParams",
"MoveToAddressableAreaForDropTipCreate",
"MoveToAddressableAreaForDropTipResult",
"MoveToAddressableAreaForDropTipCommandType",
# wait for resume command models
"WaitForResume",
"WaitForResumeParams",
Expand Down
13 changes: 13 additions & 0 deletions api/src/opentrons/protocol_engine/commands/command_unions.py
Original file line number Diff line number Diff line change
Expand Up @@ -162,6 +162,14 @@
MoveToAddressableAreaCommandType,
)

from .move_to_addressable_area_for_drop_tip import (
MoveToAddressableAreaForDropTip,
MoveToAddressableAreaForDropTipParams,
MoveToAddressableAreaForDropTipCreate,
MoveToAddressableAreaForDropTipResult,
MoveToAddressableAreaForDropTipCommandType,
)

from .wait_for_resume import (
WaitForResume,
WaitForResumeParams,
Expand Down Expand Up @@ -300,6 +308,7 @@
MoveToCoordinates,
MoveToWell,
MoveToAddressableArea,
MoveToAddressableAreaForDropTip,
PrepareToAspirate,
WaitForResume,
WaitForDuration,
Expand Down Expand Up @@ -361,6 +370,7 @@
MoveToCoordinatesParams,
MoveToWellParams,
MoveToAddressableAreaParams,
MoveToAddressableAreaForDropTipParams,
PrepareToAspirateParams,
WaitForResumeParams,
WaitForDurationParams,
Expand Down Expand Up @@ -423,6 +433,7 @@
MoveToCoordinatesCommandType,
MoveToWellCommandType,
MoveToAddressableAreaCommandType,
MoveToAddressableAreaForDropTipCommandType,
PrepareToAspirateCommandType,
WaitForResumeCommandType,
WaitForDurationCommandType,
Expand Down Expand Up @@ -484,6 +495,7 @@
MoveToCoordinatesCreate,
MoveToWellCreate,
MoveToAddressableAreaCreate,
MoveToAddressableAreaForDropTipCreate,
PrepareToAspirateCreate,
WaitForResumeCreate,
WaitForDurationCreate,
Expand Down Expand Up @@ -545,6 +557,7 @@
MoveToCoordinatesResult,
MoveToWellResult,
MoveToAddressableAreaResult,
MoveToAddressableAreaForDropTipResult,
PrepareToAspirateResult,
WaitForResumeResult,
WaitForDurationResult,
Expand Down
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
"""Move to well command request, result, and implementation models."""
"""Move to addressable area command request, result, and implementation models."""
from __future__ import annotations
from pydantic import Field
from typing import TYPE_CHECKING, Optional, Type
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,149 @@
"""Move to addressable area for drop tip command request, result, and implementation models."""
from __future__ import annotations
from pydantic import Field
from typing import TYPE_CHECKING, Optional, Type
from typing_extensions import Literal

from ..errors import LocationNotAccessibleByPipetteError
from ..types import DeckPoint, AddressableOffsetVector
from ..resources import fixture_validation
from .pipetting_common import (
PipetteIdMixin,
MovementMixin,
DestinationPositionResult,
)
from .command import AbstractCommandImpl, BaseCommand, BaseCommandCreate

if TYPE_CHECKING:
from ..execution import MovementHandler
from ..state import StateView

MoveToAddressableAreaForDropTipCommandType = Literal["moveToAddressableAreaForDropTip"]


class MoveToAddressableAreaForDropTipParams(PipetteIdMixin, MovementMixin):
"""Payload required to move a pipette to a specific addressable area.
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We could probably nuke this docstring and say something like "this is like moveToAddressableArea except it has tip alternation.


An *addressable area* is a space in the robot that may or may not be usable depending on how
the robot's deck is configured. For example, if a Flex is configured with a waste chute, it will
have additional addressable areas representing the opening of the waste chute, where tips and
labware can be dropped.

This moves the pipette so all of its nozzles are centered over the addressable area.
If the pipette is currently configured with a partial tip layout, this centering is over all
the pipette's physical nozzles, not just the nozzles that are active.

The z-position will be chosen to put the bottom of the tips---or the bottom of the nozzles,
if there are no tips---level with the top of the addressable area.

When this command is executed, Protocol Engine will make sure the robot's deck is configured
such that the requested addressable area actually exists. For example, if you request
the addressable area B4, it will make sure the robot is set up with a B3/B4 staging area slot.
If that's not the case, the command will fail.
"""

addressableAreaName: str = Field(
...,
description=(
"The name of the addressable area that you want to use."
" Valid values are the `id`s of `addressableArea`s in the"
" [deck definition](https://github.com/Opentrons/opentrons/tree/edge/shared-data/deck)."
),
)
offset: AddressableOffsetVector = Field(
AddressableOffsetVector(x=0, y=0, z=0),
description="Relative offset of addressable area to move pipette's critical point.",
)
alternateDropLocation: Optional[bool] = Field(
False,
description=(
"Whether to alternate location where tip is dropped within the addressable area."
" If True, this command will ignore the offset provided and alternate"
" between dropping tips at two predetermined locations inside the specified"
" labware well."
" If False, the tip will be dropped at the top center of the area."
),
)


class MoveToAddressableAreaForDropTipResult(DestinationPositionResult):
"""Result data from the execution of a MoveToAddressableAreaForDropTip command."""

pass


class MoveToAddressableAreaForDropTipImplementation(
AbstractCommandImpl[
MoveToAddressableAreaForDropTipParams, MoveToAddressableAreaForDropTipResult
]
):
"""Move to addressable area for drop tip command implementation."""

def __init__(
self, movement: MovementHandler, state_view: StateView, **kwargs: object
) -> None:
self._movement = movement
self._state_view = state_view

async def execute(
self, params: MoveToAddressableAreaForDropTipParams
) -> MoveToAddressableAreaForDropTipResult:
"""Move the requested pipette to the requested addressable area in preperation of a drop tip."""
self._state_view.addressable_areas.raise_if_area_not_in_deck_configuration(
params.addressableAreaName
)

if fixture_validation.is_staging_slot(params.addressableAreaName):
raise LocationNotAccessibleByPipetteError(
f"Cannot move pipette to staging slot {params.addressableAreaName}"
)

if params.alternateDropLocation:
offset = self._state_view.geometry.get_next_tip_drop_location_for_addressable_area(
addressable_area_name=params.addressableAreaName,
pipette_id=params.pipetteId,
)
else:
offset = params.offset

x, y, z = await self._movement.move_to_addressable_area(
pipette_id=params.pipetteId,
addressable_area_name=params.addressableAreaName,
offset=offset,
force_direct=params.forceDirect,
minimum_z_height=params.minimumZHeight,
speed=params.speed,
)

return MoveToAddressableAreaForDropTipResult(position=DeckPoint(x=x, y=y, z=z))


class MoveToAddressableAreaForDropTip(
BaseCommand[
MoveToAddressableAreaForDropTipParams, MoveToAddressableAreaForDropTipResult
]
):
"""Move to addressable area for drop tip command model."""

commandType: MoveToAddressableAreaForDropTipCommandType = (
"moveToAddressableAreaForDropTip"
)
params: MoveToAddressableAreaForDropTipParams
result: Optional[MoveToAddressableAreaForDropTipResult]

_ImplementationCls: Type[
MoveToAddressableAreaForDropTipImplementation
] = MoveToAddressableAreaForDropTipImplementation


class MoveToAddressableAreaForDropTipCreate(
BaseCommandCreate[MoveToAddressableAreaForDropTipParams]
):
"""Move to addressable area for drop tip command creation request model."""

commandType: MoveToAddressableAreaForDropTipCommandType = (
"moveToAddressableAreaForDropTip"
)
params: MoveToAddressableAreaForDropTipParams

_CommandCls: Type[MoveToAddressableAreaForDropTip] = MoveToAddressableAreaForDropTip
6 changes: 5 additions & 1 deletion api/src/opentrons/protocol_engine/state/addressable_areas.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
LoadModuleResult,
MoveLabwareResult,
MoveToAddressableAreaResult,
MoveToAddressableAreaForDropTipResult,
)
from ..errors import (
IncompatibleAddressableAreaError,
Expand Down Expand Up @@ -203,7 +204,10 @@ def _handle_command(self, command: Command) -> None:
elif isinstance(command.result, LoadModuleResult):
self._check_location_is_addressable_area(command.params.location)

elif isinstance(command.result, MoveToAddressableAreaResult):
elif isinstance(
command.result,
(MoveToAddressableAreaResult, MoveToAddressableAreaForDropTipResult),
):
addressable_area_name = command.params.addressableAreaName
self._check_location_is_addressable_area(addressable_area_name)

Expand Down
Loading
Loading