feat(api): retract pipettes before shaking and opening latch

Closes #11210, closes #11209

api/src/opentrons/protocol_api/module_contexts.py

@@ -11,6 +11,10 @@
 from opentrons.hardware_control.types import Axis
 from opentrons.commands import module_commands as cmds
 from opentrons.commands.publisher import CommandPublisher, publish
+from opentrons.motion_planning.adjacent_slots_getters import (
+    get_adjacent_slots,
+    get_east_west_slots,
+)
 from opentrons.protocols.api_support.types import APIVersion
 
 from .module_validation_and_errors import (
@@ -925,12 +929,16 @@ def set_and_wait_for_shake_speed(self, rpm: int) -> None:
 
         Set the heater shaker's target speed and wait until the specified speed has
         reached. Delays protocol execution until the target speed has been achieved.
+
+        NOTE: Before shaking, this command will retract the pipettes up if they are
+              parked adjacent to the heater-shaker.
         """
         if (
             self._module.labware_latch_status
             == HeaterShakerLabwareLatchStatus.IDLE_CLOSED
         ):
             validated_speed = validate_heater_shaker_speed(rpm=rpm)
+            self._prepare_for_shake()
             self._module.set_speed(rpm=validated_speed)
         else:
             # TODO: Figure out whether to issue close latch behind the scenes instead
@@ -943,11 +951,13 @@ def set_and_wait_for_shake_speed(self, rpm: int) -> None:
     def open_labware_latch(self) -> None:
         """Open the Heater-Shaker's labware latch.
 
-        Note that the labware latch needs to be closed before:
-
-        * Shaking
-        * Pipetting to or from the labware on the Heater-Shaker
-        * Pipetting to or from labware to the left or right of the Heater-Shaker
+        NOTE:
+        1. This command will retract the pipettes up if they are parked east or west
+           of the heater-shaker.
+        2. The labware latch needs to be closed before:
+            * Shaking
+            * Pipetting to or from the labware on the Heater-Shaker
+            * Pipetting to or from labware to the left or right of the Heater-Shaker
 
         Raises an error when attempting to open the latch while the Heater-Shaker is shaking.
         """
@@ -956,6 +966,7 @@ def open_labware_latch(self) -> None:
             raise CannotPerformModuleAction(
                 """Cannot open labware latch while module is shaking."""
             )
+        self._prepare_for_latch_open()
         self._module.open_labware_latch()
 
     # TODO: add API version requirement
@@ -1019,3 +1030,32 @@ def flag_unsafe_move(
             is_plate_shaking=is_plate_shaking,
             is_labware_latch_closed=is_labware_latch_closed,
         )
+
+    def _prepare_for_shake(self) -> None:
+        """Retract pipettes before shaking if necessary.
+
+        Before shaking, retracts pipettes if they're parked over a slot
+        adjacent to the heater-shaker.
+        """
+
+        if cast(HeaterShakerGeometry, self.geometry).is_pipette_blocking_shake_movement(
+            pipette_location=self._ctx.location_cache
+        ):
+            ctx_implementation = self._ctx._implementation
+            hardware = ctx_implementation.get_hardware()
+            for mount in types.Mount:
+                hardware.retract(mount=mount)
+
+    def _prepare_for_latch_open(self) -> None:
+        """Retract pipettes before opening labware latch if necessary.
+
+        Before opening latch, retracts pipettes if they're parked over a slot
+        east/ west of the heater-shaker.
+        """
+        if cast(HeaterShakerGeometry, self.geometry).is_pipette_blocking_latch_movement(
+            pipette_location=self._ctx.location_cache
+        ):
+            ctx_implementation = self._ctx._implementation
+            hardware = ctx_implementation.get_hardware()
+            for mount in types.Mount:
+                hardware.retract(mount=mount)

api/src/opentrons/protocols/geometry/module_geometry.py

@@ -23,6 +23,7 @@
 from opentrons.motion_planning.adjacent_slots_getters import (
     get_north_south_slots,
     get_east_west_slots,
+    get_adjacent_slots,
 )
 
 from opentrons.hardware_control.modules.types import (
@@ -436,6 +437,68 @@ def flag_unsafe_move(
                     "Cannot move multi-channel pipette to non-tip rack labware north or south of Heater-Shaker"
                 )
 
+    def is_pipette_blocking_shake_movement(
+        self, pipette_location: Optional[Location]
+    ) -> bool:
+        """Check whether pipette is parked adjacent to heater-shaker.
+
+        Returns True if pipette's last known location was on east/west/north/south of or
+        on the heater-shaker. Also returns True if last location is not known or is
+        not associated with a slot.
+        """
+        if pipette_location is None:
+            # If we don't know the pipette's latest location then let's be extra
+            # cautious and call it blocking
+            return True
+
+        pipette_location_slot = pipette_location.labware.first_parent()
+        if pipette_location_slot is None:
+            # If a location is not associated w/ a slot (e.g., if it has labware=None)
+            # then we don't know if it's close to the h/s, so, we will be cautious
+            # and call it blocking
+            return True
+
+        heater_shaker_slot = self.parent
+
+        assert isinstance(heater_shaker_slot, str), (
+            "Could not determine module slot " "location"
+        )
+
+        return heater_shaker_slot == pipette_location_slot or int(
+            pipette_location_slot
+        ) in get_adjacent_slots(int(heater_shaker_slot))
+
+    def is_pipette_blocking_latch_movement(
+        self, pipette_location: Optional[Location]
+    ) -> bool:
+        """Check whether pipette is parked east or west of heater-shaker.
+
+        Returns True is pipette's last known location was on east/west of or on the
+        heater-shaker. Also returns True if last location is not known or is not
+        associated with a slot.
+        """
+        if pipette_location is None:
+            # If we don't know the pipette's latest location then let's be extra
+            # cautious and call it blocking
+            return True
+
+        pipette_location_slot = pipette_location.labware.first_parent()
+        if pipette_location_slot is None:
+            # If a location is not associated w/ a slot (e.g., if it has labware=None)
+            # then we don't know if it's close to the h/s, so, we will be cautious
+            # and call it blocking
+            return True
+
+        heater_shaker_slot = self.parent
+
+        assert isinstance(heater_shaker_slot, str), (
+            "Could not determine module slot " "location"
+        )
+
+        return heater_shaker_slot == pipette_location_slot or int(
+            pipette_location_slot
+        ) in get_east_west_slots(int(heater_shaker_slot))
+
 
 def _load_from_v1(
     definition: "ModuleDefinitionV1", parent: Location, api_level: APIVersion

api/tests/opentrons/protocol_api/test_module_context.py

@@ -7,7 +7,7 @@
 import opentrons.protocol_api as papi
 import opentrons.protocols.geometry as papi_geometry
 
-from opentrons.types import Point, Location
+from opentrons.types import Point, Location, Mount
 from opentrons.drivers.types import HeaterShakerLabwareLatchStatus
 from opentrons.hardware_control.modules.magdeck import OFFSET_TO_LABWARE_BOTTOM
 from opentrons.hardware_control.modules.types import (
@@ -30,10 +30,13 @@
 from opentrons.protocols.context.protocol_api.protocol_context import (
     ProtocolContextImplementation,
 )
+from opentrons.protocols.geometry.deck import Deck
 from opentrons.protocols.geometry.module_geometry import (
     PipetteMovementRestrictedByHeaterShakerError,
+    HeaterShakerGeometry,
 )
 from opentrons.protocols.api_support import util as api_util
+from opentrons.protocols.api_support.types import APIVersion
 from opentrons_shared_data import load_shared_data
 
 
@@ -47,6 +50,12 @@ def mock_module_controller() -> mock.MagicMock:
     return mock.MagicMock()
 
 
+@pytest.fixture
+def mock_pipette_location() -> mock.MagicMock:
+    # mock_labware = papi.labware.Labware()
+    return mock.MagicMock(return_value=Location(point=Point(1, 2, 3), labware=None))
+
+
 # Async because ProtocolContext.__init__() needs an event loop,
 # so this fixture needs to run in an event loop.
 @pytest.fixture
@@ -116,10 +125,33 @@ def find_modules(resolved_model: ModuleModel, resolved_type: ModuleType):
     )
 
 
+@pytest.fixture
+def mock_heater_shaker_geometry() -> mock.MagicMock:
+    """Get a Heater-Shaker Geometry fixture."""
+    heater_shaker_slot_location = Deck().position_for(5)
+    hs_geometry = mock.MagicMock(
+        return_value=HeaterShakerGeometry(
+            display_name="A new shiny module!",
+            model=HeaterShakerModuleModel.HEATER_SHAKER_V1,
+            module_type=ModuleType.HEATER_SHAKER,
+            offset=Point(0, 0, 0),
+            overall_height=111,
+            height_over_labware=222,
+            parent=heater_shaker_slot_location,
+            api_level=APIVersion.from_string("22.22"),
+        )
+    )
+    hs_geometry.highest_z = 100
+    hs_geometry.parent = 5
+    return hs_geometry
+
+
 @pytest.fixture
 async def ctx_with_heater_shaker(
     mock_hardware: mock.AsyncMock,
     mock_module_controller: mock.MagicMock,
+    mock_pipette_location: mock.MagicMock,
+    mock_heater_shaker_geometry: mock.MagicMock,
     enable_heater_shaker_python_api: AsyncGenerator[None, None],
 ) -> ProtocolContext:
     """Context fixture with a mock heater-shaker."""
@@ -134,9 +166,12 @@ def find_modules(resolved_model: ModuleModel, resolved_type: ModuleType):
         return []
 
     mock_hardware.find_modules.side_effect = find_modules
-    return ProtocolContext(
+
+    ctx = ProtocolContext(
         implementation=ProtocolContextImplementation(sync_hardware=mock_hardware),
     )
+    ctx.location_cache = mock_pipette_location
+    return ctx
 
 
 # ______ load_module tests _______
@@ -677,8 +712,9 @@ def test_heater_shaker_temperature_properties(
 
     hs_mod = ctx_with_heater_shaker.load_module("heaterShakerModuleV1", 1)
 
-    assert hs_mod.current_temperature == 123.45  # type: ignore[union-attr]
-    assert hs_mod.target_temperature == 234.56  # type: ignore[union-attr]
+    assert isinstance(hs_mod, HeaterShakerContext)
+    assert hs_mod.current_temperature == 123.45
+    assert hs_mod.target_temperature == 234.56
 
 
 def test_heater_shaker_speed_properties(
@@ -693,8 +729,9 @@ def test_heater_shaker_speed_properties(
 
     hs_mod = ctx_with_heater_shaker.load_module("heaterShakerModuleV1", 1)
 
-    assert hs_mod.current_speed == 12  # type: ignore[union-attr]
-    assert hs_mod.target_speed == 34  # type: ignore[union-attr]
+    assert isinstance(hs_mod, HeaterShakerContext)
+    assert hs_mod.current_speed == 12
+    assert hs_mod.target_speed == 34
 
 
 def test_heater_shaker_temp_and_speed_status(
@@ -708,8 +745,9 @@ def test_heater_shaker_temp_and_speed_status(
     type(mock_module_controller).speed_status = mock_speed_status
     hs_mod = ctx_with_heater_shaker.load_module("heaterShakerModuleV1", 1)
 
-    assert hs_mod.temperature_status == "holding at target"  # type: ignore[union-attr]
-    assert hs_mod.speed_status == "slowing down"  # type: ignore[union-attr]
+    assert isinstance(hs_mod, HeaterShakerContext)
+    assert hs_mod.temperature_status == "holding at target"
+    assert hs_mod.speed_status == "slowing down"
 
 
 def test_heater_shaker_latch_status(
@@ -728,8 +766,11 @@ def test_heater_shaker_latch_status(
 def test_heater_shaker_set_and_wait_for_shake_speed(
     ctx_with_heater_shaker: ProtocolContext,
     mock_module_controller: mock.MagicMock,
+    mock_pipette_location: mock.MagicMock,
+    mock_hardware: mock.AsyncMock,
 ) -> None:
     """It should issue a blocking set target shake speed."""
+    # Mock setup
     mock_latch_status = mock.PropertyMock(
         return_value=HeaterShakerLabwareLatchStatus.IDLE_CLOSED
     )
@@ -740,8 +781,22 @@ def test_heater_shaker_set_and_wait_for_shake_speed(
     ) as mock_validator:
         mock_validator.return_value = 10
         hs_mod = ctx_with_heater_shaker.load_module("heaterShakerModuleV1", 1)
-        hs_mod.set_and_wait_for_shake_speed(rpm=400)  # type: ignore[union-attr]
+        assert isinstance(hs_mod, HeaterShakerContext)
+        # assert isinstance(hs_mod.geometry, HeaterShakerGeometry)
+        hs_mod.geometry.is_pipette_blocking_shake_movement = mock.MagicMock(  # type: ignore[attr-defined]
+            return_value=True
+        )
+
+        # Call subject method
+        hs_mod.set_and_wait_for_shake_speed(rpm=400)
+
+        # Assert expected calls
         mock_validator.assert_called_once_with(rpm=400)
+        hs_mod.geometry.is_pipette_blocking_shake_movement.assert_called_with(  # type: ignore[attr-defined]
+            pipette_location=mock_pipette_location
+        )
+        calls = [mock.call(mount=Mount.LEFT), mock.call(mount=Mount.RIGHT)]
+        mock_hardware.retract.assert_has_calls(calls, any_order=True)
         mock_module_controller.set_speed.assert_called_once_with(rpm=10)
 
 
@@ -770,14 +825,31 @@ def test_heater_shaker_set_and_wait_for_shake_speed_raises(
 
 
 def test_heater_shaker_open_labware_latch(
-    ctx_with_heater_shaker: ProtocolContext, mock_module_controller: mock.MagicMock
+    ctx_with_heater_shaker: ProtocolContext,
+    mock_module_controller: mock.MagicMock,
+    mock_pipette_location: mock.MagicMock,
+    mock_hardware: mock.MagicMock,
 ) -> None:
     """It should issue a labware latch open command."""
+    # Mocks
     mock_speed_status = mock.PropertyMock(return_value=SpeedStatus.IDLE)
     type(mock_module_controller).speed_status = mock_speed_status
 
+    # Get subject
     hs_mod = ctx_with_heater_shaker.load_module("heaterShakerModuleV1", 1)
-    hs_mod.open_labware_latch()  # type: ignore[union-attr]
+    assert isinstance(hs_mod, HeaterShakerContext)
+    hs_mod.geometry.is_pipette_blocking_latch_movement = mock.MagicMock(  # type: ignore[attr-defined]
+        return_value=True
+    )
+
+    # Call subject method
+    hs_mod.open_labware_latch()
+    # Assert calls
+    hs_mod.geometry.is_pipette_blocking_latch_movement.assert_called_with(  # type: ignore[attr-defined]
+        pipette_location=mock_pipette_location
+    )
+    calls = [mock.call(mount=Mount.LEFT), mock.call(mount=Mount.RIGHT)]
+    mock_hardware.retract.assert_has_calls(calls, any_order=True)
     mock_module_controller.open_labware_latch.assert_called_once()