File size: 8,383 Bytes
91dd2ed | 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 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 | import os
import json
import argparse
from typing import List
from tabulate import tabulate
from abc import ABC, abstractmethod
class Result(ABC):
def __init__(self, res_path, com_path_len, end_flag_len, metric_key:List[str]=[]) -> None:
super().__init__()
self.ab_path = res_path
self.rl_path = res_path[com_path_len:-end_flag_len]
self.metric_key = metric_key
@abstractmethod
def read_metric(self):
pass
@abstractmethod
def process_metric(self, metric):
pass
def get(self):
# read and process
metric = self.read_metric()
processed_metric = self.process_metric(metric)
# set result
res = {
'ab_path': self.ab_path,
'rl_path': self.rl_path,
'metric': processed_metric
}
return res
def get_all_res_path(paths, flag):
all_files = []
for path in paths:
for root, dirs, files in os.walk(path):
for file in files:
if file.endswith(flag):
all_files.append(os.path.join(root, file))
return all_files
def get_len_tuple(paths:List[str], end_flag:str) -> tuple:
if len(paths) > 1:
common_path_len = len(os.path.commonpath(paths))
end_flag_len = len(end_flag) + 1 # +1 for the '/'
max_path_len = max(len(path) for path in paths) - common_path_len - end_flag_len
elif len(paths) == 1:
path = paths[0]
p_name_len = len(path.split('/')[-2])
end_flag_len = len(end_flag) + 1 # +1 for the '/'
common_path_len = len(path) - p_name_len - end_flag_len
max_path_len = p_name_len
else:
print('No path found! Maybe you should specify the path for [baseline] by --base !')
exit(-1)
return [common_path_len, end_flag_len, max_path_len]
class AllResult:
def __init__(
self,
paths:List[str],
baseline_paths:List[str],
end_flag:str,
res_cls:Result,
incl:List[str] = None,
excl:List[str] = None,
metric_key:List[str] = []
) -> None:
super().__init__()
self.incl = incl
self.excl = excl
self.res_cls = res_cls
self.all_res = []
self.base_paths, self.base_len_tuple = self.get_base_paths(baseline_paths, end_flag)
self.res_paths, self.len_tuple = self.get_res_paths(paths, end_flag)
if self.base_len_tuple[2] + 2 > self.len_tuple[2]:
self.len_tuple[2] = self.base_len_tuple[2] + 2
self.metric_key = metric_key
def filter_paths(self, paths:List[str]):
if self.incl:
paths = [path
for path in paths
if any(tag in path for tag in self.incl)]
if self.excl:
paths = [path
for path in paths
if all(tag not in path for tag in self.excl)]
if len(paths) == 0:
print('No result found with the given tags (incl/excl)')
exit(-1)
return paths
def get_res_paths(self, paths, end_flag):
all_paths = get_all_res_path(paths, end_flag)
res_paths = self.filter_paths(all_paths)
res_paths = [path
for path in res_paths
if path not in self.base_paths]
len_tuple = get_len_tuple(res_paths, end_flag)
return res_paths, len_tuple
def get_base_paths(self, paths, end_flag):
all_paths = get_all_res_path(paths, end_flag)
len_tuple = get_len_tuple(all_paths, end_flag)
return all_paths, len_tuple
def color_best(self):
all_res = self.get()
max_info = {
key: {'idx': 0, 'value': 0, 'len': len(key)}
for key in all_res[0]['metric'].keys()
}
for i, res in enumerate(all_res):
for key, value in res['metric'].items():
key_len = max_info[key]['len']
self.all_res[i]['metric'][key] = f'{value:^{key_len}.1f}'
if value > max_info[key]['value']:
max_info[key]['value'] = value
max_info[key]['idx'] = i
for key, item in max_info.items():
idx = item['idx']
value = item['value']
key_len = max_info[key]['len']
self.all_res[idx]['metric'][key] = f'\033m\033[31m{value:^{key_len}.2f}\033[0m'
return max_info
def sorted_by(self, key:str):
all_res = self.get()
self.all_res = sorted(
all_res,
key=lambda x: x['metric'][key],
reverse=True
)
def header(self, all_res, delimiter='\t'):
if not all_res:
return 'name\tmetric'
first = '{:^{}}'.format('name', self.len_tuple[2])
metric_key = self.metric_key or all_res[0]['metric'].keys()
others = f'{delimiter}'.join(metric_key)
return f'{first} | {others}'
def print(self, delimiter='\t'):
all_res = self.get()
print(self.header(all_res, delimiter=delimiter))
for res in all_res:
rl_path = f'{res["rl_path"]:{self.len_tuple[2]}}'
if self.metric_key:
metric = f'{delimiter}'.join([f'{res["metric"][k]}' for k in self.metric_key])
else:
metric = f'{delimiter}'.join([f'{v}' for v in res['metric'].values()])
print(f'{rl_path} | {metric}')
def print_table(self):
all_res = self.get()
data = []
headers = ['name', *self.metric_key]
for res in all_res:
data.append([res['rl_path'], *res['metric'].values()])
print(tabulate(data, headers=headers, tablefmt='plain', floatfmt='.2f', numalign='center'))
def get(self):
if self.all_res:
return self.all_res
for res_path in self.res_paths:
res = self.res_cls(
res_path,
self.len_tuple[0],
self.len_tuple[1],
self.metric_key
).get()
self.all_res.append(res)
# baseline res
for i, res_path in enumerate(self.base_paths):
res = self.res_cls(
res_path,
self.base_len_tuple[0],
self.base_len_tuple[1],
self.metric_key
).get()
res['rl_path'] = f'[{res["rl_path"]}]'
self.all_res.append(res)
return self.all_res
class CapRes(Result):
sub_metric_key = 'CIDEr'
def read_metric(self):
with open(self.ab_path, 'r') as infile:
metrics = json.load(infile)
return metrics
def process_metric(self, metric):
processed_metric = {}
for key, value in metric.items():
processed_metric[key] = round(value[CapRes.sub_metric_key] * 100, 2)
for key in self.metric_key:
if key not in processed_metric:
processed_metric[key] = 0
return processed_metric
def main(args):
CapRes.sub_metric_key = args.sub_metric_key
all_res = AllResult(
args.paths,
args.base,
args.end_flag,
CapRes,
incl=args.incl,
excl=args.excl,
metric_key=['coco', 'indomain', 'neardomain', 'outdomain', 'overall', 'msrvtt', 'spotdiff', 'flickr30k'],
)
all_res.sorted_by(args.sort_by)
all_res.color_best()
all_res.print(delimiter='& ')
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser.add_argument('paths_', nargs='+', default=['./checkpoints/reproduce'], help='path of result files')
parser.add_argument('--paths', nargs='+', default=[], help='path of result files')
parser.add_argument('--base', nargs='+', default=['checkpoints/main/1.0-msrvtt/'], help='path of baseline files')
parser.add_argument('--end_flag', type=str, default='eval.json')
parser.add_argument('-e', '--excl', type=str, nargs='*', default=[])
parser.add_argument('-i', '--incl', type=str, nargs='*', default=[])
parser.add_argument('-s', '--sort_by', type=str, default='coco')
parser.add_argument('-m', '--sub_metric_key', type=str, default='CIDEr')
args = parser.parse_args()
args.paths += args.paths_
main(args)
|