| | 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) |
| |
|
