feat(api): allow custom user offsets for deck configured trash bins a…

…nd waste chute (#14560)

Adds the ability to dispense, blow out, drop tip, and move to deck configured (api level 2.16 and above) trash bins and waste chutes

---------

Co-authored-by: Edward Cormany <[email protected]>

api/docs/v2/new_protocol_api.rst

@@ -35,9 +35,10 @@ Labware
  signatures, since users should never construct these directly.
 
 .. autoclass:: opentrons.protocol_api.TrashBin()
+ :members:
 
 .. autoclass:: opentrons.protocol_api.WasteChute()
-
+ :members:
 
 Wells and Liquids
 =================

api/src/opentrons/commands/commands.py

@@ -6,8 +6,7 @@
 from . import types as command_types
 
 from opentrons.types import Location
-from opentrons.protocol_api._trash_bin import TrashBin
-from opentrons.protocol_api._waste_chute import WasteChute
+from opentrons.protocol_api.disposal_locations import TrashBin, WasteChute
 
 if TYPE_CHECKING:
  from opentrons.protocol_api import InstrumentContext

api/src/opentrons/commands/helpers.py

@@ -2,8 +2,7 @@
 
 from opentrons.protocol_api.labware import Well, Labware
 from opentrons.protocol_api.module_contexts import ModuleContext
-from opentrons.protocol_api._trash_bin import TrashBin
-from opentrons.protocol_api._waste_chute import WasteChute
+from opentrons.protocol_api.disposal_locations import TrashBin, WasteChute
 from opentrons.protocol_api._types import OffDeckType
 from opentrons.types import Location, DeckLocation
 

api/src/opentrons/commands/types.py

@@ -7,8 +7,7 @@
 if TYPE_CHECKING:
  from opentrons.protocol_api import InstrumentContext
  from opentrons.protocol_api.labware import Well
- from opentrons.protocol_api._trash_bin import TrashBin
- from opentrons.protocol_api._waste_chute import WasteChute
+ from opentrons.protocol_api.disposal_locations import TrashBin, WasteChute
 
 from opentrons.types import Location
 

api/src/opentrons/motion_planning/deck_conflict.py

@@ -65,6 +65,7 @@ class TrashBin:
  """A non-labware trash bin (loaded via api level 2.16 and above)."""
 
  name_for_errors: str
+ highest_z: float
 
 
 @dataclass
@@ -138,9 +139,7 @@ def is_allowed(self, item: DeckItem) -> bool:
  elif isinstance(item, _Module):
  return item.highest_z_including_labware < self.max_height
  elif isinstance(item, TrashBin):
- # Since this is a restriction for OT-2 only and OT-2 trashes exceeded the height limit, always return False
- # TODO(jbl 2024-01-16) Include trash height and use that for check for more robustness
- return False
+ return item.highest_z < self.max_height
 
 
 class _NoModule(NamedTuple):

api/src/opentrons/protocol_api/__init__.py

@@ -22,10 +22,9 @@
  HeaterShakerContext,
  MagneticBlockContext,
 )
+from .disposal_locations import TrashBin, WasteChute
 from ._liquid import Liquid
 from ._types import OFF_DECK
-from ._trash_bin import TrashBin
-from ._waste_chute import WasteChute
 from ._nozzle_layout import (
  COLUMN,
  ALL,

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

@@ -590,7 +590,9 @@ def _map_disposal_location(
  if isinstance(disposal_location, TrashBin):
  return (
  disposal_location.location,
- wrapped_deck_conflict.TrashBin(name_for_errors="trash bin"),
+ wrapped_deck_conflict.TrashBin(
+ name_for_errors="trash bin", highest_z=disposal_location.height
+ ),
  )
  else:
  return None

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

@@ -38,8 +38,7 @@
 from ..instrument import AbstractInstrument
 from .well import WellCore
 
-from ..._trash_bin import TrashBin
-from ..._waste_chute import WasteChute
+from ...disposal_locations import TrashBin, WasteChute
 
 if TYPE_CHECKING:
  from .protocol import ProtocolCore
@@ -478,13 +477,16 @@ def drop_tip(
  self._protocol_core.set_last_location(location=location, mount=self.get_mount())
 
  def drop_tip_in_disposal_location(
- self, disposal_location: Union[TrashBin, WasteChute], home_after: Optional[bool]
+ self,
+ disposal_location: Union[TrashBin, WasteChute],
+ home_after: Optional[bool],
+ alternate_tip_drop: bool = False,
  ) -> None:
  self._move_to_disposal_location(
  disposal_location,
  force_direct=False,
  speed=None,
- alternate_tip_drop=True,
+ alternate_tip_drop=alternate_tip_drop,
  )
  self._drop_tip_in_place(home_after=home_after)
  self._protocol_core.set_last_location(location=None, mount=self.get_mount())
@@ -498,10 +500,14 @@ def _move_to_disposal_location(
  ) -> None:
  # TODO (nd, 2023-11-30): give appropriate offset when finalized
  # https://opentrons.atlassian.net/browse/RSS-391
- offset = AddressableOffsetVector(x=0, y=0, z=0)
+
+ disposal_offset = disposal_location.offset
+ offset = AddressableOffsetVector(
+ x=disposal_offset.x, y=disposal_offset.y, z=disposal_offset.z
+ )
 
  if isinstance(disposal_location, TrashBin):
- addressable_area_name = disposal_location._addressable_area_name
+ addressable_area_name = disposal_location.area_name
  self._engine_client.move_to_addressable_area_for_drop_tip(
  pipette_id=self._pipette_id,
  addressable_area_name=addressable_area_name,

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

@@ -51,8 +51,7 @@
 from ... import validation
 from ..._types import OffDeckType
 from ..._liquid import Liquid
-from ..._trash_bin import TrashBin
-from ..._waste_chute import WasteChute
+from ...disposal_locations import TrashBin, WasteChute
 from ..protocol import AbstractProtocol
 from ..labware import LabwareLoadParams
 from .labware import LabwareCore
@@ -138,14 +137,14 @@ def append_disposal_location(
  """Append a disposal location object to the core."""
  self._disposal_locations.append(disposal_location)
 
- def add_disposal_location_to_engine(
+ def _add_disposal_location_to_engine(
  self, disposal_location: Union[TrashBin, WasteChute]
  ) -> None:
  """Verify and add disposal location to engine store and append it to the core."""
+ self._engine_client.state.addressable_areas.raise_if_area_not_in_deck_configuration(
+ disposal_location.area_name
+ )
  if isinstance(disposal_location, TrashBin):
- self._engine_client.state.addressable_areas.raise_if_area_not_in_deck_configuration(
- disposal_location.area_name
- )
  deck_conflict.check(
  engine_state=self._engine_client.state,
  new_trash_bin=disposal_location,
@@ -157,20 +156,7 @@ def add_disposal_location_to_engine(
  existing_labware_ids=list(self._labware_cores_by_id.keys()),
  existing_module_ids=list(self._module_cores_by_id.keys()),
  )
- self._engine_client.add_addressable_area(disposal_location.area_name)
- elif isinstance(disposal_location, WasteChute):
- # TODO(jbl 2024-01-25) hardcoding this specific addressable area should be refactored
- # when analysis is fixed up
- #
- # We want to tell Protocol Engine that there's a waste chute in the waste chute location when it's loaded,
- # so analysis can prevent the user from doing anything that would collide with it. At the same time, we
- # do not want to create a false negative when it comes to addressable area conflict. We therefore use the
- # addressable area `1ChannelWasteChute` because every waste chute cutout fixture provides it and it will
- # provide the engine with the information it needs.
- self._engine_client.state.addressable_areas.raise_if_area_not_in_deck_configuration(
- "1ChannelWasteChute"
- )
- self._engine_client.add_addressable_area("1ChannelWasteChute")
+ self._engine_client.add_addressable_area(disposal_location.area_name)
  self.append_disposal_location(disposal_location)
 
  def get_disposal_locations(self) -> List[Union[Labware, TrashBin, WasteChute]]:
@@ -524,6 +510,51 @@ def load_instrument(
  default_movement_speed=400,
  )
 
+ def load_trash_bin(self, slot_name: DeckSlotName, area_name: str) -> TrashBin:
+ """Load a deck configuration based trash bin.
+
+ Args:
+ slot_name: the slot the trash is being loaded into.
+ area_name: the addressable area name of the trash.
+
+ Returns:
+ A trash bin object.
+ """
+ trash_bin = TrashBin(
+ location=slot_name,
+ addressable_area_name=area_name,
+ api_version=self._api_version,
+ engine_client=self._engine_client,
+ )
+ self._add_disposal_location_to_engine(trash_bin)
+ return trash_bin
+
+ def load_ot2_fixed_trash_bin(self) -> None:
+ """Load a deck configured OT-2 fixed trash in Slot 12."""
+ _fixed_trash_trash_bin = TrashBin(
+ location=DeckSlotName.FIXED_TRASH,
+ addressable_area_name="fixedTrash",
+ api_version=self._api_version,
+ engine_client=self._engine_client,
+ )
+ # We are just appending the fixed trash to the core's internal list here, not adding it to the engine via
+ # the core, since that method works through the SyncClient and if called from here, will cause protocols
+ # to deadlock. Instead, that method is called in protocol engine directly in create_protocol_context after
+ # ProtocolContext is initialized.
+ self.append_disposal_location(_fixed_trash_trash_bin)
+
+ def load_waste_chute(self) -> WasteChute:
+ """Load a deck configured waste chute into Slot D3.
+
+ Returns:
+ A waste chute object.
+ """
+ waste_chute = WasteChute(
+ engine_client=self._engine_client, api_version=self._api_version
+ )
+ self._add_disposal_location_to_engine(waste_chute)
+ return waste_chute
+
  def pause(self, msg: Optional[str]) -> None:
  """Pause the protocol."""
  self._engine_client.wait_for_resume(message=msg)

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

@@ -10,8 +10,7 @@
 from opentrons.protocols.api_support.util import FlowRates
 from opentrons.protocol_api._nozzle_layout import NozzleLayout
 
-from .._trash_bin import TrashBin
-from .._waste_chute import WasteChute
+from ..disposal_locations import TrashBin, WasteChute
 from .well import WellCoreType
 
 
@@ -137,13 +136,17 @@ def drop_tip(
 
  @abstractmethod
  def drop_tip_in_disposal_location(
- self, disposal_location: Union[TrashBin, WasteChute], home_after: Optional[bool]
+ self,
+ disposal_location: Union[TrashBin, WasteChute],
+ home_after: Optional[bool],
+ alternate_tip_drop: bool = False,
  ) -> None:
  """Move to and drop tip into a TrashBin or WasteChute.
 
  Args:
  disposal_location: The disposal location object we're dropping to.
  home_after: Whether to home the pipette after the tip is dropped.
+ alternate_tip_drop: Whether to alternate tip drop location in a trash bin.
  """
  ...
 

api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py

@@ -19,8 +19,7 @@
 from opentrons.protocols.geometry import planning
 from opentrons.protocol_api._nozzle_layout import NozzleLayout
 
-from ..._trash_bin import TrashBin
-from ..._waste_chute import WasteChute
+from ...disposal_locations import TrashBin, WasteChute
 from ..instrument import AbstractInstrument
 from .legacy_well_core import LegacyWellCore
 from .legacy_module_core import LegacyThermocyclerCore, LegacyHeaterShakerCore
@@ -295,7 +294,10 @@ def drop_tip(
  )
 
  def drop_tip_in_disposal_location(
- self, disposal_location: Union[TrashBin, WasteChute], home_after: Optional[bool]
+ self,
+ disposal_location: Union[TrashBin, WasteChute],
+ home_after: Optional[bool],
+ alternate_tip_drop: bool = False,
  ) -> None:
  raise APIVersionError(
  "Dropping tips in a trash bin or waste chute is not supported in this API Version."

api/src/opentrons/protocol_api/core/legacy/legacy_protocol_core.py

@@ -16,10 +16,9 @@
 from opentrons.protocols import labware as labware_definition
 
 from ...labware import Labware
+from ...disposal_locations import TrashBin, WasteChute
 from ..._liquid import Liquid
 from ..._types import OffDeckType
-from ..._trash_bin import TrashBin
-from ..._waste_chute import WasteChute
 from ..protocol import AbstractProtocol
 from ..labware import LabwareLoadParams
 
@@ -143,11 +142,6 @@ def append_disposal_location(
  )
  self._disposal_locations.append(disposal_location)
 
- def add_disposal_location_to_engine(
- self, disposal_location: Union[TrashBin, WasteChute]
- ) -> None:
- assert False, "add_disposal_location_to_engine only supported on engine core"
-
  def add_labware_definition(
  self,
  definition: LabwareDefinition,
@@ -384,6 +378,21 @@ def load_instrument(
 
  return new_instr
 
+ def load_trash_bin(self, slot_name: DeckSlotName, area_name: str) -> TrashBin:
+ raise APIVersionError(
+ "Loading deck configured trash bin is not supported in this API version."
+ )
+
+ def load_ot2_fixed_trash_bin(self) -> None:
+ raise APIVersionError(
+ "Loading deck configured OT-2 fixed trash bin is not supported in this API version."
+ )
+
+ def load_waste_chute(self) -> WasteChute:
+ raise APIVersionError(
+ "Loading waste chute is not supported in this API version."
+ )
+
  def get_loaded_instruments(
  self,
  ) -> Dict[Mount, Optional[LegacyInstrumentCore]]:

api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py

@@ -21,8 +21,7 @@
  UnexpectedTipAttachError,
 )
 
-from ..._trash_bin import TrashBin
-from ..._waste_chute import WasteChute
+from ...disposal_locations import TrashBin, WasteChute
 from opentrons.protocol_api._nozzle_layout import NozzleLayout
 
 from ..instrument import AbstractInstrument
@@ -263,7 +262,10 @@ def drop_tip(
  )
 
  def drop_tip_in_disposal_location(
- self, disposal_location: Union[TrashBin, WasteChute], home_after: Optional[bool]
+ self,
+ disposal_location: Union[TrashBin, WasteChute],
+ home_after: Optional[bool],
+ alternate_tip_drop: bool = False,
  ) -> None:
  raise APIVersionError(
  "Dropping tips in a trash bin or waste chute is not supported in this API Version."