refactor(api): check both TOF sensors before raising error when retrieve labware from stacker (#19123)

This pull request refactors the `dispense_labware` logic in the Flex
Stacker module to improve labware detection.

## Changelog
1. temporarily disables `verify_hopper_presence` call until we implement
further calibration function for the TOF sensors.
2. directly calls `labware_detected` on the Z axis, hold on to the
result and use it only if the X TOF sensor fails to detect a labware
after retrieving.
3. updates tests to ensure all error scenarios are properly handled and
validated.

Author: ahiuchingau

api/src/opentrons/hardware_control/modules/flex_stacker.py

@@ -534,7 +534,12 @@ async def dispense_labware(
         await self._prepare_for_action()
 
         if enforce_hopper_lw_sensing:
-            await self.verify_hopper_labware_presence(Direction.EXTEND, True)
+            # TODO: re-enable this function after TOF calibration is implemented.
+            # Until then, we should also check the TOF X sensor before raising the error
+            # await self.verify_hopper_labware_presence(Direction.EXTEND, True)
+            hopper_empty = not await self.labware_detected(
+                StackerAxis.Z, Direction.EXTEND
+            )
 
         # Move platform along the X and make sure we DONT detect labware
         await self._move_and_home_axis(StackerAxis.X, Direction.RETRACT, HOME_OFFSET_MD)
@@ -556,7 +561,22 @@ async def dispense_labware(
         await self.home_axis(StackerAxis.Z, Direction.RETRACT)
 
         if enforce_shuttle_lw_sensing:
-            await self.verify_shuttle_labware_presence(Direction.RETRACT, True)
+            try:
+                await self.verify_shuttle_labware_presence(Direction.RETRACT, True)
+            except FlexStackerShuttleLabwareError:
+                # No labware detected on the shuttle, so we need to check what the Z TOF
+                # sensor says about the hopper
+                if hopper_empty:
+                    # homing here so we don't have to modify the error recovery flow
+                    await self._move_and_home_axis(
+                        StackerAxis.X, Direction.EXTEND, HOME_OFFSET_MD
+                    )
+                    raise FlexStackerHopperLabwareError(
+                        self.device_info["serial"],
+                        labware_expected=True,
+                    ) from None
+                raise
+
         await self._move_and_home_axis(StackerAxis.X, Direction.EXTEND, HOME_OFFSET_MD)
 
     async def store_labware(

api/tests/opentrons/hardware_control/modules/test_hc_flexstacker.py

@@ -1,7 +1,8 @@
 import asyncio
 import pytest
 import mock
-from typing import AsyncGenerator
+from contextlib import nullcontext as does_not_raise
+from typing import AsyncGenerator, ContextManager, Any
 from opentrons.drivers.flex_stacker.simulator import SimulatingDriver
 from opentrons.drivers.flex_stacker.types import (
     Direction,
@@ -26,6 +27,11 @@
 from opentrons.hardware_control.modules.types import PlatformState
 from opentrons.hardware_control.poller import Poller
 from opentrons.hardware_control.types import StatusBarState, StatusBarUpdateEvent
+from opentrons_shared_data.errors.exceptions import (
+    FlexStackerShuttleLabwareError,
+    FlexStackerHopperLabwareError,
+    FlexStackerShuttleNotEmptyError,
+)
 
 
 @pytest.fixture
@@ -401,31 +407,38 @@ async def test_dispense_labware_motion_sequence(
             subject, "_move_and_home_axis", mock.AsyncMock()
         ) as _move_and_home_axis,
         mock.patch.object(
-            subject, "verify_shuttle_labware_presence", mock.AsyncMock()
+            subject, "labware_detected", mock.AsyncMock()
+        ) as labware_detected,
+        mock.patch.object(
+            subject,
+            "verify_shuttle_labware_presence",
+            autospec=True,
+            side_effect=subject.verify_shuttle_labware_presence,
         ) as verify_shuttle_labware_presence,
         mock.patch.object(
-            subject, "verify_hopper_labware_presence", mock.AsyncMock()
+            subject, "verify_hopper_labware_presence", autospec=True
         ) as verify_hopper_labware_presence,
         mock.patch.object(subject, "move_axis", mock.AsyncMock()) as move_axis,
         mock.patch.object(subject, "home_axis", mock.AsyncMock()) as home_axis,
         mock.patch.object(subject, "open_latch", mock.AsyncMock()) as open_latch,
         mock.patch.object(subject, "close_latch", mock.AsyncMock()) as close_latch,
     ):
+        labware_detected.side_effect = [True, False, True]
         # Test valid labware height
         await subject.dispense_labware(
             labware_height=labware_height,
         )
 
         # We need to verify the move sequence
-        verify_hopper_labware_presence.assert_called_once_with(Direction.EXTEND, True)
+        # verify_hopper_labware_presence.assert_called_once_with(Direction.EXTEND, True)
+        verify_hopper_labware_presence.assert_not_called()
         _prepare_for_action.assert_called()
 
         _move_and_home_axis.assert_any_call(
             StackerAxis.X, Direction.RETRACT, HOME_OFFSET_MD
         )
         # Verify labware presence
         verify_shuttle_labware_presence.assert_any_call(Direction.RETRACT, False)
-
         _move_and_home_axis.assert_any_call(
             StackerAxis.Z, Direction.EXTEND, HOME_OFFSET_SM
         )
@@ -452,9 +465,77 @@ async def test_dispense_labware_motion_sequence(
         )
 
         # Make sure labware presense check on the X retract was called twice
+        labware_detected.assert_has_calls(
+            [
+                mock.call(StackerAxis.Z, Direction.EXTEND),
+                mock.call(StackerAxis.X, Direction.RETRACT),
+                mock.call(StackerAxis.X, Direction.RETRACT),
+            ]
+        )
         assert verify_shuttle_labware_presence.call_count == 2
 
 
+@pytest.mark.parametrize("hopper_lw_detected", [True, False])
+@pytest.mark.parametrize("shuttle_lw_detected_before", [True, False])
+@pytest.mark.parametrize("shuttle_lw_detected_after", [True, False])
+async def test_dispense_labware_error_handling(
+    subject: modules.FlexStacker,
+    hopper_lw_detected: bool,
+    shuttle_lw_detected_before: bool,
+    shuttle_lw_detected_after: bool,
+) -> None:
+    """
+    Test different error handling scenarios when dispensing labware.
+    """
+    expected_raise: ContextManager[Any]
+    if shuttle_lw_detected_before is True:
+        # If the shuttle is occupied, it should always raise an error
+        expected_raise = pytest.raises(FlexStackerShuttleNotEmptyError)
+    elif shuttle_lw_detected_after is True:
+        # if the shuttle has labware after dispensing, no need to raise any error
+        expected_raise = does_not_raise()
+    elif hopper_lw_detected is False:
+        # if the shuttle has no labware after dispensing, and the hopper has no labware,
+        # it should raise a FlexStackerHopperLabwareError
+        expected_raise = pytest.raises(FlexStackerHopperLabwareError)
+    else:
+        # if the shuttle has no labware after dispensing, but the hopper has labware,
+        # it should raise a FlexStackerShuttleLabwareError letting user know
+        # the labware latch is properly jammed
+        expected_raise = pytest.raises(FlexStackerShuttleLabwareError)
+
+    with (
+        mock.patch.object(
+            subject,
+            "labware_detected",
+            side_effect=[
+                hopper_lw_detected,
+                shuttle_lw_detected_before,
+                shuttle_lw_detected_after,
+            ],
+            autospec=True,
+        ) as labware_detected,
+    ):
+
+        with expected_raise:
+            # Test valid labware height
+            await subject.dispense_labware(
+                labware_height=100.0,
+            )
+
+        expected_calls = [
+            mock.call(StackerAxis.Z, Direction.EXTEND),
+            mock.call(StackerAxis.X, Direction.RETRACT),
+            mock.call(StackerAxis.X, Direction.RETRACT),
+        ]
+        if shuttle_lw_detected_before:
+            assert labware_detected.call_count == 2
+            labware_detected.assert_has_calls(expected_calls[:2])
+        else:
+            assert labware_detected.call_count == 3
+            labware_detected.assert_has_calls(expected_calls)
+
+
 @pytest.mark.parametrize(
     ("labware_height"),
     [(0), (-10), (200)],