Create eval_ans_therapy.py

src/eval_ans_therapy.py

@@ -0,0 +1,156 @@
+
+import json
+import random
+import os
+import time
+from random import randint
+from os import listdir
+import numpy as np
+from sklearn.metrics import average_precision_score, f1_score,recall_score
+#delete cv2
+index_list = [728, 542, 519, 1357, 1007, 1380, 783, 733, 1327, 760, 684, 795, 1075, 917, 638, 1020, 1180, 1225, 914, 1236, 871, 1231, 1330, 602, 870, 639, 1187, 934, 626, 1255, 1064, 1281, 689, 1078, 1043, 735, 1159, 1189, 1140, 579, 554, 752, 1095, 574, 841, 1054, 622, 547, 908, 1267, 767, 1215, 739, 664, 1365, 984, 1243, 1165, 634, 972, 1329, 804, 855, 1249, 311, 415, 448, 319, 14, 472, 356, 407, 265, 241, 309, 176, 105, 383, 32, 349, 149, 436, 343, 21, 147, 31, 240, 314, 9, 100, 190, 11, 218, 20, 346, 89, 25, 112, 243, 357]
+#
+current_dir_path = os.path.dirname(os.path.realpath(__file__))
+
+def All_scores():
+    All_scores = F1_score(All_gt_binary,All_predict_binary)
+    VQA_scores = F1_score(VQA_gt_binary,VQA_predict_binary)
+    VizWiz_scores = F1_score(VizWiz_gt_binary,VizWiz_predict_binary)
+    return All_scores, VizWiz_scores, VQA_scores
+
+def F1_score(gt_binary,predict_binary):
+    # predict_binary/gt_binary is np.array
+    length_diff = len(gt_binary)-len(predict_binary)
+    assert len(gt_binary) == len(predict_binary)
+    acc = 1- np.mean(abs(gt_binary - predict_binary))
+    print("accuracy: " + str(acc) + "\n")
+
+    # consider positive is diff, negative is same
+    print("consider positive is diff, negative is same")
+    relevant_set = [gt_binary[i] for i in range(len(gt_binary)) if predict_binary[i]==1]
+    precision = np.mean(relevant_set)
+    print("precision: " + str(precision))
+
+    relevant_element = [predict_binary[i] for i in range(len(gt_binary)) if gt_binary[i]==1]
+    recall = np.mean(relevant_element)
+    print("recall: " + str(recall))
+
+    scores={}
+    scores["Overall_f1_score"] = 2*precision*recall/(precision+recall)
+    scores["Oveall_precision"]=precision
+    scores["Overall_recall"] = recall
+    print("scores",scores)
+    # scores["VizWiz_f1_score"] = 
+    # scores["VizWiz_precision"]=
+    # scores["VizWiz_recall"] = 
+
+    
+    # scores["VQA_f1_score"] = 
+    # scores["VQA_precision"]=
+    # scores["VQA_recall"] = 
+    return scores
+
+def evaluate_unanswerability(self, imgs=None):
+    if imgs == None:
+        imgs = [img for img in self.params['images']]
+    pred = []
+    gt_labels = []
+    for img in imgs:
+        gt_labels.append(self.vqa.imgToQA[img]['answerable'])
+        pred.append(self.vqaRes.imgToQA[img]['answerable'])
+    gt_labels = np.array(gt_labels)
+    pred = np.array(pred)
+    
+    gt_labels_n = 1 - gt_labels
+    pred_n = 1.0 - pred
+    average_precision = average_precision_score(gt_labels_n, pred_n)
+    one_f1_score = f1_score(gt_labels_n, pred_n > 0.5)
+    
+    self.unanswerability['average_precision'] = round(100*average_precision, self.n)
+    self.unanswerability['f1_score'] = round(100*one_f1_score, self.n)
+
+def get_overlap_index(list1, list2):
+    index_list=[]
+    for i in range(len(list1)):
+        if list1[i] in list2:
+            index_list.append(list2.index(list1[i]))
+    return index_list
+def GroundingDifference(annFile, resFile):
+    Anns_vqa = []
+    Anns_vizwiz = []
+    Ress_vqa=[]
+    Ress_vizwiz=[]
+    
+    with open(annFile,'r') as annF:
+        with open(resFile,'r') as resF:
+            anns = json.load(annF)
+            test_ress = json.load(resF)
+            
+            if len(anns) == 100:
+                # print(get_overlap_index(anns,test_ress))
+                ress = [test_ress[i] for i in index_list]
+            else:
+                ress = test_ress 
+
+            Anns_labels = np.array([ann["single_grounding"] for ann in anns])
+            Res_labels = np.array([res["single_grounding"] for res in ress])
+            for ann in anns:
+                if ann["question_id"].startswith("Viz"):
+                    Anns_vizwiz.append(ann)
+                else:
+                    Anns_vqa.append(ann)
+            for res in ress:
+                if res["question_id"].startswith("Viz"):
+                    Ress_vizwiz.append(res)
+                else:
+                    Ress_vqa.append(res)
+            # All_Ress = Ress_vizwiz+Ress_vqa
+
+            Anns_vizwiz_labels = np.array([ann["single_grounding"] for ann in Anns_vizwiz])
+            Anns_vqa_labels = np.array([ann["single_grounding"] for ann in Anns_vqa])
+
+            Ress_vizwiz_labels = np.array([res["single_grounding"] for res in Ress_vizwiz])
+            Ress_vqa_labels = np.array([res["single_grounding"] for res in Ress_vqa])
+            if len(Anns_labels) !=len(Res_labels):
+                print("unsucessful submission! The number of files you generated is not equal to the number of ground-truth files.")
+            else:
+                results ={}
+                # F1_score(Anns_labels, Res_labels)
+                # F1_score(Anns_vizwiz_labels, Ress_vizwiz_labels)
+                # F1_score(Anns_vqa_labels, Ress_vqa_labels)
+                results["overall_f1"] = round(100*f1_score(Anns_labels,Res_labels>0.5),2)
+                results['overall_precision'] = round(100*average_precision_score(Anns_labels,Res_labels>0.5), 2)
+                results['overall_recall'] = round(100*recall_score(Anns_labels,Res_labels>0.5), 2)
+                results['vqav2_f1'] = round(100*f1_score(Anns_vqa_labels,Ress_vqa_labels>0.5),2)
+                results['vqa_precision'] = round(100*average_precision_score(Anns_vqa_labels,Ress_vqa_labels>0.5), 2)
+                results['vqa_recall'] = round(100*recall_score(Anns_vqa_labels,Ress_vqa_labels>0.5), 2)
+                results['vizwiz_f1'] = round(100*f1_score(Anns_vizwiz_labels,Ress_vizwiz_labels>0.5),2)
+                results['vizwiz_precision'] = round(100*average_precision_score(Anns_vizwiz_labels,Ress_vizwiz_labels>0.5), 2)
+                results['vizwiz_recall'] = round(100*recall_score(Anns_vizwiz_labels,Ress_vizwiz_labels>0.5), 2)
+                return results
+
+
+phase_splits = {
+    "test-dev2025": ["test-dev"],
+    "test-standard2025": ["test"],
+    "test-challenge2025": ["test"]
+}
+
+
+def evaluate(test_annotation_file, resFile, phase_codename, **kwargs):
+    result = []
+    splits = phase_splits[phase_codename]
+    for split in splits:
+        annFile = os.path.join(current_dir_path, "Annotations", split + ".json")
+        # if "dev" in phase_codename:
+        print(phase_codename)
+        result.append({split: GroundingDifference(annFile,resFile)}) # return a dict from groundingDifference()
+        output = {"result": result}
+        output["submission_result"] = result
+        # output["submission_result"] = output["result"][0]
+        print(result)
+        print("Completed evaluation for Test Phase")
+        return output
+
+if __name__=="__main__":
+    evaluate("t","Results/test.json","test-dev2025")