fix(hardware-testing): Add Flex stacker PVT QC StallGuard thresholds + Use labware_detect from FlexStacker module. (#18741)

Author: djgomez-opentrons

hardware-testing/Makefile

@@ -152,7 +152,7 @@ test-integration: test-production-qc test-examples test-scripts test-gravimetric
 
 .PHONY: test-stacker
 test-stacker:
-	$(python) -m hardware_testing.modules.flex_stacker_dvt_qc --simulate
+	$(python) -m hardware_testing.modules.flex_stacker_qc --simulate
 
 .PHONY: lint
 lint:

hardware-testing/hardware_testing/modules/flex_stacker_qc/__main__.py

@@ -18,7 +18,6 @@
 from .utils import find_stacker_port, get_estop
 from opentrons.drivers.rpi_drivers.types import USBPort
 from opentrons.hardware_control.modules.flex_stacker import FlexStacker
-from opentrons.hardware_control.modules.types import PlatformState
 
 
 async def build_stacker_report(
@@ -60,16 +59,8 @@ async def _main(cfg: TestConfig) -> None:
                 ui.print_error("ESTOP is still pressed, cannot start tests")
                 return
 
-        # TODO: This check should be in the basic axes tests
         await stacker.home_all()
         await stacker._reader.read()
-        if stacker.platform_state is not PlatformState.UNKNOWN:
-            ui.print_error("Platform must be removed from the carrier before starting")
-            ui.get_user_ready(f"Remove platform from {stacker.platform_state.value}")
-            await stacker._reader.get_platform_sensor_state()
-            if stacker.platform_state is not PlatformState.UNKNOWN:
-                ui.print_error("Platform is still detected, cannot start tests")
-                return
 
     device_info = await stacker._driver.get_device_info()
     report.set_tag(device_info.sn if device_info.sn else "UNKNOWN")
@@ -87,11 +78,6 @@ async def _main(cfg: TestConfig) -> None:
     report.save_to_disk()
     report.print_results()
 
-    # Restart the robot server
-    if not cfg.simulate:
-        print("Starting the robot server")
-        subprocess.run(["systemctl restart opentrons-robot-server &"], shell=True)
-
 
 if __name__ == "__main__":
     parser = argparse.ArgumentParser()

hardware-testing/hardware_testing/modules/flex_stacker_qc/test_stallguard.py

@@ -12,59 +12,76 @@
 from opentrons_shared_data.errors.exceptions import FlexStackerStallError
 from opentrons.drivers.flex_stacker.driver import (
     STACKER_MOTION_CONFIG,
+    STALLGUARD_CONFIG,
+)
+from opentrons.drivers.flex_stacker.types import (
+    StackerAxis,
+    Direction,
 )
-from opentrons.drivers.flex_stacker.types import StackerAxis, Direction
 
 # The distance from limit switch to limit switch, mm
 TEST_DISTANCE = {StackerAxis.X: 193.5, StackerAxis.Z: 137}
 
 
+def generate_test_thresholds(test_axis: StackerAxis) -> List[int]:
+    """Generate test thresholds."""
+    thresholds = []
+    default_threshold = STALLGUARD_CONFIG[test_axis].threshold
+    for sgt in range(default_threshold - 2, default_threshold + 3):
+        thresholds.append(sgt)
+    return thresholds
+
+
+def _get_test_tag(test_axis: StackerAxis, sgt: int) -> str:
+    """Get test tag."""
+    return f"stallguard-{test_axis}-SGT-{sgt}"
+
+
 def build_csv_lines() -> List[Union[CSVLine, CSVLineRepeating]]:
     """Build CSV Lines."""
-    return [
-        CSVLine(f"stallguard-{StackerAxis.X}", [CSVResult]),
-        CSVLine(f"stallguard-{StackerAxis.Z}", [CSVResult]),
-    ]
+    lines: List[Union[CSVLine, CSVLineRepeating]] = []
+    for axis in [StackerAxis.X, StackerAxis.Z]:
+        for sgt in generate_test_thresholds(axis):
+            lines.append(CSVLine(_get_test_tag(axis, sgt), [CSVResult]))
+    return lines
 
 
 async def test_stallguard(
     stacker: FlexStacker, test_axis: StackerAxis, report: CSVReport, section: str
 ) -> None:
     """Test Stall-Guard."""
     ui.print_header(f"Testing {test_axis} Axis")
-    stall_detected = False
-    try:
-        await stacker.move_axis(
-            test_axis,
-            Direction.RETRACT,
-            TEST_DISTANCE[test_axis],
-            STACKER_MOTION_CONFIG[test_axis]["move"].move_params.max_speed,
-            STACKER_MOTION_CONFIG[test_axis]["move"].move_params.acceleration,
-            STACKER_MOTION_CONFIG[test_axis]["move"].run_current,
-        )
-    except FlexStackerStallError:
-        ui.print_info("Stall Detected")
-        stall_detected = True
 
-    axis_reset = False
-    while not axis_reset:
+    for sgt in generate_test_thresholds(test_axis):
+        ui.print_header(f"Testing Stallguard Threshold: {sgt}")
+        await stacker._driver.set_stallguard_threshold(test_axis, True, sgt)
+        stall_detected = False
         try:
-            # Move the axis off the crash block before re-homing
             await stacker.move_axis(
                 test_axis,
-                Direction.EXTEND,
-                20,
-                STACKER_MOTION_CONFIG[test_axis]["home"].move_params.max_speed,
-                STACKER_MOTION_CONFIG[test_axis]["home"].move_params.acceleration,
-                STACKER_MOTION_CONFIG[test_axis]["home"].run_current,
+                Direction.RETRACT,
+                TEST_DISTANCE[test_axis],
+                STACKER_MOTION_CONFIG[test_axis]["move"].move_params.max_speed,
+                STACKER_MOTION_CONFIG[test_axis]["move"].move_params.acceleration,
+                STACKER_MOTION_CONFIG[test_axis]["move"].run_current,
             )
-            axis_reset = True
         except FlexStackerStallError:
-            axis_reset = False
+            ui.print_info("Stall Detected")
+            stall_detected = True
+            await stacker._driver.set_stallguard_threshold(test_axis, False, 0)
 
-    await stacker.home_axis(test_axis, Direction.EXTEND)
+        await stacker.home_axis(test_axis, Direction.EXTEND)
 
-    report(section, f"stallguard-{test_axis}", [CSVResult.from_bool(stall_detected)])
+        report(
+            section,
+            _get_test_tag(test_axis, sgt),
+            [CSVResult.from_bool(stall_detected)],
+        )
+
+    # Restore default stallguard threshold
+    await stacker._driver.set_stallguard_threshold(
+        test_axis, True, STALLGUARD_CONFIG[test_axis].threshold
+    )
 
     return
 
@@ -88,13 +105,6 @@ async def run(stacker: FlexStacker, report: CSVReport, section: str) -> None:
     # Test Axes
     await test_stallguard(stacker, StackerAxis.X, report, section)
     await test_stallguard(stacker, StackerAxis.Z, report, section)
-    # ui.print_header(f"Testing {StackerAxis.X} Axis")
-    # x_result = await test_stallguard(stacker, StackerAxis.X)
-    # report(section, f"stallguard-x", [CSVResult.from_bool(x_result)])
-
-    # ui.print_header("Testing Z Axis")
-    # z_result = await test_stallguard(stacker, StackerAxis.Z)
-    # report(section, f"stallguard-z", [CSVResult.from_bool(z_result)])
 
     # Prompt operator to remove the crash block
     if not stacker.is_simulated:
@@ -109,7 +119,13 @@ async def run(stacker: FlexStacker, report: CSVReport, section: str) -> None:
             test_reset = True
         except FlexStackerStallError:
             # Double check if crash block was actually removed
-            await stacker.home_axis(StackerAxis.Z, Direction.EXTEND)
-            await stacker.home_axis(StackerAxis.X, Direction.EXTEND)
+            await stacker._driver.set_stallguard_threshold(StackerAxis.Z, False, 0)
+            await stacker._driver.set_stallguard_threshold(StackerAxis.X, False, 0)
+            await stacker._driver.set_stallguard_threshold(
+                StackerAxis.Z, True, STALLGUARD_CONFIG[StackerAxis.Z].threshold
+            )
+            await stacker._driver.set_stallguard_threshold(
+                StackerAxis.X, True, STALLGUARD_CONFIG[StackerAxis.X].threshold
+            )
             if not stacker.is_simulated:
                 ui.get_user_ready("Remove the crash block from the stacker")

hardware-testing/hardware_testing/modules/flex_stacker_qc/test_tof_functional.py

@@ -9,10 +9,8 @@
     CSVLineRepeating,
     CSVResult,
 )
-from hardware_testing.modules.flex_stacker_dvt_qc.utils import labware_detected
 
 from opentrons.hardware_control.modules.flex_stacker import FlexStacker
-from .utils import NUMBER_OF_BINS, NUMBER_OF_ZONES
 from opentrons.drivers.flex_stacker.types import (
     Direction,
     StackerAxis,
@@ -56,6 +54,8 @@ async def test_tof_sensors_labware_detection(
     section: str,
     sensor: TOFSensor,
     labware: str,
+    axis: StackerAxis,
+    direction: Direction = Direction.EXTEND,
 ) -> None:
     """Test that we can detect labware with the TOF sensor."""
     open = not await stacker._driver.get_hopper_door_closed()
@@ -74,12 +74,13 @@ async def test_tof_sensors_labware_detection(
         return
 
     print(f"Getting histogram for {sensor}.")
-    bins = list(range(NUMBER_OF_BINS))
-    zones = list(range(NUMBER_OF_ZONES))
     histogram = await stacker._driver.get_tof_histogram(sensor)
-    diff = labware_detected(histogram.bins, sensor, bins, zones)
+
+    print(f"Verifying Labware Presence for {sensor}.")
     labware_expected = labware != "empty"
-    result = labware_expected == bool(diff)
+    lbw_detected = await stacker.labware_detected(axis, direction)
+    result = labware_expected == bool(lbw_detected)
+
     report(
         section,
         f"tof-{sensor.name}-histogram-{labware}",
@@ -105,15 +106,21 @@ async def run(stacker: FlexStacker, report: CSVReport, section: str) -> None:
     ui.get_user_ready("Make sure there is no labware on the stacker gripper position")
     await stacker.home_axis(StackerAxis.X, Direction.RETRACT)
     await test_tof_sensors_labware_detection(
-        stacker, report, section, TOFSensor.X, "empty"
+        stacker, report, section, TOFSensor.X, "empty", StackerAxis.X, Direction.RETRACT
     )
 
     print("Test that we detect tiprack on the X home position")
     await stacker.home_axis(StackerAxis.X, Direction.EXTEND)
     ui.get_user_ready("Add 1 tiprack to the stacker X")
     await stacker.home_axis(StackerAxis.X, Direction.RETRACT)
     await test_tof_sensors_labware_detection(
-        stacker, report, section, TOFSensor.X, "tiprack"
+        stacker,
+        report,
+        section,
+        TOFSensor.X,
+        "tiprack",
+        StackerAxis.X,
+        Direction.RETRACT,
     )
     await stacker.home_axis(StackerAxis.X, Direction.EXTEND)
 
@@ -123,12 +130,12 @@ async def run(stacker: FlexStacker, report: CSVReport, section: str) -> None:
     )
     await stacker.close_latch()
     await test_tof_sensors_labware_detection(
-        stacker, report, section, TOFSensor.Z, "empty"
+        stacker, report, section, TOFSensor.Z, "empty", StackerAxis.Z
     )
 
     print("Test that we detect tiprack on the Z")
     ui.get_user_ready("Add 1 tiprack to the stacker Z and close the hopper door")
     await test_tof_sensors_labware_detection(
-        stacker, report, section, TOFSensor.Z, "tiprack"
+        stacker, report, section, TOFSensor.Z, "tiprack", StackerAxis.Z
     )
     ui.get_user_ready("Please remove all labware from the stacker.")

hardware-testing/hardware_testing/modules/flex_stacker_qc/test_x_axis_basic.py

@@ -55,26 +55,12 @@ async def test_platform_sensors_for_direction(
     )
 
 
-async def platform_is_removed(stacker: FlexStacker) -> bool:
-    """Check if the platform is removed from the carrier."""
-    plus_side = await stacker._driver.get_platform_sensor(Direction.EXTEND)
-    minus_side = await stacker._driver.get_platform_sensor(Direction.RETRACT)
-    return not plus_side and not minus_side
-
-
 async def run(stacker: FlexStacker, report: CSVReport, section: str) -> None:
     """Run."""
-    if not stacker.is_simulated and not await platform_is_removed(stacker):
-        print("FAILURE - Cannot start tests with platform on the carrier")
-        return
-
     await test_limit_switches_per_direction(
         stacker, StackerAxis.X, Direction.EXTEND, report, section
     )
 
-    if not stacker.is_simulated:
-        ui.get_user_ready("Place the platform on the X carrier")
-
     await test_platform_sensors_for_direction(
         stacker, Direction.EXTEND, report, section
     )