format, lint, test, and style fixes

hardware-testing/Makefile

@@ -159,9 +159,6 @@ test-scripts:
 test-liquid-sense:
 	$(python) -m hardware_testing.liquid_sense --simulate --pipette 1000 --channels 1
 	$(python) -m hardware_testing.liquid_sense --simulate --pipette 50 --channels 1
-	$(python) -m hardware_testing.liquid_sense --simulate --pipette 1000 --channels 8
-	$(python) -m hardware_testing.liquid_sense --simulate --pipette 50 --channels 8
-	$(python) -m hardware_testing.liquid_sense --simulate --pipette 1000 --channels 96
 
 .PHONY: test-integration
 test-integration: test-production-qc test-examples test-scripts test-gravimetric

hardware-testing/hardware_testing/liquid_sense/__main__.py

@@ -38,11 +38,7 @@
     liquid_sense_ot3_p1000_single_vial,
 )
 
-try:
-    from . import google_sheets_tool  # type: ignore[import]
-except Exception:
-    print("WARNING: unable to import google_sheets_tool, "
-          "if not simulating check your environment")
+from . import google_sheets_tool
 
 
 CREDENTIALS_PATH = "/var/lib/jupyter/notebooks/abr.json"
@@ -162,7 +158,7 @@ def build_run_args(cls, args: argparse.Namespace) -> "RunArgs":
         environment_sensor = asair_sensor.BuildAsairSensor(simulate=True)
         git_description = get_git_description()
         protocol_cfg = LIQUID_SENSE_CFG[args.pipette][args.channels]
-        name = protocol_cfg.metadata["protocolName"]  # type: ignore[attr-defined]
+        name = protocol_cfg.metadata["protocolName"]  # type: ignore[union-attr]
         ui.print_header("LOAD PIPETTE")
         pipette = _ctx.load_instrument(
             f"flex_{args.channels}channel_{args.pipette}", args.mount
@@ -221,7 +217,7 @@ def build_run_args(cls, args: argparse.Namespace) -> "RunArgs":
             trials,
             "aspirate" if args.aspirate else "dispense",
             args.liquid,
-            protocol_cfg.LABWARE_ON_SCALE,  # type: ignore[attr-defined]
+            protocol_cfg.LABWARE_ON_SCALE,  # type: ignore[union-attr]
             args.z_speed,
             args.probe_seconds_before_contact,
         )
@@ -277,7 +273,7 @@ def build_run_args(cls, args: argparse.Namespace) -> "RunArgs":
         0.0 < args.probe_seconds_before_contact <= MAX_PROBE_SECONDS
     ), f"'--probe-seconds-before-contact' must be between 0.0-{MAX_PROBE_SECONDS}"
     run_args = RunArgs.build_run_args(args)
-    exit_error = os.X_OK
+    exit_error = 0
     serial_logger: Optional[subprocess.Popen] = None
     data_dir = get_testing_data_directory()
     data_file = f"/{data_dir}/{run_args.name}/{run_args.run_id}/serial.log"
@@ -290,21 +286,24 @@ def build_run_args(cls, args: argparse.Namespace) -> "RunArgs":
             )
             sleep(1)
             # Connect to Google Sheet
-            print(os.path.exists(CREDENTIALS_PATH))
-            google_sheet = google_sheets_tool.google_sheet(
+            ui.print_info(f"robot has credentials: {os.path.exists(CREDENTIALS_PATH)}")
+            google_sheet: Optional[
+                google_sheets_tool.google_sheet
+            ] = google_sheets_tool.google_sheet(
                 CREDENTIALS_PATH, args.google_sheet_name, 0
             )
-            sheet_id = google_sheet.create_worksheet(run_args.run_id)
+            sheet_id = google_sheet.create_worksheet(run_args.run_id)  # type: ignore[union-attr]
         else:
             google_sheet = None
+            sheet_id = None
         hw = run_args.ctx._core.get_hardware()
         ui.print_info("homing...")
         run_args.ctx.home()
         for tip in run_args.tip_volumes:
             execute.run(tip, run_args, google_sheet, sheet_id, args.starting_tip)
     except Exception as e:
-        ui.print_info(f"got error {e}")
-        ui.print_info(traceback.format_exc())
+        ui.print_error(f"got error {e}")
+        ui.print_error(traceback.format_exc())
         exit_error = 1
     finally:
         if run_args.recorder is not None:
@@ -344,8 +343,11 @@ def build_run_args(cls, args: argparse.Namespace) -> "RunArgs":
             ]
             try:
                 process_google_sheet(google_sheet, run_args, test_info, sheet_id)
-            except:
-                print('error making graphs or logging data on google sheet')
+            except Exception as e:
+                ui.print_error("error making graphs or logging data on google sheet")
+                ui.print_error(f"got error {e}")
+                ui.print_error(traceback.format_exc())
+                exit_error = 2
 
         run_args.ctx.cleanup()
         if not args.simulate:

hardware-testing/hardware_testing/liquid_sense/execute.py

@@ -28,6 +28,8 @@
 
 from opentrons_shared_data.errors.exceptions import LiquidNotFoundError
 
+from . import google_sheets_tool
+
 
 PROBE_MAX_TIME: Dict[int, float] = {
     1: 2.75,
@@ -168,8 +170,8 @@ def _load_scale(
 def run(
     tip: int,
     run_args: RunArgs,
-    google_sheet: Optional[Any],
-    sheet_id: str,
+    google_sheet: Optional[google_sheets_tool.google_sheet],
+    sheet_id: Optional[str],
     starting_tip: str = "A1",
 ) -> None:
     """Run a liquid probe test."""

hardware-testing/hardware_testing/liquid_sense/google_sheets_tool.py

@@ -1,13 +1,15 @@
 """Google Sheet Tool."""
-import gspread  # type: ignore[import]
+import gspread
 import socket
-import httplib2
+import httplib2  # type: ignore[import]
 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, Optional
 
+from hardware_testing.data import ui
+
 """Google Sheets Tool.
 
 This module requires a credentials.json file before getting started.
@@ -34,9 +36,9 @@ def __init__(self, credentials: Any, file_name: str, tab_number: int) -> None:
             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}")
+            ui.print_info(f"Connected to google sheet: {self.file_name}")
         except gspread.exceptions.APIError:
-            print("ERROR: Check google sheet name. Check credentials file.")
+            ui.print_error("ERROR: Check google sheet name. Check credentials file.")
             sys.exit()
 
     def open_google_sheet(self) -> Any:
@@ -54,7 +56,8 @@ def create_worksheet(self, title: str) -> Optional[str]:
             new_sheet = self.spread_sheet.add_worksheet(title, rows="2500", cols="40")
             return new_sheet.id
         except gspread.exceptions.APIError:
-            print("Sheet already exists.")
+            ui.print_error("Sheet already exists.")
+            return None
 
     def write_header(self, header: List) -> None:
         """Write Header to first row if not present."""
@@ -76,11 +79,11 @@ def write_to_row(self, data: List, title: str = "Sheet1") -> None:
         except socket.gaierror:
             pass
         except httplib2.ServerNotFoundError:
-            print("UNABLE TO CONNECT TO SERVER!!, CHECK CONNECTION")
+            ui.print_error("UNABLE TO CONNECT TO SERVER!!, CHECK CONNECTION")
         except Exception as error:
-            print(error.__traceback__)
+            ui.print_error(str(error.__traceback__))
         except gspread.exceptions.APIError:
-            print("Write quotes exceeded. Waiting 30 sec before writing.")
+            ui.print_error("Write quotes exceeded. Waiting 30 sec before writing.")
             t.sleep(30)
             self.worksheet.insert_row(data, index=self.row_index)
 
@@ -106,61 +109,64 @@ def batch_delete_rows(self, row_indices: List[int]) -> None:
             ]
         }
         self.spread_sheet.batch_update(body=delete_body)
-
 
     def batch_update_cells(
-        self, sheet_title: str, data: List[List[str]], start_column: str, start_row: int, sheet_id: str
+        self,
+        sheet_title: str,
+        data: List[List[str]],
+        start_column: str,
+        start_row: int,
+        sheet_id: str,
     ) -> None:
         """Writes to multiple cells at once in a specific sheet."""
+
         def column_letter_to_index(column_letter: str) -> int:
             """Convert a column letter (e.g., 'A') to a 1-based column index (e.g., 1)."""
             index = 0
             for char in column_letter.upper():
-                index = index * 26 + (ord(char) - ord('A') + 1)
+                index = index * 26 + (ord(char) - ord("A") + 1)
             return index
-    
+
         def index_to_column_letter(index: int) -> str:
             """Convert a zero-based column index to a column letter."""
             result = []
             while index >= 0:
                 index, remainder = divmod(index, 26)
-                result.append(chr(remainder + ord('A')))
+                result.append(chr(remainder + ord("A")))
                 index -= 1
-            return ''.join(result[::-1])
+            return "".join(result[::-1])
+
         requests = []
         start_column_index = column_letter_to_index(start_column) - 1
-        
+
         for col_offset, col_values in enumerate(data):
             column_index = start_column_index + col_offset
-            column_letter = index_to_column_letter(column_index)
             for row_offset, value in enumerate(col_values):
                 row_index = start_row + row_offset
                 try:
                     float_value = float(value)
-                    user_entered_value = {'numberValue': float_value}
+                    user_entered_value = {"numberValue": str(float_value)}
                 except ValueError:
-                    user_entered_value = {'stringValue': str(value)}
-                requests.append({
-                    'updateCells': {
-                        'range': {
-                            'sheetId': sheet_id,
-                            'startRowIndex': row_index - 1,
-                            'endRowIndex': row_index,
-                            'startColumnIndex': column_index,
-                            'endColumnIndex': column_index + 1,
-                        },
-                        'rows': [{
-                            'values': [{
-                                'userEnteredValue': user_entered_value
-                            }]
-                        }],
-                        'fields': 'userEnteredValue',
+                    user_entered_value = {"stringValue": str(value)}
+                requests.append(
+                    {
+                        "updateCells": {
+                            "range": {
+                                "sheetId": sheet_id,
+                                "startRowIndex": row_index - 1,
+                                "endRowIndex": row_index,
+                                "startColumnIndex": column_index,
+                                "endColumnIndex": column_index + 1,
+                            },
+                            "rows": [
+                                {"values": [{"userEnteredValue": user_entered_value}]}
+                            ],
+                            "fields": "userEnteredValue",
+                        }
                     }
-                })
+                )
 
-        body = {
-            'requests': requests
-        }
+        body = {"requests": requests}
         self.spread_sheet.batch_update(body=body)
 
     def update_cell(
@@ -190,7 +196,7 @@ def get_single_col_range(self, sheet_name: str, range: str) -> List:
     def get_index_row(self) -> int:
         """Check for the next available row to write too."""
         row_index = len(self.get_column(1))
-        print(f"Row Index: {row_index} recorded on google sheet.")
+        ui.print_info(f"Row Index: {row_index} recorded on google sheet.")
         return row_index
 
     def update_row_index(self) -> None:
@@ -215,15 +221,15 @@ def get_sheet_by_name(self, title: str) -> None:
     def token_check(self) -> None:
         """Check if still credentials are still logged in."""
         if self.credentials.access_token_expired:
-            self.gc.login()
+            self.gc = gspread.authorize(self.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."""
         cell = self.worksheet.find(some_string, in_column=col_num)
         try:
             row_index = int(cell.row)
         except AttributeError:
-            print("Row not found.")
+            ui.print_error("Row not found.")
             return None
         return row_index