diff --git a/hardware-testing/hardware_testing/abr_tools/abr_command_data.py b/hardware-testing/hardware_testing/abr_tools/abr_command_data.py index 55bd499bd47..99fa68a0a51 100644 --- a/hardware-testing/hardware_testing/abr_tools/abr_command_data.py +++ b/hardware-testing/hardware_testing/abr_tools/abr_command_data.py @@ -5,7 +5,7 @@ import sys import json from datetime import datetime, timedelta -from read_robot_logs import ReadRobotLogs +from . import read_robot_logs def set_up_data_sheet( @@ -20,7 +20,7 @@ def set_up_data_sheet( except FileNotFoundError: print("No google sheets credentials. Add credentials to storage notebook.") local_file_str = google_sheet_name + "-" + commandTypes - csv_name = ReadRobotLogs.create_abr_data_sheet( + csv_name = read_robot_logs.create_abr_data_sheet( storage_directory, local_file_str, headers ) @@ -305,7 +305,7 @@ def command_data_dictionary( error_code, error_instrument, error_level, - ) = ReadRobotLogs.get_error_info(file_results) + ) = read_robot_logs.get_error_info(file_results) all_pipette_commands_list = pipette_commands(file_results) all_module_commands_list = module_commands(file_results) @@ -487,33 +487,33 @@ def command_data_dictionary( google_sheet_movement, csv_movement = set_up_data_sheet( 3, google_sheet_name, "Movement", movement_headers ) - runs_from_storage = ReadRobotLogs.get_run_ids_from_storage(storage_directory) + runs_from_storage = read_robot_logs.get_run_ids_from_storage(storage_directory) i = 0 n = 0 m = 0 p = 0 - runs_in_sheet = ReadRobotLogs.read_abr_data_sheet( + runs_in_sheet = read_robot_logs.read_abr_data_sheet( storage_directory, csv_instruments, google_sheet_instruments ) - runs_to_save = ReadRobotLogs.get_unseen_run_ids(runs_from_storage, runs_in_sheet) + runs_to_save = read_robot_logs.get_unseen_run_ids(runs_from_storage, runs_in_sheet) ( runs_and_instrument_commands, runs_and_module_commands, runs_and_setup_commands, runs_and_move_commands, ) = command_data_dictionary(runs_to_save, storage_directory, i, m, n, p) - ReadRobotLogs.write_to_abr_sheet( + read_robot_logs.write_to_abr_sheet( runs_and_instrument_commands, storage_directory, csv_instruments, google_sheet_instruments, ) - ReadRobotLogs.write_to_abr_sheet( + read_robot_logs.write_to_abr_sheet( runs_and_module_commands, storage_directory, csv_modules, google_sheet_modules ) - ReadRobotLogs.write_to_abr_sheet( + read_robot_logs.write_to_abr_sheet( runs_and_setup_commands, storage_directory, csv_setup, google_sheet_setup ) - ReadRobotLogs.write_to_abr_sheet( + read_robot_logs.write_to_abr_sheet( runs_and_move_commands, storage_directory, csv_movement, google_sheet_movement ) diff --git a/hardware-testing/hardware_testing/abr_tools/abr_read_logs.py b/hardware-testing/hardware_testing/abr_tools/abr_read_logs.py index f7b4c98ae79..a0b43fad2c0 100644 --- a/hardware-testing/hardware_testing/abr_tools/abr_read_logs.py +++ b/hardware-testing/hardware_testing/abr_tools/abr_read_logs.py @@ -5,7 +5,7 @@ import json import sys from datetime import datetime, timedelta -from hardware_testing.abr_tools.read_robot_logs import ReadRobotLogs +from . import read_robot_logs def get_modules(file_results: Dict[str, str]) -> Dict[str, Any]: @@ -53,7 +53,7 @@ def create_data_dictionary( error_code, error_instrument, error_level, - ) = ReadRobotLogs.get_error_info(file_results) + ) = read_robot_logs.get_error_info(file_results) all_modules = get_modules(file_results) start_time_str, complete_time_str, start_date, run_time_min = ( @@ -171,15 +171,15 @@ def create_data_dictionary( "magneticBlockV1", "thermocyclerModuleV2", ] - runs_from_storage = ReadRobotLogs.get_run_ids_from_storage(storage_directory) - file_name_csv = ReadRobotLogs.create_abr_data_sheet( + runs_from_storage = read_robot_logs.get_run_ids_from_storage(storage_directory) + file_name_csv = read_robot_logs.create_abr_data_sheet( storage_directory, file_name, headers ) - runs_in_sheet = ReadRobotLogs.read_abr_data_sheet( + runs_in_sheet = read_robot_logs.read_abr_data_sheet( storage_directory, file_name_csv, google_sheet ) - runs_to_save = ReadRobotLogs.get_unseen_run_ids(runs_from_storage, runs_in_sheet) + runs_to_save = read_robot_logs.get_unseen_run_ids(runs_from_storage, runs_in_sheet) runs_and_robots = create_data_dictionary(runs_to_save, storage_directory) - ReadRobotLogs.write_to_abr_sheet( + read_robot_logs.write_to_abr_sheet( runs_and_robots, storage_directory, file_name_csv, google_sheet ) diff --git a/hardware-testing/hardware_testing/abr_tools/abr_run_logs.py b/hardware-testing/hardware_testing/abr_tools/abr_run_logs.py index d1e09d812db..c9362571f2b 100644 --- a/hardware-testing/hardware_testing/abr_tools/abr_run_logs.py +++ b/hardware-testing/hardware_testing/abr_tools/abr_run_logs.py @@ -5,7 +5,7 @@ import json import traceback import requests -from hardware_testing.abr_tools.read_robot_logs import ReadRobotLogs +from . import read_robot_logs def get_run_ids_from_robot(ip: str) -> Set[str]: @@ -95,11 +95,11 @@ def get_all_run_logs(storage_directory: str) -> None: ip_address_list = ip_file["ip_address_list"] print(ip_address_list) - runs_from_storage = ReadRobotLogs.get_run_ids_from_storage(storage_directory) + runs_from_storage = read_robot_logs.get_run_ids_from_storage(storage_directory) for ip in ip_address_list: try: runs = get_run_ids_from_robot(ip) - runs_to_save = ReadRobotLogs.get_unseen_run_ids(runs, runs_from_storage) + runs_to_save = read_robot_logs.get_unseen_run_ids(runs, runs_from_storage) save_runs(runs_to_save, ip, storage_directory) except Exception: print(f"Failed to read IP address: {ip}.") diff --git a/hardware-testing/hardware_testing/abr_tools/abr_scale.py b/hardware-testing/hardware_testing/abr_tools/abr_scale.py index 7fdd84563a1..b9d5c35715f 100644 --- a/hardware-testing/hardware_testing/abr_tools/abr_scale.py +++ b/hardware-testing/hardware_testing/abr_tools/abr_scale.py @@ -7,7 +7,7 @@ from typing import Any, List import argparse import csv -from hardware_testing.abr_tools.read_robot_logs import ReadRobotLogs +from . import read_robot_logs def write_to_sheets(file_name_csv: str, google_sheet: Any, row_list: List) -> None: @@ -77,7 +77,7 @@ def write_to_sheets(file_name_csv: str, google_sheet: Any, row_list: List) -> No is_stable = False # Set up csv sheet headers = ["Robot", "Date", "Timestamp", "Labware", "Mass (g)", "Measurement Step"] - all_data_csv = ReadRobotLogs.create_abr_data_sheet( + all_data_csv = read_robot_logs.create_abr_data_sheet( storage_directory, file_name, headers ) # Set up google sheet diff --git a/hardware-testing/hardware_testing/abr_tools/read_robot_logs.py b/hardware-testing/hardware_testing/abr_tools/read_robot_logs.py index 336e8f27929..8f28d392140 100644 --- a/hardware-testing/hardware_testing/abr_tools/read_robot_logs.py +++ b/hardware-testing/hardware_testing/abr_tools/read_robot_logs.py @@ -12,126 +12,122 @@ import json -class ReadRobotLogs: - """Functions to read robot logs and add to google sheet.""" +def create_abr_data_sheet(storage_directory: str, file_name: str, headers: List) -> str: + """Creates csv file to log ABR data.""" + file_name_csv = file_name + ".csv" + print(file_name_csv) + sheet_location = os.path.join(storage_directory, file_name_csv) + if os.path.exists(sheet_location): + print(f"File {sheet_location} located. Not overwriting.") + else: + with open(sheet_location, "w") as csvfile: + writer = csv.DictWriter(csvfile, fieldnames=headers) + writer.writeheader() + print(f"Created file. Located: {sheet_location}.") + return file_name_csv - def create_abr_data_sheet( - storage_directory: str, file_name: str, headers: List - ) -> str: - """Creates csv file to log ABR data.""" - file_name_csv = file_name + ".csv" - print(file_name_csv) - sheet_location = os.path.join(storage_directory, file_name_csv) - if os.path.exists(sheet_location): - print(f"File {sheet_location} located. Not overwriting.") - else: - with open(sheet_location, "w") as csvfile: - writer = csv.DictWriter(csvfile, fieldnames=headers) - writer.writeheader() - print(f"Created file. Located: {sheet_location}.") - return file_name_csv - def get_error_info(file_results: Dict[str, Any]) -> Tuple[int, str, str, str, str]: - """Determines if errors exist in run log and documents them.""" - error_levels = [] - # Read error levels file - with open(ERROR_LEVELS_PATH, "r") as error_file: - error_levels = list(csv.reader(error_file)) - num_of_errors = len(file_results["errors"]) - if num_of_errors == 0: - error_type = "" - error_code = "" - error_instrument = "" - error_level = "" - return 0, error_type, error_code, error_instrument, error_level - commands_of_run: List[Dict[str, Any]] = file_results.get("commands", []) - run_command_error: Dict[str, Any] = commands_of_run[-1] - error_str: int = len(run_command_error.get("error", "")) - if error_str > 1: - error_type = run_command_error["error"].get("errorType", "") - error_code = run_command_error["error"].get("errorCode", "") - try: - # Instrument Error - error_instrument = run_command_error["error"]["errorInfo"]["node"] - except KeyError: - # Module Error - error_instrument = run_command_error["error"]["errorInfo"].get( - "port", "" - ) - else: - error_type = file_results["errors"][0]["errorType"] - print(error_type) - error_code = file_results["errors"][0]["errorCode"] - error_instrument = file_results["errors"][0]["detail"] - for error in error_levels: - code_error = error[1] - if code_error == error_code: - error_level = error[4] +def get_error_info(file_results: Dict[str, Any]) -> Tuple[int, str, str, str, str]: + """Determines if errors exist in run log and documents them.""" + error_levels = [] + # Read error levels file + with open(ERROR_LEVELS_PATH, "r") as error_file: + error_levels = list(csv.reader(error_file)) + num_of_errors = len(file_results["errors"]) + if num_of_errors == 0: + error_type = "" + error_code = "" + error_instrument = "" + error_level = "" + return 0, error_type, error_code, error_instrument, error_level + commands_of_run: List[Dict[str, Any]] = file_results.get("commands", []) + run_command_error: Dict[str, Any] = commands_of_run[-1] + error_str: int = len(run_command_error.get("error", "")) + if error_str > 1: + error_type = run_command_error["error"].get("errorType", "") + error_code = run_command_error["error"].get("errorCode", "") + try: + # Instrument Error + error_instrument = run_command_error["error"]["errorInfo"]["node"] + except KeyError: + # Module Error + error_instrument = run_command_error["error"]["errorInfo"].get("port", "") + else: + error_type = file_results["errors"][0]["errorType"] + print(error_type) + error_code = file_results["errors"][0]["errorCode"] + error_instrument = file_results["errors"][0]["detail"] + for error in error_levels: + code_error = error[1] + if code_error == error_code: + error_level = error[4] - return num_of_errors, error_type, error_code, error_instrument, error_level + return num_of_errors, error_type, error_code, error_instrument, error_level - def write_to_abr_sheet( - runs_and_robots: Dict[Any, Dict[str, Any]], - storage_directory: str, - file_name_csv: str, - google_sheet: Any, - ) -> None: - """Write dict of data to abr csv.""" - sheet_location = os.path.join(storage_directory, file_name_csv) - list_of_runs = list(runs_and_robots.keys()) - with open(sheet_location, "a", newline="") as f: - writer = csv.writer(f) - for run in range(len(list_of_runs)): - row = runs_and_robots[list_of_runs[run]].values() - row_list = list(row) - writer.writerow(row_list) - if google_sheet.creditals.access_token_expired: - google_sheet.gc.login() - google_sheet.update_row_index() - google_sheet.write_to_row(row_list) - t.sleep(3) - def read_abr_data_sheet( - storage_directory: str, file_name_csv: str, google_sheet: Any - ) -> Set[str]: - """Reads current run sheet to determine what new run data should be added.""" - print(file_name_csv) - sheet_location = os.path.join(storage_directory, file_name_csv) - runs_in_sheet = set() - # Read the CSV file - with open(sheet_location, "r") as csv_start: - data = csv.DictReader(csv_start) - headers = data.fieldnames - if headers is not None: - for row in data: - run_id = row[headers[1]] - runs_in_sheet.add(run_id) - print( - f"There are {str(len(runs_in_sheet))} runs documented in the ABR sheet." - ) - # Read Google Sheet - if google_sheet.creditals.access_token_expired: - google_sheet.gc.login() - google_sheet.write_header(headers) - google_sheet.update_row_index() - return runs_in_sheet +def write_to_abr_sheet( + runs_and_robots: Dict[Any, Dict[str, Any]], + storage_directory: str, + file_name_csv: str, + google_sheet: Any, +) -> None: + """Write dict of data to abr csv.""" + sheet_location = os.path.join(storage_directory, file_name_csv) + list_of_runs = list(runs_and_robots.keys()) + with open(sheet_location, "a", newline="") as f: + writer = csv.writer(f) + for run in range(len(list_of_runs)): + row = runs_and_robots[list_of_runs[run]].values() + row_list = list(row) + writer.writerow(row_list) + if google_sheet.creditals.access_token_expired: + google_sheet.gc.login() + google_sheet.update_row_index() + google_sheet.write_to_row(row_list) + t.sleep(3) - def get_run_ids_from_storage(storage_directory: str) -> Set[str]: - """Read all files in storage directory, extracts run id, adds to set.""" - os.makedirs(storage_directory, exist_ok=True) - list_of_files = os.listdir(storage_directory) - run_ids = set() - for this_file in list_of_files: - read_file = os.path.join(storage_directory, this_file) - if read_file.endswith(".json"): - file_results = json.load(open(read_file)) - run_id = file_results.get("run_id", "") - if len(run_id) > 0: - run_ids.add(run_id) - return run_ids - def get_unseen_run_ids(runs: Set[str], runs_from_storage: Set[str]) -> Set[str]: - """Subtracts runs from storage from current runs being read.""" - runs_to_save = runs - runs_from_storage - print(f"There are {str(len(runs_to_save))} new run(s) to save.") - return runs_to_save +def read_abr_data_sheet( + storage_directory: str, file_name_csv: str, google_sheet: Any +) -> Set[str]: + """Reads current run sheet to determine what new run data should be added.""" + print(file_name_csv) + sheet_location = os.path.join(storage_directory, file_name_csv) + runs_in_sheet = set() + # Read the CSV file + with open(sheet_location, "r") as csv_start: + data = csv.DictReader(csv_start) + headers = data.fieldnames + if headers is not None: + for row in data: + run_id = row[headers[1]] + runs_in_sheet.add(run_id) + print(f"There are {str(len(runs_in_sheet))} runs documented in the ABR sheet.") + # Read Google Sheet + if google_sheet.creditals.access_token_expired: + google_sheet.gc.login() + google_sheet.write_header(headers) + google_sheet.update_row_index() + return runs_in_sheet + + +def get_run_ids_from_storage(storage_directory: str) -> Set[str]: + """Read all files in storage directory, extracts run id, adds to set.""" + os.makedirs(storage_directory, exist_ok=True) + list_of_files = os.listdir(storage_directory) + run_ids = set() + for this_file in list_of_files: + read_file = os.path.join(storage_directory, this_file) + if read_file.endswith(".json"): + file_results = json.load(open(read_file)) + run_id = file_results.get("run_id", "") + if len(run_id) > 0: + run_ids.add(run_id) + return run_ids + + +def get_unseen_run_ids(runs: Set[str], runs_from_storage: Set[str]) -> Set[str]: + """Subtracts runs from storage from current runs being read.""" + runs_to_save = runs - runs_from_storage + print(f"There are {str(len(runs_to_save))} new run(s) to save.") + return runs_to_save