start of flex stacker PVT cycling test

Author: djgomez-opentrons

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

@@ -21,60 +21,76 @@
 from opentrons.hardware_control.modules import FlexStacker
 
 async def build_stacker_report(
-    is_simulating: bool, port: str = ""
+    is_simulating: bool, operator: str, port: str = "",
 ) -> Tuple[CSVReport, FlexStacker]:
     """Report setup for FLEX Stacker Cycle QC Test."""
     test_name = Path(__file__).parent.name.replace("_", "-")
-    ui.print_title(test_name.upper())
-
     stacker = await FlexStacker.build(port=port,
                                 usb_port=None,
                                 execution_manager=None,
-                                loop=asyncio.get_running_loop(),
+                                hw_control_loop=asyncio.get_running_loop(),
                                 simulating=is_simulating)
 
 
     report = build_report(test_name)
-    report.set_operator(
-        "simulating" if is_simulating else input("enter OPERATOR name: ")
-    )
+    report.set_operator("simulating" if is_simulating else operator)
     return report, stacker
 
 
 async def _main(cfg: TestConfig) -> None:
     # BUILD REPORT
-    port = []
+    ports = []
     for i in comports():
+        print(i)
         if i.vid == STACKER_VID and i.pid == STACKER_PID:
-            port.append(i.device)
-            break
-    assert port, "could not find connected FLEX Stacker"
-    report, stacker = await build_stacker_report(cfg.simulate, port=port[0])
+            ports.append(i.device)
+            print(f"Found FLEX Stacker on {i.device}")
+    assert ports, "could not find connected FLEX Stacker"
 
     if not cfg.simulate:
         print("Stopping the robot server")
         subprocess.run(["systemctl stop opentrons-robot-server"], shell=True)
 
-    device_info = await stacker._driver.get_device_info()
-    report.set_tag(device_info.sn if device_info.sn else "UNKNOWN")
+    test_name = Path(__file__).parent.name.replace("_", "-")
+    ui.print_title(test_name.upper())
+    operator = "simulating" if cfg.simulate else input("enter OPERATOR name: ")
+
+    stackers = {}
+    for p in ports:
+        report, stacker = await build_stacker_report(cfg.simulate, operator=operator, port=p)
+        device_sn = stacker.device_info['serial']
+        report.set_tag(device_sn if device_sn else "UNKNOWN")
+        stackers[device_sn] = (report, stacker)
 
-    # RUN TESTS
+    # # RUN TESTS
     try:
         for section, test_run in cfg.tests.items():
             ui.print_title(section.value)
-            await test_run(stacker, report, section.value)
+            tasks = []
+            for sn, (report, stacker) in stackers.items():
+                tasks.append(test_run(stacker, report, section.value))
+            await asyncio.gather(*tasks)  # Run all tasks concurrently
     except Exception as e:
         ui.print_error(f"An error occurred: {e}")
 
+    # try:
+    #     for section, test_run in cfg.tests.items():
+    #         ui.print_title(section.value)
+    #         await test_run(stacker, report, section.value)
+    # except Exception as e:
+    #     ui.print_error(f"An error occurred: {e}")
+
+
     # SAVE REPORT
     ui.print_title("DONE")
-    report.save_to_disk()
-    report.print_results()
+    for sn, (report, stacker) in stackers.items():
+        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 not cfg.simulate:
+    #     print("Starting the robot server")
+    #     subprocess.run(["systemctl restart opentrons-robot-server &"], shell=True)
 
 
 if __name__ == "__main__":

hardware-testing/hardware_testing/modules/flex_stacker_cycle_qc/test_cycles.py

@@ -1,5 +1,5 @@
 """Test Cycling."""
-from typing import List, Union
+from typing import List, Union, Tuple
 
 from hardware_testing.data import ui
 from hardware_testing.data.csv_report import (
@@ -9,34 +9,46 @@
     CSVResult,
 )
 
-from opentrons.drivers.flex_stacker.types import HardwareRevision
 from opentrons.hardware_control.modules import FlexStacker
 
+NUM_CYCLES = 10
 
 def build_csv_lines() -> List[Union[CSVLine, CSVLineRepeating]]:
     """Build CSV Lines."""
     return [
-        CSVLine("cycle", [CSVResult]),
+        CSVLine("cycle", [int, CSVResult]),
     ]
 
+def print_cycle(stacker: FlexStacker, cycle: int, total_cycles: int) -> None:
+    """Print formater for cycle progress"""
+    out_str = f"{stacker.device_info['serial']}: Cycle {cycle}/{total_cycles} Completed"
+    ui.print_info(out_str)
 
-# async def test_gcode(stacker: FlexStacker, report: CSVReport) -> None:
-#     """Send and receive response for GCODE M115."""
-#     success = True
-#     info = await stacker._driver.get_device_info()
-#     # TODO: update this with PVT/MP revisions
-#     if info.hw != HardwareRevision.DVT:
-#         ui.print_warning(f"Hardware Revision is {info.hw}, expected DVT")
-#     report(
-#         "CONNECTIVITY",
-#         "usb-get-device-info",
-#         [info.fw, info.hw, info.sn, CSVResult.from_bool(success)],
-#     )
+async def test_cycles(stacker: FlexStacker) -> Tuple[bool, int]:
+    """Test Cycling Labware on Stacker"""
+    cycles_completed = 0
+
+    for cycle in range(NUM_CYCLES):
+        try:
+            await stacker.dispense_labware(15)
+            await stacker.store_labware(15)
+        except Exception as e:
+            ui.print_error(f"{stacker.device_info['serial']}: An error occurred: {e}")
+
+        cycles_completed += 1
+        if (cycles_completed % 5 == 0) and (not cycles_completed==NUM_CYCLES):
+            print_cycle(stacker, cycles_completed, NUM_CYCLES)
+    
+    print_cycle(stacker, NUM_CYCLES, NUM_CYCLES)   
+    return (cycles_completed==NUM_CYCLES, cycles_completed)
 
 
 
 async def run(stacker: FlexStacker, report: CSVReport, section: str) -> None:
     """Run."""
-    ui.print_header("Cycle Test")
-    model = await stacker.get_model()
-    ui.print_info(f"Model: {model}")
+    ui.print_header(f"Cycle Test - {stacker.device_info['serial']}")
+
+    await stacker.home_all()
+    cycle_result, cycles_completed = await test_cycles(stacker)     
+    
+    report(section, "cycle", [cycles_completed, CSVResult.from_bool(cycle_result)])