Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Module and Pipette Run Data #15306

Merged
merged 3 commits into from
Jun 3, 2024
Merged

Module and Pipette Run Data #15306

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b84d996
relabel time to 4c col
rclarke0 May 30, 2024
75abbeb
added time to tc and gripper moves
rclarke0 May 31, 2024
8be5f14
fixed temp mod total time error
rclarke0 May 31, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions abr-testing/abr_testing/data_collection/abr_google_drive.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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())
Expand Down
87 changes: 78 additions & 9 deletions abr-testing/abr_testing/data_collection/read_robot_logs.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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]:
Expand All @@ -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
Expand Down Expand Up @@ -152,20 +204,25 @@ 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
commandData = file_results.get("commands", "")
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"
)
Expand All @@ -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

Expand All @@ -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
Expand All @@ -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"
Expand Down Expand Up @@ -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
Expand Down
Loading