ABR LPC Automated Logging (#15163)

<!--
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

<!--
Use this section to describe your pull-request at a high level. If the
PR addresses any open issues, please tag the issues here.
-->

# 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/automation/google_sheets_tool.py

@@ -87,14 +87,33 @@ def delete_row(self, row_index: int) -> None:
         """Delete Row from google sheet."""
         self.worksheet.delete_rows(row_index)
 
+    def batch_delete_rows(self, row_indices: List[int]) -> None:
+        """Batch delete rows in list of indices."""
+        delete_body = {
+            "requests": [
+                {
+                    "deleteDimension": {
+                        "range": {
+                            "sheetId": 0,
+                            "dimension": "ROWS",
+                            "startIndex": index,
+                            "endIndex": index + 1,
+                        }
+                    }
+                }
+                for index in row_indices
+            ]
+        }
+        self.spread_sheet.batch_update(body=delete_body)
+
     def update_cell(
         self, row: int, column: int, single_data: Any
     ) -> Tuple[int, int, Any]:
         """Update ONE individual cell according to a row and column."""
         self.worksheet.update_cell(row, column, single_data)
         return row, column, single_data
 
-    def get_all_data(self) -> Dict[str, Any]:
+    def get_all_data(self) -> List[Dict[str, Any]]:
         """Get all the Data recorded from worksheet."""
         return self.worksheet.get_all_records()
 
@@ -141,3 +160,45 @@ def get_row_index_with_value(self, some_string: str, col_num: int) -> Any:
             print("Row not found.")
             return None
         return row_index
+
+    def create_line_chart(
+        self,
+        titles: List[str],
+        series: List[Dict[str, Any]],
+        domains: List[Dict[str, Any]],
+    ) -> None:
+        """Create chart of data on google sheet."""
+        request_body = {
+            "requests": [
+                {
+                    "addChart": {
+                        "chart": {
+                            "spec": {
+                                "title": titles[0],
+                                "basicChart": {
+                                    "chartType": "LINE",
+                                    "legendPosition": "RIGHT_LEGEND",
+                                    "axis": [
+                                        {"position": "BOTTOM_AXIS", "title": titles[1]},
+                                        {"position": "LEFT_AXIS", "title": titles[2]},
+                                    ],
+                                    "domains": domains,
+                                    "series": series,
+                                    "headerCount": 1,
+                                },
+                            },
+                            "position": {
+                                "overlayPosition": {
+                                    "anchorCell": {
+                                        "sheetId": 0,
+                                        "rowIndex": 1,
+                                        "columnIndex": 1,
+                                    }
+                                }
+                            },
+                        }
+                    }
+                }
+            ]
+        }
+        self.spread_sheet.batch_update(body=request_body)

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_lpc.py

@@ -1 +1,56 @@
-"""Get Unique LPC Values from Run logs."""
+"""Automated LPC Data Analysis."""
+import os
+import argparse
+from abr_testing.automation import google_sheets_tool
+import sys
+
+
+def remove_duplicate_data() -> None:
+    """Determine unique sets of data."""
+    seen = set()
+    new_values = []
+    row_indices = []
+    sheet_data = google_sheet_lpc.get_all_data()
+    for i, row in enumerate(sheet_data):
+        key = (
+            row["Robot"],
+            row["Software Version"],
+            row["Errors"],
+            row["Slot"],
+            row["Module"],
+            row["Adapter"],
+            row["X"],
+            row["Y"],
+            row["Z"],
+        )
+
+        if key not in seen:
+            seen.add(key)
+            new_values.append(row)
+        else:
+            row_indices.append(i)
+    if len(row_indices) > 0:
+        google_sheet_lpc.batch_delete_rows(row_indices)
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Read run logs on google drive.")
+    parser.add_argument(
+        "storage_directory",
+        metavar="STORAGE_DIRECTORY",
+        type=str,
+        nargs=1,
+        help="Path to long term storage directory for run logs.",
+    )
+    args = parser.parse_args()
+    storage_directory = args.storage_directory[0]
+    try:
+        credentials_path = os.path.join(storage_directory, "credentials.json")
+    except FileNotFoundError:
+        print(f"Add credentials.json file to: {storage_directory}.")
+        sys.exit()
+    google_sheet_lpc = google_sheets_tool.google_sheet(credentials_path, "ABR-LPC", 0)
+    print(len(google_sheet_lpc.get_all_data()))
+    remove_duplicate_data()
+    num_of_rows = print(len(google_sheet_lpc.get_all_data()))
+    # TODO: automate data analysis

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]:
@@ -145,6 +143,7 @@ def get_error_info_from_robot(
         whole_description_str,
         run_log_file_path,
     ) = get_error_info_from_robot(ip, one_run, storage_directory)
+    affects_version = "internal release - any"
     # Get Calibration Data
     saved_file_path_calibration, calibration = read_robot_logs.get_calibration_offsets(
         ip, storage_directory
@@ -183,35 +182,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())