added functions to count module commands per run (#14797)

<!--
Thanks for taking the time to open a pull request! 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

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

# Overview

Functions to Count Module commands per run

# Test Plan

- looked at run logs and used cmd f to double check command counts/times

# Changelog

Added a function for the thermocycler, temperature module, and heater
shaker to count values of interest for lifetime test comparison
Added those dictionaries to larger dictionary to be included on run
sheet
# Review requests

<!--
Describe any requests for your reviewers here.
-->

# Risk assessment

- These functions are not set up to handle multiples of the same module
in a protocol. It will group total commands together
- some modules do not deactivate at the end of the run. To get total on
time, the protocol completedAt timestamp is used.

abr-testing/abr_testing/automation/jira_tool.py

@@ -44,6 +44,7 @@ def issues_on_board(self, board_id: str) -> List[str]:
     def open_issue(self, issue_key: str) -> None:
         """Open issue on web browser."""
         url = f"{self.url}/browse/{issue_key}"
+        print(f"Opening at {url}.")
         webbrowser.open(url)
 
     def create_ticket(

abr-testing/abr_testing/data_collection/abr_google_drive.py

@@ -6,7 +6,7 @@
 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
+from typing import Set, Dict, Any, Tuple, List
 from abr_testing.automation import google_drive_tool, google_sheets_tool
 
 
@@ -31,7 +31,7 @@ def get_modules(file_results: Dict[str, str]) -> Dict[str, Any]:
 
 def create_data_dictionary(
     runs_to_save: Set[str], storage_directory: str
-) -> Dict[Any, Dict[str, Any]]:
+) -> Tuple[Dict[Any, Dict[str, Any]], List]:
     """Pull data from run files and format into a dictionary."""
     runs_and_robots = {}
     for filename in os.listdir(storage_directory):
@@ -100,12 +100,17 @@ def create_data_dictionary(
                     "Right Mount": right_pipette,
                     "Extension": extension,
                 }
-                row_2 = {**row, **all_modules}
+                tc_dict = read_robot_logs.thermocycler_commands(file_results)
+                hs_dict = read_robot_logs.hs_commands(file_results)
+                tm_dict = read_robot_logs.temperature_module_commands(file_results)
+                notes = {"Note1": "", "Note2": ""}
+                row_2 = {**row, **all_modules, **notes, **hs_dict, **tm_dict, **tc_dict}
+                headers = list(row_2.keys())
                 runs_and_robots[run_id] = row_2
             else:
                 os.remove(file_path)
                 print(f"Run ID: {run_id} has a run time of 0 minutes. Run removed.")
-    return runs_and_robots
+    return runs_and_robots, headers
 
 
 if __name__ == "__main__":
@@ -175,29 +180,9 @@ def create_data_dictionary(
         run_ids_on_gd, run_ids_on_gs
     )
     # Add missing runs to google sheet
-    runs_and_robots = create_data_dictionary(missing_runs_from_gs, storage_directory)
-    headers = [
-        "Robot",
-        "Run_ID",
-        "Protocol_Name",
-        "Software Version",
-        "Date",
-        "Start_Time",
-        "End_Time",
-        "Run_Time (min)",
-        "Errors",
-        "Error_Code",
-        "Error_Type",
-        "Error_Instrument",
-        "Error_Level",
-        "Left Mount",
-        "Right Mount",
-        "Extension",
-        "heaterShakerModuleV1",
-        "temperatureModuleV2",
-        "magneticBlockV1",
-        "thermocyclerModuleV2",
-    ]
+    runs_and_robots, headers = create_data_dictionary(
+        missing_runs_from_gs, storage_directory
+    )
     read_robot_logs.write_to_local_and_google_sheet(
         runs_and_robots, storage_directory, google_sheet_name, google_sheet, headers
     )

abr-testing/abr_testing/data_collection/abr_robot_error.py

@@ -44,6 +44,7 @@ def get_error_info_from_robot(
     # JIRA Ticket Fields
     failure_level = "Level " + str(error_level) + " Failure"
     components = [failure_level, "Flex-RABR"]
+    components = ["Flex-RABR"]
     affects_version = results["API_Version"]
     parent = results.get("robot_name", "")
     print(parent)
@@ -141,10 +142,15 @@ def get_error_info_from_robot(
         whole_description_str,
         saved_file_path,
     ) = get_error_info_from_robot(ip, one_run, storage_directory)
+    # get calibration data
+    saved_file_path_calibration, calibration = read_robot_logs.get_calibration_offsets(
+        ip, storage_directory
+    )
     print(f"Making ticket for run: {one_run} on robot {robot}.")
     # TODO: make argument or see if I can get rid of with using board_id.
     project_key = "RABR"
     parent_key = project_key + "-" + robot[-1]
+    issues_ids = ticket.issues_on_board(board_id)
     issue_url, issue_key = ticket.create_ticket(
         summary,
         whole_description_str,
@@ -158,8 +164,4 @@ def get_error_info_from_robot(
     )
     ticket.open_issue(issue_key)
     ticket.post_attachment_to_ticket(issue_key, saved_file_path)
-    # get calibration data
-    saved_file_path_calibration, calibration = read_robot_logs.get_calibration_offsets(
-        ip, storage_directory
-    )
     ticket.post_attachment_to_ticket(issue_key, saved_file_path_calibration)

abr-testing/abr_testing/data_collection/error_levels.csv

@@ -20,7 +20,7 @@ Prefix,Error Code,Description,Categories,Level of Failure,
 2,2009,Early Capactivive Sense Trigger,A Robot Action Failed,4,
 2,2010,Innacrruate Non Contact Sweep,A Robot Action Failed,3,
 2,2011,Misaligned Gantry,A Robot Action Failed,3,
-2,2012,Unmatched Tip Presence States,A Robot Action Failed,3-4,
+2,2012,Unmatched Tip Presence States,A Robot Action Failed, 4,
 2,2013,Position Unknown,A Robot Action Failed,4,
 2,2014,Execution Cancelled,A Robot Action Failed, 4,
 2,2015,Failed Gripper Pickup Error,A Robot Action Failed,3,
@@ -31,18 +31,18 @@ Prefix,Error Code,Description,Categories,Level of Failure,
 3,3004,Tip Drop Failed,A Robot Interaction Failed,4,
 3,3005,Unexpeted Tip Removal,A Robot Interaction Failed,4,
 3,3006,Pipette Overpressure,A Robot Interaction Failed,3,
-3,3008,E-Stop Activated,A Robot Interaction Failed,Not an error,
+3,3008,E-Stop Activated,A Robot Interaction Failed,5, Not an error,
 3,3009,E-Stop Not Present,A Robot Interaction Failed,5,
 3,3010,Pipette Not Present,A Robot Interaction Failed,5,
 3,3011,Gripper Not Present,A Robot Interaction Failed,5,
 3,3012,Unexpected Tip Attach,A Robot Interaction Failed,4,
-3,3013,Firmware Update Required,A Robot Interaction Failed,Not an error,
+3,3013,Firmware Update Required,A Robot Interaction Failed,5, Not an error,
 3,3014,Invalid ID Actuator,A Robot Interaction Failed,3,
 3,3015,Module Not Pesent,A Robot Interaction Failed,5,Not an error
 3,3016,Invalid Instrument Data,A Robot Interaction Failed,3,
 3,3017,Invalid Liquid Class Name,A Robot Interaction Failed,5,Not an error
 3,3018,Tip Detector Not Found,A Robot Interaction Failed,3,
-4,4000,General Error,A Software Error Occured,2-4,How severe does a general error get
+4,4000,General Error,A Software Error Occured,4,How severe does a general error get
 4,4001,Robot In Use,A Software Error Occured,5,Not an error
 4,4002,API Removed,A Software Error Occured,5,used an old app on a new robot
 4,4003,Not Supported On Robot Type,A Software Error Occured,5,Not an error

abr-testing/abr_testing/data_collection/read_robot_logs.py

@@ -5,7 +5,7 @@
 saved in a local directory.
 """
 import csv
-import datetime
+from datetime import datetime
 import os
 from abr_testing.data_collection.error_levels import ERROR_LEVELS_PATH
 from typing import List, Dict, Any, Tuple, Set
@@ -14,6 +14,210 @@
 import requests
 
 
+def command_time(command: Dict[str, str]) -> Tuple[float, float]:
+    """Calculate total create and complete time per command."""
+    try:
+        create_time = datetime.strptime(
+            command.get("createdAt", ""), "%Y-%m-%dT%H:%M:%S.%f%z"
+        )
+        start_time = datetime.strptime(
+            command.get("startedAt", ""), "%Y-%m-%dT%H:%M:%S.%f%z"
+        )
+        complete_time = datetime.strptime(
+            command.get("completedAt", ""), "%Y-%m-%dT%H:%M:%S.%f%z"
+        )
+        create_to_start = (start_time - create_time).total_seconds()
+        start_to_complete = (complete_time - start_time).total_seconds()
+    except ValueError:
+        create_to_start = 0
+        start_to_complete = 0
+    return create_to_start, start_to_complete
+
+
+def hs_commands(file_results: Dict[str, Any]) -> Dict[str, float]:
+    """Gets total latch engagements, homes, rotations and total on time (sec) for heater shaker."""
+    # TODO: modify for cases that have more than 1 heater shaker.
+    commandData = file_results.get("commands", "")
+    hs_latch_count: float = 0.0
+    hs_temp: float = 0.0
+    hs_home_count: float = 0.0
+    hs_speed: float = 0.0
+    hs_rotations: Dict[str, float] = dict()
+    hs_temps: Dict[str, float] = dict()
+    temp_time = None
+    shake_time = None
+    for command in commandData:
+        commandType = command["commandType"]
+        # Heatershaker
+        # Latch count
+        if (
+            commandType == "heaterShaker/closeLabwareLatch"
+            or commandType == "heaterShaker/openLabwareLatch"
+        ):
+            hs_latch_count += 1
+        # Home count
+        elif commandType == "heaterShaker/deactivateShaker":
+            hs_home_count += 1
+            deactivate_time = datetime.strptime(
+                command.get("startedAt", ""), "%Y-%m-%dT%H:%M:%S.%f%z"
+            )
+            if temp_time is not None and deactivate_time > temp_time:
+                temp_duration = (deactivate_time - temp_time).total_seconds()
+                hs_temps[hs_temp] = hs_temps.get(hs_temp, 0.0) + temp_duration
+            if shake_time is not None and deactivate_time > shake_time:
+                shake_duration = (deactivate_time - shake_time).total_seconds()
+                hs_rotations[hs_speed] = hs_rotations.get(hs_speed, 0.0) + (
+                    (hs_speed * shake_duration) / 60
+                )
+        # of Rotations
+        elif commandType == "heaterShaker/setAndWaitForShakeSpeed":
+            hs_speed = command["params"]["rpm"]
+            shake_time = datetime.strptime(
+                command.get("completedAt", ""), "%Y-%m-%dT%H:%M:%S.%f%z"
+            )
+        # On Time
+        elif commandType == "heaterShaker/setTargetTemperature":
+            # if heater shaker temp is not deactivated.
+            hs_temp = command["params"]["celsius"]
+            temp_time = datetime.strptime(
+                command.get("completedAt", ""), "%Y-%m-%dT%H:%M:%S.%f%z"
+            )
+
+    hs_total_rotations = sum(hs_rotations.values())
+    hs_total_temp_time = sum(hs_temps.values())
+    hs_dict = {
+        "Heatershaker # of Latch Engagements": hs_latch_count,
+        "Heatershaker # of Homes": hs_home_count,
+        "Heatershaker # of Rotations": hs_total_rotations,
+        "Heatershaker Temp On Time (sec)": hs_total_temp_time,
+    }
+    return hs_dict
+
+
+def temperature_module_commands(file_results: Dict[str, Any]) -> Dict[str, float]:
+    """Get # of temp changes and total temp on time for temperature module from run log."""
+    # TODO: modify for cases that have more than 1 temperature module.
+    tm_temp_change = 0
+    tm_temps: Dict[str, float] = dict()
+    temp_time = None
+    deactivate_time = None
+    commandData = file_results.get("commands", "")
+    for command in commandData:
+        commandType = command["commandType"]
+        if commandType == "temperatureModule/setTargetTemperature":
+            tm_temp = command["params"]["celsius"]
+            tm_temp_change += 1
+        if commandType == "temperatureModule/waitForTemperature":
+            temp_time = datetime.strptime(
+                command.get("completedAt", ""), "%Y-%m-%dT%H:%M:%S.%f%z"
+            )
+        if commandType == "temperatureModule/deactivate":
+            deactivate_time = datetime.strptime(
+                command.get("completedAt", ""), "%Y-%m-%dT%H:%M:%S.%f%z"
+            )
+            if temp_time is not None and deactivate_time > temp_time:
+                temp_duration = (deactivate_time - temp_time).total_seconds()
+                tm_temps[tm_temp] = tm_temps.get(tm_temp, 0.0) + temp_duration
+    if temp_time is not None and deactivate_time is None:
+        # If temperature module is not deactivated, protocol completedAt time stamp used.
+        protocol_end = datetime.strptime(
+            file_results.get("completedAt", ""), "%Y-%m-%dT%H:%M:%S.%f%z"
+        )
+        temp_duration = (protocol_end - temp_time).total_seconds()
+        tm_temps[tm_temp] = tm_temps.get(tm_temp, 0.0) + temp_duration
+    tm_total_temp_time = sum(tm_temps.values())
+    tm_dict = {
+        "Temp Module # of Temp Changes": tm_temp_change,
+        "Temp Module Temp On Time (sec)": tm_total_temp_time,
+    }
+    return tm_dict
+
+
+def thermocycler_commands(file_results: Dict[str, Any]) -> Dict[str, float]:
+    """Counts # of lid engagements, temp changes, and temp sustaining mins."""
+    # TODO: modify for cases that have more than 1 thermocycler.
+    commandData = file_results.get("commands", "")
+    lid_engagements: float = 0.0
+    block_temp_changes: float = 0.0
+    lid_temp_changes: float = 0.0
+    lid_temps: Dict[str, float] = dict()
+    block_temps: Dict[str, float] = dict()
+    lid_on_time = None
+    lid_off_time = None
+    block_on_time = None
+    block_off_time = None
+    for command in commandData:
+        commandType = command["commandType"]
+        if (
+            commandType == "thermocycler/openLid"
+            or commandType == "thermocycler/closeLid"
+        ):
+            lid_engagements += 1
+        if commandType == "thermocycler/setTargetBlockTemperature":
+            block_temp = command["params"]["celsius"]
+            block_temp_changes += 1
+            block_on_time = datetime.strptime(
+                command.get("completedAt", ""), "%Y-%m-%dT%H:%M:%S.%f%z"
+            )
+        if commandType == "thermocycler/setTargetLidTemperature":
+            lid_temp_changes += 1
+            lid_temp = command["params"]["celsius"]
+            lid_on_time = datetime.strptime(
+                command.get("completedAt", ""), "%Y-%m-%dT%H:%M:%S.%f%z"
+            )
+        if commandType == "thermocycler/deactivateLid":
+            lid_off_time = datetime.strptime(
+                command.get("completedAt", ""), "%Y-%m-%dT%H:%M:%S.%f%z"
+            )
+            if lid_on_time is not None and lid_off_time > lid_on_time:
+                lid_duration = (lid_off_time - lid_on_time).total_seconds()
+                lid_temps[lid_temp] = lid_temps.get(lid_temp, 0.0) + lid_duration
+        if commandType == "thermocycler/deactivateBlock":
+            block_off_time = datetime.strptime(
+                command.get("completedAt", ""), "%Y-%m-%dT%H:%M:%S.%f%z"
+            )
+            if block_on_time is not None and block_off_time > block_on_time:
+                block_duration = (block_off_time - block_on_time).total_seconds()
+                block_temps[block_temp] = (
+                    block_temps.get(block_temp, 0.0) + block_duration
+                )
+        if commandType == "thermocycler/runProfile":
+            profile = command["params"]["profile"]
+            total_changes = len(profile)
+            block_temp_changes += total_changes
+            for cycle in profile:
+                block_temp = cycle["celsius"]
+                block_time = cycle["holdSeconds"]
+                block_temps[block_temp] = block_temps.get(block_temp, 0.0) + block_time
+    if block_on_time is not None and block_off_time is None:
+        # If thermocycler block not deactivated protocol completedAt time stamp used
+        protocol_end = datetime.strptime(
+            file_results.get("completedAt", ""), "%Y-%m-%dT%H:%M:%S.%f%z"
+        )
+        temp_duration = (protocol_end - block_on_time).total_seconds()
+        block_temps[block_temp] = block_temps.get(block_temp, 0.0) + temp_duration
+    if lid_on_time is not None and lid_off_time is None:
+        # If thermocycler lid not deactivated protocol completedAt time stamp used
+        protocol_end = datetime.strptime(
+            file_results.get("completedAt", ""), "%Y-%m-%dT%H:%M:%S.%f%z"
+        )
+        temp_duration = (protocol_end - lid_on_time).total_seconds()
+        lid_temps[lid_temp] = block_temps.get(lid_temp, 0.0) + temp_duration
+
+    block_total_time = sum(block_temps.values())
+    lid_total_time = sum(lid_temps.values())
+
+    tc_dict = {
+        "Thermocycler # of Lid Engagements": lid_engagements,
+        "Thermocycler Block # of Temp Changes": block_temp_changes,
+        "Thermocycler Block Temp On Time (sec)": block_total_time,
+        "Thermocycler Lid # of Temp Changes": lid_temp_changes,
+        "Thermocycler Lid Temp On Time (sec)": lid_total_time,
+    }
+
+    return tc_dict
+
+
 def create_abr_data_sheet(
     storage_directory: str, file_name: str, headers: List[str]
 ) -> str:
@@ -112,7 +316,7 @@ def read_abr_data_sheet(
                 runs_in_sheet.add(run_id)
         print(f"There are {str(len(runs_in_sheet))} runs documented in the ABR sheet.")
     # Read Google Sheet
-    google_sheet.check_token()
+    google_sheet.token_check()
     google_sheet.write_header(headers)
     google_sheet.update_row_index()
     return runs_in_sheet
@@ -189,7 +393,7 @@ def get_calibration_offsets(
     health_data = response.json()
     robot_name = health_data.get("name", "")
     api_version = health_data.get("api_version", "")
-    pull_date_timestamp = datetime.datetime.now()
+    pull_date_timestamp = datetime.now()
     date = pull_date_timestamp.date().isoformat()
     file_date = str(pull_date_timestamp).replace(":", "").split(".")[0]
     calibration["Robot"] = robot_name
@@ -219,5 +423,7 @@ def get_calibration_offsets(
     )
     deck: Dict[str, Any] = response.json()
     calibration["Deck"] = deck.get("deckCalibration", "")
-    saved_file_path = save_run_log_to_json(ip, calibration, storage_directory)
+    save_name = ip + "_calibration.json"
+    saved_file_path = os.path.join(storage_directory, save_name)
+    json.dump(calibration, open(saved_file_path, mode="w"))
     return saved_file_path, calibration