| import argparse |
| import json |
| import os |
| import random |
| import re |
|
|
| import numpy as np |
| import requests |
| from data.data_loader import * |
|
|
| random.seed(42) |
|
|
|
|
| def find_dataset_results(dataset_name, split, model_name): |
| """ |
| logs/eval/{model_name}/{dataset_name} |
| e.g., /logs/eval/kl_cot_gaussian_03_iouv2_2500/tvgbench |
| """ |
| data_dirs = [] |
| eval_root = f"./logs/eval/{model_name}" |
| for data_dir in os.listdir(eval_root): |
| if dataset_name in data_dir: |
| data_dirs.append(os.path.join(eval_root, data_dir)) |
| return sorted(data_dirs) |
|
|
|
|
| def get_args(): |
| parser = argparse.ArgumentParser( |
| description="Evaluation for training-free video temporal grounding (Single GPU Version)" |
| ) |
| parser.add_argument( |
| "--dataset", |
| default=[ |
| "charades", |
| "activitynet", |
| "mvbench", |
| "tvgbench", |
| "videomme", |
| "tempcompass", |
| "egoschema", |
| ], |
| help="Specify the dataset.", |
| choices=[ |
| "charades", |
| "activitynet", |
| "mvbench", |
| "videomme", |
| "tvgbench", |
| "videomme", |
| "egoschema", |
| "tempcompass", |
| ], |
| nargs="+", |
| ) |
| parser.add_argument("--split", type=str, default="test", help="dataset type") |
| parser.add_argument( |
| "--model_name", |
| type=str, |
| default="kl_cot_gaussian_03_iouv2_2500", |
| help="model name", |
| ) |
| return parser.parse_args() |
|
|
|
|
| def compute_IoU(pred, gt): |
| """Compute the IoU given predicted and ground truth windows.""" |
| assert isinstance(pred, list) and isinstance(gt, list) |
| pred_is_list = isinstance(pred[0], list) |
| gt_is_list = isinstance(gt[0], list) |
| if not pred_is_list: |
| pred = [pred] |
| if not gt_is_list: |
| gt = [gt] |
| pred, gt = np.array(pred), np.array(gt) |
| inter_left = np.maximum(pred[:, 0, None], gt[None, :, 0]) |
| inter_right = np.minimum(pred[:, 1, None], gt[None, :, 1]) |
| inter = np.maximum(0.0, inter_right - inter_left) |
| union_left = np.minimum(pred[:, 0, None], gt[None, :, 0]) |
| union_right = np.maximum(pred[:, 1, None], gt[None, :, 1]) |
| union = np.maximum(0.0, union_right - union_left) |
| overlap = 1.0 * inter / union |
| if not gt_is_list: |
| overlap = overlap[:, 0] |
| if not pred_is_list: |
| overlap = overlap[0] |
| return overlap |
|
|
|
|
| def mcq_is_correct(pred, gt): |
| gt = chr(gt + ord("A")) |
| matches = re.findall(r"\(([A-Z])\)", pred) |
| if matches: |
| return int(matches[-1] == gt) |
| return int(pred[0] == gt) |
|
|
|
|
| def load_scored_data(data_dir, datasetname): |
| data = {} |
| cnt = 0 |
| for file in os.listdir(data_dir): |
| if "jsonl" not in file: |
| continue |
| file_path = os.path.join(data_dir, file) |
| for line in open(file_path): |
| tmp = json.loads(line) |
| cnt += 1 |
| if datasetname in ["activitynet", "charades", "tvgbench"]: |
| score = 0.0 |
| if None not in tmp["pred"]: |
| score = compute_IoU(tmp["pred"], tmp["target"]) |
| else: |
| if tmp["pred"] is not None: |
| score = int(tmp["pred"] == tmp["target"]) |
| else: |
| score = mcq_is_correct(tmp["output_text"], tmp["target"]) |
| data[tmp["qid"]] = score |
| return data |
|
|
|
|
| def calc_score(difficulty_data_dict, datasetname): |
| data = list(difficulty_data_dict.values()) |
| if datasetname in ["activitynet", "charades", "tvgbench"]: |
| scores = {} |
| scores["mIoU"] = np.mean([itm for itm in data]) * 100 |
| for i in [0.3, 0.5, 0.7]: |
| cnt = len([itm for itm in data if itm > i]) |
| score = cnt / len(difficulty_data_dict) * 100.0 |
| scores[i] = score |
| scores["avg"] = sum(scores.values()) / len(scores) |
| else: |
| correct = sum([itm for itm in data]) |
| scores = { |
| "correct": correct, |
| "total": len(data), |
| "avg": round(correct / len(data) * 100, 2), |
| } |
| return scores |
|
|
|
|
| def upload_json_to_server( |
| data, api_url="https://validation-server.onrender.com/api/upload/" |
| ): |
| headers = {"Content-Type": "application/json"} |
| try: |
| response = requests.post(url=api_url, headers=headers, json=data) |
| response.raise_for_status() |
| try: |
| return response.json() |
| except ValueError: |
| return {"status": "success", "response_text": response.text} |
|
|
| except requests.exceptions.RequestException as e: |
| return { |
| "status": "error", |
| "message": str(e), |
| "details": f"Failed to upload data to {api_url}", |
| } |
|
|
|
|
| def eval_egoschema_online(data_dir, original_data): |
| qid_to_vid = {} |
| for itm in original_data: |
| qid, vid = itm["qid"], itm["video"].split("/")[-1].split(".")[0] |
| qid_to_vid[qid] = vid |
|
|
| data = {} |
| for file in os.listdir(data_dir): |
| if "jsonl" not in file: |
| continue |
| file_path = os.path.join(data_dir, file) |
| for line in open(file_path): |
| tmp = json.loads(line) |
| matches = re.findall(r"\(([A-Z])\)", tmp["output_text"]) |
| if matches: |
| pred = ord(matches[-1]) - ord("A") |
| else: |
| pred = ord(random.choice(["A", "B", "C", "D", "E"])) - ord("A") |
| data[qid_to_vid[tmp["qid"]]] = pred |
|
|
| return upload_json_to_server(data) |
|
|
|
|
| def main(args): |
| for dataset in args.dataset: |
| if dataset == "charades": |
| load_func = load_charades |
| if dataset == "activitynet": |
| load_func = load_activitynet |
| if dataset == "mvbench": |
| load_func = load_mvbench |
| if dataset == "videomme": |
| load_func = load_videomme |
| if dataset == "tvgbench": |
| load_func = load_tvgbench |
| if dataset == "egoschema": |
| load_func = load_egoschema |
| if dataset == "tempcompass": |
| load_func = load_tempcompass |
|
|
| for split in ["multi-choice"]: |
| original_data = load_func(split) |
| print(f"==========={dataset} {split}===========") |
| print(f"Original data length: {len(original_data)}") |
| for data_dir in find_dataset_results( |
| dataset, args.split, args.model_name |
| ): |
|
|
| print(f"data_dir: {data_dir}") |
| if "captioning" in data_dir: |
| continue |
| difficulty_data_dict = load_scored_data(data_dir, dataset) |
| if len(difficulty_data_dict) == 0: |
| continue |
| print(f"len(difficulty_data_dict): {len(difficulty_data_dict)}") |
| for k, v in calc_score(difficulty_data_dict, dataset).items(): |
| print(v) |
| with open(data_dir + "/scores.json", "w") as f: |
| json.dump( |
| calc_score(difficulty_data_dict, dataset), f, indent=4 |
| ) |
| continue |
|
|
| original_data = None |
| if dataset == "egoschema": |
| original_data = load_func() |
|
|
| print(f"==========={dataset}===========") |
|
|
| if original_data is not None: |
| print(f"Original data length: {len(original_data)}") |
| for data_dir in find_dataset_results(dataset, args.split, args.model_name): |
|
|
| print(f"data_dir: {data_dir}") |
| if dataset == "egoschema": |
| results_ego = eval_egoschema_online(data_dir, original_data) |
| print(results_ego) |
| with open(data_dir + "/scores.json", "w") as f: |
| json.dump(results_ego, f, indent=4) |
| continue |
|
|
| difficulty_data_dict = load_scored_data(data_dir, dataset) |
| if len(difficulty_data_dict) == 0: |
| continue |
| print(f"len(difficulty_data_dict): {len(difficulty_data_dict)}") |
| for k, v in calc_score(difficulty_data_dict, dataset).items(): |
| print(f"IoU R1@ {k}: {v}") |
| with open(data_dir + "/scores.json", "w") as f: |
| json.dump(calc_score(difficulty_data_dict, dataset), f, indent=4) |
|
|
|
|
| if __name__ == "__main__": |
| args = get_args() |
| main(args) |
|
|
