feat (abr-testing): simulate protocols with csv parameter (#16685)

# Overview

Enabled support for simulating protocols with csv parameters

abr-testing/abr_testing/data_collection/abr_calibration_logs.py

@@ -363,3 +363,4 @@ def run(
     folder_name = args.folder_name[0]
     google_sheet_name = args.google_sheet_name[0]
     email = args.email[0]
+    run(storage_directory, folder_name, google_sheet_name, email)

abr-testing/abr_testing/data_collection/read_robot_logs.py

@@ -110,6 +110,7 @@ def identify_labware_ids(
     file_results: Dict[str, Any], labware_name: Optional[str]
 ) -> List[str]:
     """Determine what type of labware is being picked up."""
+    list_of_labware_ids: List[str] = []
     if labware_name:
         labwares = file_results.get("labware", "")
         list_of_labware_ids = []
@@ -341,8 +342,9 @@ def hs_commands(file_results: Dict[str, Any]) -> Dict[str, float]:
             )
     if temp_time is not None and deactivate_time is None:
         # If heater shaker module is not deactivated, protocol completedAt time stamp used.
+        default = commandData[len(commandData) - 1].get("completedAt")
         protocol_end = datetime.strptime(
-            file_results.get("completedAt", ""), "%Y-%m-%dT%H:%M:%S.%f%z"
+            file_results.get("completedAt", default), "%Y-%m-%dT%H:%M:%S.%f%z"
         )
         temp_duration = (protocol_end - temp_time).total_seconds()
         hs_temps[hs_temp] = hs_temps.get(hs_temp, 0.0) + temp_duration
@@ -389,8 +391,9 @@ def temperature_module_commands(file_results: Dict[str, Any]) -> Dict[str, Any]:
                 tm_temps[tm_temp] = tm_temps.get(tm_temp, 0.0) + temp_duration
     if temp_time is not None and deactivate_time is None:
         # If temperature module is not deactivated, protocol completedAt time stamp used.
+        default = commandData[len(commandData) - 1].get("completedAt")
         protocol_end = datetime.strptime(
-            file_results.get("completedAt", ""), "%Y-%m-%dT%H:%M:%S.%f%z"
+            file_results.get("completedAt", default), "%Y-%m-%dT%H:%M:%S.%f%z"
         )
         temp_duration = (protocol_end - temp_time).total_seconds()
         tm_temps[tm_temp] = tm_temps.get(tm_temp, 0.0) + temp_duration
@@ -473,15 +476,17 @@ def thermocycler_commands(file_results: Dict[str, Any]) -> Dict[str, float]:
                 block_temps[block_temp] = block_temps.get(block_temp, 0.0) + block_time
     if block_on_time is not None and block_off_time is None:
         # If thermocycler block not deactivated protocol completedAt time stamp used
+        default = commandData[len(commandData) - 1].get("completedAt")
         protocol_end = datetime.strptime(
-            file_results.get("completedAt", ""), "%Y-%m-%dT%H:%M:%S.%f%z"
+            file_results.get("completedAt", default), "%Y-%m-%dT%H:%M:%S.%f%z"
         )
         temp_duration = (protocol_end - block_on_time).total_seconds()
-        block_temps[block_temp] = block_temps.get(block_temp, 0.0) + temp_duration
+
     if lid_on_time is not None and lid_off_time is None:
         # If thermocycler lid not deactivated protocol completedAt time stamp used
+        default = commandData[len(commandData) - 1].get("completedAt")
         protocol_end = datetime.strptime(
-            file_results.get("completedAt", ""), "%Y-%m-%dT%H:%M:%S.%f%z"
+            file_results.get("completedAt", default), "%Y-%m-%dT%H:%M:%S.%f%z"
         )
         temp_duration = (protocol_end - lid_on_time).total_seconds()
         lid_temps[lid_temp] = block_temps.get(lid_temp, 0.0) + temp_duration

abr-testing/abr_testing/protocol_simulation/abr_sim_check.py

@@ -5,9 +5,9 @@
 from pathlib import Path
 
 
-def run(file_to_simulate: str) -> None:
+def run(file_to_simulate: Path) -> None:
     """Simulate protocol and raise errors."""
-    protocol_name = Path(file_to_simulate).stem
+    protocol_name = file_to_simulate.stem
     try:
         simulation_metrics.main(file_to_simulate, False)
     except Exception:
@@ -29,6 +29,6 @@ def run(file_to_simulate: str) -> None:
             if file.endswith(".py"):  # If it's a Python file
                 if file in exclude:
                     continue
-                file_path = os.path.join(root, file)
-                print(f"Simulating protocol: {file_path}")
+                file_path = Path(os.path.join(root, file))
+                print(f"Simulating protocol: {file_path.stem}")
                 run(file_path)

abr-testing/abr_testing/protocol_simulation/simulation_metrics.py

@@ -241,6 +241,7 @@ def parse_results_volume(
         "Right Pipette Total Aspirates",
         "Right Pipette Total Dispenses",
         "Gripper Pick Ups",
+        "Gripper Pick Ups of opentrons_tough_pcr_auto_sealing_lid",
         "Total Liquid Probes",
         "Average Liquid Probe Time (sec)",
     ]
@@ -302,6 +303,7 @@ def parse_results_volume(
     total_time_row.append(str(end_time - start_time))
 
     for metric in metrics:
+        print(f"Dictionary: {metric}\n\n")
         for cmd in metric.keys():
             values_row.append(str(metric[cmd]))
     return (
@@ -320,15 +322,16 @@ def parse_results_volume(
 
 
 def main(
-    protocol_file_path_name: str,
+    protocol_file_path: Path,
     save: bool,
     storage_directory: str = os.curdir,
     google_sheet_name: str = "",
+    parameters: List[str] = [],
 ) -> None:
     """Main module control."""
     sys.exit = mock_exit  # Replace sys.exit with the mock function
     # Read file path from arguments
-    protocol_file_path = Path(protocol_file_path_name)
+    # protocol_file_path = Path(protocol_file_path_name)
     protocol_name = protocol_file_path.stem
     print("Simulating", protocol_name)
     file_date = datetime.now()
@@ -344,26 +347,58 @@ def main(
                 )
                 json_file_output = open(json_file_path, "wb+")
                 # log_output_file = f"{protocol_name}_log"
-                ctx.invoke(
-                    analyze,
-                    files=[protocol_file_path],
-                    json_output=json_file_output,
-                    human_json_output=None,
-                    log_output=error_output,
-                    log_level="ERROR",
-                    check=False,
-                )
+                if parameters:
+                    print(f"Parameter: {parameters[0]}\n")
+                    csv_params = {}
+                    csv_params["parameters_csv"] = parameters[0]
+                    rtp_json = json.dumps(csv_params)
+                    ctx.invoke(
+                        analyze,
+                        files=[protocol_file_path],
+                        rtp_files=rtp_json,
+                        json_output=json_file_output,
+                        human_json_output=None,
+                        log_output=error_output,
+                        log_level="ERROR",
+                        check=False,
+                    )
+
+                else:
+                    ctx.invoke(
+                        analyze,
+                        files=[protocol_file_path],
+                        json_output=json_file_output,
+                        human_json_output=None,
+                        log_output=error_output,
+                        log_level="ERROR",
+                        check=False,
+                    )
                 json_file_output.close()
             else:
-                ctx.invoke(
-                    analyze,
-                    files=[protocol_file_path],
-                    json_output=None,
-                    human_json_output=None,
-                    log_output=error_output,
-                    log_level="ERROR",
-                    check=True,
-                )
+                if parameters:
+                    csv_params = {}
+                    csv_params["parameters_csv"] = parameters[0]
+                    rtp_json = json.dumps(csv_params)
+                    ctx.invoke(
+                        analyze,
+                        files=[protocol_file_path],
+                        rtp_files=rtp_json,
+                        json_output=None,
+                        human_json_output=None,
+                        log_output=error_output,
+                        log_level="ERROR",
+                        check=True,
+                    )
+                else:
+                    ctx.invoke(
+                        analyze,
+                        files=[protocol_file_path],
+                        json_output=None,
+                        human_json_output=None,
+                        log_output=error_output,
+                        log_level="ERROR",
+                        check=True,
+                    )
 
     except SystemExit as e:
         print(f"SystemExit caught with code: {e}")
@@ -395,6 +430,7 @@ def main(
             credentials_path, google_sheet_name, 0
         )
         google_sheet.write_to_row([])
+
         for row in parse_results_volume(
             json_file_path,
             protocol_name,
@@ -428,13 +464,15 @@ def main(
         "protocol_file_path",
         metavar="PROTOCOL_FILE_PATH",
         type=str,
-        nargs=1,
+        nargs="*",
         help="Path to protocol file",
     )
     args = parser.parse_args()
     storage_directory = args.storage_directory[0]
     sheet_name = args.sheet_name[0]
     protocol_file_path: str = args.protocol_file_path[0]
+    parameters: List[str] = args.protocol_file_path[1:]
+    print(parameters)
     SETUP = True
     while SETUP:
         print(
@@ -445,7 +483,7 @@ def main(
             choice = ""
             while not choice:
                 choice = input(
-                    "Remove air_gap commands to ensure accurate results? (Y/N): "
+                    "Remove air_gap commands to ensure accurate results: (continue)? (Y/N): "
                 )
                 if choice.upper() == "Y":
                     SETUP = False
@@ -462,11 +500,20 @@ def main(
     # Change api level
     if CLEAN_PROTOCOL:
         set_api_level(protocol_file_path)
-        main(
-            protocol_file_path,
-            True,
-            storage_directory,
-            sheet_name,
-        )
+        if parameters:
+            main(
+                Path(protocol_file_path),
+                True,
+                storage_directory,
+                sheet_name,
+                parameters=parameters,
+            )
+        else:
+            main(
+                protocol_file_path=Path(protocol_file_path),
+                save=True,
+                storage_directory=storage_directory,
+                google_sheet_name=sheet_name,
+            )
     else:
         sys.exit(0)