fix(api, hardware-testing): Fix the flex stacker PVT diagnostics script + increase stallguard 1 -> 2 (#18971)

Author: vegano1

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

@@ -123,7 +123,7 @@
 # Stallguard defaults
 STALLGUARD_CONFIG = {
     StackerAxis.X: StallGuardParams(StackerAxis.X, True, 0),
-    StackerAxis.Z: StallGuardParams(StackerAxis.Z, True, 0),
+    StackerAxis.Z: StallGuardParams(StackerAxis.Z, True, 2),
 }
 
 # Motion Parameter defaults

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

@@ -12,22 +12,26 @@
     FlexStacker,
     STACKER_MOTION_CONFIG,
     STALLGUARD_CONFIG,
+    MAX_TRAVEL,
 )
 from opentrons_shared_data.errors.exceptions import FlexStackerStallError
 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}
+# SGT Test range will be -n 1st index and +n 2nd index
+SGT_TEST_RANGE = (2, 3)
+# SGT threshold above this value should be ignored
+SGT_TEST_CUTOFF = 2
 
 
 def generate_test_thresholds(test_axis: StackerAxis) -> List[int]:
     """Generate test thresholds."""
     thresholds = []
+    low, high = SGT_TEST_RANGE
     default_threshold = STALLGUARD_CONFIG[test_axis].threshold
-    for sgt in range(default_threshold - 2, default_threshold + 3):
+    for sgt in range(default_threshold - low, default_threshold + high):
         thresholds.append(sgt)
     return thresholds
 
@@ -42,7 +46,7 @@ def build_csv_lines() -> List[Union[CSVLine, CSVLineRepeating]]:
     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]))
+            lines.append(CSVLine(_get_test_tag(axis, sgt), [bool, CSVResult]))
     return lines
 
 
@@ -60,7 +64,7 @@ async def test_stallguard(
             await stacker.move_axis(
                 test_axis,
                 Direction.RETRACT,
-                TEST_DISTANCE[test_axis],
+                MAX_TRAVEL[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,
@@ -72,10 +76,11 @@ async def test_stallguard(
 
         await stacker.home_axis(test_axis, Direction.EXTEND)
 
+        result = stall_detected or sgt > SGT_TEST_CUTOFF
         report(
             section,
             _get_test_tag(test_axis, sgt),
-            [CSVResult.from_bool(stall_detected)],
+            [stall_detected, CSVResult.from_bool(result)],
         )
 
     # Restore default stallguard threshold

hardware-testing/hardware_testing/scripts/stacker_tof_data_collection.py

@@ -1,18 +1,19 @@
 """Flex Stacker TOF Data Collection Script."""
 
-# TODO: Make the output of this match tools/tof-analysis/data/raw_data_frame.csv
 import argparse
 import asyncio
 import subprocess
 import re
 import time
 from datetime import datetime
 from typing import Any, Dict, List, Optional
+import uuid
 
 from hardware_testing import data
 from hardware_testing.opentrons_api.types import OT3Mount, Axis, Point
 from hardware_testing.opentrons_api.helpers_ot3 import build_async_ot3_hardware_api
 from opentrons.drivers.flex_stacker.types import StackerAxis, Direction, TOFSensor
+from opentrons.drivers.flex_stacker.utils import NUMBER_OF_BINS
 from opentrons.hardware_control.ot3api import OT3API
 
 
@@ -70,18 +71,33 @@ def __init__(
         self.axes = [Axis.X, Axis.Y, Axis.Z_L, Axis.Z_R]
         self.stackers: List[str] = []
         self.test_files: List[str] = []
+
         self.test_data = {
-            "Time": "None",
+            "Hash_id": "None",
+            "Date": "None",
+            "Test": "None",
+            "Labware_Name": "None",
+            "Stacker_SN": "None",
+            "Axis": "None",
+            "Platform_Position": "None",
+            "Labware_Num_X": "None",
+            "Labware_Num_Z": "None",
             "Sample": "None",
             "Zone": "None",
+            "Time": "None",
         }
+        # Bins 1-128
+        self.test_data.update(
+            {str(bin): "None" for bin in range(1, NUMBER_OF_BINS + 1)}
+        )
+
         self.tof_axes = {
-            "X-Axis": TOFSensor.X,
-            "Z-Axis": TOFSensor.Z,
+            "x": TOFSensor.X,
+            "z": TOFSensor.Z,
         }
         self.directions = {
-            "Retract": Direction.RETRACT,
-            "Extend": Direction.EXTEND,
+            "retract": Direction.RETRACT,
+            "extend": Direction.EXTEND,
         }
 
     async def test_setup(self) -> None:
@@ -113,66 +129,32 @@ def file_setup(self) -> None:
         self.test_date = "run-" + datetime.utcnow().strftime("%y-%m-%d")
         self.test_path = data.create_folder_for_test_data(self.test_name)
         if self.labware_amount == 0:
-            self.labware_name = "LW=baseline"
+            self.labware_name = "baseline"
             self.labware_amount_z = self.labware_amount
         elif self.labware_amount == 1:
-            self.labware_name = "LW=nest-96-pcr"
+            self.labware_name = "nest-96-pcr"
             self.labware_amount_z = self.labware_amount
         elif self.labware_amount == 3:
-            self.labware_name = "LW=tiprack"
+            self.labware_name = "tiprack"
             self.labware_amount = 1
             self.labware_amount_z = 3
         print("FILE PATH = ", self.test_path)
         print("FILE NAMES = ")
         for stacker in self.stackers:
-            self.test_tag_x_ret = f"x-axis_labx{self.labware_amount}_labz{self.labware_amount_z}_retract_{stacker}"
-            self.test_tag_x_ext = f"x-axis_labx{self.labware_amount}_labz{self.labware_amount_z}_extend_{stacker}"
-            self.test_tag_z_ret = f"z-axis_labx{self.labware_amount}_labz{self.labware_amount_z}_retract_{stacker}"
-            self.test_tag_z_ext = f"z-axis_labx{self.labware_amount}_labz{self.labware_amount_z}_extend_{stacker}"
-            test_file_x_ret = data.create_file_name(
-                self.labware_name, self.test_id, self.test_tag_x_ret
-            )
-            test_file_x_ext = data.create_file_name(
-                self.labware_name, self.test_id, self.test_tag_x_ext
-            )
-            test_file_z_ret = data.create_file_name(
-                self.labware_name, self.test_id, self.test_tag_z_ret
-            )
-            test_file_z_ext = data.create_file_name(
-                self.labware_name, self.test_id, self.test_tag_z_ext
-            )
-            data.append_data_to_file(
-                test_name=self.test_name,
-                run_id=self.test_date,
-                file_name=test_file_x_ret,
-                data=self.test_header,
+            self.test_tag = (
+                f"labx{self.labware_amount}_labz{self.labware_amount_z}_{stacker}"
             )
-            data.append_data_to_file(
-                test_name=self.test_name,
-                run_id=self.test_date,
-                file_name=test_file_x_ext,
-                data=self.test_header,
+            test_file = data.create_file_name(
+                self.labware_name, self.test_id, self.test_tag
             )
             data.append_data_to_file(
                 test_name=self.test_name,
                 run_id=self.test_date,
-                file_name=test_file_z_ret,
+                file_name=test_file,
                 data=self.test_header,
             )
-            data.append_data_to_file(
-                test_name=self.test_name,
-                run_id=self.test_date,
-                file_name=test_file_z_ext,
-                data=self.test_header,
-            )
-            self.test_files.append(test_file_x_ret)
-            self.test_files.append(test_file_x_ext)
-            self.test_files.append(test_file_z_ret)
-            self.test_files.append(test_file_z_ext)
-            print(test_file_x_ret)
-            print(test_file_x_ext)
-            print(test_file_z_ret)
-            print(test_file_z_ext)
+            self.test_files.append(test_file)
+            print(test_file)
 
     def dict_keys_to_line(self, dict: Dict[str, Any]) -> str:
         """Convert dict keys to CSV line."""
@@ -185,6 +167,7 @@ def dict_values_to_line(self, dict: Dict[str, Any]) -> str:
     async def read_stacker_tof(self) -> None:
         """Read the stacker TOF Sensor data."""
         for i in range(len(self.stackers)):
+            await self.api.attached_modules[i].home_all()  # type: ignore
             print(f"\n>> Stacker = {self.stackers[i]}")
             for axis, tof_axis in self.tof_axes.items():
                 for pos, direction in self.directions.items():
@@ -198,30 +181,36 @@ async def read_stacker_tof(self) -> None:
                             )
                             hist = await self.api.attached_modules[  # type: ignore
                                 i
-                            ]._driver.get_tof_histogram(tof_axis)
+                            ]._driver.get_tof_histogram(  # type: ignore
+                                tof_axis
+                            )
                             for zone, bins_list in hist.bins.items():
+                                date = datetime.now().strftime("%Y-%m-%d-%H-%M-%S")
                                 test_data = self.test_data.copy()
-                                test_data["Time"] = str(elapsed_time)
+                                test_data["Hash_id"] = str(uuid.uuid4())
+                                test_data["Date"] = str(date)
+                                test_data["Test"] = self.test_name
+                                test_data["Labware_Name"] = self.labware_name
+                                test_data["Stacker_SN"] = self.stackers[i]
+                                test_data["Axis"] = str(axis.lower())
+                                test_data["Platform_Position"] = pos.lower()
+                                test_data["Labware_Num_X"] = str(self.labware_amount)
+                                test_data["Labware_Num_Z"] = str(self.labware_amount_z)
                                 test_data["Sample"] = str(sample)
                                 test_data["Zone"] = str(zone)
-                                bins_dict = {
-                                    index: str(value)
-                                    for index, value in enumerate(bins_list)
-                                }
-                                test_data.update(bins_dict)  # type: ignore
+                                test_data["Time"] = str(elapsed_time)
+                                # Add the bin values
+                                test_data.update({str(i): str(v) for i, v in enumerate(bins_list, start=1)})  # type: ignore
+
+                                # Update the csv with new values
                                 test_data_str = self.dict_values_to_line(test_data)
                                 for test_file in self.test_files:
-                                    if (
-                                        self.stackers[i] in test_file
-                                        and axis.lower() in test_file
-                                        and pos.lower() in test_file
-                                    ):
-                                        data.append_data_to_file(
-                                            test_name=self.test_name,
-                                            run_id=self.test_date,
-                                            file_name=test_file,
-                                            data=test_data_str,
-                                        )
+                                    data.append_data_to_file(
+                                        test_name=self.test_name,
+                                        run_id=self.test_date,
+                                        file_name=test_file,
+                                        data=test_data_str,
+                                    )
                             time.sleep(self.interval)
                 print("")
 

hardware-testing/hardware_testing/tools/tof-analysis/tof_analysis.py

@@ -83,10 +83,10 @@ def _parse_tuple(arg: str) -> Tuple[int, int]:
 
 def _get_value_from_index(deviations: List[int], axis: str, platform: str) -> int:
     index_map = {
-        ("X", "extend"): 0,
-        ("X", "retract"): 1,
-        ("Z", "extend"): 2,
-        ("Z", "retract"): 3,
+        ("x", "extend"): 0,
+        ("x", "retract"): 1,
+        ("z", "extend"): 2,
+        ("z", "retract"): 3,
     }
     return deviations[index_map.get((axis, platform), 0)]
 
@@ -95,17 +95,20 @@ def _get_tuple_from_index(
     bins_list: List[Tuple[int, int]], axis: str, platform: str
 ) -> Tuple[int, int]:
     index_map = {
-        ("X", "extend"): 0,
-        ("X", "retract"): 1,
-        ("Z", "extend"): 2,
-        ("Z", "retract"): 3,
+        ("x", "extend"): 0,
+        ("x", "retract"): 1,
+        ("z", "extend"): 2,
+        ("z", "retract"): 3,
     }
     return bins_list[index_map.get((axis, platform), 0)]
 
 
 def convert_to_dict(obj: DefaultDict[Any, Any]) -> Dict[Any, Any]:
     """Convert a defaultdict to dict."""
-    return {k: convert_to_dict(v) for k, v in obj.items()}
+    return {
+        k: convert_to_dict(v) if isinstance(v, defaultdict) else v
+        for k, v in obj.items()
+    }
 
 
 def is_valid_row(axis: str, platform: str, zone: int, config: Dict[Any, Any]) -> bool:
@@ -339,7 +342,7 @@ def _get_visibility_mask(
                     ],
                 ]
                 fig.update_layout(
-                    title=f"TOF Sensor Baseline: {config['labware_list']} {axis}",
+                    title=f"TOF Sensor Baseline: {config['labware_list']} {axis} {args.graph_name}",
                     xaxis_title="Bins",
                     yaxis_title="Photon Count",
                     template="plotly_white",
@@ -376,7 +379,7 @@ def generate_baseline(args: argparse.Namespace) -> None:
 
     baselines = defaultdict(dict)  # type: ignore
     for axis, platform_data in histograms.items():
-        zone_count = zone_count_x if axis == "X" else zone_count_z
+        zone_count = zone_count_x if axis == "x" else zone_count_z
         for platform, zone_data in platform_data.items():
             deviation = _get_value_from_index(deviations, axis, platform)
             baseline = create_baseline(zone_data, zone_count, bin_count, deviation)
@@ -639,6 +642,12 @@ def main(args: argparse.Namespace) -> None:
         type=int,
         default=DEFAULT_MAX_SAMPLES,
     )
+    parser.add_argument(
+        "--graph-name",
+        help="Optional graph tag to add to the name",
+        type=str,
+        default="",
+    )
 
     args = parser.parse_args()
     main(args)