feat(api): allow custom user offsets for deck configured trash bins and waste chute

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

@@ -32,8 +32,7 @@
 from opentrons.protocol_engine.errors.exceptions import LabwareNotLoadedOnModuleError
 from opentrons.protocol_engine.types import StagingSlotLocation, Dimensions
 from opentrons.types import DeckSlotName, StagingSlotName, Point
-from ..._trash_bin import TrashBin
-from ..._waste_chute import WasteChute
+from ...disposal_locations import TrashBin, WasteChute
 
 if TYPE_CHECKING:
     from ...labware import Labware
@@ -521,7 +520,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]]: