File size: 6,693 Bytes
1da285f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 | from collections import defaultdict
import os
import json
import pandas as pd
import argparse
from tqdm import tqdm
split = os.getenv('split', '')
suf_split = f'-{split}' if split else ''
ROUND = int(os.getenv('ROUND', 30))
CATEGORY_APP_MAP_FILE = f'./cat_apps{suf_split}.json'
def parse_imgid(img_id):
if isinstance(img_id, int):
img_id = str(img_id)
return int(img_id[:-3]), int(img_id[-3:])
def cal_metrics(anno_data, interact_data):
# Group annotations by image id
img_annos = {}
for anno in anno_data['annotations']:
img_id = anno['image_id']
if img_id not in img_annos:
img_annos[img_id] = []
img_annos[img_id].append(anno)
effective_interacts_rate = 0
effective_interacts_cnt = 0
coverage_rate = 0
# for sk in interact_data:
# Add current batch interact points
# Group interact points by image id
img_interact = {}
for interact in interact_data:
img_id = interact['img_id']
if img_id not in img_interact:
img_interact[img_id] = []
img_interact[img_id].append(interact)
# Calculate effective interact rate
for img_id in img_annos:
if img_id not in img_interact:
continue
effective_interacts_rate += cnt_effective_interact(img_annos[img_id], img_interact[img_id]) / len(img_interact[img_id])
effective_interacts_cnt += cnt_effective_interact(img_annos[img_id], img_interact[img_id])
effective_interacts_rate /= len(img_annos)
# Calculate coverage rate
for img_id in img_annos:
if img_id not in img_interact:
continue
coverage_rate += cnt_covered_anno(img_annos[img_id], img_interact[img_id]) / len(img_annos[img_id])
coverage_rate /= len(img_annos)
return effective_interacts_cnt, effective_interacts_rate, coverage_rate
def cnt_effective_interact(annos, interacts):
effective_interacts = 0
for interact in interacts:
for anno in annos:
if interact['X'] - anno['bbox'][0] >= 0 and \
interact['Y'] - anno['bbox'][1] >= 0 and \
interact['X'] - anno['bbox'][0] <= anno['bbox'][2] and \
interact['Y'] - anno['bbox'][1] <= anno['bbox'][3]:
effective_interacts += 1
break
return effective_interacts
def cnt_covered_anno(annos, interacts):
coverage = 0
# each interact point can only cover one interactable object
t_interacts = []
t_interacts.extend(interacts)
for anno in annos:
for interact in t_interacts:
if interact['X'] - anno['bbox'][0] >= 0 and \
interact['Y'] - anno['bbox'][1] >= 0 and \
interact['X'] - anno['bbox'][0] <= anno['bbox'][2] and \
interact['Y'] - anno['bbox'][1] <= anno['bbox'][3]:
coverage += 1
t_interacts.remove(interact)
break
return coverage
def main(args):
if (os.path.exists(os.path.join(args.output_dir, 'effective_interacts_cnt.csv')) and
os.path.exists(os.path.join(args.output_dir, 'effective_interacts_rate.csv')) and
os.path.exists(os.path.join(args.output_dir, 'coverage_rate.csv'))):
print("Evaluation results already exist. Loading existing results.")
return
with open(args.annotation, 'r') as f:
anno_data = json.load(f)
with open(args.category, 'r') as f:
category_data = json.load(f)
interact_series = args.interact_series
interact_series = [f'{interact_series}{i}.json' for i in range(0, ROUND)]
effective_interacts_cnt = defaultdict(list)
effective_interacts_rate = defaultdict(list)
coverage_rate = defaultdict(list)
for interact_file in tqdm(interact_series, desc='Processing interact files'):
with open(interact_file, 'r') as f:
interact_data = json.load(f)
cat_interact_data = {}
for cat in category_data:
cat_interact_data[cat] = []
for K in interact_data:
for interact in interact_data[K]:
app_id, img_id = parse_imgid(interact['img_id'])
if app_id in category_data[cat]:
cat_interact_data[cat].append(interact)
if args.prediction:
with open(args.prediction, 'r') as f:
pred_data = json.load(f)
pred_imgs = [pred['image_id'] for pred in pred_data]
anno_data['images'] = [img for img in anno_data['images'] if img['id'] in pred_imgs]
anno_data['annotations'] = [anno for anno in anno_data['annotations'] if anno['image_id'] in pred_imgs]
for cat in cat_interact_data:
if len(cat_interact_data[cat]) == 0:
continue
result = cal_metrics(anno_data, cat_interact_data[cat])
effective_interacts_cnt[cat].append(result[0])
effective_interacts_rate[cat].append(result[1])
coverage_rate[cat].append(result[2])
os.makedirs(args.output_dir, exist_ok=True)
eff_int_cnt_df = pd.DataFrame(effective_interacts_cnt)
eff_int_rate_df = pd.DataFrame(effective_interacts_rate)
cov_rate_df = pd.DataFrame(coverage_rate)
# Calculate average metrics for sorting
avg_effective_interacts_rate = eff_int_rate_df.mean().sort_values(ascending=False)
avg_coverage_rate = cov_rate_df.mean().sort_values(ascending=False)
# Sort columns by average effective interacts rate
eff_int_cnt_df = eff_int_cnt_df[avg_effective_interacts_rate.index]
eff_int_rate_df = eff_int_rate_df[avg_effective_interacts_rate.index]
cov_rate_df = cov_rate_df[avg_coverage_rate.index]
eff_int_cnt_df.to_csv(os.path.join(args.output_dir, 'effective_interacts_cnt.csv'), index_label='Round')
eff_int_rate_df.to_csv(os.path.join(args.output_dir, 'effective_interacts_rate.csv'), index_label='Round')
cov_rate_df.to_csv(os.path.join(args.output_dir, 'coverage_rate.csv'), index_label='Round')
if __name__ == '__main__':
parser = argparse.ArgumentParser(description='Evaluate the interact points')
parser.add_argument('-a', '--annotation', type=str, help='Path to the annotation file')
parser.add_argument('-is', '--interact_series', type=str, help='Path to the interact points file series')
parser.add_argument('-o', '--output_dir', type=str, help='Path to the result directory')
parser.add_argument('-p', '--prediction', type=str, help='Path to the prediction file')
parser.add_argument('-c', '--category', type=str, help='Path to the app category file', default=CATEGORY_APP_MAP_FILE)
args = parser.parse_args()
main(args)
|