move trash loading more to core

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

@@ -524,6 +524,30 @@ 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)
+ self.add_disposal_location_to_engine(trash_bin)
+ return 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()
+ 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/legacy/legacy_protocol_core.py

@@ -384,6 +384,16 @@ 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_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/protocol.py

@@ -136,6 +136,14 @@ def load_instrument(
  ) -> InstrumentCoreType:
  ...
 
+ @abstractmethod
+ def load_trash_bin(self, slot_name: DeckSlotName, area_name: str) -> TrashBin:
+ ...
+
+ @abstractmethod
+ def load_waste_chute(self) -> WasteChute:
+ ...
+
  @abstractmethod
  def pause(self, msg: Optional[str]) -> None:
  ...

api/src/opentrons/protocol_api/protocol_context.py

@@ -165,14 +165,14 @@ def __init__(
  elif should_load_fixed_trash_area_for_python_protocol(
  self._api_version, self._core.robot_type
  ):
- _fixed_trash_trashbin = TrashBin(
+ _fixed_trash_trash_bin = TrashBin(
  location=DeckSlotName.FIXED_TRASH, addressable_area_name="fixedTrash"
  )
  # 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._core.append_disposal_location(_fixed_trash_trashbin)
+ self._core.append_disposal_location(_fixed_trash_trash_bin)
 
  self._commands: List[str] = []
  self._unsubscribe_commands: Optional[Callable[[], None]] = None
@@ -512,10 +512,7 @@ def load_trash_bin(self, location: DeckLocation) -> TrashBin:
  api_version=self._api_version,
  robot_type=self._core.robot_type,
  )
- trash_bin = TrashBin(
- location=slot_name, addressable_area_name=addressable_area_name
- )
- self._core.add_disposal_location_to_engine(trash_bin)
+ trash_bin = self._core.load_trash_bin(slot_name, addressable_area_name)
  return trash_bin
 
  @requires_version(2, 16)
@@ -531,9 +528,7 @@ def load_waste_chute(
  load another item in slot D3 after loading the waste chute, or vice versa, the
  API will raise an error.
  """
- waste_chute = WasteChute()
- self._core.add_disposal_location_to_engine(waste_chute)
- return waste_chute
+ return self._core.load_waste_chute()
 
  @requires_version(2, 15)
  def load_adapter(