refactor(abr-testing): Adding additional functionality to ABR data collection (#18956)

<!--
Thanks for taking the time to open a Pull Request (PR)! 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

GitHub provides robust markdown to format your PR. Links, diagrams,
pictures, and videos along with text formatting make it possible to
create a rich and informative PR. For more information on GitHub
markdown, see:


https://docs.github.com/en/get-started/writing-on-github/getting-started-with-writing-and-formatting-on-github/basic-writing-and-formatting-syntax

To ensure your code is reviewed quickly and thoroughly, please fill out
the sections below to the best of your ability!
-->

# Overview

Including HEPA/UV to Data Collection Script

## Test Plan and Hands on Testing

- Ran test script 

## Changelog

- Added timestamping of hepa / uv turning on and off
- Changed the ABR Temperature humidity sensors to reference storage
directory in config file
- changed batch_delete_cells to include sheet id as an input

## Review requests

<!--
- What do you need from reviewers to feel confident this PR is ready to
merge?
- Ask questions.
-->

## Risk assessment
- hard coded google sheet ids, but not really a problem because these
scripts are used by ABR

Author: rclarke0

abr-testing/abr_testing/automation/google_sheets_tool.py

@@ -95,21 +95,21 @@ 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:
+    def batch_delete_rows(self, row_indices: List[int], sheet_id: str) -> None:
         """Batch delete rows in list of indices."""
         delete_body = {
             "requests": [
                 {
                     "deleteDimension": {
                         "range": {
-                            "sheetId": 0,
+                            "sheetId": sheet_id,
                             "dimension": "ROWS",
                             "startIndex": index,
                             "endIndex": index + 1,
                         }
                     }
                 }
-                for index in row_indices
+                for index in sorted(row_indices, reverse=True)
             ]
         }
         self.spread_sheet.batch_update(body=delete_body)

abr-testing/abr_testing/data_collection/abr_hepauv.py

@@ -0,0 +1,217 @@
+"""Record HEPA UV Testing."""
+import datetime
+import argparse
+from typing import List, Any
+from abr_testing.automation import google_sheets_tool
+import os
+import sys
+import json
+
+
+def get_hepa_serials(storage_directory: str) -> List[List[Any]]:
+    """Get HEPA / UV Serial Number."""
+    try:
+        ips_path = os.path.join(storage_directory, "IPs.json")
+        ip_file = json.load(open(ips_path))
+        robot_dict = ip_file.get("ip_address_list")
+    except FileNotFoundError:
+        print(f"Add IPs.json file to: {storage_directory}.")
+        sys.exit()
+    robot_name_and_hepa_serial = list(map(list, (zip(*robot_dict.values()))))
+    return robot_name_and_hepa_serial
+
+
+def turning_uv_and_hepa_on(
+    google_sheet: google_sheets_tool.google_sheet, all_robots: List[List[Any]]
+) -> None:
+    """Create set of lines to add to google sheet."""
+    today = datetime.date.today()
+    formatted_date = today.strftime("%m/%d/%Y")
+    timestamp = datetime.datetime.now()
+    formatted_timestamp = timestamp.strftime("%m/%d/%Y %H:%M:%S")
+    num_robots = len(all_robots[0])
+    blank_column = [""] * num_robots
+    date_column = [formatted_date] * num_robots
+    fan_start_column = [formatted_timestamp] * num_robots
+    uv_cycles_column = [1] * num_robots
+    uv_duration_column = [15] * num_robots
+
+    all_columns = [
+        date_column,
+        all_robots[0],
+        all_robots[1],
+        fan_start_column,
+        blank_column,
+        blank_column,
+        blank_column,
+        uv_cycles_column,
+        uv_duration_column,
+    ]
+
+    start_row = google_sheet.get_index_row() + 1
+    google_sheet.batch_update_cells(all_columns, "A", start_row, "1790601450")
+
+
+def turning_hepa_off_and_uv_on(
+    google_sheet: google_sheets_tool.google_sheet, all_robots: List[List[str]]
+) -> None:
+    """Records when Hepa is Turned Off and UV on."""
+    # List of Expected Headers
+    expected_headers = {
+        "Date",
+        "Robot",
+        "Hepa Filter Serial",
+        "Fan Start Time",
+        "Fan Stop Time",
+        "Fan On Duration (timestamp)",
+        "Fan On Duration (min)",
+        "UV # of Cycles",
+        "UV Duration (min)",
+        "Errors",
+        "Level",
+        "Description",
+    }
+    all_data = google_sheet.get_all_data(expected_headers)
+    todays_rows = []
+
+    # Initialize column-wise lists
+    dates, robots, hepa_serials = [], [], []
+    fan_start_times, fan_stop_times, durations, durations_min = [], [], [], []
+    uv_cycles, uv_durations, errors, levels, descriptions = [], [], [], [], []
+
+    # Timestamp formatting
+    today = datetime.date.today()
+    formatted_date = today.strftime("%m/%d/%Y")
+    timestamp = datetime.datetime.now()
+    formatted_timestamp = timestamp.strftime("%m/%d/%Y %H:%M:%S")
+
+    # Process matching rows
+    for row in all_data:
+        if (row["Date"] == formatted_date) and (row["Robot"] in all_robots[0]):
+            start_time_str = row["Fan Start Time"]
+            start_time = datetime.datetime.strptime(start_time_str, "%m/%d/%Y %H:%M:%S")
+
+            duration = timestamp - start_time
+            duration_str = str(duration)
+            duration_min = duration.total_seconds() / 60
+
+            # Update row with new info
+            row["Fan Stop Time"] = formatted_timestamp
+            row["Fan On Duration (timestamp)"] = duration_str
+            row["Fan On Duration (min)"] = duration_min
+            row["UV # of Cycles"] = 2
+            row["UV Duration (min)"] = 30
+
+            todays_rows.append(row)
+
+            # Populate columns
+            dates.append(row["Date"])
+            robots.append(row["Robot"])
+            hepa_serials.append(row["Hepa Filter Serial"])
+            fan_start_times.append(row["Fan Start Time"])
+            fan_stop_times.append(row["Fan Stop Time"])
+            durations.append(duration_str)
+            durations_min.append(duration_min)
+            uv_cycles.append(2)
+            uv_durations.append(30)
+            errors.append(row.get("Errors", ""))
+            levels.append(row.get("Level", ""))
+            descriptions.append(row.get("Description", ""))
+
+    # Combine all columns into a nested list
+    nested_list = [
+        dates,
+        robots,
+        hepa_serials,
+        fan_start_times,
+        fan_stop_times,
+        durations,
+        durations_min,
+        uv_cycles,
+        uv_durations,
+        errors,
+        levels,
+        descriptions,
+    ]
+    last_row_index = google_sheet.get_index_row() + 1
+    starting_row_index = last_row_index - len(all_robots[0])
+    row_nums = list(range(starting_row_index, last_row_index))
+    row_indices = [i - 1 for i in row_nums]  # API needs 0-based
+    google_sheet.batch_delete_rows(row_indices, "1790601450")
+    google_sheet.update_row_index()
+    google_sheet.batch_update_cells(nested_list, "A", starting_row_index, "1790601450")
+
+
+def run(
+    turning_hepa_fan: str,
+    google_sheet_name: str,
+    storage_directory: str,
+) -> None:
+    """Main control function."""
+    try:
+        credentials_path = os.path.join(storage_directory, "credentials.json")
+    except FileNotFoundError:
+        print(f"Add credentials.json file to: {storage_directory}.")
+        sys.exit()
+    hepa_uv_sheet = google_sheets_tool.google_sheet(
+        credentials_path, google_sheet_name, 5
+    )
+    robot_names_and_hepa_serials = get_hepa_serials(storage_directory)
+    if turning_hepa_fan == "on":
+        print("𖣘 TIME STAMPING START OF HEPA FANS & UV LIGHT.")
+    else:
+        print("𖣘 TIME STAMPING END OF HEPA FANS & UV LIGHT.")
+
+    answer = input("Do you want to exclude any robots (y/n)?")
+    if answer.lower == "y":
+        robots_to_exclude = input(
+            "Enter robots you want to exclude in a comma separated list: "
+        )
+        exclude_list = [r.strip() for r in robots_to_exclude.split(",")]
+        robots = robot_names_and_hepa_serials[0]
+        serials = robot_names_and_hepa_serials[1]
+        # Zip robots and serials into pairs, filter, then unzip back
+        filtered_pairs = [
+            (r, s) for r, s in zip(robots, serials) if r not in exclude_list
+        ]
+        # Unzip into separate lists again
+        filtered_robots, filtered_serials = (
+            zip(*filtered_pairs) if filtered_pairs else ([], [])
+        )
+        robot_names_and_hepa_serials = [list(filtered_robots), list(filtered_serials)]
+    else:
+        robot_names_and_hepa_serials = robot_names_and_hepa_serials
+    if turning_hepa_fan == "on":
+        turning_uv_and_hepa_on(hepa_uv_sheet, robot_names_and_hepa_serials)
+    else:
+        turning_hepa_off_and_uv_on(hepa_uv_sheet, robot_names_and_hepa_serials)
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Record HEPA/UV Actions")
+    parser.add_argument(
+        "turning_hepa_fan",
+        metavar="-hepa",
+        type=str,
+        nargs=1,
+        help="'on' or 'off'",
+    )
+    parser.add_argument(
+        "google_sheet_name",
+        metavar="-google_sheet",
+        type=str,
+        nargs=1,
+        help="Google sheet name.",
+    )
+    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()
+    turning_hepa_fan = args.turning_hepa_fan[0]
+    google_sheet_name = args.google_sheet_name[0]
+    storage_directory = args.storage_directory[0]
+    run(turning_hepa_fan, google_sheet_name, storage_directory)

abr-testing/abr_testing/tools/abr_lpc.py

@@ -31,7 +31,7 @@ def remove_duplicate_data() -> None:
         else:
             row_indices.append(i)
     if len(row_indices) > 0:
-        google_sheet_lpc.batch_delete_rows(row_indices)
+        google_sheet_lpc.batch_delete_rows(row_indices, "0")
 
 
 if __name__ == "__main__":

abr-testing/abr_testing/tools/abr_setup.py

@@ -12,6 +12,7 @@
     get_run_logs,
     abr_google_drive,
     abr_calibration_logs,
+    abr_hepauv,
 )
 from abr_testing.tools import sync_abr_sheet
 
@@ -50,7 +51,7 @@ def clean_sheet(sheet_name: str, credentials: str) -> Any:
         return
     print(f"Rows to be removed: {rem_rows}")
     try:
-        sheet.batch_delete_rows(rem_rows)
+        sheet.batch_delete_rows(rem_rows, "0")
         print("deleted rows")
     except Exception:
         print("could not delete rows")
@@ -66,11 +67,22 @@ def run_sync_abr_sheet(
     sync_abr_sheet.run(storage_directory, abr_data_sheet, room_conditions_sheet)
 
 
-def run_temp_sensor(ambient_conditions_sheet: str, credentials: str) -> None:
+def run_hepa_uv(
+    turning_hepa_fan: str,
+    google_sheet_name: str,
+    storage_directory: str,
+) -> None:
+    """Record HEPA UVs."""
+    abr_hepauv.run(turning_hepa_fan, google_sheet_name, storage_directory)
+
+
+def run_temp_sensor(
+    ambient_conditions_sheet: str, credentials: str, storage_directory: str
+) -> None:
     """Run temperature sensors on all robots."""
     # Remove entries > 60 days
     clean_sheet(ambient_conditions_sheet, credentials)
-    processes = ABRAsairScript.run()
+    processes = ABRAsairScript.run(storage_directory)
     for process in processes:
         process.start()
         time.sleep(20)
@@ -127,26 +139,36 @@ def main(configurations: configparser.ConfigParser) -> None:
             credentials = default["Credentials"]
         except KeyError as e:
             print("Cannot read config file\n" + str(e))
-
+    # Record HEPA/UV Tracking
+    storage_directory = configurations["RUN-LOG"]["Storage"]
+    sheet_name = configurations["RUN-LOG"]["Sheet_Name"]
+    hepa_on = input("Are you turning HEPA Fans on or off? ")
+    run_hepa_uv(hepa_on.lower(), sheet_name, storage_directory)
+    if hepa_on == "off":
+        continue_str = input(
+            "Type 'continue' to continue with the rest of the set up script."
+        )
+        if continue_str == "continue":
+            print("Script will continue.")
+        else:
+            sys.exit()
     # Run Temperature Sensors
     ambient_conditions_sheet = configurations["TEMP-SENSOR"]["Sheet_Url"]
     ambient_conditions_sheet_name = configurations["TEMP-SENSOR"]["Sheet_Name"]
-    print("Starting temp sensors...")
-    run_temp_sensor(ambient_conditions_sheet_name, credentials)
-    print("Temp Sensors Started")
+    print("Starting 🌡️ temp sensors 🌡️...")
+    run_temp_sensor(ambient_conditions_sheet_name, credentials, storage_directory)
     # Get Run Logs and Record
-    storage_directory = configurations["RUN-LOG"]["Storage"]
     email = configurations["RUN-LOG"]["Email"]
     drive_folder = configurations["RUN-LOG"]["Drive_Folder"]
     sheet_name = configurations["RUN-LOG"]["Sheet_Name"]
     sheet_url = configurations["RUN-LOG"]["Sheet_Url"]
     print(sheet_name)
     if storage_directory and drive_folder and sheet_name and email:
-        print("Retrieving robot run logs...")
+        print("🤖 Retrieving robot run logs...")
         get_abr_logs(storage_directory, drive_folder, email)
-        print("Recording robot run logs...")
+        print("📄 Recording robot run logs...")
         record_abr_logs(storage_directory, drive_folder, sheet_name, email)
-        print("Run logs updated")
+        print("✅ Run logs updated")
     else:
         print("Storage, Email, or Drive Folder is missing, please fix configs")
         sys.exit(1)

hardware-testing/hardware_testing/scripts/ABRAsairScript.py

@@ -3,6 +3,7 @@
 import time
 import json
 import multiprocessing
+import os
 from typing import Optional, List, Any
 
 
@@ -56,32 +57,26 @@ def run_command_on_ip(
         ssh = connect_ssh(curr_ip)
         status = execute(ssh, cd + cmd, [robot_names[index], "540", "5"])
         if status == 0:
-            print(f"Envrironmental sensors for {curr_ip}, are now running")
+            print(f"Environmental sensors for {curr_ip}, are now running")
     except Exception as e:
         print(f"Error running command on {curr_ip}: {e}")
 
 
-def run() -> List[Any]:
+def run(storage_directory: str) -> List[Any]:
     """Run asair script module."""
     # Load Robot IPs
     cmd = "nohup python3 -m hardware_testing.scripts.abr_asair_sensor {name} {duration} {frequency}"
     cd = "cd /opt/opentrons-robot-server && "
     robot_ips = []
     robot_names = []
 
-    robot = input("Enter IP of robot (type 'all' to run on all robots): ")
-    if robot.lower() == "all":
-        ip_file = input("Path of IPs.json: ")
-        with open(ip_file) as file:
-            file_dict = json.load(file)
-            robot_dict = file_dict.get("ip_address_list")
-            robot_ips = list(robot_dict.keys())
-            robot_names = list(robot_dict.values())
-    else:
-        robot_name = input("What is the name of the robot? ")
-        robot_ips.append(robot)
-        robot_names.append(robot_name)
-    print("Executing Script on Robot(s):")
+    ip_file = os.path.join(storage_directory, "IPs.json")
+    with open(ip_file) as file:
+        file_dict = json.load(file)
+        robot_dict = file_dict.get("ip_address_list")
+        robot_ips = list(robot_dict.keys())
+        robot_names = list(map(list, (zip(*robot_dict.values()))))[0]
+    print("Executing Script on Robots:")
     # Launch the processes for each robot.
     processes = []
     for index in range(len(robot_ips)):
@@ -94,7 +89,8 @@ def run() -> List[Any]:
 
 if __name__ == "__main__":
     # Wait for all processes to finish.
-    processes = run()
+    storage_directory = ""
+    processes = run(storage_directory)
     for process in processes:
         process.start()
         time.sleep(20)