fix(api): associate thermocycler with all slots it occupies (#14491)

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

@@ -34,7 +34,6 @@
     LabwareOffsetVector,
     ModuleOffsetVector,
     ModuleOffsetData,
-    DeckType,
     CurrentWell,
     CurrentPipetteLocation,
     TipGeometry,
@@ -153,17 +152,20 @@ def get_all_obstacle_highest_z(self) -> float:
     def get_highest_z_in_slot(
         self, slot: Union[DeckSlotLocation, StagingSlotLocation]
     ) -> float:
-        """Get the highest Z-point of all items stacked in the given deck slot."""
+        """Get the highest Z-point of all items stacked in the given deck slot.
+
+        This height includes the height of any module that occupies the given slot
+        even if it wasn't loaded in that slot (e.g., thermocycler).
+        """
         slot_item = self.get_slot_item(slot.slotName)
         if isinstance(slot_item, LoadedModule):
             # get height of module + all labware on it
             module_id = slot_item.id
             try:
                 labware_id = self._labware.get_id_by_module(module_id=module_id)
             except LabwareNotLoadedOnModuleError:
-                deck_type = DeckType(self._labware.get_deck_definition()["otId"])
                 return self._modules.get_module_highest_z(
-                    module_id=module_id, deck_type=deck_type
+                    module_id=module_id,
                 )
             else:
                 return self.get_highest_z_of_labware_stack(labware_id)
@@ -244,10 +246,7 @@ def _get_labware_position_offset(
             return LabwareOffsetVector(x=0, y=0, z=0)
         elif isinstance(labware_location, ModuleLocation):
             module_id = labware_location.moduleId
-            deck_type = DeckType(self._labware.get_deck_definition()["otId"])
-            module_offset = self._modules.get_nominal_module_offset(
-                module_id=module_id, deck_type=deck_type
-            )
+            module_offset = self._modules.get_nominal_module_offset(module_id=module_id)
             module_model = self._modules.get_connected_model(module_id)
             stacking_overlap = self._labware.get_module_overlap_offsets(
                 labware_id, module_model
@@ -698,7 +697,11 @@ def get_extra_waypoints(
     def get_slot_item(
         self, slot_name: Union[DeckSlotName, StagingSlotName]
     ) -> Union[LoadedLabware, LoadedModule, CutoutFixture, None]:
-        """Get the item present in a deck slot, if any."""
+        """Get the top-most item present in a deck slot, if any.
+
+        This includes any module that occupies the given slot even if it wasn't loaded
+        in that slot (e.g., thermocycler).
+        """
         maybe_labware = self._labware.get_by_slot(
             slot_name=slot_name,
         )
@@ -717,7 +720,7 @@ def get_slot_item(
 
             maybe_module = self._modules.get_by_slot(
                 slot_name=slot_name,
-            )
+            ) or self._modules.get_overflowed_module_in_slot(slot_name=slot_name)
         else:
             # Modules and fixtures can't be loaded on staging slots
             maybe_fixture = None

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

@@ -69,6 +69,7 @@
     MagneticBlockId,
     ModuleSubStateType,
 )
+from .config import Config
 
 
 ModuleSubStateT = TypeVar("ModuleSubStateT", bound=ModuleSubStateType)
@@ -107,6 +108,14 @@ class SlotTransit(NamedTuple):
     _OT2_THERMOCYCLER_SLOT_TRANSITS_TO_DODGE | _OT3_THERMOCYCLER_SLOT_TRANSITS_TO_DODGE
 )
 
+_THERMOCYCLER_SLOT = DeckSlotName.SLOT_B1
+_OT2_THERMOCYCLER_ADDITIONAL_SLOTS = [
+    DeckSlotName.SLOT_8,
+    DeckSlotName.SLOT_10,
+    DeckSlotName.SLOT_11,
+]
+_OT3_THERMOCYCLER_ADDITIONAL_SLOTS = [DeckSlotName.SLOT_A1]
+
 
 @dataclass(frozen=True)
 class HardwareModule:
@@ -127,6 +136,17 @@ class ModuleState:
     ProtocolEngine.use_attached_modules() instead of an explicit loadModule command.
     """
 
+    additional_slots_occupied_by_module_id: Dict[str, List[DeckSlotName]]
+    """List of additional slots occupied by each module.
+
+    The thermocycler (both GENs), occupies multiple slots on both OT-2 and the Flex
+    but only one slot is associated with the location of the thermocycler.
+    In order to check for deck conflicts with other items, we will keep track of any
+    additional slots occupied by a module here.
+
+    This will be None when a module occupies only one slot.
+    """
+
     requested_model_by_id: Dict[str, Optional[ModuleModel]]
     """The model by which each loaded module was requested.
 
@@ -147,23 +167,31 @@ class ModuleState:
     module_offset_by_serial: Dict[str, ModuleOffsetData]
     """Information about each modules offsets."""
 
+    deck_type: DeckType
+    """Type of deck that the modules are on."""
+
 
 class ModuleStore(HasState[ModuleState], HandlesActions):
     """Module state container."""
 
     _state: ModuleState
 
     def __init__(
-        self, module_calibration_offsets: Optional[Dict[str, ModuleOffsetData]] = None
+        self,
+        config: Config,
+        module_calibration_offsets: Optional[Dict[str, ModuleOffsetData]] = None,
     ) -> None:
         """Initialize a ModuleStore and its state."""
         self._state = ModuleState(
             slot_by_module_id={},
+            additional_slots_occupied_by_module_id={},
             requested_model_by_id={},
             hardware_by_module_id={},
             substate_by_module_id={},
             module_offset_by_serial=module_calibration_offsets or {},
+            deck_type=config.deck_type,
         )
+        self._robot_type = config.robot_type
 
     def handle_action(self, action: Action) -> None:
         """Modify state in reaction to an action."""
@@ -284,11 +312,32 @@ def _add_module_substate(
                 target_block_temperature=live_data["targetTemp"] if live_data else None,  # type: ignore[arg-type]
                 target_lid_temperature=live_data["lidTarget"] if live_data else None,  # type: ignore[arg-type]
             )
+            self._update_additional_slots_occupied_by_thermocycler(
+                module_id=module_id, slot_name=slot_name
+            )
         elif ModuleModel.is_magnetic_block(actual_model):
             self._state.substate_by_module_id[module_id] = MagneticBlockSubState(
                 module_id=MagneticBlockId(module_id)
             )
 
+    def _update_additional_slots_occupied_by_thermocycler(
+        self,
+        module_id: str,
+        slot_name: Optional[
+            DeckSlotName
+        ],  # addModuleAction will not have a slot location
+    ) -> None:
+        if slot_name != _THERMOCYCLER_SLOT.to_equivalent_for_robot_type(
+            self._robot_type
+        ):
+            return
+
+        self._state.additional_slots_occupied_by_module_id[module_id] = (
+            _OT3_THERMOCYCLER_ADDITIONAL_SLOTS
+            if self._state.deck_type == DeckType.OT3_STANDARD
+            else _OT2_THERMOCYCLER_ADDITIONAL_SLOTS
+        )
+
     def _update_module_calibration(
         self,
         module_id: str,
@@ -656,7 +705,8 @@ def get_dimensions(self, module_id: str) -> ModuleDimensions:
         return self.get_definition(module_id).dimensions
 
     def get_nominal_module_offset(
-        self, module_id: str, deck_type: DeckType
+        self,
+        module_id: str,
     ) -> LabwareOffsetVector:
         """Get the module's nominal offset vector computed with slot transform."""
         definition = self.get_definition(module_id)
@@ -670,7 +720,9 @@ def get_nominal_module_offset(
                 1,
             )
         )
-        xforms_ser = definition.slotTransforms.get(str(deck_type.value), {}).get(
+        xforms_ser = definition.slotTransforms.get(
+            str(self._state.deck_type.value), {}
+        ).get(
             slot,
             {"labwareOffset": [[1, 0, 0, 0], [0, 1, 0, 0], [0, 0, 1, 0], [0, 0, 0, 1]]},
         )
@@ -703,7 +755,7 @@ def get_height_over_labware(self, module_id: str) -> float:
         """Get the height of module parts above module labware base."""
         return self.get_dimensions(module_id).overLabwareHeight
 
-    def get_module_highest_z(self, module_id: str, deck_type: DeckType) -> float:
+    def get_module_highest_z(self, module_id: str) -> float:
         """Get the highest z point of the module, as placed on the robot.
 
         The highest Z of a module, unlike the bare overall height, depends on
@@ -729,7 +781,7 @@ def get_module_highest_z(self, module_id: str, deck_type: DeckType) -> float:
         z_difference = module_height - default_lw_offset_point
 
         nominal_transformed_lw_offset_z = self.get_nominal_module_offset(
-            module_id=module_id, deck_type=deck_type
+            module_id=module_id
         ).z
         calibration_offset = self.get_module_calibration_offset(module_id)
         return (
@@ -980,3 +1032,20 @@ def get_default_gripper_offsets(
         """Get the deck's default gripper offsets."""
         offsets = self.get_definition(module_id).gripperOffsets
         return offsets.get("default") if offsets else None
+
+    def get_overflowed_module_in_slot(
+        self, slot_name: DeckSlotName
+    ) -> Optional[LoadedModule]:
+        """Get the module that's not loaded in the given slot, but still occupies the slot.
+
+        For example, if there's a thermocycler loaded in B1,
+        `get_overflowed_module_in_slot(DeckSlotName.Slot_A1)` will return the loaded
+        thermocycler module.
+        """
+        slots_by_id = self._state.additional_slots_occupied_by_module_id
+
+        for module_id, module_slots in slots_by_id.items():
+            if module_slots and slot_name in module_slots:
+                return self.get(module_id)
+
+        return None

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

@@ -174,7 +174,8 @@ def __init__(
             deck_definition=deck_definition,
         )
         self._module_store = ModuleStore(
-            module_calibration_offsets=module_calibration_offsets
+            config=config,
+            module_calibration_offsets=module_calibration_offsets,
         )
         self._liquid_store = LiquidStore()
         self._tip_store = TipStore()

api/tests/opentrons/protocol_api_integration/test_pipette_movement_deck_conflicts.py

@@ -16,7 +16,6 @@ def test_deck_conflicts_for_96_ch_a12_column_configuration() -> None:
     trash_labware = protocol_context.load_labware(
         "opentrons_1_trash_3200ml_fixed", "A3"
     )
-
     badly_placed_tiprack = protocol_context.load_labware(
         "opentrons_flex_96_tiprack_50ul", "C2"
     )
@@ -30,7 +29,7 @@ def test_deck_conflicts_for_96_ch_a12_column_configuration() -> None:
     )
 
     thermocycler = protocol_context.load_module("thermocyclerModuleV2")
-    partially_accessible_plate = thermocycler.load_labware(
+    accessible_plate = thermocycler.load_labware(
         "opentrons_96_wellplate_200ul_pcr_full_skirt"
     )
 
@@ -75,7 +74,7 @@ def test_deck_conflicts_for_96_ch_a12_column_configuration() -> None:
 
     # Will NOT raise error since first column of TC labware is accessible
     # (it is just a few mm away from the left bound)
-    instrument.dispense(25, partially_accessible_plate.wells_by_name()["A1"])
+    instrument.dispense(25, accessible_plate.wells_by_name()["A1"])
 
     instrument.drop_tip()
 
@@ -88,8 +87,8 @@ def test_deck_conflicts_for_96_ch_a12_column_configuration() -> None:
     # No error NOW because of full config
     instrument.aspirate(50, badly_placed_labware.wells_by_name()["A1"])
 
-    # No error NOW because of full config
-    instrument.dispense(50, partially_accessible_plate.wells_by_name()["A1"])
+    # No error
+    instrument.dispense(50, accessible_plate.wells_by_name()["A1"])
 
 
 @pytest.mark.ot3_only

api/tests/opentrons/protocol_engine/state/test_geometry_view.py

@@ -178,9 +178,7 @@ def test_get_labware_parent_position_on_module(
     ).then_return(Point(1, 2, 3))
     decoy.when(labware_view.get_deck_definition()).then_return(ot2_standard_deck_def)
     decoy.when(
-        module_view.get_nominal_module_offset(
-            module_id="module-id", deck_type=DeckType.OT2_STANDARD
-        )
+        module_view.get_nominal_module_offset(module_id="module-id")
     ).then_return(LabwareOffsetVector(x=4, y=5, z=6))
     decoy.when(module_view.get_connected_model("module-id")).then_return(
         ModuleModel.THERMOCYCLER_MODULE_V2
@@ -242,9 +240,7 @@ def test_get_labware_parent_position_on_labware(
 
     decoy.when(labware_view.get_deck_definition()).then_return(ot2_standard_deck_def)
     decoy.when(
-        module_view.get_nominal_module_offset(
-            module_id="module-id", deck_type=DeckType.OT2_STANDARD
-        )
+        module_view.get_nominal_module_offset(module_id="module-id")
     ).then_return(LabwareOffsetVector(x=1, y=2, z=3))
 
     decoy.when(module_view.get_connected_model("module-id")).then_return(
@@ -424,9 +420,7 @@ def test_get_module_labware_highest_z(
     )
     decoy.when(labware_view.get_deck_definition()).then_return(ot2_standard_deck_def)
     decoy.when(
-        module_view.get_nominal_module_offset(
-            module_id="module-id", deck_type=DeckType.OT2_STANDARD
-        )
+        module_view.get_nominal_module_offset(module_id="module-id")
     ).then_return(LabwareOffsetVector(x=4, y=5, z=6))
     decoy.when(module_view.get_height_over_labware("module-id")).then_return(0.5)
     decoy.when(module_view.get_module_calibration_offset("module-id")).then_return(
@@ -711,11 +705,9 @@ def test_get_highest_z_in_slot_with_single_module(
         errors.LabwareNotLoadedOnModuleError("only module")
     )
     decoy.when(labware_view.get_deck_definition()).then_return(ot2_standard_deck_def)
-    decoy.when(
-        module_view.get_module_highest_z(
-            module_id="only-module", deck_type=DeckType("ot2_standard")
-        )
-    ).then_return(12345)
+    decoy.when(module_view.get_module_highest_z(module_id="only-module")).then_return(
+        12345
+    )
 
     assert (
         subject.get_highest_z_in_slot(DeckSlotLocation(slotName=DeckSlotName.SLOT_3))
@@ -889,9 +881,7 @@ def test_get_highest_z_in_slot_with_labware_stack_on_module(
         DeckSlotLocation(slotName=DeckSlotName.SLOT_3)
     )
     decoy.when(
-        module_view.get_nominal_module_offset(
-            module_id="module-id", deck_type=DeckType("ot2_standard")
-        )
+        module_view.get_nominal_module_offset(module_id="module-id")
     ).then_return(LabwareOffsetVector(x=40, y=50, z=60))
     decoy.when(module_view.get_connected_model("module-id")).then_return(
         ModuleModel.TEMPERATURE_MODULE_V2
@@ -1095,9 +1085,7 @@ def test_get_module_labware_well_position(
     )
     decoy.when(labware_view.get_deck_definition()).then_return(ot2_standard_deck_def)
     decoy.when(
-        module_view.get_nominal_module_offset(
-            module_id="module-id", deck_type=DeckType.OT2_STANDARD
-        )
+        module_view.get_nominal_module_offset(module_id="module-id")
     ).then_return(LabwareOffsetVector(x=4, y=5, z=6))
     decoy.when(module_view.get_module_calibration_offset("module-id")).then_return(
         ModuleOffsetData(
@@ -1636,9 +1624,7 @@ def test_get_labware_grip_point_for_labware_on_module(
     )
     decoy.when(labware_view.get_deck_definition()).then_return(ot2_standard_deck_def)
     decoy.when(
-        module_view.get_nominal_module_offset(
-            module_id="module-id", deck_type=DeckType.OT2_STANDARD
-        )
+        module_view.get_nominal_module_offset(module_id="module-id")
     ).then_return(LabwareOffsetVector(x=1, y=2, z=3))
     decoy.when(module_view.get_connected_model("module-id")).then_return(
         ModuleModel.MAGNETIC_MODULE_V2
@@ -1744,6 +1730,22 @@ def test_get_slot_item(
     assert subject.get_slot_item(DeckSlotName.SLOT_3) == module
 
 
+def test_get_slot_item_that_is_overflowed_module(
+    decoy: Decoy,
+    labware_view: LabwareView,
+    module_view: ModuleView,
+    subject: GeometryView,
+) -> None:
+    """It should return the module that occupies the slot, even if not loaded on it."""
+    module = LoadedModule.construct(id="cool-module")  # type: ignore[call-arg]
+    decoy.when(labware_view.get_by_slot(DeckSlotName.SLOT_3)).then_return(None)
+    decoy.when(module_view.get_by_slot(DeckSlotName.SLOT_3)).then_return(None)
+    decoy.when(
+        module_view.get_overflowed_module_in_slot(DeckSlotName.SLOT_3)
+    ).then_return(module)
+    assert subject.get_slot_item(DeckSlotName.SLOT_3) == module
+
+
 @pytest.mark.parametrize(
     argnames=["slot_name", "expected_column"],
     argvalues=[