feat (abr-testing): Lengthen ABR Protocols, Add Protocol Versioning, Add Liquid Waste Probing and Recording

abr-testing/abr_testing/data_collection/abr_google_drive.py

@@ -34,16 +34,13 @@ def create_data_dictionary(
     runs_to_save: Union[Set[str], str],
     storage_directory: str,
     issue_url: str,
-    plate: str,
-    accuracy: Any,
     hellma_plate_standards: List[Dict[str, Any]],
-) -> Tuple[List[List[Any]], List[str], List[List[Any]], List[str], List[List[Any]]]:
+) -> Tuple[List[List[Any]], List[str], List[List[Any]], List[str]]:
     """Pull data from run files and format into a dictionary."""
     runs_and_robots: List[Any] = []
     runs_and_lpc: List[Dict[str, Any]] = []
     headers: List[str] = []
     headers_lpc: List[str] = []
-    list_of_heights: List[List[Any]] = [[], [], [], [], [], [], [], []]
     hellma_plate_orientation = False  # default hellma plate is not rotated.
     for filename in os.listdir(storage_directory):
         file_path = os.path.join(storage_directory, filename)
@@ -103,12 +100,15 @@ def create_data_dictionary(
                 run_time_min = run_time.total_seconds() / 60
             except ValueError:
                 pass  # Handle datetime parsing errors if necessary
+            # Get protocol version #
+            version_number = read_robot_logs.get_protocol_version_number(file_results)
 
             if run_time_min > 0:
                 run_row = {
                     "Robot": robot,
                     "Run_ID": run_id,
                     "Protocol_Name": protocol_name,
+                    "Protocol Version": version_number,
                     "Software Version": software_version,
                     "Date": start_date,
                     "Start_Time": start_time_str,
@@ -130,13 +130,10 @@ def create_data_dictionary(
                 plate_reader_dict = read_robot_logs.plate_reader_commands(
                     file_results, hellma_plate_standards, hellma_plate_orientation
                 )
-                list_of_heights = read_robot_logs.liquid_height_commands(
-                    file_results, list_of_heights
-                )
                 notes = {"Note1": "", "Jira Link": issue_url}
+                liquid_height = read_robot_logs.get_liquid_waste_height(file_results)
                 plate_measure = {
-                    "Plate Measured": plate,
-                    "End Volume Accuracy (%)": accuracy,
+                    "Liquid Waste Height (mm)": liquid_height,
                     "Average Temp (oC)": "",
                     "Average RH(%)": "",
                 }
@@ -173,7 +170,6 @@ def create_data_dictionary(
         headers,
         transposed_runs_and_lpc,
         headers_lpc,
-        list_of_heights,
     )
 
 
@@ -211,26 +207,15 @@ def run(
         headers,
         transposed_runs_and_lpc,
         headers_lpc,
-        list_of_heights,
     ) = create_data_dictionary(
         missing_runs_from_gs,
         storage_directory,
         "",
-        "",
-        "",
-        hellma_plate_standards=file_values,
+        file_values,
     )
     start_row = google_sheet.get_index_row() + 1
     google_sheet.batch_update_cells(transposed_runs_and_robots, "A", start_row, "0")
-    # Record Liquid Heights Found
-    google_sheet_ldf = google_sheets_tool.google_sheet(
-        credentials_path, google_sheet_name, 2
-    )
-    google_sheet_ldf.get_row(1)
-    start_row_lhd = google_sheet_ldf.get_index_row() + 1
-    google_sheet_ldf.batch_update_cells(
-        list_of_heights, "A", start_row_lhd, "2075262446"
-    )
+
     # Add LPC to google sheet
     google_sheet_lpc = google_sheets_tool.google_sheet(credentials_path, "ABR-LPC", 0)
     start_row_lpc = google_sheet_lpc.get_index_row() + 1

abr-testing/abr_testing/data_collection/abr_robot_error.py

@@ -614,28 +614,16 @@ def get_run_error_info_from_robot(
             headers,
             runs_and_lpc,
             headers_lpc,
-            list_of_heights,
         ) = abr_google_drive.create_data_dictionary(
             run_id,
             error_folder_path,
             issue_url,
-            "",
-            "",
-            hellma_plate_standards=file_values,
+            file_values,
         )
 
         start_row = google_sheet.get_index_row() + 1
         google_sheet.batch_update_cells(runs_and_robots, "A", start_row, "0")
         print("Wrote run to ABR-run-data")
-        # Record Liquid Heights Found
-        google_sheet_ldf = google_sheets_tool.google_sheet(
-            credentials_path, google_sheet_name, 4
-        )
-        start_row_lhd = google_sheet_ldf.get_index_row() + 1
-        google_sheet_ldf.batch_update_cells(
-            list_of_heights, "A", start_row_lhd, "1795535088"
-        )
-        print("wrote liquid heights found.")
         # Add LPC to google sheet
         google_sheet_lpc = google_sheets_tool.google_sheet(
             credentials_path, "ABR-LPC", 0

abr-testing/abr_testing/data_collection/read_robot_logs.py

@@ -134,7 +134,8 @@ def match_pipette_to_action(
     left_pipette_add = 0
     for command in commandTypes:
         command_type = command_dict["commandType"]
-        command_pipette = command_dict.get("pipetteId", "")
+        command_params = command_dict.get("params", "")
+        command_pipette = command_params.get("pipetteId", "")
         if command_type == command and command_pipette == right_pipette:
             right_pipette_add = 1
         elif command_type == command and command_pipette == left_pipette:
@@ -213,13 +214,48 @@ def instrument_commands(
     return pipette_dict
 
 
+def get_comment_result_by_string(file_results: Dict[str, Any], key_phrase: str) -> str:
+    """Get comment string based off ky phrase."""
+    commandData = file_results.get("commands", "")
+    result_str = command_str = ""
+    for command in commandData:
+        commandType = command["commandType"]
+        if commandType == "comment":
+            command_str = command["params"].get("message", "")
+        try:
+            result_str = command_str.split(key_phrase)[1]
+        except IndexError:
+            continue
+    return result_str
+
+
+def get_protocol_version_number(file_results: Dict[str, Any]) -> str:
+    """Get protocol version number."""
+    return get_comment_result_by_string(file_results, "Protocol Version: ")
+
+
+def get_liquid_waste_height(file_results: Dict[str, Any]) -> float:
+    """Find liquid waste height."""
+    result_str = get_comment_result_by_string(
+        file_results, "Liquid Waste Total Height: "
+    )
+    try:
+        height = float(result_str)
+    except ValueError:
+        height = 0.0
+    return height
+
+
 def liquid_height_commands(
     file_results: Dict[str, Any], all_heights_list: List[List[Any]]
 ) -> List[List[Any]]:
     """Record found liquid heights during a protocol."""
     commandData = file_results.get("commands", "")
     robot = file_results.get("robot_name", "")
     run_id = file_results.get("run_id", "")
+    list_of_heights = []
+    print(robot)
+    liquid_waste_height = 0.0
     for command in commandData:
         commandType = command["commandType"]
         if commandType == "comment":
@@ -236,16 +272,24 @@ def liquid_height_commands(
                     well_location = str(entry.split(", ")[1].split(" ")[0])
                     slot_location = str(entry.split("slot ")[1].split(")")[0])
                     labware_name = str(entry.split("of ")[1].split(" on")[0])
-                    all_heights_list[0].append(robot)
-                    all_heights_list[1].append(run_id)
-                    all_heights_list[2].append(comment_time)
-                    all_heights_list[3].append(labware_type)
-                    all_heights_list[4].append(labware_name)
-                    all_heights_list[5].append(slot_location)
-                    all_heights_list[6].append(well_location)
-                    all_heights_list[7].append(height)
+                    if labware_name == "Liquid Waste":
+                        liquid_waste_height += height
+                    one_entry = {
+                        "Timestamp": comment_time,
+                        "Labware Name": labware_name,
+                        "Labware Type": labware_type,
+                        "Slot Location": slot_location,
+                        "Well Location": well_location,
+                        "All Heights (mm)": height,
+                    }
+                    list_of_heights.append(one_entry)
             except (IndexError, ValueError):
                 continue
+    if len(list_of_heights) > 0:
+        all_heights_list[0].append(robot)
+        all_heights_list[1].append(run_id)
+        all_heights_list[2].append(list_of_heights)
+        all_heights_list[3].append(liquid_waste_height)
     return all_heights_list
 
 
@@ -281,10 +325,13 @@ def plate_reader_commands(
             read = "yes"
         elif read == "yes" and commandType == "comment":
             result = command["params"].get("message", "")
-            if "result:" in result:
-                plate_name = result.split("result:")[0]
-                formatted_result = result.split("result: ")[1]
-                print(formatted_result)
+            if "result:" in result or "Result:" in result:
+                try:
+                    plate_name = result.split("result:")[0]
+                    formatted_result = result.split("result: ")[1]
+                except IndexError:
+                    plate_name = result.split("Result:")[0]
+                    formatted_result = result.split("Result: ")[1]
                 result_dict = eval(formatted_result)
                 result_dict_keys = list(result_dict.keys())
                 if len(result_dict_keys) > 1:

abr-testing/abr_testing/data_collection/single_run_log_reader.py

@@ -34,14 +34,11 @@
         header,
         runs_and_lpc,
         lpc_headers,
-        list_of_heights,
     ) = abr_google_drive.create_data_dictionary(
         run_ids_in_storage,
         run_log_file_path,
         "",
-        "",
-        "",
-        hellma_plate_standards=file_values,
+        file_values,
     )
     print("list_of_heights not recorded.")
     transposed_list = list(zip(*runs_and_robots))