From 190b0b372d1a112e7d6103ba33ccb33150066180 Mon Sep 17 00:00:00 2001 From: Rhyann Clarke <146747548+rclarke0@users.noreply.github.com> Date: Mon, 3 Jun 2024 15:20:28 -0400 Subject: [PATCH] Module and Pipette Run Data (#15306) # Overview Adds instrument action counts and module "reach temp" times to abr data sheet. # Test Plan # Changelog # Review requests # Risk assessment --- .../data_collection/abr_google_drive.py | 2 + .../data_collection/read_robot_logs.py | 87 +++++++++++++++++-- 2 files changed, 80 insertions(+), 9 deletions(-) diff --git a/abr-testing/abr_testing/data_collection/abr_google_drive.py b/abr-testing/abr_testing/data_collection/abr_google_drive.py index 29af9269459..e2d4826b098 100644 --- a/abr-testing/abr_testing/data_collection/abr_google_drive.py +++ b/abr-testing/abr_testing/data_collection/abr_google_drive.py @@ -110,6 +110,7 @@ def create_data_dictionary( 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) + pipette_dict = read_robot_logs.instrument_commands(file_results) notes = {"Note1": "", "Jira Link": issue_url} plate_measure = { "Plate Measured": plate, @@ -123,6 +124,7 @@ def create_data_dictionary( **hs_dict, **tm_dict, **tc_dict, + **pipette_dict, **plate_measure, } headers: List[str] = list(row_2.keys()) diff --git a/abr-testing/abr_testing/data_collection/read_robot_logs.py b/abr-testing/abr_testing/data_collection/read_robot_logs.py index 97c6072d9b1..a41688ac8fa 100644 --- a/abr-testing/abr_testing/data_collection/read_robot_logs.py +++ b/abr-testing/abr_testing/data_collection/read_robot_logs.py @@ -60,24 +60,75 @@ def lpc_data( return runs_and_lpc, headers_lpc -def command_time(command: Dict[str, str]) -> Tuple[float, float]: +def command_time(command: Dict[str, str]) -> 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 + return start_to_complete + + +def instrument_commands(file_results: Dict[str, Any]) -> Dict[str, float]: + """Count number of pipette and gripper commands per run.""" + pipettes = file_results.get("pipettes", "") + commandData = file_results.get("commands", "") + left_tip_pick_up = 0.0 + left_aspirate = 0.0 + right_tip_pick_up = 0.0 + right_aspirate = 0.0 + left_dispense = 0.0 + right_dispense = 0.0 + right_pipette_id = "" + left_pipette_id = "" + gripper_pickups = 0.0 + # Match pipette mount to id + for pipette in pipettes: + if pipette["mount"] == "right": + right_pipette_id = pipette["id"] + elif pipette["mount"] == "left": + left_pipette_id = pipette["id"] + for command in commandData: + commandType = command["commandType"] + # Count tip pick ups + if commandType == "pickUpTip": + if command["params"].get("pipetteId", "") == right_pipette_id: + right_tip_pick_up += 1 + elif command["params"].get("pipetteId", "") == left_pipette_id: + left_tip_pick_up += 1 + # Count aspirates + elif commandType == "aspirate": + if command["params"].get("pipetteId", "") == right_pipette_id: + right_aspirate += 1 + elif command["params"].get("pipetteId", "") == left_pipette_id: + left_aspirate += 1 + # count dispenses/blowouts + elif commandType == "dispense" or commandType == "blowout": + if command["params"].get("pipetteId", "") == right_pipette_id: + right_dispense += 1 + elif command["params"].get("pipetteId", "") == left_pipette_id: + left_dispense += 1 + elif ( + commandType == "moveLabware" + and command["params"]["strategy"] == "usingGripper" + ): + gripper_pickups += 1 + pipette_dict = { + "Left Pipette Total Tip Pick Up(s)": left_tip_pick_up, + "Left Pipette Total Aspirates": left_aspirate, + "Left Pipette Total Dispenses": left_dispense, + "Right Pipette Total Tip Pick Up(s)": right_tip_pick_up, + "Right Pipette Total Aspirates": right_aspirate, + "Right Pipette Total Dispenses": right_dispense, + "Gripper Pick Ups": gripper_pickups, + } + return pipette_dict def hs_commands(file_results: Dict[str, Any]) -> Dict[str, float]: @@ -93,6 +144,7 @@ def hs_commands(file_results: Dict[str, Any]) -> Dict[str, float]: temp_time = None shake_time = None deactivate_time = None + for command in commandData: commandType = command["commandType"] # Heatershaker @@ -152,10 +204,11 @@ def hs_commands(file_results: Dict[str, Any]) -> Dict[str, float]: return hs_dict -def temperature_module_commands(file_results: Dict[str, Any]) -> Dict[str, float]: +def temperature_module_commands(file_results: Dict[str, Any]) -> Dict[str, Any]: """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 + time_to_4c = 0.0 tm_temps: Dict[str, float] = dict() temp_time = None deactivate_time = None @@ -163,9 +216,13 @@ def temperature_module_commands(file_results: Dict[str, Any]) -> Dict[str, float for command in commandData: commandType = command["commandType"] if commandType == "temperatureModule/setTargetTemperature": + temp_time = datetime.strptime( + command.get("completedAt", ""), "%Y-%m-%dT%H:%M:%S.%f%z" + ) tm_temp = command["params"]["celsius"] tm_temp_change += 1 - if commandType == "temperatureModule/waitForTemperature": + if commandType == "temperatureModule/waitForTemperature" and int(tm_temp) == 4: + time_to_4c = command_time(command) temp_time = datetime.strptime( command.get("completedAt", ""), "%Y-%m-%dT%H:%M:%S.%f%z" ) @@ -187,6 +244,7 @@ def temperature_module_commands(file_results: Dict[str, Any]) -> Dict[str, float tm_dict = { "Temp Module # of Temp Changes": tm_temp_change, "Temp Module Temp On Time (sec)": tm_total_temp_time, + "Temp Mod Time to 4C (sec)": time_to_4c, } return tm_dict @@ -198,6 +256,8 @@ def thermocycler_commands(file_results: Dict[str, Any]) -> Dict[str, float]: lid_engagements: float = 0.0 block_temp_changes: float = 0.0 lid_temp_changes: float = 0.0 + block_to_4c = 0.0 + lid_to_105c = 0.0 lid_temps: Dict[str, float] = dict() block_temps: Dict[str, float] = dict() lid_on_time = None @@ -217,12 +277,19 @@ def thermocycler_commands(file_results: Dict[str, Any]) -> Dict[str, float]: block_on_time = datetime.strptime( command.get("completedAt", ""), "%Y-%m-%dT%H:%M:%S.%f%z" ) + if ( + commandType == "thermocycler/waitForBlockTemperature" + and int(block_temp) == 4 + ): + block_to_4c = command_time(command) 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/waitForLidTemperature" and int(lid_temp) == 105: + lid_to_105c = command_time(command) if commandType == "thermocycler/deactivateLid": lid_off_time = datetime.strptime( command.get("completedAt", ""), "%Y-%m-%dT%H:%M:%S.%f%z" @@ -269,8 +336,10 @@ def thermocycler_commands(file_results: Dict[str, Any]) -> Dict[str, float]: "Thermocycler # of Lid Open/Close": lid_sets, "Thermocycler Block # of Temp Changes": block_temp_changes, "Thermocycler Block Temp On Time (sec)": block_total_time, + "Thermocycler Block Time to 4C (sec)": block_to_4c, "Thermocycler Lid # of Temp Changes": lid_temp_changes, "Thermocycler Lid Temp On Time (sec)": lid_total_time, + "Thermocycler Lid Time to 105C (sec)": lid_to_105c, } return tc_dict