| 10 | |
| 11 | |
| 12 | def main(result_folder: str, config_json: str) -> None: |
| 13 | all_data = {} |
| 14 | template_to_id: dict[str, Any] = defaultdict(lambda: len(template_to_id)) |
| 15 | |
| 16 | with open(config_json, "r") as f: |
| 17 | data_configs = json.load(f) |
| 18 | data_configs = {int(item["task_id"]): item for item in data_configs} |
| 19 | for k, v in data_configs.items(): |
| 20 | v.pop("require_login") |
| 21 | v.pop("storage_state") |
| 22 | v.pop("start_url") |
| 23 | v.pop("geolocation") |
| 24 | v.pop("require_reset") |
| 25 | v.pop("intent_template_id") |
| 26 | v["intent_template_id"] = template_to_id[v["intent_template"]] |
| 27 | v["eval_types"] = v["eval"].pop("eval_types") |
| 28 | if v["eval"]["reference_answers"]: |
| 29 | v["reference_answers"] = v["eval"].pop("reference_answers") |
| 30 | if v["eval"]["reference_url"]: |
| 31 | v["reference_url"] = v["eval"].pop("reference_url") |
| 32 | v.pop("eval") |
| 33 | if v.get("reference_answers", {}).get("exact_match", "") == "N/A": |
| 34 | v["achievable"] = False |
| 35 | else: |
| 36 | v["achievable"] = True |
| 37 | |
| 38 | with open(f"{result_folder}/merged_log.txt", "r") as f: |
| 39 | results = {} |
| 40 | for line in f: |
| 41 | if "[Result]" in line: |
| 42 | id = line.strip().split(".")[-2].split("/")[-1] |
| 43 | results[int(id)] = True if "(PASS)" in line else False |
| 44 | |
| 45 | files = list(glob.glob(f"{result_folder}/render_*.html")) |
| 46 | files = [x for x in files if os.path.exists(x)] |
| 47 | print(f"Total number of files: {len(files)}") |
| 48 | |
| 49 | for render_file in files: |
| 50 | task_id = int(render_file.split("_")[-1].split(".")[0]) |
| 51 | with open(render_file, "r") as f: |
| 52 | try: |
| 53 | content = f.read() |
| 54 | soup = BeautifulSoup(content, "html.parser") |
| 55 | observations = [ |
| 56 | obv.find("pre").text |
| 57 | for obv in soup.find_all("div", {"class": "state_obv"}) |
| 58 | ] |
| 59 | base64_images = [ |
| 60 | img["src"].split(",")[1] for img in soup.find_all("img") |
| 61 | ] |
| 62 | image_observations = [] |
| 63 | # save image to file and change the value to be path |
| 64 | image_folder = f"images/{os.path.basename(result_folder)}" |
| 65 | os.makedirs(image_folder, exist_ok=True) |
| 66 | for i, image in enumerate(base64_images): |
| 67 | image_data = base64.b64decode(image) |
| 68 | filename = f"{image_folder}/image_{task_id}_{i}.png" |
| 69 | with open(filename, "wb") as f: # type: ignore[assignment] |