Function merge_logs

webarena/scripts/check_error_runs.py:11–84

(result_folder: str, args: argparse.Namespace)

def merge_logs(result_folder: str, args: argparse.Namespace) -> str:
    if not os.path.exists(f"{result_folder}/log_files.txt"):
        sys.exit(1)

    with open(f"{result_folder}/log_files.txt", "r") as f:
        log_files = f.readlines()

    merged_results = {}
    for file in log_files:
        with open(file.strip(), "r") as f:
            lines = f.readlines()

        cur_log: list[str] = []
        index = None
        for line in lines:
            if "[Config file]" in line:
                if (
                    cur_log
                    and index
                    and os.path.exists(f"{result_folder}/render_{index}.html")
                    and len(cur_log) >= 3
                ):
                    merged_results[index] = cur_log
                # update index and log
                index = line.split("/")[-1].split(".")[0]
                cur_log = [line]
            else:
                cur_log.append(line)

        if (
            cur_log
            and index
            and os.path.exists(f"{result_folder}/render_{index}.html")
            and len(cur_log) >= 3
        ):

            merged_results[index] = cur_log

    # sort by the key
    merged_results = dict(
        sorted(merged_results.items(), key=lambda x: int(x[0]))
    )

    merged_log_path = f"{result_folder}/tmp_merged_log.txt"
    with open(merged_log_path, "w") as f:
        for k, v in merged_results.items():
            for line in v:
                f.write(line)
    print(f"Number of examples: {len(merged_results)}")

    unlog_examples = []
    for i in range(812):
        if (
            os.path.exists(f"{result_folder}/render_{i}.html")
            and str(i) not in merged_results
        ):
            unlog_examples.append(i)