Merge branch 'edge' into app_odd-anonymous-localization-provider

.github/CODEOWNERS

@@ -8,18 +8,6 @@
 *.d.ts @Opentrons/js
 /webpack-config @Opentrons/js
 
-# subprojects - those with clear team ownership should have appropriate notifications
-/protocol-designer @Opentrons/app-and-uis
-/labware-designer @Opentrons/app-and-uis
-/labware-library @Opentrons/app-and-uis
-/update-server @Opentrons/robot-svcs
-/discovery-client @Opentrons/robot-svcs @Opentrons/app-and-uis
-/shared-data/pipette @Opentrons/embedded-sw
-/shared-data/protocol @Opentrons/robot-svcs @Opentrons/app-and-uis
-/shared-data/module @Opentrons/embedded-sw
-/shared-data/deck @Opentrons/embedded-sw
-/shared-data/labware @Opentrons/embedded-sw
-
 # subprojects by language - some subprojects are shared by teams but united by a
 # language community (including makefiles and config) so mark them appropriately
 /app @Opentrons/js

abr-testing/Pipfile

@@ -5,20 +5,30 @@ name = "pypi"
 
 [packages]
 abr-testing = { editable = true, path = "." }
+google-api-python-client = "==2.41.0"
+httplib2 = "==0.22.0"
+types-httplib2 = "*"
+oauth2client = "==4.1.3"
+gspread = "==6.0.2"
+hardware-testing = {editable = true, path = "../hardware-testing"}
+opentrons-shared-data = {editable = true, path = "./../shared-data/python"}
+opentrons-hardware = {editable = true, path = "./../hardware", extras=['FLEX']}
+opentrons = {editable = true, path = "./../api", extras=['flex-hardware']}
 
 [dev-packages]
 atomicwrites = "==1.4.1"
 colorama = "==0.4.4"
 pytest = "==7.1.1"
 pytest-cov = "==2.10.1"
-mypy = "==0.981"
+mypy = "==1.8.0"
 black = "==22.3.0"
 flake8 = "~=3.9.0"
 flake8-annotations = "~=2.6.2"
 flake8-docstrings = "~=1.6.0"
 flake8-noqa = "~=1.2.1"
 requests = "==2.27.1"
 types-requests = "==2.25.6"
+google-api-python-client-stubs = "*"
 
 [requires]
 python_version = "3.10"

abr-testing/abr_testing/automation/__init__.py

@@ -0,0 +1 @@
+"""Tools for Google API Interaction."""

abr-testing/abr_testing/automation/google_drive_tool.py

@@ -0,0 +1,152 @@
+"""Google Drive Tool."""
+import os
+from typing import Set, Any, Optional
+import webbrowser
+import mimetypes
+from oauth2client.service_account import ServiceAccountCredentials  # type: ignore[import]
+from googleapiclient.discovery import build
+from googleapiclient.http import MediaFileUpload
+
+"""Google Drive Tool.
+
+This module requires a credentials.json file before getting started.
+Retrieve from https://console.cloud.google.com/apis/credentials."""
+
+
+class google_drive:
+    """Google Drive Tool."""
+
+    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
+        self.folder = self.open_folder()
+
+    def list_folder(self, delete: Any = False) -> Set[str]:
+        """List folders and files in Google Drive."""
+        file_names = set()
+        page_token: str = ""
+        while True:
+            results = (
+                self.drive_service.files()
+                .list(
+                    q=f"'{self.parent_folder}' in parents and trashed=false"
+                    if self.parent_folder
+                    else ""  # type: ignore
+                    if self.parent_folder
+                    else None,
+                    pageSize=1000,
+                    fields="nextPageToken, files(id, name, mimeType)",
+                    pageToken=page_token,
+                )
+                .execute()
+            )
+            items = results.get("files", [])
+            if not items:
+                break
+            for item in items:
+                item_name = item["name"]
+                if delete:
+                    self.delete_files(item["id"])
+                file_names.add(item_name)
+            page_token = results.get("nextPageToken", "")
+            if len(page_token) < 1:
+                break
+            if not file_names:
+                print("No folders or files found in Google Drive.")
+        print(f"{len(file_names)} item(s) in Google Drive")
+        return file_names
+
+    def delete_files(self, file_or_folder_id: str) -> None:
+        """Delete a file or folder in Google Drive by ID."""
+        try:
+            self.drive_service.files().delete(fileId=file_or_folder_id).execute()
+            print(f"Successfully deleted file/folder with ID: {file_or_folder_id}")
+        except Exception as e:
+            print(f"Error deleting file/folder with ID: {file_or_folder_id}")
+            print(f"Error details: {str(e)}")
+
+    def upload_file(self, file_path: str) -> str:
+        """Upload file to Google Drive."""
+        file_metadata = {
+            "name": os.path.basename(file_path),
+            "mimeType": str(mimetypes.guess_type(file_path)[0]),
+            "parents": [self.parent_folder],
+        }
+        media = MediaFileUpload(file_path, resumable=True)
+
+        uploaded_file = (
+            self.drive_service.files()
+            .create(body=file_metadata, media_body=media, fields="id")  # type: ignore
+            .execute()
+        )
+        return uploaded_file["id"]
+
+    def upload_missing_files(self, storage_directory: str) -> None:
+        """Upload missing files to Google Drive."""
+        # Read Google Drive .json files.
+        google_drive_files = self.list_folder()
+        google_drive_files_json = [
+            file for file in google_drive_files if file.endswith(".json")
+        ]
+        # Read local directory.
+        local_files_json = set(
+            file for file in os.listdir(storage_directory) if file.endswith(".json")
+        )
+        missing_files = local_files_json - set(google_drive_files_json)
+        print(f"Missing files: {len(missing_files)}")
+        # Upload missing files.
+        uploaded_files = []
+        for file in missing_files:
+            file_path = os.path.join(storage_directory, file)
+            uploaded_file_id = google_drive.upload_file(self, file_path)
+            self.share_permissions(uploaded_file_id)
+            uploaded_files.append(
+                {"name": os.path.basename(file_path), "id": uploaded_file_id}
+            )
+        # Fetch the updated file list after all files are uploaded
+        files = google_drive.list_folder(self)
+        file_names = [file for file in files]
+        for uploaded_file in uploaded_files:
+            this_name = uploaded_file["name"]
+            if this_name in file_names:
+                print(
+                    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."
+                )
+
+    def open_folder(self) -> Optional[str]:
+        """Open folder in web browser."""
+        folder_metadata = (
+            self.drive_service.files()
+            .get(fileId=self.parent_folder, fields="webViewLink")
+            .execute()
+        )
+        folder_link = folder_metadata.get("webViewLink")
+        if folder_link:
+            print(f"Folder link: {folder_link}")
+            webbrowser.open(
+                folder_link
+            )  # Open the folder link in the default web browser
+        else:
+            print("Folder link not found.")
+        return folder_link
+
+    def share_permissions(self, file_id: str) -> None:
+        """Share permissions with self."""
+        new_permission = {
+            "type": "user",
+            "role": "writer",
+            "emailAddress": self.email,
+        }
+        self.drive_service.permissions().create(
+            fileId=file_id, body=new_permission, transferOwnership=False  # type: ignore
+        ).execute()

abr-testing/abr_testing/automation/google_sheets_tool.py

@@ -0,0 +1,115 @@
+"""Google Sheet Tool."""
+import gspread  # type: ignore[import]
+import socket
+import httplib2
+from oauth2client.service_account import ServiceAccountCredentials  # type: ignore[import]
+from typing import Dict, List, Any, Set, Tuple
+
+"""Google Sheets Tool.
+
+This module requires a credentials.json file before getting started.
+Retrieve from https://console.cloud.google.com/apis/credentials.
+"""
+
+
+class google_sheet:
+    """Google Sheets Tool."""
+
+    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
+
+    def open_google_sheet(self) -> Any:
+        """Open Google Spread Sheet."""
+        sheet = self.gc.open(self.file_name)
+        return sheet
+
+    def open_worksheet(self, tab_number: int) -> Any:
+        """Open individual worksheet within a googlesheet."""
+        return self.spread_sheet.get_worksheet(tab_number)
+
+    def create_worksheet(self, tab_name: int) -> None:
+        """Create a worksheet with tab name. Existing spreadsheet needed."""
+        try:
+            self.spread_sheet.add_worksheet(tab_name, rows="1000", cols="26")
+        except gspread.exceptions.APIError:
+            print("Work Sheet already exists")
+
+    def write_header(self, header: List) -> None:
+        """Write Header to first row if not present."""
+        header_list = self.worksheet.row_values(1)
+        if header_list != header:
+            self.worksheet.insert_row(header, self.row_index)
+
+    def write_to_row(self, data: List) -> None:
+        """Write data into a row in a List[] format."""
+        try:
+            self.row_index += 1
+            self.worksheet.insert_row(data, index=self.row_index)
+        except socket.gaierror:
+            pass
+        except httplib2.ServerNotFoundError:
+            print("UNABLE TO CONNECT TO SERVER!!, CHECK CONNECTION")
+        except Exception as error:
+            print(error.__traceback__)
+
+    def delete_row(self, row_index: int) -> None:
+        """Delete Row from google sheet."""
+        self.worksheet.delete_row(row_index)
+
+    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]:
+        """Get all the Data recorded from worksheet."""
+        return self.worksheet.get_all_records()
+
+    def get_column(self, column_number: int) -> Set[str]:
+        """Get all values in column."""
+        return self.worksheet.col_values(column_number)
+
+    def get_index_row(self) -> int:
+        """Check for the next available row to write too."""
+        row_index = len(self.get_column(1))
+        print("Row Index: ", row_index)
+        return row_index
+
+    def update_row_index(self) -> None:
+        """Update self.row_index instance variable."""
+        self.row_index = self.get_index_row()
+
+    def get_all_sheets(self) -> List[str]:
+        """List all tabs in the spreadsheets."""
+        worksheets = self.spread_sheet.worksheets()
+        return worksheets
+
+    def get_sheet_by_name(self, title: str) -> None:
+        """Reference sheet by name."""
+        try:
+            worksheet = self.spread_sheet.worksheet(title)
+            return worksheet
+        except gspread.exceptions.WorksheetNotFound:
+            raise gspread.exceptions.WorksheetNotFound(
+                "Worksheet does not exist!!, Use create_worksheet() function first."
+            )
+
+    def token_check(self) -> None:
+        """Check if still credentials are still logged in."""
+        if self.credentials.access_token_expired:
+            self.gc.login()