From 4ed944ad08c04401661f256c6410e20c03d63427 Mon Sep 17 00:00:00 2001 From: Christian Huitema Date: Thu, 28 Mar 2024 17:28:03 -0700 Subject: [PATCH] Finalize cluster computing --- src/imrs_cluster.py | 46 +++++++++++++++++++++++++++++++-------------- 1 file changed, 32 insertions(+), 14 deletions(-) diff --git a/src/imrs_cluster.py b/src/imrs_cluster.py index 04bd3cb..0a6b1d8 100644 --- a/src/imrs_cluster.py +++ b/src/imrs_cluster.py @@ -75,26 +75,44 @@ def check_or_create_dir(dir_path): return False return True -def process_cluster(cluster_id, result_folder, tmp_folder, ithitool, dates): +def process_cluster(cluster_id, result_folder, tmp_folder, ithitool, dates, do_debug): cluster_folder = join(result_folder, cluster_id) dates = clusters[cluster_id] if check_or_create_dir(cluster_folder): for one_date in dates: file_list = dates[one_date] - ipstats_file_name = one_date + "-ipstats.csv" + ipstats_file_name = one_date + "-ipstats.csv" + report_name = cluster_id + ipstats_file_name ipstats_file = join(cluster_folder, ipstats_file_name) - tmp_file_name = cluster_id + "-" + one_date + ".txt" - tmp_file = join(tmp_folder, tmp_file_name) - with open(tmp_file_name,"wt") as F: - for file_name in file_list: - F.write(file_name + "\n") - merge_cmd = ithitool + ' -I ' + ipstats_file + " " + tmp_file_name - cmd_ret = os.system(merge_cmd) - if cmd_ret == 0: - print("Computation of " + ipstats_file + " succeeds.") + if isfile(ipstats_file): + if do_debug: + print(report_name + ": already computed.") + elif len(file_list) == 0: + if do_debug: + print(report_name + ": no cbor file.") + elif len(file_list) == 1: + cp_cmd = "cp " + file_list[0] + " " + ipstats_file_name + cp_ret = os.system(cp_cmd) + if cp_ret == 0: + if do_debug: + print(report_name + " copied.") + else: + print(report_name + " copy failed, error:" + str(cp_ret)) + return False else: - print("Computation of " + ipstats_file + " failed, error:" + str(cmd_ret)) - return False + tmp_file_name = cluster_id + "-" + one_date + ".txt" + tmp_file = join(tmp_folder, tmp_file_name) + with open(tmp_file_name,"wt") as F: + for file_name in file_list: + F.write(file_name + "\n") + merge_cmd = ithitool + ' -I ' + ipstats_file + " " + tmp_file_name + cmd_ret = os.system(merge_cmd) + if cmd_ret == 0: + if do_debug: + print(report_name + ": computed.") + else: + print(report_name + ": computation failed, error:" + str(cmd_ret)) + return False return True # main @@ -119,7 +137,7 @@ def process_cluster(cluster_id, result_folder, tmp_folder, ithitool, dates): for cluster_id in clusters: dates = clusters[cluster_id] if len(dates) > 0: - if not process_cluster(cluster_id, result_folder, tmp_folder, ithitool, dates): + if not process_cluster(cluster_id, result_folder, tmp_folder, ithitool, dates, do_debug): exit(1) except Exception as exc: traceback.print_exc()