| import itertools |
| import json |
| import os |
| import re |
| from collections import namedtuple |
|
|
| import torch |
| from tqdm import tqdm |
|
|
|
|
| class InferenceSampler(torch.utils.data.sampler.Sampler): |
|
|
| def __init__(self, size): |
| self._size = int(size) |
| assert size > 0 |
| self._rank = torch.distributed.get_rank() |
| self._world_size = torch.distributed.get_world_size() |
| self._local_indices = self._get_local_indices(size, self._world_size, |
| self._rank) |
|
|
| @staticmethod |
| def _get_local_indices(total_size, world_size, rank): |
| shard_size = total_size // world_size |
| left = total_size % world_size |
| shard_sizes = [shard_size + int(r < left) for r in range(world_size)] |
|
|
| begin = sum(shard_sizes[:rank]) |
| end = min(sum(shard_sizes[:rank + 1]), total_size) |
| return range(begin, end) |
|
|
| def __iter__(self): |
| yield from self._local_indices |
|
|
| def __len__(self): |
| return len(self._local_indices) |
| |
| def collate_fn_vqa(batches): |
| ''' |
| ''' |
| image_paths = [_['image_path'] for _ in batches] |
| questions = [_['question'] for _ in batches] |
| gt_answers = [_['gt_answers'] for _ in batches] |
| ocr_tokens = [_['ocr_tokens'] if 'ocr_tokens' in _ else None for _ in batches] |
| question_ids = [_['question_id'] if 'question_id' in _ else None for _ in batches] |
| question_type = [_['question_type'] if 'question_type' in _ else None for _ in batches] |
|
|
| return image_paths, questions, gt_answers, ocr_tokens, question_ids, question_type |
|
|
| def has_word(sentence, word): |
| if word[0].isalnum(): |
| start_pattern = r"\b" |
| else: |
| start_pattern = r"" |
|
|
| if word[-1].isalnum(): |
| end_pattern = r"\b" |
| else: |
| end_pattern = r"" |
|
|
| pattern = start_pattern + re.escape(word) + end_pattern |
| match = re.search(pattern, sentence) |
| return bool(match) |
|
|
| def remove_special_chars(s): |
| pattern = r"[^a-zA-Z0-9\s]" |
| s = re.sub(pattern, "", s) |
| return s |
|
|
| def levenshtein_distance(s1, s2): |
| if len(s1) > len(s2): |
| s1, s2 = s2, s1 |
|
|
| distances = range(len(s1) + 1) |
| for i2, c2 in enumerate(s2): |
| distances_ = [i2+1] |
| for i1, c1 in enumerate(s1): |
| if c1 == c2: |
| distances_.append(distances[i1]) |
| else: |
| distances_.append(1 + min((distances[i1], distances[i1 + 1], distances_[-1]))) |
| distances = distances_ |
| return distances[-1] |
|
|
| class VQAEval: |
| def __init__(self): |
| self.contractions = { |
| "aint": "ain't", |
| "arent": "aren't", |
| "cant": "can't", |
| "couldve": "could've", |
| "couldnt": "couldn't", |
| "couldn'tve": "couldn't've", |
| "couldnt've": "couldn't've", |
| "didnt": "didn't", |
| "doesnt": "doesn't", |
| "dont": "don't", |
| "hadnt": "hadn't", |
| "hadnt've": "hadn't've", |
| "hadn'tve": "hadn't've", |
| "hasnt": "hasn't", |
| "havent": "haven't", |
| "hed": "he'd", |
| "hed've": "he'd've", |
| "he'dve": "he'd've", |
| "hes": "he's", |
| "howd": "how'd", |
| "howll": "how'll", |
| "hows": "how's", |
| "Id've": "I'd've", |
| "I'dve": "I'd've", |
| "Im": "I'm", |
| "Ive": "I've", |
| "isnt": "isn't", |
| "itd": "it'd", |
| "itd've": "it'd've", |
| "it'dve": "it'd've", |
| "itll": "it'll", |
| "let's": "let's", |
| "maam": "ma'am", |
| "mightnt": "mightn't", |
| "mightnt've": "mightn't've", |
| "mightn'tve": "mightn't've", |
| "mightve": "might've", |
| "mustnt": "mustn't", |
| "mustve": "must've", |
| "neednt": "needn't", |
| "notve": "not've", |
| "oclock": "o'clock", |
| "oughtnt": "oughtn't", |
| "ow's'at": "'ow's'at", |
| "'ows'at": "'ow's'at", |
| "'ow'sat": "'ow's'at", |
| "shant": "shan't", |
| "shed've": "she'd've", |
| "she'dve": "she'd've", |
| "she's": "she's", |
| "shouldve": "should've", |
| "shouldnt": "shouldn't", |
| "shouldnt've": "shouldn't've", |
| "shouldn'tve": "shouldn't've", |
| "somebody'd": "somebodyd", |
| "somebodyd've": "somebody'd've", |
| "somebody'dve": "somebody'd've", |
| "somebodyll": "somebody'll", |
| "somebodys": "somebody's", |
| "someoned": "someone'd", |
| "someoned've": "someone'd've", |
| "someone'dve": "someone'd've", |
| "someonell": "someone'll", |
| "someones": "someone's", |
| "somethingd": "something'd", |
| "somethingd've": "something'd've", |
| "something'dve": "something'd've", |
| "somethingll": "something'll", |
| "thats": "that's", |
| "thered": "there'd", |
| "thered've": "there'd've", |
| "there'dve": "there'd've", |
| "therere": "there're", |
| "theres": "there's", |
| "theyd": "they'd", |
| "theyd've": "they'd've", |
| "they'dve": "they'd've", |
| "theyll": "they'll", |
| "theyre": "they're", |
| "theyve": "they've", |
| "twas": "'twas", |
| "wasnt": "wasn't", |
| "wed've": "we'd've", |
| "we'dve": "we'd've", |
| "weve": "we've", |
| "werent": "weren't", |
| "whatll": "what'll", |
| "whatre": "what're", |
| "whats": "what's", |
| "whatve": "what've", |
| "whens": "when's", |
| "whered": "where'd", |
| "wheres": "where's", |
| "whereve": "where've", |
| "whod": "who'd", |
| "whod've": "who'd've", |
| "who'dve": "who'd've", |
| "wholl": "who'll", |
| "whos": "who's", |
| "whove": "who've", |
| "whyll": "why'll", |
| "whyre": "why're", |
| "whys": "why's", |
| "wont": "won't", |
| "wouldve": "would've", |
| "wouldnt": "wouldn't", |
| "wouldnt've": "wouldn't've", |
| "wouldn'tve": "wouldn't've", |
| "yall": "y'all", |
| "yall'll": "y'all'll", |
| "y'allll": "y'all'll", |
| "yall'd've": "y'all'd've", |
| "y'alld've": "y'all'd've", |
| "y'all'dve": "y'all'd've", |
| "youd": "you'd", |
| "youd've": "you'd've", |
| "you'dve": "you'd've", |
| "youll": "you'll", |
| "youre": "you're", |
| "youve": "you've", |
| } |
| self.manualMap = { |
| "none": "0", |
| "zero": "0", |
| "one": "1", |
| "two": "2", |
| "three": "3", |
| "four": "4", |
| "five": "5", |
| "six": "6", |
| "seven": "7", |
| "eight": "8", |
| "nine": "9", |
| "ten": "10", |
| } |
| self.articles = ["a", "an", "the"] |
|
|
| self.periodStrip = re.compile("(?!<=\d)(\.)(?!\d)") |
| self.commaStrip = re.compile("(\d)(\,)(\d)") |
| self.punct = [ |
| ";", |
| r"/", |
| "[", |
| "]", |
| '"', |
| "{", |
| "}", |
| "(", |
| ")", |
| "=", |
| "+", |
| "\\", |
| "_", |
| "-", |
| ">", |
| "<", |
| "@", |
| "`", |
| ",", |
| "?", |
| "!", |
| ] |
| def clean_text(self, text): |
| text = text.replace("\n", " ").replace("\t", " ").strip() |
| text = self.processPunctuation(text) |
| text = self.processDigitArticle(text) |
| return text |
| |
| def evaluate_vqa_human(self, answer, gt_answers): |
| '''TextVQA, VQAv2, OKVQA, vizwiz''' |
| answer = answer.replace("\n", " ").replace("\t", " ").strip() |
| answer = self.processPunctuation(answer) |
| answer = self.processDigitArticle(answer) |
| gt_answers = [self.processPunctuation(ans) for ans in gt_answers] |
| gt_answers = [self.processDigitArticle(ans) for ans in gt_answers] |
|
|
| gtAcc = [] |
|
|
| for idx, gtAnsDatum in enumerate(gt_answers): |
| otherGTAns = gt_answers[:idx] + gt_answers[idx+1:] |
|
|
| matchingAns = [item for item in otherGTAns if answer == item] |
| |
| acc = min(1, float(len(matchingAns)) / 3) |
| gtAcc.append(acc) |
|
|
| avgGTAcc = float(sum(gtAcc)) / len(gtAcc) if gtAcc else 0 |
|
|
| return avgGTAcc |
|
|
| def evaluate_anls(self, answer, gt_answers, threshold=0.5): |
| '''DOcVQA, InfographicsVQA, STVQA''' |
| answer = ' '.join(answer.strip().lower().split()) |
| if not isinstance(gt_answers, list): |
| gt_answers = [gt_answers] |
| gt_answers = [' '.join(gt_answer.strip().lower().split()) for gt_answer in gt_answers] |
|
|
| values = [] |
| for gt_answer in gt_answers: |
| dist = levenshtein_distance(answer, gt_answer) |
| length = max(len(answer), len(gt_answer)) |
| values.append(0.0 if length == 0 else float(dist) / float(length)) |
|
|
| score = 1 - min(values) |
| |
| score = 0 if score < threshold else score |
| |
| return score |
|
|
| def processPunctuation(self, inText): |
| outText = inText |
| for p in self.punct: |
| if (p + " " in inText or " " + p in inText) or ( |
| re.search(self.commaStrip, inText) != None |
| ): |
| outText = outText.replace(p, "") |
| else: |
| outText = outText.replace(p, " ") |
| outText = self.periodStrip.sub("", outText, re.UNICODE) |
| return outText |
|
|
| def processDigitArticle(self, inText): |
| outText = [] |
| tempText = inText.lower().split() |
| for word in tempText: |
| word = self.manualMap.setdefault(word, word) |
| if word not in self.articles: |
| outText.append(word) |
| else: |
| pass |
| for wordId, word in enumerate(outText): |
| if word in self.contractions: |
| outText[wordId] = self.contractions[word] |
| outText = " ".join(outText) |
| return outText |
|
|
|
|
| def evaluate_dataset(dataset_name, answer_file_path, model_name, method = None): |
| with open(answer_file_path, 'r', encoding='utf-8') as f: |
| predictions = json.load(f) |
|
|
| eval = VQAEval() |
| total_accuracy = 0 |
| num = 0 |
| Entry = namedtuple('Entry', ['text', 'bbox']) |
|
|
| for item in predictions: |
| gt_answers = item['gt_answers'] |
| answer = item['answer'] |
| if method is not None: |
| pass |
| if dataset_name in ["textVQA"]: |
| if num == 0: |
| print(f"evaluating vqa...") |
| accuracy = eval.evaluate_vqa_human(answer, gt_answers) |
| elif dataset_name in ['docVQA']: |
| if num == 0: |
| print(f"evaluating anls...") |
| accuracy = eval.evaluate_anls(answer, gt_answers) |
| else: |
| accuracy = eval.evaluate_has(answer, gt_answers) |
| item['accuracy'] = accuracy |
|
|
| total_accuracy += accuracy |
| num += 1 |
|
|
| average_accuracy = total_accuracy / num |
| print(f'{dataset_name}:{average_accuracy}') |
| |
| answer_model_method_path = answer_file_path.replace('.json', f'_{model_name}_{method}.json') |
| with open(answer_model_method_path, "w", encoding='utf-8') as f: |
| json.dump(predictions, f, indent=4, ensure_ascii=False) |
|
|
| return average_accuracy |
|
|
|
|
| def evaluate_VQA( |
| model, |
| dataset, |
| model_name, |
| dataset_name, |
| time, |
| batch_size=1, |
| generate_method="interleave", |
| answer_path='./answers', |
| ): |
| print(f"answer path:{answer_path}") |
|
|
| sampler = None |
| if torch.distributed.is_initialized(): |
| sampler=InferenceSampler(len(dataset)) |
|
|
| dataloader = torch.utils.data.DataLoader( |
| dataset=dataset, |
| batch_size=batch_size, |
| sampler=sampler, |
| collate_fn=collate_fn_vqa |
| ) |
| |
| now_rank = torch.distributed.get_rank() |
| |
| answer_dir = os.path.join(answer_path, model_name, time) |
| os.makedirs(answer_dir, exist_ok=True) |
| |
| image_list = [] |
| for item in dataset: |
| image_list.append(item["image_path"]) |
| |
| predictions = [] |
| |
| for batch in tqdm(dataloader, desc="Running inference"): |
| image_paths, questions, gt_answers, ocr_tokens_list, question_ids, question_type = batch |
|
|
| with torch.no_grad(): |
| if model_name != "minicpm": |
| if model_name != "codellama": |
| outputs = model.generate(images=image_paths, questions=questions, datasetname=dataset_name) |
| else: |
| outputs = model.generate() |
| elif model_name == "minicpm": |
| if generate_method == "old": |
| outputs = model.generate(images=image_paths, questions=questions, datasetname=dataset_name) |
| elif generate_method == "interleave": |
| outputs = model.generate_with_interleaved(images=image_paths, questions=questions, datasetname=dataset_name) |
| else: |
| raise Exception(f"Wrong generate paradigm {generate_method}!") |
| |
| for i in range(len(outputs)): |
| answer_dict = { |
| 'question_id': question_ids[i], |
| 'question': questions[i], |
| 'answer': outputs[i], |
| 'gt_answers': gt_answers[i], |
| 'image_path': image_paths[i], |
| 'model_name': model_name, |
| 'question_type': question_type[i] |
| } |
| predictions.append(answer_dict) |
| |
| if torch.distributed.is_initialized(): |
| torch.distributed.barrier() |
| if torch.distributed.is_initialized(): |
| world_size = torch.distributed.get_world_size() |
| merged_predictions = [None for _ in range(world_size)] |
| torch.distributed.all_gather_object(merged_predictions, predictions) |
| predictions = [_ for _ in itertools.chain.from_iterable(merged_predictions)] |
|
|
| if torch.distributed.is_initialized() and torch.distributed.get_rank() != 0: |
| return None |
| |
| answer_file_path = os.path.join(answer_dir, f"{dataset_name}.json") |
| print(f"answer_file_path:{answer_file_path}") |
| |
| with open(answer_file_path, "w", encoding='utf-8') as f: |
| json.dump(predictions, f, indent=4, ensure_ascii=False) |
|
|
| if dataset_name in ["docVQATest"]: |
| return -1.0 |
|
|
| return evaluate_dataset(answer_file_path=answer_file_path, dataset_name=dataset_name, model_name=model_name) |
|
|
