Robot accuracy recording (#15254)

<!--
Thanks for taking the time to open a pull request! Please make sure
you've read the "Opening Pull Requests" section of our Contributing
Guide:


https://github.com/Opentrons/opentrons/blob/edge/CONTRIBUTING.md#opening-pull-requests

To ensure your code is reviewed quickly and thoroughly, please fill out
the sections below to the best of your ability!
-->

# Overview

Writes final plate measurement to abr data sheet referencing run id.

# Test Plan

<!--
Use this section to describe the steps that you took to test your Pull
Request.
If you did not perform any testing provide justification why.

OT-3 Developers: You should default to testing on actual physical
hardware.
Once again, if you did not perform testing against hardware, justify
why.

Note: It can be helpful to write a test plan before doing development

Example Test Plan (HTTP API Change)

- Verified that new optional argument `dance-party` causes the robot to
flash its lights, move the pipettes,
then home.
- Verified that when you omit the `dance-party` option the robot homes
normally
- Added protocol that uses `dance-party` argument to G-Code Testing
Suite
- Ran protocol that did not use `dance-party` argument and everything
was successful
- Added unit tests to validate that changes to pydantic model are
correct

-->

# Changelog

<!--
List out the changes to the code in this PR. Please try your best to
categorize your changes and describe what has changed and why.

Example changelog:
- Fixed app crash when trying to calibrate an illegal pipette
- Added state to API to track pipette usage
- Updated API docs to mention only two pipettes are supported

IMPORTANT: MAKE SURE ANY BREAKING CHANGES ARE PROPERLY COMMUNICATED
-->

# Review requests

<!--
Describe any requests for your reviewers here.
-->

# Risk assessment

<!--
Carefully go over your pull request and look at the other parts of the
codebase it may affect. Look for the possibility, even if you think it's
small, that your change may affect some other part of the system - for
instance, changing return tip behavior in protocol may also change the
behavior of labware calibration.

Identify the other parts of the system your codebase may affect, so that
in addition to your own review and testing, other people who may not
have the system internalized as much as you can focus their attention
and testing there.
-->

abr-testing/abr_testing/data_collection/abr_google_drive.py

@@ -32,6 +32,8 @@ def create_data_dictionary(
     runs_to_save: Union[Set[str], str],
     storage_directory: str,
     issue_url: str,
+    plate: str,
+    accuracy: 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] = []
@@ -109,6 +111,10 @@ 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}
+                plate_measure = {
+                    "Plate Measured": plate,
+                    "End Volume Accuracy (%)": accuracy,
+                }
                 row_for_lpc = {**row, **all_modules, **notes}
                 row_2 = {
                     **row,
@@ -117,6 +123,7 @@ def create_data_dictionary(
                     **hs_dict,
                     **tm_dict,
                     **tc_dict,
+                    **plate_measure,
                 }
                 headers: List[str] = list(row_2.keys())
                 # runs_and_robots[run_id] = row_2
@@ -191,7 +198,7 @@ def create_data_dictionary(
         headers,
         transposed_runs_and_lpc,
         headers_lpc,
-    ) = create_data_dictionary(missing_runs_from_gs, storage_directory, "")
+    ) = create_data_dictionary(missing_runs_from_gs, storage_directory, "", "", "")
 
     start_row = google_sheet.get_index_row() + 1
     google_sheet.batch_update_cells(transposed_runs_and_robots, "A", start_row, "0")

abr-testing/abr_testing/data_collection/abr_robot_error.py

@@ -201,7 +201,9 @@ def get_error_info_from_robot(
         headers,
         runs_and_lpc,
         headers_lpc,
-    ) = abr_google_drive.create_data_dictionary(run_id, error_folder_path, issue_url)
+    ) = abr_google_drive.create_data_dictionary(
+        run_id, error_folder_path, issue_url, "", ""
+    )
 
     start_row = google_sheet.get_index_row() + 1
     google_sheet.batch_update_cells(runs_and_robots, "A", start_row, "0")

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,10 +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
 
 
@@ -307,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/data_collection/single_run_log_reader.py

@@ -31,7 +31,7 @@
         runs_and_lpc,
         lpc_headers,
     ) = abr_google_drive.create_data_dictionary(
-        run_ids_in_storage, run_log_file_path, ""
+        run_ids_in_storage, run_log_file_path, "", "", ""
     )
     transposed_list = list(zip(*runs_and_robots))
     # Adds Run to local csv

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