fix(api): Match the deck definition to the robot type in protocol ana…

…lysis (#12691)

api/src/opentrons/protocol_api/create_protocol_context.py

@@ -59,6 +59,7 @@ def create_protocol_context(
         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.
+            This must match the deck type in `protocol_engine`'s config, if there is one.
         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.
@@ -96,6 +97,7 @@ def create_protocol_context(
         sync_hardware = SynchronousAdapter(hardware_api)
 
     if protocol_engine is not None:
+        assert deck_type == protocol_engine.state_view.config.deck_type.value
         labware_offset_provider = LabwareOffsetProvider(engine=protocol_engine)
     else:
         labware_offset_provider = NullLabwareOffsetProvider()

api/src/opentrons/protocol_engine/__init__.py

@@ -26,6 +26,7 @@
     LabwareOffsetLocation,
     LabwareMovementStrategy,
     DeckPoint,
+    DeckType,
     DeckSlotLocation,
     ModuleLocation,
     OFF_DECK_LOCATION,
@@ -76,6 +77,7 @@
     "LabwareMovementStrategy",
     "DeckSlotLocation",
     "DeckPoint",
+    "DeckType",
     "ModuleLocation",
     "OFF_DECK_LOCATION",
     "Dimensions",
@@ -93,7 +95,6 @@
     "ModuleModel",
     "ModuleDefinition",
     "Liquid",
-    "StaticPipetteConfig",
     # plugins
     "AbstractPlugin",
 ]

api/src/opentrons/protocol_engine/create_protocol_engine.py

@@ -18,7 +18,7 @@ async def create_protocol_engine(
         hardware_api: Hardware control API to pass down to dependencies.
         config: ProtocolEngine configuration.
     """
-    deck_data = DeckDataProvider()
+    deck_data = DeckDataProvider(config.deck_type)
     deck_definition = await deck_data.get_deck_definition()
     deck_fixed_labware = await deck_data.get_deck_fixed_labware(deck_definition)
     module_calibration_offsets = ModuleDataProvider.load_module_calibrations()

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

@@ -3,15 +3,17 @@
 from typing import List, Optional, cast
 from typing_extensions import final
 
-from opentrons_shared_data.deck import load as load_deck
+import anyio
+
+from opentrons_shared_data.deck import (
+    load as load_deck,
+    DEFAULT_DECK_DEFINITION_VERSION,
+)
 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.deck_type import (
-    guess_from_global_config as guess_deck_type_from_global_config,
-)
 
-from ..types import DeckSlotLocation
+from ..types import DeckSlotLocation, DeckType
 from .labware_data_provider import LabwareDataProvider
 
 
@@ -30,23 +32,29 @@ class DeckDataProvider:
 
     _labware_data: LabwareDataProvider
 
-    def __init__(self, labware_data: Optional[LabwareDataProvider] = None) -> None:
+    def __init__(
+        self, deck_type: DeckType, labware_data: Optional[LabwareDataProvider] = None
+    ) -> None:
         """Initialize a DeckDataProvider."""
+        self._deck_type = deck_type
         self._labware_data = labware_data or LabwareDataProvider()
 
-    # NOTE(mc, 2020-11-18): async to allow file reading and parsing to be
-    # async on a worker thread in the future
-    @staticmethod
-    async def get_deck_definition() -> DeckDefinitionV3:
+    async def get_deck_definition(self) -> DeckDefinitionV3:
         """Get a labware definition given the labware's identification."""
-        return load_deck(guess_deck_type_from_global_config(), 3)
+
+        def sync() -> DeckDefinitionV3:
+            return load_deck(
+                name=self._deck_type.value, version=DEFAULT_DECK_DEFINITION_VERSION
+            )
+
+        return await anyio.to_thread.run_sync(sync)
 
     async def get_deck_fixed_labware(
         self,
         deck_definition: DeckDefinitionV3,
     ) -> List[DeckFixedLabware]:
         """Get a list of all labware fixtures from a given deck definition."""
-        labware = []
+        labware: List[DeckFixedLabware] = []
 
         for fixture in deck_definition["locations"]["fixtures"]:
             labware_id = fixture["id"]

api/src/opentrons/protocol_engine/state/config.py

@@ -3,6 +3,8 @@
 
 from opentrons_shared_data.robot.dev_types import RobotType
 
+from opentrons.protocol_engine.types import DeckType
+
 
 @dataclass(frozen=True)
 class Config:
@@ -11,6 +13,8 @@ class Config:
     Params:
         robot_type: What kind of robot the engine is controlling,
             or pretending to control.
+        deck_type: The type of deck on the robot that the engine is controlling,
+            or pretending to control.
         ignore_pause: The engine should no-op instead of waiting
             for pauses and delays to complete.
         use_virtual_modules: The engine should no-op instead of calling
@@ -22,6 +26,7 @@ class Config:
     """
 
     robot_type: RobotType
+    deck_type: DeckType
     ignore_pause: bool = False
     use_virtual_pipettes: bool = False
     use_virtual_modules: bool = False

api/src/opentrons/protocol_runner/create_simulating_runner.py

@@ -2,8 +2,10 @@
 
 from opentrons.config import feature_flags
 from opentrons.hardware_control import API as OT2API, HardwareControlAPI
+from opentrons.protocols.api_support import deck_type
 from opentrons.protocol_engine import (
     Config as ProtocolEngineConfig,
+    DeckType,
     create_protocol_engine,
 )
 from opentrons.protocol_reader.protocol_source import ProtocolConfig
@@ -50,6 +52,7 @@ async def create_simulating_runner(
         hardware_api=simulating_hardware_api,
         config=ProtocolEngineConfig(
             robot_type=robot_type,
+            deck_type=DeckType(deck_type.for_simulation(robot_type)),
             ignore_pause=True,
             use_virtual_modules=True,
             use_virtual_gripper=True,

api/src/opentrons/protocol_runner/legacy_wrappers.py

@@ -20,9 +20,6 @@
     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
 
@@ -144,11 +141,7 @@ 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(),
+            deck_type=self._protocol_engine.state_view.config.deck_type.value,
             protocol_engine=self._protocol_engine,
             protocol_engine_loop=asyncio.get_running_loop(),
             broker=broker,

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

@@ -1,3 +1,5 @@
+from opentrons_shared_data.robot.dev_types import RobotType
+
 from opentrons.config import feature_flags
 
 
@@ -9,21 +11,32 @@
 
 
 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:
+    """Return the deck type that the host device physically has.
 
-          * 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.
+    This only makes sense when the software is running on a real robot.
 
-        If you need to know the deck type, derive it from explicit sources,
-        like the protocol's declared robot type, instead.
+    When simulating or analyzing a protocol, especially off-robot, don't use this, because it may
+    not match the protocol's declared robot type. Use `for_analysis` 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
+
+
+def for_simulation(robot_type: RobotType) -> str:
+    """Return the deck type that should be used for simulating and analyzing a protocol.
+
+    Params:
+        robot_type: The robot type that the protocol is meant to run on.
+    """
+    if robot_type == "OT-2 Standard":
+        # OT-2 protocols don't have a way of defining whether they're meant to run on a short-trash
+        # or standard deck. So when we're simulating an OT-2 protocol, we need to make an
+        # arbitrary choice for which deck type to use.
+        return STANDARD_OT2_DECK
+    elif robot_type == "OT-3 Standard":
+        # OT-3s currently only have a single deck type.
+        return STANDARD_OT3_DECK

api/tests/opentrons/cli/test_cli.py

@@ -1,7 +1,9 @@
 """Test cli execution."""
 import json
+import tempfile
+import textwrap
 
-from typing import Iterator
+from typing import Any, Iterator, List, Tuple
 from pathlib import Path
 
 import pytest
@@ -16,25 +18,112 @@ def _list_fixtures(version: int) -> Iterator[Path]:
     )
 
 
+def _get_analysis_result(protocol_files: List[Path]) -> Tuple[int, Any]:
+    """Run `protocol_files` as a single protocol through the analysis CLI.
+
+    Returns:
+        A tuple (exit_code, analysis_json_dict).
+    """
+    with tempfile.TemporaryDirectory() as temp_dir:
+        analysis_output_file = Path(temp_dir) / "analysis_output.json"
+        runner = CliRunner()
+        result = runner.invoke(
+            analyze,
+            [
+                "--json-output",
+                str(analysis_output_file),
+                *[str(p.resolve()) for p in protocol_files],
+            ],
+        )
+        if result.exception is not None:
+            raise result.exception
+        else:
+            return result.exit_code, json.loads(analysis_output_file.read_bytes())
+
+
 @pytest.mark.parametrize("fixture_path", _list_fixtures(6))
 def test_analyze(
     fixture_path: Path,
-    tmp_path: Path,
 ) -> None:
     """Should return with no errors and a non-empty output."""
-    analysis_output_path = tmp_path / "analysis_output.json"
-    runner = CliRunner()
-    result = runner.invoke(
-        analyze,
-        [str(fixture_path.resolve()), "--json-output", str(analysis_output_path)],
-    )
+    exit_code, analysis_output_json = _get_analysis_result([fixture_path])
 
-    assert result.exit_code == 0
+    assert exit_code == 0
 
-    analysis_output_json = json.loads(analysis_output_path.read_bytes())
     assert "robotType" in analysis_output_json
     assert "pipettes" in analysis_output_json
     assert "commands" in analysis_output_json
     assert "labware" in analysis_output_json
     assert "liquids" in analysis_output_json
     assert "modules" in analysis_output_json
+
+
+_DECK_DEFINITION_TEST_SLOT = 2
+_DECK_DEFINITION_TEST_LABWARE = "agilent_1_reservoir_290ml"
+_DECK_DEFINITION_TEST_WELL = "A1"
+
+
+def _get_deck_definition_test_source(api_level: str, robot_type: str) -> str:
+    return textwrap.dedent(
+        f"""\
+        requirements = {{
+            "apiLevel": "{api_level}",
+            "robotType": "{robot_type}",
+        }}
+        def run(protocol):
+            labware = protocol.load_labware(
+                "{_DECK_DEFINITION_TEST_LABWARE}",
+                "{_DECK_DEFINITION_TEST_SLOT}",
+            )
+            test_point = labware["{_DECK_DEFINITION_TEST_WELL}"].top().point
+            protocol.comment(str(test_point))
+        """
+    )
+
+
+@pytest.mark.parametrize(
+    ("api_level", "robot_type", "expected_point"),
+    [
+        # These expected_point values were copied from known-good analysis outputs.
+        # The exact values don't matter much for this test, since we're not checking positional
+        # accuracy here. They just need to be clearly different between the OT-2 and OT-3.
+        ("2.13", "OT-2", "(196.38, 42.785, 44.04)"),
+        ("2.14", "OT-2", "(196.38, 42.785, 44.04)"),
+        pytest.param(
+            "2.14",
+            "OT-3",
+            "(227.88, 42.785, 44.04)",
+            marks=pytest.mark.ot3_only,  # Analyzing an OT-3 protocol requires an OT-3 hardware API.
+        ),
+    ],
+)
+def test_analysis_deck_definition(
+    api_level: str,
+    robot_type: str,
+    expected_point: str,
+    tmp_path: Path,
+) -> None:
+    """Test that the analysis uses the appropriate deck definition for the protocol's robot type.
+
+    At the time of writing, the only official, public, documented way to observe the deck definition
+    that a protocol uses is for the protocol to load a labware and inspect the deck coordinates of
+    its wells.
+    """
+    protocol_source_file = Path(tmp_path) / "protocol.py"
+    protocol_source_file.write_text(
+        _get_deck_definition_test_source(
+            api_level=api_level,
+            robot_type=robot_type,
+        )
+    )
+
+    exit_code, analysis_output_json = _get_analysis_result([protocol_source_file])
+
+    assert exit_code == 0
+
+    [load_labware_command, comment_command] = analysis_output_json["commands"]
+    _ = load_labware_command
+
+    # todo(mm, 2023-05-12): When protocols emit true Protocol Engine comment commands instead
+    # of legacy commands, "legacyCommandText" should change to "message".
+    assert comment_command["params"]["legacyCommandText"] == expected_point