feat(abr-testing): Add recording for API 2.20 features - error recovery and liquid probing

abr-testing/abr_testing/automation/google_sheets_tool.py

@@ -117,7 +117,7 @@ def batch_delete_rows(self, row_indices: List[int]) -> None:
     def batch_update_cells(
         self,
         data: List[List[Any]],
-        start_column: str,
+        start_column_index: Any,
         start_row: int,
         sheet_id: str,
     ) -> None:
@@ -132,7 +132,8 @@ def column_letter_to_index(column_letter: str) -> int:
 
         requests = []
         user_entered_value: Dict[str, Any] = {}
-        start_column_index = column_letter_to_index(start_column) - 1
+        if type(start_column_index) == str:
+            start_column_index = column_letter_to_index(start_column_index) - 1
 
         for col_offset, col_values in enumerate(data):
             column_index = start_column_index + col_offset
@@ -223,9 +224,9 @@ def get_sheet_by_name(self, title: str) -> None:
             )
 
     def token_check(self) -> None:
-        """Check if still credentials are still logged in."""
-        if self.credentials.access_token_expired:
-            self.gc.login()
+        """Check if credentials are still valid and refresh if expired."""
+        if self.credentials.expired:
+            self.credentials.refresh()  # Refresh the credentials
 
     def get_row_index_with_value(self, some_string: str, col_num: int) -> Any:
         """Find row index of string by looking in specific column."""

abr-testing/abr_testing/automation/jira_tool.py

@@ -51,8 +51,12 @@ def issues_on_board(self, project_key: str) -> List[List[Any]]:
     def match_issues(self, issue_ids: List[List[str]], ticket_summary: str) -> List:
         """Matches related ticket ID's."""
         to_link = []
-        error = ticket_summary.split("_")[3]
-        robot = ticket_summary.split("_")[0]
+        try:
+            error = ticket_summary.split("_")[3]
+            robot = ticket_summary.split("_")[0]
+        except IndexError:
+            error = ""
+            robot = ""
         # for every issue see if both match, if yes then grab issue ID and add it to a list
         for issue in issue_ids:
             summary = issue[1]

abr-testing/abr_testing/data_collection/abr_google_drive.py

@@ -65,13 +65,7 @@ def create_data_dictionary(
             left_pipette = file_results.get("left", "")
             right_pipette = file_results.get("right", "")
             extension = file_results.get("extension", "")
-            (
-                num_of_errors,
-                error_type,
-                error_code,
-                error_instrument,
-                error_level,
-            ) = read_robot_logs.get_error_info(file_results)
+            error_dict = read_robot_logs.get_error_info(file_results)
 
             all_modules = get_modules(file_results)
 
@@ -99,7 +93,7 @@ def create_data_dictionary(
                 pass  # Handle datetime parsing errors if necessary
 
             if run_time_min > 0:
-                row = {
+                run_row = {
                     "Robot": robot,
                     "Run_ID": run_id,
                     "Protocol_Name": protocol_name,
@@ -108,15 +102,13 @@ def create_data_dictionary(
                     "Start_Time": start_time_str,
                     "End_Time": complete_time_str,
                     "Run_Time (min)": run_time_min,
-                    "Errors": num_of_errors,
-                    "Error_Code": error_code,
-                    "Error_Type": error_type,
-                    "Error_Instrument": error_instrument,
-                    "Error_Level": error_level,
+                }
+                instrument_row = {
                     "Left Mount": left_pipette,
                     "Right Mount": right_pipette,
                     "Extension": extension,
                 }
+                row = {**run_row, **error_dict, **instrument_row}
                 tc_dict = read_robot_logs.thermocycler_commands(file_results)
                 hs_dict = read_robot_logs.hs_commands(file_results)
                 tm_dict = read_robot_logs.temperature_module_commands(file_results)
@@ -128,7 +120,11 @@ def create_data_dictionary(
                     "Average Temp (oC)": "",
                     "Average RH(%)": "",
                 }
-                row_for_lpc = {**row, **all_modules, **notes}
+                row_for_lpc = {
+                    **row,
+                    **all_modules,
+                    **notes,
+                }
                 row_2 = {
                     **row,
                     **all_modules,

abr-testing/abr_testing/data_collection/abr_robot_error.py

@@ -62,15 +62,17 @@ def compare_current_trh_to_average(
     df_all_run_data["Start_Time"] = pd.to_datetime(
         df_all_run_data["Start_Time"], format="mixed", utc=True
     ).dt.tz_localize(None)
-    df_all_run_data["Errors"] = pd.to_numeric(df_all_run_data["Errors"])
+    df_all_run_data["Run Ending Error"] = pd.to_numeric(
+        df_all_run_data["Run Ending Error"]
+    )
     df_all_run_data["Average Temp (oC)"] = pd.to_numeric(
         df_all_run_data["Average Temp (oC)"]
     )
     common_filters = (
         (df_all_run_data["Robot"] == robot)
         & (df_all_run_data["Start_Time"] >= weeks_ago_3)
         & (df_all_run_data["Start_Time"] <= start_time)
-        & (df_all_run_data["Errors"] < 1)
+        & (df_all_run_data["Run Ending Error"] < 1)
         & (df_all_run_data["Average Temp (oC)"] > 1)
     )
 
@@ -122,7 +124,7 @@ def compare_lpc_to_historical_data(
         & (df_lpc_data["Robot"] == robot)
         & (df_lpc_data["Module"] == labware_dict["Module"])
         & (df_lpc_data["Adapter"] == labware_dict["Adapter"])
-        & (df_lpc_data["Errors"] < 1)
+        & (df_lpc_data["Run Ending Error"] < 1)
     ]
     # Converts coordinates to floats and finds averages.
     x_float = [float(value) for value in relevant_lpc["X"]]
@@ -330,18 +332,17 @@ def get_run_error_info_from_robot(
         ip, results, storage_directory
     )
     # Error Printout
-    (
-        num_of_errors,
-        error_type,
-        error_code,
-        error_instrument,
-        error_level,
-    ) = read_robot_logs.get_error_info(results)
+    error_dict = read_robot_logs.get_error_info(results)
+    error_level = error_dict["Error_Level"]
+    error_type = error_dict["Error_Type"]
+    error_code = error_dict["Error_Code"]
+    error_instrument = error_dict["Error_Instrument"]
     # JIRA Ticket Fields
+
     failure_level = "Level " + str(error_level) + " Failure"
 
     components = [failure_level, "Flex-RABR"]
-    components = match_error_to_component("RABR", error_type, components)
+    components = match_error_to_component("RABR", str(error_type), components)
     print(components)
     affects_version = results["API_Version"]
     parent = results.get("robot_name", "")

abr-testing/abr_testing/data_collection/read_robot_logs.py

@@ -76,6 +76,36 @@ def command_time(command: Dict[str, str]) -> float:
     return start_to_complete
 
 
+def count_command_in_run_data(
+    commands: List[Dict[str, Any]], command_of_interest: str, find_avg_time: bool
+) -> Tuple[int, float]:
+    """Count number of times command occurs in a run."""
+    total_command = 0
+    total_time = 0.0
+    for command in commands:
+        command_type = command["commandType"]
+        if command_type == command_of_interest:
+            total_command += 1
+            if find_avg_time:
+                started_at = command.get("startedAt", "")
+                completed_at = command.get("completedAt", "")
+
+                if started_at and completed_at:
+                    try:
+                        start_time = datetime.strptime(
+                            started_at, "%Y-%m-%dT%H:%M:%S.%f%z"
+                        )
+                        end_time = datetime.strptime(
+                            completed_at, "%Y-%m-%dT%H:%M:%S.%f%z"
+                        )
+                        total_time += (end_time - start_time).total_seconds()
+                    except ValueError:
+                        # Handle case where date parsing fails
+                        pass
+    avg_time = total_time / total_command if total_command > 0 else 0.0
+    return total_command, avg_time
+
+
 def instrument_commands(file_results: Dict[str, Any]) -> Dict[str, float]:
     """Count number of pipette and gripper commands per run."""
     pipettes = file_results.get("pipettes", "")
@@ -89,6 +119,7 @@ def instrument_commands(file_results: Dict[str, Any]) -> Dict[str, float]:
     right_pipette_id = ""
     left_pipette_id = ""
     gripper_pickups = 0.0
+    avg_liquid_probe_time_sec = 0.0
     # Match pipette mount to id
     for pipette in pipettes:
         if pipette["mount"] == "right":
@@ -120,6 +151,9 @@ def instrument_commands(file_results: Dict[str, Any]) -> Dict[str, float]:
             and command["params"]["strategy"] == "usingGripper"
         ):
             gripper_pickups += 1
+    liquid_probes, avg_liquid_probe_time_sec = count_command_in_run_data(
+        commandData, "liquidProbe", True
+    )
     pipette_dict = {
         "Left Pipette Total Tip Pick Up(s)": left_tip_pick_up,
         "Left Pipette Total Aspirates": left_aspirate,
@@ -128,6 +162,8 @@ def instrument_commands(file_results: Dict[str, Any]) -> Dict[str, float]:
         "Right Pipette Total Aspirates": right_aspirate,
         "Right Pipette Total Dispenses": right_dispense,
         "Gripper Pick Ups": gripper_pickups,
+        "Total Liquid Probes": liquid_probes,
+        "Average Liquid Probe Time (sec)": avg_liquid_probe_time_sec,
     }
     return pipette_dict
 
@@ -362,50 +398,58 @@ def create_abr_data_sheet(
     return sheet_location
 
 
-def get_error_info(file_results: Dict[str, Any]) -> Tuple[int, str, str, str, str]:
+def get_error_info(file_results: Dict[str, Any]) -> Dict[str, Any]:
     """Determines if errors exist in run log and documents them."""
-    error_levels = []
-    error_level = ""
     # Read error levels file
     with open(ERROR_LEVELS_PATH, "r") as error_file:
-        error_levels = list(csv.reader(error_file))
-    num_of_errors = len(file_results["errors"])
-    if num_of_errors == 0:
-        error_type = ""
-        error_code = ""
-        error_instrument = ""
-        error_level = ""
-        return 0, error_type, error_code, error_instrument, error_level
+        error_levels = {row[1]: row[4] for row in csv.reader(error_file)}
+    # Initialize Variables
+    recoverable_errors: Dict[str, int] = dict()
+    total_recoverable_errors = 0
+    end_run_errors = len(file_results["errors"])
     commands_of_run: List[Dict[str, Any]] = file_results.get("commands", [])
+    error_recovery = file_results.get("hasEverEnteredErrorRecovery", False)
+    # Count recoverable errors
+    if error_recovery:
+        for command in commands_of_run:
+            error_info = command.get("error", {})
+            if error_info.get("isDefined"):
+                total_recoverable_errors += 1
+                error_type = error_info.get("errorType", "")
+                recoverable_errors[error_type] = (
+                    recoverable_errors.get(error_type, 0) + 1
+                )
+    # Get run-ending error info
     try:
-        run_command_error: Dict[str, Any] = commands_of_run[-1]
-        error_str: int = len(run_command_error.get("error", ""))
-    except IndexError:
-        error_str = 0
-    if error_str > 1:
-        error_type = run_command_error["error"].get("errorType", "")
+        run_command_error = commands_of_run[-1]["error"]
+        error_type = run_command_error.get("errorType", "")
         if error_type == "PythonException":
-            # Reassign error_type to be more descriptive
-            error_type = run_command_error.get("detail", "").split(":")[0]
-        error_code = run_command_error["error"].get("errorCode", "")
+            error_type = commands_of_run[-1].get("detail", "").split(":")[0]
+        error_code = run_command_error.get("errorCode", "")
+        error_instrument = run_command_error.get("errorInfo", {}).get(
+            "node", run_command_error.get("errorInfo", {}).get("port", "")
+        )
+    except (IndexError, KeyError):
         try:
-            # Instrument Error
-            error_instrument = run_command_error["error"]["errorInfo"]["node"]
-        except KeyError:
-            # Module
-            error_instrument = run_command_error["error"]["errorInfo"].get("port", "")
-    else:
-        error_type = file_results["errors"][0]["errorType"]
-        error_code = file_results["errors"][0]["errorCode"]
-        error_instrument = file_results["errors"][0]["detail"]
-    for error in error_levels:
-        code_error = error[1]
-        if code_error == error_code:
-            error_level = error[4]
-    if len(error_level) < 1:
-        error_level = str(4)
-
-    return num_of_errors, error_type, error_code, error_instrument, error_level
+            error_details = file_results.get("errors", [{}])[0]
+        except IndexError:
+            error_details = {}
+        error_type = error_details.get("errorType", "")
+        error_code = error_details.get("errorCode", "")
+        error_instrument = error_details.get("detail", "")
+    # Determine error level
+    error_level = error_levels.get(error_code, "4")
+    # Create dictionary with all error descriptions
+    error_dict = {
+        "Total Recoverable Error(s)": total_recoverable_errors,
+        "Recoverable Error(s) Description": recoverable_errors,
+        "Run Ending Error": end_run_errors,
+        "Error_Code": error_code,
+        "Error_Type": error_type,
+        "Error_Instrument": error_instrument,
+        "Error_Level": error_level,
+    }
+    return error_dict
 
 
 def write_to_local_and_google_sheet(
@@ -570,10 +614,10 @@ def get_calibration_offsets(
 def get_logs(storage_directory: str, ip: str) -> List[str]:
     """Get Robot logs."""
     log_types: List[Dict[str, Any]] = [
-        {"log type": "api.log", "records": 1000},
+        {"log type": "api.log", "records": 10000},
         {"log type": "server.log", "records": 10000},
         {"log type": "serial.log", "records": 10000},
-        {"log type": "touchscreen.log", "records": 1000},
+        {"log type": "touchscreen.log", "records": 10000},
     ]
     all_paths = []
     for log_type in log_types:

abr-testing/abr_testing/tools/sync_abr_sheet.py

@@ -15,6 +15,7 @@ def determine_lifetime(abr_google_sheet: Any) -> None:
     """Record lifetime % of robot, pipettes, and gripper per run."""
     # Get all data
     headers = abr_google_sheet.get_row(1)
+    lifetime_index = headers.index("Robot Lifetime (%)")
     all_google_data = abr_google_sheet.get_all_data(expected_headers=headers)
     # Convert dictionary to pandas dataframe
     df_sheet_data = pd.DataFrame.from_dict(all_google_data)
@@ -92,7 +93,9 @@ def determine_lifetime(abr_google_sheet: Any) -> None:
                 [right_pipette_percent_lifetime],
                 [gripper_percent_lifetime],
             ]
-            abr_google_sheet.batch_update_cells(update_list, "AV", row_num, "0")
+            abr_google_sheet.batch_update_cells(
+                update_list, lifetime_index, row_num, "0"
+            )
             print(f"Updated row {row_num} for run: {run_id}")
 
 
@@ -101,6 +104,10 @@ def compare_run_to_temp_data(
 ) -> None:
     """Read ABR Data and compare robot and timestamp columns to temp data."""
     row_update = 0
+    # Get column number for average temp and rh
+    headers = google_sheet.get_row(1)
+    temp_index = headers.index("Average Temp (oC)") + 1
+    rh_index = headers.index("Average RH(%)") + 1
     for run in abr_data:
         run_id = run["Run_ID"]
         try:
@@ -129,9 +136,9 @@ def compare_run_to_temp_data(
                 avg_humidity = mean(rel_hums)
                 row_num = google_sheet.get_row_index_with_value(run_id, 2)
                 # Write average temperature
-                google_sheet.update_cell("Sheet1", row_num, 46, avg_temps)
+                google_sheet.update_cell("Sheet1", row_num, temp_index, avg_temps)
                 # Write average humidity
-                google_sheet.update_cell("Sheet1", row_num, 47, avg_humidity)
+                google_sheet.update_cell("Sheet1", row_num, rh_index, avg_humidity)
                 row_update += 1
                 # TODO: Write averages to google sheet
                 print(f"Updated row {row_num}.")