fix(api, hardware-testing): Flex stacker PVT diagnostics test fixes + Check the X for labware when dispensing. (#19014)

Author: vegano1

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

@@ -2,7 +2,17 @@
 
 import asyncio
 import logging
-from typing import Any, Awaitable, Callable, Dict, Literal, Optional, Mapping, cast
+from typing import (
+    Any,
+    Awaitable,
+    Callable,
+    Dict,
+    List,
+    Literal,
+    Optional,
+    Mapping,
+    cast,
+)
 
 from opentrons.drivers.flex_stacker.types import (
     AxisParams,
@@ -14,6 +24,7 @@
     StackerAxis,
     StallGuardParams,
     TOFDetection,
+    TOFMeasurementResult,
     TOFSensor,
     HardwareRevision,
     TOFSensorMode,
@@ -520,11 +531,16 @@ async def dispense_labware(
         """Dispenses the next labware in the stacker."""
         self.verify_labware_height(labware_height)
         await self._prepare_for_action()
+
         if enforce_hopper_lw_sensing:
             await self.verify_hopper_labware_presence(Direction.EXTEND, True)
 
-        # Move platform along the X then Z axis
+        # 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)
+        if enforce_shuttle_lw_sensing:
+            await self.verify_shuttle_labware_presence(Direction.RETRACT, False)
+
+        # Move platform along the Z axis
         await self._move_and_home_axis(StackerAxis.Z, Direction.EXTEND, HOME_OFFSET_SM)
 
         # Transfer
@@ -556,7 +572,7 @@ async def store_labware(
         if enforce_shuttle_lw_sensing:
             await self.verify_shuttle_labware_presence(Direction.RETRACT, True)
 
-        # Move the Z so the labware sits right under any labware already stored
+        # Move the Z so the labware sits right under the labware already stored
         latch_clear_distance = labware_height + PLATFORM_OFFSET - LATCH_CLEARANCE
         distance = MAX_TRAVEL[StackerAxis.Z] - latch_clear_distance
         await self.move_axis(StackerAxis.Z, Direction.EXTEND, distance)
@@ -638,7 +654,13 @@ async def home_all(self, ignore_latch: bool = False) -> None:
         await self.home_axis(StackerAxis.Z, Direction.RETRACT)
         await self.home_axis(StackerAxis.X, Direction.EXTEND)
 
-    async def labware_detected(self, axis: StackerAxis, direction: Direction) -> bool:
+    async def labware_detected(
+        self,
+        axis: StackerAxis,
+        direction: Direction,
+        histogram: Optional[TOFMeasurementResult] = None,
+        baseline: Optional[Dict[int, List[float]]] = None,
+    ) -> bool:
         """Detect labware on the TOF sensor using the `baseline` method
 
         NOTE: This method is still under development and is inconsistent when detecting
@@ -648,11 +670,13 @@ async def labware_detected(self, axis: StackerAxis, direction: Direction) -> boo
         """
         dir_str = cast(Literal["extend", "retract"], str(direction))
         sensor = TOFSensor.X if axis == StackerAxis.X else TOFSensor.Z
-        baseline = load_tof_baseline_data(self.model())[sensor.value][dir_str]
+        baseline = (
+            baseline or load_tof_baseline_data(self.model())[sensor.value][dir_str]
+        )
         config = TOF_DETECTION_CONFIG[sensor][direction]
 
         # Take a histogram reading and determine if labware was detected
-        histogram = await self._driver.get_tof_histogram(sensor)
+        histogram = histogram or await self._driver.get_tof_histogram(sensor)
         for zone in config.zones:
             raw_data = histogram.bins[zone]
             baseline_data = baseline[zone]

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

@@ -419,9 +419,13 @@ async def test_dispense_labware_motion_sequence(
         # We need to verify the move sequence
         verify_hopper_labware_presence.assert_called_once_with(Direction.EXTEND, True)
         _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
         )
@@ -440,13 +444,16 @@ async def test_dispense_labware_motion_sequence(
         home_axis.assert_any_call(StackerAxis.Z, Direction.RETRACT)
 
         # Verify labware presence
-        verify_shuttle_labware_presence.assert_called_once_with(Direction.RETRACT, True)
+        verify_shuttle_labware_presence.assert_any_call(Direction.RETRACT, True)
 
         # Then finally the x is moved to the gripper position
         _move_and_home_axis.assert_any_call(
             StackerAxis.X, Direction.EXTEND, HOME_OFFSET_MD
         )
 
+        # Make sure labware presense check on the X retract was called twice
+        assert verify_shuttle_labware_presence.call_count == 2
+
 
 @pytest.mark.parametrize(
     ("labware_height"),

hardware-testing/hardware_testing/modules/flex_stacker/flex_stacker_qc/config.py

@@ -10,6 +10,7 @@
     test_z_axis_basic,
     test_x_axis_basic,
     test_l_axis_basic,
+    test_l_axis_current_speed,
     test_z_axis_current_speed,
     test_x_axis_current_speed,
     test_stallguard,
@@ -38,6 +39,7 @@ class TestSection(enum.Enum):
     INSTALL_DETECTION = "INSTALL_DETECTION"
     TOF_BASIC = "TOF_BASIC"
     TOF_FUNCTIONAL = "TOF_FUNCTIONAL"
+    L_AXIS_CURRENT_SPEED = "L_AXIS_CURRENT_SPEED"
     Z_AXIS_CURRENT_SPEED = "Z_AXIS_CURRENT_SPEED"
     X_AXIS_CURRENT_SPEED = "X_AXIS_CURRENT_SPEED"
 
@@ -99,6 +101,10 @@ class TestConfig:
         TestSection.TOF_FUNCTIONAL,
         test_tof_functional.run,
     ),
+    (
+        TestSection.L_AXIS_CURRENT_SPEED,
+        test_l_axis_current_speed.run,
+    ),
     (
         TestSection.Z_AXIS_CURRENT_SPEED,
         test_z_axis_current_speed.run,
@@ -163,6 +169,10 @@ def build_report(test_name: str) -> CSVReport:
                 title=TestSection.TOF_FUNCTIONAL.value,
                 lines=test_tof_functional.build_csv_lines(),
             ),
+            CSVSection(
+                title=TestSection.L_AXIS_CURRENT_SPEED.value,
+                lines=test_l_axis_current_speed.build_csv_lines(),
+            ),
             CSVSection(
                 title=TestSection.Z_AXIS_CURRENT_SPEED.value,
                 lines=test_z_axis_current_speed.build_csv_lines(),

hardware-testing/hardware_testing/modules/flex_stacker/flex_stacker_qc/test_l_axis_current_speed.py

@@ -0,0 +1,163 @@
+"""Test L Axis."""
+from typing import List, Union, Tuple, Optional
+from hardware_testing.data import ui
+from hardware_testing.data.csv_report import (
+    CSVReport,
+    CSVLine,
+    CSVLineRepeating,
+    CSVResult,
+)
+
+from opentrons.hardware_control.modules.flex_stacker import (
+    MAX_TRAVEL,
+    FlexStacker,
+    STACKER_MOTION_CONFIG,
+)
+from opentrons.drivers.flex_stacker.types import StackerAxis, Direction
+from opentrons_shared_data.errors.exceptions import FlexStackerStallError
+
+TEST_AXIS = StackerAxis.L
+HOME_SPEED = STACKER_MOTION_CONFIG[TEST_AXIS]["home"].move_params.max_speed
+HOME_CURRENT = STACKER_MOTION_CONFIG[TEST_AXIS]["home"].run_current
+
+TEST_SPEEDS = [50, 100, 150]  # mm/s
+TEST_CURRENTS = [1.5, 1.2, 0.8, 0.3]  # A rms
+TEST_ACCELERATION = STACKER_MOTION_CONFIG[TEST_AXIS]["move"].move_params.acceleration
+CURRENT_THRESHOD = 0.8  # A rms
+TEST_TRIALS = 5
+
+# All units in mm
+
+# The distance from retracted to extended limit switch
+AXIS_TRAVEL = MAX_TRAVEL[TEST_AXIS]
+OFFSET = 2  # The distance to be off the springs from the limit switch
+AXIS_TOLERANCE = 0.5  # Distance tolerance of AXIS_TRAVEL in ONE direction
+LIMIT_SWICH_CHECK = 0.1
+MOVEMENT_TOLERANCE = 0.5  # Maximum allowed movement error in ONE direction
+
+
+def build_csv_lines() -> List[Union[CSVLine, CSVLineRepeating]]:
+    """Build CSV Lines."""
+    lines: List[Union[CSVLine, CSVLineRepeating]] = []
+    for speed in TEST_SPEEDS:
+        for current in TEST_CURRENTS:
+            tag = f"speed-{speed}-current-{current}"
+            lines.append(
+                CSVLine(f"{tag}-success-failed-pass%", [int, int, float, CSVResult])
+            )
+            lines.append(CSVLine(f"{tag}-extend-distance", [float] * TEST_TRIALS))
+            lines.append(CSVLine(f"{tag}-retract-distance", [float] * TEST_TRIALS))
+    return lines
+
+
+async def test_cycle_per_direction(
+    stacker: FlexStacker,
+    direction: Direction,
+    speed: int,
+    current: float,
+) -> Tuple[bool, float]:
+    """Test one cycle."""
+    # latch does not have extend limit switch, so we have to cycle the test
+    await stacker.open_latch()
+
+    # Move at homing speed off the springs
+    await stacker.move_axis(
+        TEST_AXIS, direction, OFFSET, HOME_SPEED, None, HOME_CURRENT
+    )
+
+    try:
+        # moving at the testing speed and current to just before the springs
+        # minus the tolerances
+        test_distance = AXIS_TRAVEL - (2 * OFFSET) - AXIS_TOLERANCE - MOVEMENT_TOLERANCE
+        await stacker.move_axis(
+            TEST_AXIS, direction, test_distance, speed, TEST_ACCELERATION, current
+        )
+
+        # Move to the farthest position the limit switch could be
+        check_distance = OFFSET + 2 * AXIS_TOLERANCE + 2 * MOVEMENT_TOLERANCE
+        try:
+            await stacker.move_axis(
+                TEST_AXIS,
+                direction,
+                check_distance,
+                HOME_SPEED,
+                0,
+                HOME_CURRENT,
+            )
+        except Exception:
+            pass
+
+        # Limit switch only has a retract limit switch
+        if await stacker._driver.get_limit_switch(TEST_AXIS, Direction.RETRACT):
+            # The limit switch was triggered within this amount of distance
+            movement_distance = round(
+                (AXIS_TRAVEL + OFFSET + AXIS_TOLERANCE + MOVEMENT_TOLERANCE), 1
+            )
+            ui.print_info(
+                f"{TEST_AXIS.name} Axis, {direction}, PASS, {speed}mm/s, {current}A, {movement_distance}mm"
+            )
+            return True, movement_distance
+    except FlexStackerStallError:
+        ui.print_error("unexpected axis stall!")
+    # If we reach this point, limit switch did not trigger in expected distance
+    # Probable stall, Movement distance is unknown, return 0
+    return False, 0
+
+
+async def run(stacker: FlexStacker, report: CSVReport, section: str) -> None:
+    """Run."""
+    # Home to closed position
+    await stacker.home_axis(TEST_AXIS, Direction.RETRACT)
+
+    for speed in TEST_SPEEDS:
+        for current in TEST_CURRENTS:
+            tag = f"speed-{speed}-current-{current}"
+            ui.print_header(
+                f"{TEST_AXIS.name} Speed: {speed} mm/s, Current: {current} A"
+            )
+            trial = 0
+            failures = 0
+            extend_data: List[Optional[float]] = [None] * TEST_TRIALS
+            retract_data: List[Optional[float]] = [None] * TEST_TRIALS
+            # Home to closed position
+            await stacker.home_axis(TEST_AXIS, Direction.RETRACT)
+            while trial < TEST_TRIALS:
+                # Can only test retract direction
+                retract, dist = await test_cycle_per_direction(
+                    stacker, Direction.RETRACT, speed, current
+                )
+                extend_data[trial] = dist
+                if not retract:
+                    ui.print_error(
+                        f"{TEST_AXIS.name} Axis retract failed at speed {speed} mm/s, "
+                        f"current {current} A, Distance {dist} mm"
+                    )
+                    failures += 1
+                    trial += 1
+                    continue
+                trial += 1
+
+            success_trials = trial - failures
+            success_rate = (1 - failures / trial) * 100
+            if current >= CURRENT_THRESHOD:
+                # If current is above threshold, all trials must pass
+                result = CSVResult.from_bool(success_rate == 100.0)
+            else:
+                result = CSVResult.PASS
+            report(
+                section,
+                f"{tag}-success-failed-pass%",
+                [success_trials, failures, success_rate, result],
+            )
+            report(section, f"{tag}-extend-distance", extend_data)
+            report(section, f"{tag}-retract-distance", retract_data)
+
+            # Stop the test if any trial fails
+            if result == CSVResult.FAIL:
+                ui.print_error(
+                    f"{TEST_AXIS.name} Axis failed at speed {speed} mm/s, current {current} A"
+                )
+                return
+
+    # End test with the latch closed
+    await stacker.home_axis(TEST_AXIS, Direction.RETRACT)

hardware-testing/hardware_testing/modules/flex_stacker/flex_stacker_qc/test_tof_basic.py

@@ -16,6 +16,10 @@
     TOFSensor,
 )
 
+from hardware_testing.modules.flex_stacker.flex_stacker_qc.utils import (
+    convert_histogram_bins_to_csv_string,
+)
+
 
 def build_csv_lines() -> List[Union[CSVLine, CSVLineRepeating]]:
     """Build CSV Lines."""
@@ -71,12 +75,13 @@ async def test_get_tof_sensor_histogram(
     """Test that we can request and store histogram measurements from this TOF sensor."""
     print(f"Getting histogram for {sensor}.")
     histogram = await stacker._driver.get_tof_histogram(sensor)
+    histogram_string = convert_histogram_bins_to_csv_string(histogram.bins)
     report(
         section,
         f"tof-{sensor.name}-histogram",
         [
             CSVResult.PASS,
-            histogram.bins,
+            histogram_string,
         ],
     )