added script to read individual files

abr-testing/abr_testing/automation/google_drive_tool.py

@@ -1,12 +1,16 @@
 """Google Drive Tool."""
 import os
-from typing import Set, Any, Optional
+import io
+import json
+import sys
+from typing import Set, Any, Optional, List, Dict
 import webbrowser
 import mimetypes
 from oauth2client.service_account import ServiceAccountCredentials  # type: ignore[import]
 import googleapiclient  # type: ignore[import]
 from googleapiclient.discovery import build
 from googleapiclient.http import MediaFileUpload
+from googleapiclient.http import MediaIoBaseDownload
 
 """Google Drive Tool.
 
@@ -19,13 +23,17 @@ class google_drive:
 
     def __init__(self, credentials: Any, folder_name: str, email: str) -> None:
         """Connects to google drive via credentials file."""
-        self.scope = ["https://www.googleapis.com/auth/drive"]
-        self.credentials = ServiceAccountCredentials.from_json_keyfile_name(
-            credentials, self.scope
-        )
-        self.drive_service = build("drive", "v3", credentials=self.credentials)
-        self.parent_folder = folder_name
-        self.email = email
+        try:
+            self.scope = ["https://www.googleapis.com/auth/drive"]
+            self.credentials = ServiceAccountCredentials.from_json_keyfile_name(
+                credentials, self.scope
+            )
+            self.drive_service = build("drive", "v3", credentials=self.credentials)
+            self.parent_folder = folder_name
+            self.email = email
+        except json.decoder.JSONDecodeError:
+            print("Error! Get file: https://console.cloud.google.com/apis/credentials")
+            sys.exit()
 
     def list_folder(self, delete: Any = False) -> Set[str]:
         """List folders and files in Google Drive."""
@@ -88,7 +96,7 @@ def upload_file(self, file_path: str) -> str:
 
     def upload_missing_files(self, storage_directory: str) -> None:
         """Upload missing files to Google Drive."""
-        # Read Google Drive .json files.
+        # Read .json files.
         google_drive_files = self.list_folder()
         google_drive_files_json = [
             file for file in google_drive_files if file.endswith(".json")
@@ -111,7 +119,7 @@ def upload_missing_files(self, storage_directory: str) -> None:
             except googleapiclient.errors.HttpError:
                 continue
 
-        # Fetch the updated file list after all files are uploaded
+        # Fetch the updated file list after all are uploaded
         files = google_drive.list_folder(self)
 
         file_names = [file for file in files]
@@ -122,9 +130,7 @@ def upload_missing_files(self, storage_directory: str) -> None:
                     f"File '{this_name}' was successfully uploaded with ID: {uploaded_file['id']}"
                 )
             else:
-                print(
-                    f"File '{this_name}' was not found in the list of files after uploading."
-                )
+                print(f"File '{this_name}' was not found after uploading.")
         print(f"{len(files)} item(s) in Google Drive")
 
     def open_folder(self) -> Optional[str]:
@@ -154,3 +160,32 @@ def share_permissions(self, file_id: str) -> None:
         self.drive_service.permissions().create(
             fileId=file_id, body=new_permission, transferOwnership=False  # type: ignore
         ).execute()
+
+    def download_files(
+        self, files_to_download: List[Dict[str, Any]], save_directory: str
+    ) -> None:
+        """Download files to a specified directory."""
+        for file in files_to_download:
+            id = file["id"]
+            file_name = file["name"]
+            file_path = os.path.join(save_directory, file_name)
+            request = self.drive_service.files().get_media(fileId=id)  # type: ignore[attr-defined]
+            fh = io.FileIO(file_path, "wb")
+            downloader = MediaIoBaseDownload(fh, request)
+            done = False
+            while done is False:
+                status, done = downloader.next_chunk()
+                print(f"Downloading {file_name}... {int(status.progress() * 100)}%")
+
+    def search_folder(self, search_strings: List[str], folder_id: str) -> List[Any]:
+        """Search folder for files containing string from list."""
+        files_found = []
+        for search_string in search_strings:
+            query = f"'{folder_id}' in parents and name contains '{search_string}'"
+            response = (
+                self.drive_service.files()
+                .list(q=query, fields="files(id,name)")
+                .execute()
+            )
+            files_found.extend(response.get("files", []))
+        return files_found

abr-testing/abr_testing/automation/google_sheets_tool.py

@@ -3,6 +3,7 @@
 import socket
 import httplib2
 import time as t
+import sys
 from datetime import datetime
 from oauth2client.service_account import ServiceAccountCredentials  # type: ignore[import]
 from typing import Dict, List, Any, Set, Tuple
@@ -19,19 +20,24 @@ class google_sheet:
 
     def __init__(self, credentials: Any, file_name: str, tab_number: int) -> None:
         """Connects to google sheet via credentials file."""
-        self.scope = [
-            "https://spreadsheets.google.com/feeds",
-            "https://www.googleapis.com/auth/drive",
-        ]
-        self.credentials = ServiceAccountCredentials.from_json_keyfile_name(
-            credentials, self.scope
-        )
-        self.gc = gspread.authorize(self.credentials)
-        self.file_name = file_name
-        self.tab_number = tab_number
-        self.spread_sheet = self.open_google_sheet()
-        self.worksheet = self.open_worksheet(self.tab_number)
-        self.row_index = 1
+        try:
+            self.scope = [
+                "https://spreadsheets.google.com/feeds",
+                "https://www.googleapis.com/auth/drive",
+            ]
+            self.credentials = ServiceAccountCredentials.from_json_keyfile_name(
+                credentials, self.scope
+            )
+            self.gc = gspread.authorize(self.credentials)
+            self.file_name = file_name
+            self.tab_number = tab_number
+            self.spread_sheet = self.open_google_sheet()
+            self.worksheet = self.open_worksheet(self.tab_number)
+            self.row_index = 1
+            print(f"Connected to google sheet: {self.file_name}")
+        except gspread.exceptions.APIError:
+            print("ERROR: Check google sheet name. Check credentials file.")
+            sys.exit()
 
     def open_google_sheet(self) -> Any:
         """Open Google Spread Sheet."""
@@ -79,7 +85,7 @@ def write_to_row(self, data: List) -> None:
 
     def delete_row(self, row_index: int) -> None:
         """Delete Row from google sheet."""
-        self.worksheet.delete_row(row_index)
+        self.worksheet.delete_rows(row_index)
 
     def update_cell(
         self, row: int, column: int, single_data: Any
@@ -125,3 +131,13 @@ def token_check(self) -> None:
         """Check if still credentials are still logged in."""
         if self.credentials.access_token_expired:
             self.gc.login()
+
+    def get_row_index_with_value(self, some_string: str, col_num: int) -> Any:
+        """Find row index of string by looking in specific column."""
+        cell = self.worksheet.find(some_string, in_column=col_num)
+        try:
+            row_index = int(cell.row)
+        except AttributeError:
+            print("Row not found.")
+            return None
+        return row_index

abr-testing/abr_testing/data_collection/abr_calibration_logs.py

@@ -3,7 +3,6 @@
 import argparse
 import os
 import json
-import gspread  # type: ignore[import]
 import sys
 import time as t
 from abr_testing.data_collection import read_robot_logs
@@ -175,32 +174,17 @@ def upload_calibration_offsets(
     except FileNotFoundError:
         print(f"Add credentials.json file to: {storage_directory}.")
         sys.exit()
-    try:
-        google_drive = google_drive_tool.google_drive(
-            credentials_path, folder_name, email
-        )
-        # Upload calibration logs to google drive.
-        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, folder_name, email)
     # Connect to google sheet
-    try:
-        google_sheet_instruments = google_sheets_tool.google_sheet(
-            credentials_path, google_sheet_name, 0
-        )
-        google_sheet_modules = google_sheets_tool.google_sheet(
-            credentials_path, google_sheet_name, 1
-        )
-        google_sheet_deck = google_sheets_tool.google_sheet(
-            credentials_path, google_sheet_name, 2
-        )
-        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_instruments = google_sheets_tool.google_sheet(
+        credentials_path, google_sheet_name, 0
+    )
+    google_sheet_modules = google_sheets_tool.google_sheet(
+        credentials_path, google_sheet_name, 1
+    )
+    google_sheet_deck = google_sheets_tool.google_sheet(
+        credentials_path, google_sheet_name, 2
+    )
     ip_json_file = os.path.join(storage_directory, "IPs.json")
     try:
         ip_file = json.load(open(ip_json_file))

abr-testing/abr_testing/data_collection/abr_google_drive.py

@@ -3,7 +3,6 @@
 import os
 import sys
 import json
-import gspread  # type: ignore[import]
 from datetime import datetime, timedelta
 from abr_testing.data_collection import read_robot_logs
 from typing import Set, Dict, Any, Tuple, List, Union
@@ -121,8 +120,6 @@ def create_data_dictionary(
                 runs_and_robots[run_id] = row_2
             else:
                 continue
-                # os.remove(file_path)
-                # print(f"Run ID: {run_id} has a run time of 0 minutes. Run removed.")
     return runs_and_robots, headers
 
 
@@ -162,33 +159,13 @@ def create_data_dictionary(
     except FileNotFoundError:
         print(f"Add credentials.json file to: {storage_directory}.")
         sys.exit()
-    try:
-        google_drive = google_drive_tool.google_drive(
-            credentials_path, folder_name, email
-        )
-        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, folder_name, email)
     # Get run ids on 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()
-    try:
-        google_sheet_lpc = google_sheets_tool.google_sheet(
-            credentials_path, "ABR-LPC", 0
-        )
-        print("Connected to google sheet ABR-LPC")
-    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
+    )
+    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)
 

abr-testing/abr_testing/data_collection/get_run_logs.py

@@ -139,14 +139,5 @@ def get_all_run_logs(storage_directory: str) -> None:
     except FileNotFoundError:
         print(f"Add credentials.json file to: {storage_directory}.")
         sys.exit()
-    try:
-        google_drive = google_drive_tool.google_drive(
-            credentials_path, folder_name, email
-        )
-        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, folder_name, email)
     get_all_run_logs(storage_directory)

abr-testing/abr_testing/data_collection/module_ramp_rates.py

@@ -1,7 +1,6 @@
 """Get ramp rates of modules."""
 from abr_testing.automation import google_sheets_tool
 from abr_testing.data_collection import read_robot_logs
-import gspread  # type: ignore[import]
 import argparse
 import os
 import sys
@@ -79,14 +78,9 @@ def ramp_rate(file_results: Dict[str, Any]) -> Dict[int, float]:
         print(f"Add credentials.json file to: {storage_directory}.")
         sys.exit()
     # CONNECT TO GOOGLE SHEET
-    try:
-        google_sheet = google_sheets_tool.google_sheet(
-            credentials_path, google_sheet_name, 1
-        )
-        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, 1
+    )
     run_ids_on_sheet = google_sheet.get_column(2)
     runs_and_robots = {}
     for filename in os.listdir(storage_directory):

abr-testing/abr_testing/data_collection/read_robot_logs.py

@@ -299,11 +299,10 @@ def get_error_info(file_results: Dict[str, Any]) -> Tuple[int, str, str, str, st
             # Instrument Error
             error_instrument = run_command_error["error"]["errorInfo"]["node"]
         except KeyError:
-            # Module Error
+            # Module
             error_instrument = run_command_error["error"]["errorInfo"].get("port", "")
     else:
         error_type = file_results["errors"][0]["errorType"]
-        print(error_type)
         error_code = file_results["errors"][0]["errorCode"]
         error_instrument = file_results["errors"][0]["detail"]
     for error in error_levels:

abr-testing/abr_testing/data_collection/single_run_log_reader.py

@@ -0,0 +1,51 @@
+"""Reads single run log retrieved by get_run_logs.py and saves to local csv."""
+import argparse
+import sys
+import os
+import csv
+from abr_testing.data_collection import read_robot_logs
+from abr_testing.data_collection import abr_google_drive
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Read single run log locally saved.")
+    parser.add_argument(
+        "run_log_file_path",
+        metavar="RUN_LOG_FILE_PATH",
+        type=str,
+        nargs=1,
+        help="Folder path that holds individual run logs of interest.",
+    )
+    parser.add_argument(
+        "google_sheet_name",
+        metavar="GOOGLE_SHEET_NAME",
+        type=str,
+        nargs=1,
+        help="Google sheet name.",
+    )
+    args = parser.parse_args()
+    run_log_file_path = args.run_log_file_path[0]
+    google_sheet_name = args.google_sheet_name[0]
+
+    try:
+        credentials_path = os.path.join(run_log_file_path, "credentials.json")
+    except FileNotFoundError:
+        print(f"Add credentials.json file to: {run_log_file_path}.")
+        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(
+        run_ids_in_storage, run_log_file_path, ""
+    )
+    list_of_runs = list(runs_and_robots.keys())
+    # Adds Run to local csv
+    sheet_location = os.path.join(run_log_file_path, "saved_data.csv")
+    file_exists = os.path.exists(sheet_location) and os.path.getsize(sheet_location) > 0
+    with open(sheet_location, "a", newline="") as f:
+        writer = csv.writer(f)
+        if not file_exists:
+            writer.writerow(header)
+        for run in list_of_runs:
+            # Add new row
+            row = runs_and_robots[run].values()
+            row_list = list(row)
+            writer.writerow(row_list)