ABR LPC Automated Logging

abr-testing/abr_testing/data_collection/abr_google_drive.py

@@ -32,9 +32,10 @@ def create_data_dictionary(
     runs_to_save: Union[Set[str], str],
     storage_directory: str,
     issue_url: str,
-) -> Tuple[Dict[Any, Dict[str, Any]], List]:
+) -> Tuple[Dict[str, Dict[str, Any]], List[str], Dict[str, Dict[str, Any]], List[str]]:
     """Pull data from run files and format into a dictionary."""
-    runs_and_robots = {}
+    runs_and_robots: Dict[Any, Dict[str, Any]] = {}
+    runs_and_lpc: Dict[Any, Dict[str, Any]] = {}
     for filename in os.listdir(storage_directory):
         file_path = os.path.join(storage_directory, filename)
         if file_path.endswith(".json"):
@@ -108,6 +109,7 @@ def create_data_dictionary(
                 hs_dict = read_robot_logs.hs_commands(file_results)
                 tm_dict = read_robot_logs.temperature_module_commands(file_results)
                 notes = {"Note1": "", "Jira Link": issue_url}
+                row_for_lpc = {**row, **all_modules, **notes}
                 row_2 = {
                     **row,
                     **all_modules,
@@ -116,11 +118,15 @@ def create_data_dictionary(
                     **tm_dict,
                     **tc_dict,
                 }
-                headers = list(row_2.keys())
+                headers: List[str] = list(row_2.keys())
                 runs_and_robots[run_id] = row_2
+                # LPC Data Recording
+                runs_and_lpc, headers_lpc = read_robot_logs.lpc_data(
+                    file_results, row_for_lpc, runs_and_lpc
+                )
             else:
                 continue
-    return runs_and_robots, headers
+    return runs_and_robots, headers, runs_and_lpc, headers_lpc
 
 
 if __name__ == "__main__":
@@ -164,7 +170,6 @@ def create_data_dictionary(
     google_sheet = google_sheets_tool.google_sheet(
         credentials_path, google_sheet_name, 0
     )
-    google_sheet_lpc = google_sheets_tool.google_sheet(credentials_path, "ABR-LPC", 0)
 
     run_ids_on_gs = google_sheet.get_column(2)
     run_ids_on_gs = set(run_ids_on_gs)
@@ -178,9 +183,14 @@ def create_data_dictionary(
         run_ids_on_gd, run_ids_on_gs
     )
     # Add missing runs to google sheet
-    runs_and_robots, headers = create_data_dictionary(
+    runs_and_robots, headers, runs_and_lpc, headers_lpc = create_data_dictionary(
         missing_runs_from_gs, storage_directory, ""
     )
     read_robot_logs.write_to_local_and_google_sheet(
         runs_and_robots, storage_directory, google_sheet_name, google_sheet, headers
     )
+    # Add LPC to google sheet
+    google_sheet_lpc = google_sheets_tool.google_sheet(credentials_path, "ABR-LPC", 0)
+    read_robot_logs.write_to_local_and_google_sheet(
+        runs_and_lpc, storage_directory, "ABR-LPC", google_sheet_lpc, headers_lpc
+    )

abr-testing/abr_testing/data_collection/abr_robot_error.py

@@ -7,9 +7,7 @@
 import shutil
 import os
 import subprocess
-import json
 import sys
-import gspread  # type: ignore[import]
 
 
 def get_error_runs_from_robot(ip: str) -> List[str]:
@@ -183,35 +181,31 @@ def get_error_info_from_robot(
     # CONNECT TO GOOGLE DRIVE
     credentials_path = os.path.join(storage_directory, "credentials.json")
     google_sheet_name = "ABR-run-data"
-    try:
-        google_drive = google_drive_tool.google_drive(
-            credentials_path,
-            "1Cvej0eadFOTZr9ILRXJ0Wg65ymOtxL4m",
-            "[email protected]",
-        )
-        print("Connected to google drive.")
-    except json.decoder.JSONDecodeError:
-        print(
-            "Credential file is damaged. Get from https://console.cloud.google.com/apis/credentials"
-        )
-        sys.exit()
+    google_drive = google_drive_tool.google_drive(
+        credentials_path,
+        "1Cvej0eadFOTZr9ILRXJ0Wg65ymOtxL4m",
+        "[email protected]",
+    )
     # CONNECT TO GOOGLE SHEET
-    try:
-        google_sheet = google_sheets_tool.google_sheet(
-            credentials_path, google_sheet_name, 0
-        )
-        print(f"Connected to google sheet: {google_sheet_name}")
-    except gspread.exceptions.APIError:
-        print("ERROR: Check google sheet name. Check credentials file.")
-        sys.exit()
+    google_sheet = google_sheets_tool.google_sheet(
+        credentials_path, google_sheet_name, 0
+    )
     # WRITE ERRORED RUN TO GOOGLE SHEET
     error_run_log = os.path.join(error_folder_path, os.path.basename(run_log_file_path))
     google_drive.upload_file(error_run_log)
     run_id = os.path.basename(error_run_log).split("_")[1].split(".")[0]
-    runs_and_robots, headers = abr_google_drive.create_data_dictionary(
-        run_id, error_folder_path, issue_url
-    )
+    (
+        runs_and_robots,
+        headers,
+        runs_and_lpc,
+        headers_lpc,
+    ) = abr_google_drive.create_data_dictionary(run_id, error_folder_path, issue_url)
     read_robot_logs.write_to_local_and_google_sheet(
         runs_and_robots, storage_directory, google_sheet_name, google_sheet, headers
     )
     print("Wrote run to ABR-run-data")
+    # Add LPC to google sheet
+    google_sheet_lpc = google_sheets_tool.google_sheet(credentials_path, "ABR-LPC", 0)
+    read_robot_logs.write_to_local_and_google_sheet(
+        runs_and_lpc, storage_directory, "ABR-LPC", google_sheet_lpc, headers_lpc
+    )

abr-testing/abr_testing/data_collection/read_robot_logs.py

@@ -15,11 +15,17 @@
 import sys
 
 
-def lpc_data(file_results: Dict[str, Any], protocol_info: Dict) -> List[Dict[str, Any]]:
+def lpc_data(
+    file_results: Dict[str, Any],
+    protocol_info: Dict[str, Any],
+    runs_and_lpc: Dict[str, Any],
+) -> Tuple[Dict[str, Dict[str, Any]], List[str]]:
     """Get labware offsets from one run log."""
     offsets = file_results.get("labwareOffsets", "")
-    all_offsets: List[Dict[str, Any]] = []
+    n = 0
+    # TODO: per UNIQUE slot AND LABWARE TYPE only keep the most recent LPC recording
     if len(offsets) > 0:
+        unique_offsets: Dict[Any, Any] = {}
         for offset in offsets:
             labware_type = offset.get("definitionUri", "")
             slot = offset["location"].get("slotName", "")
@@ -29,19 +35,32 @@ def lpc_data(file_results: Dict[str, Any], protocol_info: Dict) -> List[Dict[str
             y_offset = offset["vector"].get("y", 0.0)
             z_offset = offset["vector"].get("z", 0.0)
             created_at = offset.get("createdAt", "")
-            row = {
-                "createdAt": created_at,
-                "Labware Type": labware_type,
-                "Slot": slot,
-                "Module": module_location,
-                "Adapter": adapter,
-                "X": x_offset,
-                "Y": y_offset,
-                "Z": z_offset,
-            }
-            row2 = {**protocol_info, **row}
-            all_offsets.append(row2)
-    return all_offsets
+            if (
+                slot,
+                labware_type,
+            ) not in unique_offsets or created_at > unique_offsets[
+                (slot, labware_type)
+            ][
+                "createdAt"
+            ]:
+                unique_offsets[(slot, labware_type)] = {
+                    **protocol_info,
+                    "createdAt": created_at,
+                    "Labware Type": labware_type,
+                    "Slot": slot,
+                    "Module": module_location,
+                    "Adapter": adapter,
+                    "X": x_offset,
+                    "Y": y_offset,
+                    "Z": z_offset,
+                }
+    for item in unique_offsets:
+        run_id = protocol_info["Run_ID"] + "_" + str(n)
+        runs_and_lpc[run_id] = unique_offsets[item]
+        n += 1
+    headers_lpc = list(unique_offsets[(slot, labware_type)].keys())
+
+    return runs_and_lpc, headers_lpc
 
 
 def command_time(command: Dict[str, str]) -> Tuple[float, float]:
@@ -323,13 +342,12 @@ def write_to_local_and_google_sheet(
     """Write data dictionary to google sheet and local csv."""
     sheet_location = os.path.join(storage_directory, file_name)
     file_exists = os.path.exists(sheet_location) and os.path.getsize(sheet_location) > 0
-    list_of_runs = list(runs_and_robots.keys())
     with open(sheet_location, "a", newline="") as f:
         writer = csv.writer(f)
         if not file_exists:
             writer.writerow(header)
-        for run in range(len(list_of_runs)):
-            row = runs_and_robots[list_of_runs[run]].values()
+        for run in runs_and_robots:
+            row = runs_and_robots[run].values()
             row_list = list(row)
             writer.writerow(row_list)
             google_sheet.write_header(header)

abr-testing/abr_testing/data_collection/single_run_log_reader.py

@@ -33,7 +33,12 @@
         sys.exit()
     # Get Runs from Storage and Read Logs
     run_ids_in_storage = read_robot_logs.get_run_ids_from_storage(run_log_file_path)
-    runs_and_robots, header = abr_google_drive.create_data_dictionary(
+    (
+        runs_and_robots,
+        header,
+        runs_and_lpc,
+        lpc_headers,
+    ) = abr_google_drive.create_data_dictionary(
         run_ids_in_storage, run_log_file_path, ""
     )
     list_of_runs = list(runs_and_robots.keys())