feat(abr-testing): plate reader data collection

abr-testing/Pipfile

@@ -18,7 +18,7 @@ slackclient = "*"
 slack-sdk = "*"
 pandas = "*"
 pandas-stubs = "*"
-
+numpy = "*"
 
 [dev-packages]
 atomicwrites = "==1.4.1"

abr-testing/abr_testing/automation/google_sheets_tool.py

@@ -225,8 +225,8 @@ def get_sheet_by_name(self, title: str) -> None:
 
     def token_check(self) -> None:
         """Check if credentials are still valid and refresh if expired."""
-        if self.credentials.expired:
-            self.credentials.refresh()  # Refresh the credentials
+        if self.credentials.access_token_expired:
+            self.gc.login()  # Refresh the credentials
 
     def get_row_index_with_value(self, some_string: str, col_num: int) -> Any:
         """Find row index of string by looking in specific column."""

abr-testing/abr_testing/data_collection/abr_google_drive.py

@@ -7,8 +7,7 @@
 from abr_testing.data_collection import read_robot_logs
 from typing import Set, Dict, Any, Tuple, List, Union
 from abr_testing.automation import google_drive_tool, google_sheets_tool
-from abr_testing.tools import sync_abr_sheet
-
+from abr_testing.tools import sync_abr_sheet, plate_reader
 
 def get_modules(file_results: Dict[str, str]) -> Dict[str, Any]:
     """Get module IPs and models from run log."""
@@ -17,6 +16,7 @@ def get_modules(file_results: Dict[str, str]) -> Dict[str, Any]:
         "temperatureModuleV2",
         "magneticBlockV1",
         "thermocyclerModuleV2",
+        "absorbanceReaderV1"
     )
     all_modules = {key: "" for key in modList}
     for module in file_results.get("modules", []):
@@ -35,6 +35,7 @@ def create_data_dictionary(
     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]]:
     """Pull data from run files and format into a dictionary."""
     runs_and_robots: List[Any] = []
@@ -113,6 +114,7 @@ def create_data_dictionary(
                 hs_dict = read_robot_logs.hs_commands(file_results)
                 tm_dict = read_robot_logs.temperature_module_commands(file_results)
                 pipette_dict = read_robot_logs.instrument_commands(file_results)
+                plate_reader_dict = read_robot_logs.plate_reader_commands(file_results, hellma_plate_standards)
                 notes = {"Note1": "", "Jira Link": issue_url}
                 plate_measure = {
                     "Plate Measured": plate,
@@ -132,6 +134,7 @@ def create_data_dictionary(
                     **hs_dict,
                     **tm_dict,
                     **tc_dict,
+                    **plate_reader_dict,
                     **pipette_dict,
                     **plate_measure,
                 }
@@ -181,6 +184,8 @@ def create_data_dictionary(
     storage_directory = args.storage_directory[0]
     google_sheet_name = args.google_sheet_name[0]
     email = args.email[0]
+
+
     try:
         credentials_path = os.path.join(storage_directory, "credentials.json")
     except FileNotFoundError:
@@ -203,13 +208,15 @@ def create_data_dictionary(
     missing_runs_from_gs = read_robot_logs.get_unseen_run_ids(
         run_ids_on_gd, run_ids_on_gs
     )
+    # Read Hellma Files
+    file_values = plate_reader.read_hellma_plate_files(storage_directory, 101934)
     # Add missing runs to google sheet
     (
         transposed_runs_and_robots,
         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, "", "", "", hellma_plate_standards=file_values)
     start_row = google_sheet.get_index_row() + 1
     print(start_row)
     google_sheet.batch_update_cells(transposed_runs_and_robots, "A", start_row, "0")

abr-testing/abr_testing/data_collection/get_run_logs.py

@@ -47,7 +47,7 @@ def get_run_data(one_run: Any, ip: str) -> Dict[str, Any]:
             params={"cursor": cursor, "pageLength": page_length},
         )
         command_data = response.json()
-        commands.extend(command_data["data"])
+        commands.extend(command_data.get("data",""))
     run["commands"] = commands
     response = requests.get(
         f"http://{ip}:31950/runs/{one_run}", headers={"opentrons-version": "3"}