write accuracy recordings to google sheet

abr-testing/abr_testing/data_collection/read_robot_logs.py

@@ -23,8 +23,9 @@ def lpc_data(
     """Get labware offsets from one run log."""
     offsets = file_results.get("labwareOffsets", "")
     # TODO: per UNIQUE slot AND LABWARE TYPE only keep the most recent LPC recording
+    unique_offsets: Dict[Any, Any] = {}
+    headers_lpc = []
     if len(offsets) > 0:
-        unique_offsets: Dict[Any, Any] = {}
         for offset in offsets:
             labware_type = offset.get("definitionUri", "")
             slot = offset["location"].get("slotName", "")
@@ -53,9 +54,9 @@ def lpc_data(
                     "Y": y_offset,
                     "Z": z_offset,
                 }
-    for item in unique_offsets:
-        runs_and_lpc.append(unique_offsets[item].values())
-    headers_lpc = list(unique_offsets[(slot, labware_type)].keys())
+        for item in unique_offsets:
+            runs_and_lpc.append(unique_offsets[item].values())
+        headers_lpc = list(unique_offsets[(slot, labware_type)].keys())
 
     return runs_and_lpc, headers_lpc
 
@@ -306,8 +307,11 @@ def get_error_info(file_results: Dict[str, Any]) -> Tuple[int, str, str, str, st
         error_level = ""
         return 0, error_type, error_code, error_instrument, error_level
     commands_of_run: List[Dict[str, Any]] = file_results.get("commands", [])
-    run_command_error: Dict[str, Any] = commands_of_run[-1]
-    error_str: int = len(run_command_error.get("error", ""))
+    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", "")
         error_code = run_command_error["error"].get("errorCode", "")

abr-testing/abr_testing/tools/abr_scale.py

@@ -4,8 +4,120 @@
 from hardware_testing.drivers import find_port, list_ports_and_select  # type: ignore[import]
 from hardware_testing.drivers.radwag import RadwagScale  # type: ignore[import]
 import argparse
-from abr_testing.data_collection import read_robot_logs
+from abr_testing.data_collection import read_robot_logs, abr_google_drive, get_run_logs
 from abr_testing.automation import google_sheets_tool
+import requests
+from typing import Any, Tuple
+import sys
+
+
+def get_protocol_step_as_int() -> Tuple[int, int, str]:
+    """Get user input as integer."""
+    expected_liquid_moved = 0
+    ip = ""
+    while True:
+        try:
+            protocol_step = int(input("Measurement Step (1, 2, 3): "))
+            if protocol_step in [1, 2, 3]:
+                break
+            else:
+                print("Protocol step should be one of the values: 1, 2, or 3.")
+        except ValueError:
+            print("Protocol step should be an integer value 1, 2, or 3.")
+
+    if int(protocol_step) == 3:
+        ip = input("Robot IP: ")
+        while True:
+            try:
+                expected_liquid_moved = int(input("Expected volume moved: "))
+                if expected_liquid_moved >= 0:
+                    break
+            except ValueError:
+                print("Expected liquid moved volume should be an integer.")
+    return protocol_step, expected_liquid_moved, ip
+
+
+def get_all_plate_readings(
+    robot: str, plate: str, mass_3: float, expected_moved: int, google_sheet: Any
+) -> float:
+    """Calculate accuracy of liquid moved on final measurement step."""
+    accuracy = 0.0
+    all_data = google_sheet.get_all_data()
+    # Get mass of first reading
+    mass_1_readings = []
+    for row in all_data:
+        if (
+            row["Robot"] == robot
+            and row["Labware"] == plate
+            and (int(row["Measurement Step"]) == 1)
+        ):
+            mass_1_readings.append(row["Mass (g)"])
+    if len(mass_1_readings) > 0:
+        mass_1 = mass_1_readings[-1]
+    else:
+        print(
+            f"Initial mass for plate {plate} on robot {robot} not found. Check sheet."
+        )
+        sys.exit()
+    # Get mass of second reading
+    mass_2_readings = []
+    for row in all_data:
+        if (
+            row["Robot"] == robot
+            and row["Labware"] == plate
+            and (int(row["Measurement Step"]) == 2)
+        ):
+            mass_2_readings.append(row["Mass (g)"])
+    if len(mass_2_readings) > 0:
+        mass_2 = mass_2_readings[-1]
+        starting_liquid = 1000 * (mass_2 - mass_1)
+    else:
+        starting_liquid = 0
+    actual_moved = ((mass_3 - mass_1) * 1000) - starting_liquid
+    accuracy = ((float(expected_moved) - actual_moved) / actual_moved) * 100
+    return accuracy
+
+
+def get_most_recent_run_and_record(
+    ip: str, storage_directory: str, labware: str, accuracy: float
+) -> None:
+    """Write accuracy level to google sheet."""
+    # Get most recent run
+    try:
+        response = requests.get(
+            f"http://{ip}:31950/runs", headers={"opentrons-version": "3"}
+        )
+    except Exception:
+        print(
+            f"ERROR: Failed to read IP address {ip}. Accuracy was not recorded on sheet."
+        )
+        sys.exit()
+    run_data = response.json()
+    run_list = run_data["data"]
+    most_recent_run_id = run_list[-1]["id"]
+    results = get_run_logs.get_run_data(most_recent_run_id, ip)
+    # Save run information to local directory as .json file
+    read_robot_logs.save_run_log_to_json(ip, results, storage_directory)
+    # Record run to google sheets.
+    print(most_recent_run_id)
+    (
+        runs_and_robots,
+        headers,
+        runs_and_lpc,
+        headers_lpc,
+    ) = abr_google_drive.create_data_dictionary(
+        most_recent_run_id, storage_directory, "", labware, accuracy
+    )
+    google_sheet_abr_data = google_sheets_tool.google_sheet(
+        credentials_path, "ABR-run-data", tab_number=0
+    )
+    start_row = google_sheet_abr_data.get_index_row() + 1
+    google_sheet_abr_data.batch_update_cells(runs_and_robots, "A", start_row, "0")
+    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)
+    start_row_lpc = google_sheet_lpc.get_index_row() + 1
+    google_sheet_lpc.batch_update_cells(runs_and_lpc, "A", start_row_lpc, "0")
 
 
 if __name__ == "__main__":
@@ -47,15 +159,15 @@
     # Set up google sheet
     try:
         credentials_path = os.path.join(storage_directory, "credentials.json")
-        google_sheet = google_sheets_tool.google_sheet(
-            credentials_path, file_name, tab_number=0
-        )
-        print("Connected to google sheet.")
     except FileNotFoundError:
         print("No google sheets credentials. Add credentials to storage notebook.")
+    google_sheet = google_sheets_tool.google_sheet(
+        credentials_path, file_name, tab_number=0
+    )
     robot = input("Robot: ")
     labware = input("Labware: ")
-    protocol_step = input("Measurement Step (1,2,3): ")
+    protocol_step, expected_liquid_moved, ip = get_protocol_step_as_int()
+
     # Scale Loop
     grams, is_stable = scale.read_mass()
     grams, is_stable = scale.read_mass()
@@ -74,6 +186,14 @@
             read_robot_logs.write_to_sheets(
                 sheet_location, google_sheet, row_list, headers
             )
+            if int(protocol_step) == 3:
+                # Calculate accuracy of plate
+                accuracy = get_all_plate_readings(
+                    robot, labware, grams, expected_liquid_moved, google_sheet
+                )
+                # Connect to robot - get most recent run - write run data to google sheet.
+                get_most_recent_run_and_record(ip, storage_directory, labware, accuracy)
+
             is_stable = False
             y_or_no = input("Do you want to weigh another sample? (Y/N): ")
             if y_or_no == "Y":
@@ -82,7 +202,7 @@
                 is_stable = False
                 robot = input("Robot: ")
                 labware = input("Labware: ")
-                protocol_step = input("Measurement Step (1,2,3): ")
+                protocol_step, expected_liquid_moved, ip = get_protocol_step_as_int()
                 grams, is_stable = scale.read_mass()
             elif y_or_no == "N":
                 break_all = True