refactor(api): Choose the deck type slightly less implicitly (#11812)

api/src/opentrons/execute.py

@@ -20,6 +20,9 @@
 from opentrons.commands import types as command_types
 from opentrons.protocols import parse
 from opentrons.protocols.types import ApiDeprecationError
+from opentrons.protocols.api_support.deck_type import (
+ guess_from_global_config as guess_deck_type_from_global_config,
+)
 from opentrons.protocols.api_support.types import APIVersion
 from opentrons.hardware_control import API as OT2API, ThreadManager, HardwareControlAPI
 from opentrons.hardware_control.types import MachineType
@@ -123,6 +126,10 @@ def get_protocol_api(
  try:
  context = protocol_api.create_protocol_context(
  api_version=checked_version,
+ # FIXME(2022-12-02): Instead of guessing,
+ # match this to the robot type declared by the protocol.
+ # https://opentrons.atlassian.net/browse/RSS-156
+ deck_type=guess_deck_type_from_global_config(),
  hardware_api=_THREAD_MANAGED_HW, # type: ignore[arg-type]
  bundled_labware=bundled_labware,
  bundled_data=bundled_data,

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

@@ -15,7 +15,6 @@
 
 from opentrons.hardware_control.modules.types import ModuleType
 from opentrons.motion_planning import deck_conflict
-from opentrons.protocols.api_support.constants import deck_type
 from opentrons.protocols.api_support.labware_like import LabwareLike
 from opentrons.types import DeckLocation, Location, Mount, Point
 
@@ -48,12 +47,11 @@ def load_name(self) -> str:
 class Deck(UserDict): # type: ignore[type-arg]
  data: Dict[int, Optional[DeckItem]]
 
- def __init__(self) -> None:
+ def __init__(self, deck_type: str) -> None:
  super().__init__()
- # TODO(mc, 2022-06-17): move deck type selection
- # (and maybe deck definition loading) out of this constructor
- # to decouple from config (and environment) reading / loading
- self._definition = load_deck(deck_type(), DEFAULT_DECK_DEFINITION_VERSION)
+ self._definition = load_deck(
+ name=deck_type, version=DEFAULT_DECK_DEFINITION_VERSION
+ )
  self._positions = {}
  for slot in self._definition["locations"]["orderedSlots"]:
  self.data[int(slot["id"])] = None

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

@@ -42,8 +42,8 @@ def __init__(
  sync_hardware: SyncHardwareAPI,
  api_version: APIVersion,
  labware_offset_provider: AbstractLabwareOffsetProvider,
+ deck_layout: Deck,
  equipment_broker: Optional[EquipmentBroker[LoadInfo]] = None,
- deck_layout: Optional[Deck] = None,
  bundled_labware: Optional[Dict[str, LabwareDefinition]] = None,
  extra_labware: Optional[Dict[str, LabwareDefinition]] = None,
  ) -> None:
@@ -71,8 +71,8 @@ def __init__(
  self._sync_hardware = sync_hardware
  self._api_version = api_version
  self._labware_offset_provider = labware_offset_provider
+ self._deck_layout = deck_layout
  self._equipment_broker = equipment_broker or EquipmentBroker()
- self._deck_layout = Deck() if deck_layout is None else deck_layout
 
  self._instruments: Dict[Mount, Optional[LegacyInstrumentCore]] = {
  mount: None for mount in Mount

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

@@ -448,8 +448,9 @@ def create_geometry(
  else:
  par = str(parent.labware.object)
 
- # this needs to change to look up the current deck type if/when we add
- # that notion
+ # This ModuleGeometry class is only used in Python Protocol API versions <=2.13,
+ # which only support the OT-2, so we can ignore OT-3s and hard-code "ot2_standard."
+ # We also assume that "ot2_short_trash" has the same transforms as "ot2_standard."
  xforms_ser = (
  definition["slotTransforms"]
  .get("ot2_standard", {})

api/src/opentrons/protocol_api/create_protocol_context.py

@@ -21,6 +21,7 @@
 from .deck import Deck
 
 from .core.common import ProtocolCore as AbstractProtocolCore
+from .core.legacy.deck import Deck as LegacyDeck
 from .core.legacy.legacy_protocol_core import LegacyProtocolCore
 from .core.legacy.labware_offset_provider import (
  AbstractLabwareOffsetProvider,
@@ -42,6 +43,7 @@ def create_protocol_context(
  api_version: APIVersion,
  *,
  hardware_api: Union[HardwareControlAPI, ThreadManager[HardwareControlAPI]],
+ deck_type: str,
  protocol_engine: Optional[ProtocolEngine] = None,
  protocol_engine_loop: Optional[asyncio.AbstractEventLoop] = None,
  broker: Optional[Broker] = None,
@@ -56,6 +58,7 @@ def create_protocol_context(
  Args:
  api_version: The API version to target.
  hardware_api: Control interface to the device's hardware.
+ deck_type: What kind of deck the device has.
  protocol_engine: A ProtocolEngine to use for labware offsets
  and core protocol logic. If omitted, labware offsets will
  all be (0, 0, 0) and ProtocolEngine-based core will not work.
@@ -116,19 +119,23 @@ def create_protocol_context(
 
  # TODO(mc, 2022-8-22): remove `disable_fast_protocol_upload`
  elif use_simulating_core and not feature_flags.disable_fast_protocol_upload():
+ legacy_deck = LegacyDeck(deck_type=deck_type)
  core = LegacyProtocolCoreSimulator(
  sync_hardware=sync_hardware,
  labware_offset_provider=labware_offset_provider,
+ deck_layout=legacy_deck,
  equipment_broker=equipment_broker,
  api_version=api_version,
  bundled_labware=bundled_labware,
  extra_labware=extra_labware,
  )
 
  else:
+ legacy_deck = LegacyDeck(deck_type=deck_type)
  core = LegacyProtocolCore(
  sync_hardware=sync_hardware,
  labware_offset_provider=labware_offset_provider,
+ deck_layout=legacy_deck,
  equipment_broker=equipment_broker,
  api_version=api_version,
  bundled_labware=bundled_labware,
@@ -141,6 +148,9 @@ def create_protocol_context(
 
  return ProtocolContext(
  api_version=api_version,
+ # TODO(mm, 2023-05-11): This cast shouldn't be necessary.
+ # Fix this by making the appropriate TypeVars covariant?
+ # https://peps.python.org/pep-0484/#covariance-and-contravariance
  core=cast(AbstractProtocolCore, core),
  broker=broker,
  deck=deck,

api/src/opentrons/protocol_engine/resources/deck_data_provider.py

@@ -7,7 +7,9 @@
 from opentrons_shared_data.deck.dev_types import DeckDefinitionV3
 from opentrons.protocols.models import LabwareDefinition
 from opentrons.types import DeckSlotName
-from opentrons.protocols.api_support.constants import deck_type
+from opentrons.protocols.api_support.deck_type import (
+ guess_from_global_config as guess_deck_type_from_global_config,
+)
 
 from ..types import DeckSlotLocation
 from .labware_data_provider import LabwareDataProvider
@@ -37,7 +39,7 @@ def __init__(self, labware_data: Optional[LabwareDataProvider] = None) -> None:
  @staticmethod
  async def get_deck_definition() -> DeckDefinitionV3:
  """Get a labware definition given the labware's identification."""
- return load_deck(deck_type(), 3)
+ return load_deck(guess_deck_type_from_global_config(), 3)
 
  async def get_deck_fixed_labware(
  self,

api/src/opentrons/protocol_engine/types.py

@@ -151,6 +151,9 @@ class DeckPoint(BaseModel):
  z: float
 
 
+# TODO(mm, 2023-05-10): Deduplicate with constants in
+# opentrons.protocols.api_support.deck_type
+# and consider moving to shared-data.
 class DeckType(str, Enum):
  """Types of deck available."""
 

api/src/opentrons/protocol_runner/legacy_wrappers.py

@@ -20,6 +20,9 @@
  ThermocyclerModuleModel as LegacyThermocyclerModuleModel,
  HeaterShakerModuleModel as LegacyHeaterShakerModuleModel,
 )
+from opentrons.protocols.api_support.deck_type import (
+ guess_from_global_config as guess_deck_type_from_global_config,
+)
 from opentrons.protocol_engine import ProtocolEngine
 from opentrons.protocol_reader import ProtocolSource, ProtocolFileRole
 
@@ -141,6 +144,11 @@ def create(
  return create_protocol_context(
  api_version=protocol.api_level,
  hardware_api=self._hardware_api,
+ # FIXME(mm, 2022-12-02): deck_type, like hardware_api, needs to match the
+ # robot type that the protocol declares itself to run on.
+ # Guessing like this is wrong in in-app analysis.
+ # https://opentrons.atlassian.net/browse/RSS-156
+ deck_type=guess_deck_type_from_global_config(),
  protocol_engine=self._protocol_engine,
  protocol_engine_loop=asyncio.get_running_loop(),
  broker=broker,

api/src/opentrons/protocols/api_support/constants.py

@@ -1,22 +1,8 @@
 from pathlib import Path
 
 from opentrons.config import get_opentrons_path
-from opentrons.config import feature_flags as ff
 
 OPENTRONS_NAMESPACE = "opentrons"
 CUSTOM_NAMESPACE = "custom_beta"
 STANDARD_DEFS_PATH = Path("labware/definitions/2")
 USER_DEFS_PATH = get_opentrons_path("labware_user_definitions_dir_v2")
-
-SHORT_TRASH_DECK = "ot2_short_trash"
-STANDARD_OT2_DECK = "ot2_standard"
-STANDARD_OT3_DECK = "ot3_standard"
-
-
-def deck_type() -> str:
- if ff.enable_ot3_hardware_controller():
- return STANDARD_OT3_DECK
- elif ff.short_fixed_trash():
- return SHORT_TRASH_DECK
- else:
- return STANDARD_OT2_DECK

api/src/opentrons/protocols/api_support/deck_type.py

@@ -0,0 +1,29 @@
+from opentrons.config import feature_flags
+
+
+# TODO(mm, 2023-05-10): Deduplicate these constants with
+# opentrons.protocol_engine.types.DeckType and consider moving to shared-data.
+SHORT_TRASH_DECK = "ot2_short_trash"
+STANDARD_OT2_DECK = "ot2_standard"
+STANDARD_OT3_DECK = "ot3_standard"
+
+
+def guess_from_global_config() -> str:
+ """Return a default deck type based on global environment configuration.
+
+ Deprecated:
+ The notion of "a default deck type" doesn't make sense now that we have:
+
+ * Decks that are meaningfully different from each other (OT-2 vs. OT-3).
+ * Protocol analysis running off-robot, in environments that cannot be
+ permanently configured for any single specific deck type.
+
+ If you need to know the deck type, derive it from explicit sources,
+ like the protocol's declared robot type, instead.
+ """
+ if feature_flags.enable_ot3_hardware_controller():
+ return STANDARD_OT3_DECK
+ elif feature_flags.short_fixed_trash():
+ return SHORT_TRASH_DECK
+ else:
+ return STANDARD_OT2_DECK

api/src/opentrons/protocols/duration/estimator.py

@@ -8,6 +8,9 @@
 from dataclasses import dataclass
 
 from opentrons.commands import types
+from opentrons.protocols.api_support.deck_type import (
+ guess_from_global_config as guess_deck_type_from_global_config,
+)
 from opentrons.protocols.api_support.labware_like import LabwareLike
 from opentrons.protocols.duration.errors import DurationEstimatorException
 from opentrons.protocol_api.core.legacy.deck import Deck
@@ -60,7 +63,9 @@ def __init__(self):
  self._last_thermocycler_module_temperature = START_MODULE_TEMPERATURE
  # Per step time estimate.
  self._increments: List[TimerEntry] = []
- self._deck = Deck()
+
+ # TODO(mm, 2022-12-01): Allow the caller to configure the deck type.
+ self._deck = Deck(deck_type=guess_deck_type_from_global_config())
 
  def get_total_duration(self) -> float:
  """Return the total duration"""

api/src/opentrons/simulate.py

@@ -42,6 +42,9 @@
 
 from opentrons.protocols import parse, bundle
 from opentrons.protocols.types import PythonProtocol, BundleContents
+from opentrons.protocols.api_support.deck_type import (
+ guess_from_global_config as guess_deck_type_from_global_config,
+)
 from opentrons.protocols.api_support.types import APIVersion
 from opentrons_shared_data.labware.dev_types import LabwareDefinition
 
@@ -260,6 +263,10 @@ def _build_protocol_context(
  context = protocol_api.create_protocol_context(
  api_version=version,
  hardware_api=hardware_simulator,
+ # FIXME(2022-12-02): Instead of guessing,
+ # match this to the robot type declared by the protocol.
+ # https://opentrons.atlassian.net/browse/RSS-156
+ deck_type=guess_deck_type_from_global_config(),
  bundled_labware=bundled_labware,
  bundled_data=bundled_data,
  extra_labware=extra_labware,
@@ -646,6 +653,7 @@ def main() -> int:
  args = parser.parse_args()
  # Try to migrate api v1 containers if needed
 
+ # TODO(mm, 2022-12-01): Configure the DurationEstimator with the correct deck type.
  duration_estimator = DurationEstimator() if args.estimate_duration else None # type: ignore[no-untyped-call]
 
  runlog, maybe_bundle = simulate(

api/tests/opentrons/conftest.py

@@ -50,6 +50,7 @@
 )
 from opentrons.protocol_api import ProtocolContext, Labware, create_protocol_context
 from opentrons.protocol_api.core.legacy.legacy_labware_core import LegacyLabwareCore
+from opentrons.protocols.api_support import deck_type
 from opentrons.protocols.api_support.types import APIVersion
 from opentrons.types import Location, Point
 
@@ -230,11 +231,19 @@ async def robot_model(
 
 
 @pytest.fixture
-def deck_definition(robot_model: RobotModel) -> DeckDefinitionV3:
+def deck_definition_name(robot_model: RobotModel) -> str:
  if robot_model == "OT-3 Standard":
- return load_deck("ot3_standard", DEFAULT_DECK_DEFINITION_VERSION)
- else:
- return load_deck("ot2_standard", DEFAULT_DECK_DEFINITION_VERSION)
+ return deck_type.STANDARD_OT3_DECK
+ elif robot_model == "OT-2 Standard":
+ # There are two OT-2 deck definitions (standard and short-trash),
+ # but RobotModel does not draw such a distinction. We assume here that it's
+ # sufficient to run OT-2 tests with the standard deck definition only.
+ return deck_type.STANDARD_OT2_DECK
+
+
+@pytest.fixture
+def deck_definition(deck_definition_name: str) -> DeckDefinitionV3:
+ return load_deck(deck_definition_name, DEFAULT_DECK_DEFINITION_VERSION)
 
 
 @pytest.fixture()
@@ -257,22 +266,27 @@ async def hardware(
  yield hw
 
 
-def _make_ot2_non_pe_ctx(hardware: ThreadManagedHardware) -> ProtocolContext:
+def _make_ot2_non_pe_ctx(
+ hardware: ThreadManagedHardware, deck_type: str
+) -> ProtocolContext:
  """Return a ProtocolContext configured for an OT-2 and not backed by Protocol Engine."""
- return create_protocol_context(api_version=APIVersion(2, 13), hardware_api=hardware)
+ return create_protocol_context(
+ api_version=APIVersion(2, 13), hardware_api=hardware, deck_type=deck_type
+ )
 
 
 @contextlib.contextmanager
 def _make_ot3_pe_ctx(
  hardware: ThreadManagedHardware,
+ deck_type: str,
 ) -> Generator[ProtocolContext, None, None]:
  """Return a ProtocolContext configured for an OT-3 and backed by Protocol Engine."""
  with protocol_engine_in_thread(hardware=hardware) as (engine, loop):
  yield create_protocol_context(
  api_version=APIVersion(2, 14),
  hardware_api=hardware,
+ deck_type=deck_type,
  protocol_engine=engine,
- # TODO will this deadlock?
  protocol_engine_loop=loop,
  )
 
@@ -282,14 +296,17 @@ def ctx(
  request: pytest.FixtureRequest,
  robot_model: RobotModel,
  hardware: ThreadManagedHardware,
+ deck_definition_name: str,
 ) -> Generator[ProtocolContext, None, None]:
  if robot_model == "OT-2 Standard":
- yield _make_ot2_non_pe_ctx(hardware=hardware)
+ yield _make_ot2_non_pe_ctx(hardware=hardware, deck_type=deck_definition_name)
  elif robot_model == "OT-3 Standard":
  if request.node.get_closest_marker("apiv2_non_pe_only"):
  pytest.skip("Test requests only non-Protocol-Engine ProtocolContexts")
  else:
- with _make_ot3_pe_ctx(hardware=hardware) as ctx:
+ with _make_ot3_pe_ctx(
+ hardware=hardware, deck_type=deck_definition_name
+ ) as ctx:
  yield ctx