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(api): Experimental, barebones support for dropping stuff in the waste chute #13828

Merged
merged 13 commits into from
Oct 25, 2023
17 changes: 17 additions & 0 deletions api/drop_labware_in_waste_chute.py
SyntaxColoring marked this conversation as resolved.
Show resolved Hide resolved
Original file line number Diff line number Diff line change
@@ -0,0 +1,17 @@
from opentrons import protocol_api
SyntaxColoring marked this conversation as resolved.
Show resolved Hide resolved


requirements = {
"apiLevel": "2.16",
"robotType": "Flex",
}


def run(protocol: protocol_api.ProtocolContext) -> None:
waste_chute = protocol.load_waste_chute(orifice="wide_open")

labware_1 = protocol.load_labware("opentrons_flex_96_tiprack_1000ul", "C1")
labware_2 = protocol.load_labware("opentrons_flex_96_tiprack_1000ul", "C2")

protocol.move_labware(labware_1, waste_chute, use_gripper=True)
protocol.move_labware(labware_2, waste_chute, use_gripper=True)
19 changes: 19 additions & 0 deletions api/drop_tips_in_waste_chute.py
SyntaxColoring marked this conversation as resolved.
Show resolved Hide resolved
Original file line number Diff line number Diff line change
@@ -0,0 +1,19 @@
from opentrons import protocol_api


requirements = {
"apiLevel": "2.16",
"robotType": "Flex",
}


def run(protocol: protocol_api.ProtocolContext) -> None:
tip_rack = protocol.load_labware("opentrons_flex_96_tiprack_1000ul", "C2")

waste_chute = protocol.load_waste_chute(orifice="columnar_slit")

pipette = protocol.load_instrument("flex_8channel_1000", mount="left")

for column in tip_rack.columns():
pipette.pick_up_tip(column[0])
pipette.drop_tip(waste_chute)
4 changes: 3 additions & 1 deletion api/src/opentrons/protocol_api/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,6 @@
from .deck import Deck
from .instrument_context import InstrumentContext
from .labware import Labware, Well
from ._types import OFF_DECK
from .module_contexts import (
ModuleContext,
ThermocyclerContext,
Expand All @@ -24,6 +23,8 @@
MagneticBlockContext,
)
from ._liquid import Liquid
from ._types import OFF_DECK
from ._waste_chute import WasteChute

from .create_protocol_context import (
create_protocol_context,
Expand All @@ -44,6 +45,7 @@
"HeaterShakerContext",
"MagneticBlockContext",
"Labware",
"WasteChute",
"Well",
"Liquid",
"OFF_DECK",
Expand Down
16 changes: 16 additions & 0 deletions api/src/opentrons/protocol_api/_waste_chute.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,16 @@
from typing_extensions import Literal


class WasteChute:
"""Represents a Flex waste chute.

See :py:obj:`ProtocolContext.load_waste_chute`.
"""

def __init__(
self,
with_staging_area_slot_d4: bool,
orifice: Literal["wide_open", "columnar_slit"],
) -> None:
self._with_staging_area_slot_d4 = with_staging_area_slot_d4
self._orifice = orifice
20 changes: 20 additions & 0 deletions api/src/opentrons/protocol_api/_waste_chute_dimensions.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
"""Constants for the dimensions of the Flex waste chute.

https://docs.google.com/spreadsheets/d/1yGRN-5QOa45p-jLsi4tL_PbHmKSK-QsXqHws6oBK7do/edit#gid=258725851
SyntaxColoring marked this conversation as resolved.
Show resolved Hide resolved

These should be moved into shared-data and interpreted by Protocol Engine.
"""


from opentrons.types import Point


SLOT_ORIGIN_TO_1_OR_8_TIP_A1 = Point(64, 21.91, 144)
SLOT_ORIGIN_TO_96_TIP_A1 = Point(14.445, 42.085, 115)

# TODO: This z-coord is misleading. We need to account for the labware height and the paddle height;
# we can't define this as a single coordinate.
SLOT_ORIGIN_TO_GRIPPER_JAW_CENTER = Point(64, 29, 136.5)

# This includes the height of the optional lid.
ENVELOPE_HEIGHT = 154
66 changes: 65 additions & 1 deletion api/src/opentrons/protocol_api/core/engine/instrument.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,12 +18,16 @@
from opentrons.protocol_engine.errors.exceptions import TipNotAttachedError
from opentrons.protocol_engine.clients import SyncClient as EngineClient
from opentrons.protocols.api_support.definitions import MAX_SUPPORTED_VERSION
from opentrons.types import Point

from opentrons_shared_data.pipette.dev_types import PipetteNameType

from ..instrument import AbstractInstrument
from .well import WellCore

from ..._waste_chute import WasteChute
from ... import _waste_chute_dimensions

if TYPE_CHECKING:
from .protocol import ProtocolCore

Expand Down Expand Up @@ -373,6 +377,67 @@ def drop_tip(

self._protocol_core.set_last_location(location=location, mount=self.get_mount())

def _drop_tip_in_place(self, home_after: Optional[bool]) -> None:
self._engine_client.drop_tip_in_place(
pipette_id=self._pipette_id,
home_after=home_after,
)

def drop_tip_in_waste_chute(
self, waste_chute: WasteChute, home_after: Optional[bool]
) -> None:
# TODO: Can we get away with implementing this in two steps like this,
# or does drop_tip() need to take the waste chute location because the z-height
# depends on the intent of dropping tip? How would Protocol Designer want to implement
# this?
self._move_to_waste_chute(
waste_chute,
force_direct=False,
speed=None,
)
self._drop_tip_in_place(home_after=home_after)

def _move_to_waste_chute(
self,
waste_chute: WasteChute,
force_direct: bool,
speed: Optional[float],
) -> None:
if self.get_channels() == 96:
if waste_chute._orifice != "wide_open":
raise ValueError(
"Wrong waste chute, you silly goose."
" You're using a 96-channel pipette, but the waste chute has a lid."
)
slot_origin_to_tip_a1 = _waste_chute_dimensions.SLOT_ORIGIN_TO_96_TIP_A1
else:
slot_origin_to_tip_a1 = _waste_chute_dimensions.SLOT_ORIGIN_TO_1_OR_8_TIP_A1

# TODO: All of this logic to compute the destination coordinate belongs in Protocol Engine.
slot_d3 = next(
s
for s in self._protocol_core.get_deck_definition()["locations"][
"orderedSlots"
]
if s["id"] == "D3"
)
slot_d3_origin = Point(*slot_d3["position"])
destination_point = slot_d3_origin + slot_origin_to_tip_a1

# Normally, we use a 10 mm margin. (DEFAULT_GENERAL_ARC_Z_MARGIN.) Unfortunately, with
# 1000µL tips, we have slightly not enough room to meet that margin. We can make the margin
# as big as 7.5 mm before the motion planner raises an error. So, use that reduced margin,
# with a little more subtracted in order to leave wiggle room for pipette calibration.
minimum_z = _waste_chute_dimensions.ENVELOPE_HEIGHT + 5.0

self.move_to(
Location(destination_point, labware=None),
well_core=None,
force_direct=force_direct,
minimum_z_height=minimum_z,
speed=speed,
)

def home(self) -> None:
z_axis = self._engine_client.state.pipettes.get_z_axis(self._pipette_id)
plunger_axis = self._engine_client.state.pipettes.get_plunger_axis(
Expand All @@ -394,7 +459,6 @@ def move_to(
minimum_z_height: Optional[float],
speed: Optional[float],
) -> None:

if well_core is not None:
labware_id = well_core.labware_id
well_name = well_core.get_name()
Expand Down
104 changes: 91 additions & 13 deletions api/src/opentrons/protocol_api/core/engine/protocol.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,8 @@
"""ProtocolEngine-based Protocol API core implementation."""
from typing import Dict, Optional, Type, Union, List, Tuple
from typing_extensions import Literal

from opentrons.protocol_api import _waste_chute_dimensions

from opentrons.protocol_engine.commands import LoadModuleResult
from opentrons_shared_data.deck.dev_types import DeckDefinitionV3, SlotDefV3
Expand Down Expand Up @@ -39,8 +42,9 @@
)

from ... import validation
from ..._types import OffDeckType
from ..._types import OffDeckType, OFF_DECK
from ..._liquid import Liquid
from ..._waste_chute import WasteChute
from ..protocol import AbstractProtocol
from ..labware import LabwareLoadParams
from .labware import LabwareCore
Expand Down Expand Up @@ -248,16 +252,19 @@ def move_labware(
self,
labware_core: LabwareCore,
new_location: Union[
DeckSlotName, LabwareCore, ModuleCore, NonConnectedModuleCore, OffDeckType
DeckSlotName,
LabwareCore,
ModuleCore,
NonConnectedModuleCore,
OffDeckType,
WasteChute,
],
use_gripper: bool,
pause_for_manual_move: bool,
pick_up_offset: Optional[Tuple[float, float, float]],
drop_offset: Optional[Tuple[float, float, float]],
) -> None:
"""Move the given labware to a new location."""
to_location = self._convert_labware_location(location=new_location)

if use_gripper:
strategy = LabwareMovementStrategy.USING_GRIPPER
elif pause_for_manual_move:
Expand All @@ -272,27 +279,80 @@ def move_labware(
if pick_up_offset
else None
)

_drop_offset = (
LabwareOffsetVector(x=drop_offset[0], y=drop_offset[1], z=drop_offset[2])
if drop_offset
else None
)

# TODO(mm, 2023-02-23): Check for conflicts with other items on the deck,
# when move_labware() support is no longer experimental.
if isinstance(new_location, WasteChute):
self._move_labware_to_waste_chute(
labware_core, strategy, _pick_up_offset, _drop_offset
)
else:
to_location = self._convert_labware_location(location=new_location)

# TODO(mm, 2023-02-23): Check for conflicts with other items on the deck,
# when move_labware() support is no longer experimental.

self._engine_client.move_labware(
labware_id=labware_core.labware_id,
new_location=to_location,
strategy=strategy,
pick_up_offset=_pick_up_offset,
drop_offset=_drop_offset,
)

self._engine_client.move_labware(
labware_id=labware_core.labware_id,
new_location=to_location,
strategy=strategy,
pick_up_offset=_pick_up_offset,
drop_offset=_drop_offset,
)
if strategy == LabwareMovementStrategy.USING_GRIPPER:
# Clear out last location since it is not relevant to pipetting
# and we only use last location for in-place pipetting commands
self.set_last_location(location=None, mount=Mount.EXTENSION)

def _move_labware_to_waste_chute(
self,
labware_core: LabwareCore,
strategy: LabwareMovementStrategy,
pick_up_offset: Optional[LabwareOffsetVector],
drop_offset: Optional[LabwareOffsetVector],
) -> None:
slot = DeckSlotLocation(slotName=DeckSlotName.SLOT_D3)
slot_width = 128
slot_height = 86
drop_offset_from_slot = (
_waste_chute_dimensions.SLOT_ORIGIN_TO_GRIPPER_JAW_CENTER
- Point(x=slot_width / 2, y=slot_height / 2)
)
if drop_offset is not None:
drop_offset_from_slot += Point(
x=drop_offset.x, y=drop_offset.y, z=drop_offset.z
)

# To get the physical movement to happen, move the labware "into the slot" with a giant
# offset to dunk it in the waste chute.
self._engine_client.move_labware(
labware_id=labware_core.labware_id,
new_location=slot,
strategy=strategy,
pick_up_offset=pick_up_offset,
drop_offset=LabwareOffsetVector(
x=drop_offset_from_slot.x,
y=drop_offset_from_slot.y,
z=drop_offset_from_slot.z,
),
)

# To get the logical movement to be correct, move the labware off-deck.
# Otherwise, leaving the labware "in the slot" would mean you couldn't call this function
# again for other labware.
self._engine_client.move_labware(
labware_id=labware_core.labware_id,
new_location=self._convert_labware_location(OFF_DECK),
strategy=LabwareMovementStrategy.MANUAL_MOVE_WITHOUT_PAUSE,
pick_up_offset=None,
drop_offset=None,
)

def _resolve_module_hardware(
self, serial_number: str, model: ModuleModel
) -> AbstractModule:
Expand Down Expand Up @@ -606,3 +666,21 @@ def _get_non_stacked_location(
return OFF_DECK_LOCATION
elif isinstance(location, DeckSlotName):
return DeckSlotLocation(slotName=location)


def _waste_chute_fixture_id(
with_staging_area_slot_d4: bool, orifice: Literal["wide_open", "columnar_slit"]
) -> str:
if orifice not in {"wide_open", "columnar_slit"}:
raise ValueError(
f"orifice must be 'wide_open' or 'columnar_slit', not {repr(orifice)}."
)

ids: Dict[Tuple[bool, Literal["wide_open", "columnar_slit"]], str] = {
(False, "wide_open"): "wasteChuteRightAdapterNoCover",
(False, "columnar_slit"): "wasteChuteRightAdapterCovered",
(True, "wide_open"): "stagingAreaSlotWithWasteChuteRightAdapterNoCover",
(True, "columnar_slit"): "stagingAreaSlotWithWasteChuteRightAdapterCovered",
}

return ids[with_staging_area_slot_d4, orifice]
7 changes: 7 additions & 0 deletions api/src/opentrons/protocol_api/core/instrument.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
from opentrons.hardware_control.dev_types import PipetteDict
from opentrons.protocols.api_support.util import FlowRates

from .._waste_chute import WasteChute
from .well import WellCoreType


Expand Down Expand Up @@ -132,6 +133,12 @@ def drop_tip(
"""
...

@abstractmethod
def drop_tip_in_waste_chute(
self, waste_chute: WasteChute, home_after: Optional[bool]
) -> None:
...

@abstractmethod
def home(self) -> None:
...
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
)
from opentrons.protocols.geometry import planning

from ..._waste_chute import WasteChute
from ..instrument import AbstractInstrument
from .legacy_well_core import LegacyWellCore
from .legacy_module_core import LegacyThermocyclerCore, LegacyHeaterShakerCore
Expand Down Expand Up @@ -282,6 +283,13 @@ def drop_tip(
f"Could not return tip to {labware_core.get_display_name()}"
)

def drop_tip_in_waste_chute(
self, waste_chute: WasteChute, home_after: Optional[bool]
) -> None:
raise APIVersionError(
"Dropping tips in a waste chute is not supported in this API Version."
)

def home(self) -> None:
"""Home the mount"""
self._protocol_interface.get_hardware().home_z(
Expand Down
Loading