fix(hardware-testing): Various fixes for the flex stacker diagnostics script for Shenzhen. (#17916)

Author: vegano1

api/src/opentrons/drivers/asyncio/communication/serial_connection.py

@@ -257,7 +257,19 @@ def raise_on_error(self, response: str, request: str) -> None:
 
         Raises: SerialException
         """
+        if not response or not request:
+            return
+
         lower = response.lower()
+        res_gcode = response.split()[0]
+        req_gcode = request.split()[0]
+
+        # Make sure this is not just a normal response that happens to contain the
+        # `err` or `alarm` keyword in the message body by checking the gcode values
+        # for both the request and response. If the gcodes are the same then this
+        # is not an error response.
+        if res_gcode == req_gcode:
+            return
 
         if self._alarm_keyword in lower:
             raise AlarmResponse(port=self._port, response=response)

api/tests/opentrons/drivers/asyncio/communication/test_serial_connection.py

@@ -180,6 +180,23 @@ def test_raise_on_error(
         subject.raise_on_error(response, "fake request")
 
 
+def test_raise_on_error_no_raise_on_keyword_in_body(
+    subject: SerialKind,
+) -> None:
+    """It should not raise when there is a keyword in the response body."""
+    request = "M226 Z"
+    # This response contains `eRR` which tricks the system into thinking there is an
+    # error, we fixed this by making sure the request and response gcodes match.
+    response = "M226 Z I:12 D:gW2ACQuAAAAAAAAAAAAAAAAAAAABAQAAAAAAAAAAAAAAAAAAAAAAAA"
+    "AAAAAAAAAAAAAAAAAAAACPbxeRRikcFhINCQYFBAICAQEBAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
+    "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
+    subject.raise_on_error(response, request)
+
+    # This should still raise
+    with pytest.raises(expected_exception=ErrorResponse, match="error"):
+        subject.raise_on_error("error", request)
+
+
 def test_get_error_codes_lowercase(
     subject: SerialKind,
 ) -> None:

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

@@ -7,6 +7,7 @@
 
 import argparse
 import asyncio
+import subprocess
 from pathlib import Path
 from typing import Tuple
 
@@ -42,6 +43,8 @@ async def _main(cfg: TestConfig) -> None:
     report, stacker = await build_stacker_report(cfg.simulate)
 
     if not cfg.simulate:
+        print("Stopping the robot server")
+        subprocess.run(["systemctl stop opentrons-robot-server"], shell=True)
         # Perform initial checks before starting tests
         # 1. estop should not be pressed
         # 2. platform should be removed
@@ -76,6 +79,11 @@ 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_dvt_qc/test_estop.py

@@ -101,7 +101,7 @@ async def run(stacker: FlexStacker, report: CSVReport, section: str) -> None:
     limit_switch_triggered = await stacker._driver.get_limit_switch(
         StackerAxis.L, l_limit
     )
-    if not limit_switch_triggered:
+    if limit_switch_triggered:
         report(
             section,
             "l-move-disabled",
@@ -111,7 +111,7 @@ async def run(stacker: FlexStacker, report: CSVReport, section: str) -> None:
         print("try to move L axis off the limit switch...")
         try:
             await stacker._driver.move_in_mm(
-                StackerAxis.L, l_limit.opposite().distance(1)
+                StackerAxis.L, l_limit.opposite().distance(5)
             )
         except EStopTriggered:
             print("E-Stop Error is raised")
@@ -120,7 +120,7 @@ async def run(stacker: FlexStacker, report: CSVReport, section: str) -> None:
         report(
             section,
             "l-move-disabled",
-            [CSVResult.from_bool(triggered)],
+            [CSVResult.from_bool(not triggered)],
         )
 
     if not stacker._simulating:

hardware-testing/hardware_testing/modules/flex_stacker_dvt_qc/test_install_detection.py

@@ -25,12 +25,12 @@ async def run(stacker: FlexStacker, report: CSVReport, section: str) -> None:
     """Run."""
     ui.print_header("Window Detected")
     if not stacker._simulating:
-        ui.get_user_ready("Attach to detection pins")
+        ui.get_user_ready("Press the window detect switch")
     detected = await stacker._driver.get_installation_detected()
     report(section, "window-detected-high", [CSVResult.from_bool(detected)])
 
     ui.print_header("Window Not Detected")
     if not stacker._simulating:
-        ui.get_user_ready("Remove from detection pins")
+        ui.get_user_ready("Un-press the window detect switch")
     not_detected = not await stacker._driver.get_installation_detected()
     report(section, "window-not-detected-low", [CSVResult.from_bool(not_detected)])

hardware-testing/hardware_testing/modules/flex_stacker_dvt_qc/test_tof_basic.py

@@ -13,7 +13,6 @@
 from opentrons.drivers.flex_stacker.types import (
     Direction,
     StackerAxis,
-    LEDPattern,
     TOFSensor,
 )
 
@@ -84,11 +83,9 @@ async def test_get_tof_sensor_histogram(
 
 async def run(stacker: FlexStacker, report: CSVReport, section: str) -> None:
     """Run."""
-    # Reset LEDs to off
     if not stacker._simulating:
         ui.get_user_ready("Make sure both TOF sensors are installed.")
         ui.get_user_ready("Make sure there is no labware in the stacker.")
-        await stacker._driver.set_led(0, pattern=LEDPattern.STATIC)
 
     print("Homing stacker X and Z axis.")
     await stacker.home_axis(StackerAxis.X, Direction.EXTEND)

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

@@ -16,7 +16,6 @@
 from opentrons.drivers.flex_stacker.types import (
     Direction,
     StackerAxis,
-    LEDPattern,
     TOFSensor,
 )
 
@@ -95,10 +94,8 @@ async def test_tof_sensors_labware_detection(
 
 async def run(stacker: FlexStacker, report: CSVReport, section: str) -> None:
     """Run."""
-    # Reset LEDs to off
     if not stacker._simulating:
         ui.get_user_ready("Make sure both TOF sensors are installed.")
-        await stacker._driver.set_led(0, pattern=LEDPattern.STATIC)
 
     print("Homing stacker X and Z axis.")
     await stacker.home_axis(StackerAxis.X, Direction.EXTEND)
@@ -134,3 +131,4 @@ async def run(stacker: FlexStacker, report: CSVReport, section: str) -> None:
     await test_tof_sensors_labware_detection(
         stacker, report, section, TOFSensor.Z, "tiprack"
     )
+    ui.get_user_ready("Please remove all labware from the stacker.")

hardware-testing/hardware_testing/modules/flex_stacker_dvt_qc/test_ui_leds.py

@@ -49,3 +49,9 @@ async def run(stacker: FlexStacker, report: CSVReport, section: str) -> None:
             else:
                 led_on = True
             report(section, f"{tag}-{color}", [CSVResult.from_bool(led_on)])
+
+    # Turn LEDs back to green
+    if not stacker._simulating:
+        await stacker._driver.set_led(
+            0.5, color=LEDColor.GREEN, pattern=LEDPattern.STATIC
+        )

hardware-testing/hardware_testing/modules/flex_stacker_dvt_qc/test_uv_lockout_switch.py

@@ -28,8 +28,7 @@ async def run(stacker: FlexStacker, report: CSVReport, section: str) -> None:
         ui.get_user_ready("Open the hopper door")
         door_open = not await stacker._driver.get_hopper_door_closed()
         if door_open:
-            ui.print_info("Meausre between J1 and J4")
-            open = not ui.get_user_answer("Is there continuity (closed circuit)?")
+            open = ui.get_user_answer("Is the UV lockout LED off?")
             # circuit should be open
             report(
                 section,
@@ -45,7 +44,7 @@ async def run(stacker: FlexStacker, report: CSVReport, section: str) -> None:
         ui.get_user_ready("Close the hopper door")
         door_closed = await stacker._driver.get_hopper_door_closed()
         if door_closed:
-            closed = ui.get_user_answer("Is there continuity (closed circuit)?")
+            closed = ui.get_user_answer("Is the UV lockout LED on?")
             report(
                 section,
                 "closed-door-closed-circuit",

hardware-testing/hardware_testing/modules/flex_stacker_dvt_qc/utils.py

@@ -1,7 +1,7 @@
 """Utility functions for the Flex Stacker EVT QC module."""
 from collections import defaultdict
 import statistics
-from typing import Dict, List
+from typing import Any, Dict, List
 from opentrons.drivers.flex_stacker.driver import TOFSensor
 from opentrons.drivers.flex_stacker.utils import NUMBER_OF_BINS, NUMBER_OF_ZONES
 from hardware_testing.data import ui
@@ -14,6 +14,12 @@
 from opentrons.drivers.flex_stacker.types import StackerAxis, Direction
 
 
+TOF_DETECTION_CONFIG: Dict[TOFSensor, Dict[str, Any]] = {
+    TOFSensor.X: {"zones": [5, 6, 7], "bins": list(range(10, 20)), "threshold": 30000},
+    TOFSensor.Z: {"zones": [1], "bins": list(range(50, 56)), "threshold": 20000},
+}
+
+
 async def test_limit_switches_per_direction(
     stacker: FlexStacker,
     axis: StackerAxis,
@@ -115,39 +121,23 @@ def labware_detected(
     zones: List[int],
 ) -> Dict[int, List[int]]:
     """Detect labware by subtracting baseline from histogram."""
-    print(f"Detect labware: {sensor}")
     baseline: Dict[int, List[float]] = STACKER_TOF_BASELINE[sensor]
     diff = defaultdict(list)
-    if sensor == TOFSensor.Z:
-        for zone in zones:
-            raw_data = histogram[zone]
-            baseline_data = baseline[zone]
-            for bin in bins:
-                delta = raw_data[bin] - baseline_data[bin]
-                if delta > 0:
-                    print(
-                        f"detected: zn: {zone} bn: {bin} count: {raw_data[bin]} dt: {delta}"
-                    )
-                    diff[zone].append(delta)
-    elif sensor == TOFSensor.X:
-        for zone in zones:
-            # We only care about these zones because the X sensor is angled and
-            # most of the zones are always detecting obsticles.
-            if zone not in [5, 6, 7]:
+    config = TOF_DETECTION_CONFIG[sensor]
+    for zone in zones:
+        if zone not in config["zones"]:
+            continue
+        raw_data = histogram[zone]
+        baseline_data = baseline[zone]
+        for bin in bins:
+            if bin not in config["bins"]:
+                continue
+            if raw_data[bin] < config["threshold"]:
                 continue
-            raw_data = histogram[zone]
-            baseline_data = baseline[zone]
-            for bin in bins:
-                if bin not in range(10, 20):
-                    continue
-                # We need to ignore raw photon count below 10000 on the X as
-                # it becomes inconsistent to detect labware on the home position.
-                if raw_data[bin] < 10000:
-                    continue
-                delta = raw_data[bin] - baseline_data[bin]
-                if delta > 0:
-                    print(
-                        f"detected: zn: {zone} bn: {bin} count: {raw_data[bin]} dt: {delta}"
-                    )
-                    diff[zone].append(delta)
+            delta = raw_data[bin] - baseline_data[bin]
+            if delta > 0:
+                print(
+                    f"detected: zn: {zone} bn: {bin} count: {raw_data[bin]} dt: {delta}"
+                )
+                diff[zone].append(delta)
     return dict(diff)  # type: ignore