

"""simple docstring"""

import argparse
import collections
import json
import os
import re
import string
import sys

import numpy as np


_UpperCamelCase		     =    re.compile(r"""\b(a|an|the)\b""", re.UNICODE)

_UpperCamelCase		     =    None



def       SCREAMING_SNAKE_CASE							( )			->						Dict:
 lowerCAmelCase__   :				Optional[Any] 			=				argparse.ArgumentParser("Official evaluation script for SQuAD version 2.0."   )
 parser.add_argument("data_file"					,					metavar="data.json"					,					help="Input data JSON file."   )
 parser.add_argument("pred_file"					,					metavar="pred.json"					,					help="Model predictions."   )
 parser.add_argument(
     "--out-file"					,					"-o"					,					metavar="eval.json"					,					help="Write accuracy metrics to file (default is stdout)."   )
 parser.add_argument(
     "--na-prob-file"					,					"-n"					,					metavar="na_prob.json"					,					help="Model estimates of probability of no answer."   )
 parser.add_argument(
     "--na-prob-thresh"					,					"-t"					,					type=lowercase__					,					default=1.0					,					help="Predict \"\" if no-answer probability exceeds this (default = 1.0)."					,					)
 parser.add_argument(
     "--out-image-dir"					,					"-p"					,					metavar="out_images"					,					default=lowercase__					,					help="Save precision-recall curves to directory."   )
 parser.add_argument("--verbose"					,					"-v"					,					action="store_true"   )
 if len(sys.argv   ) == 1:
  parser.print_help()
  sys.exit(1   )
 return parser.parse_args()



def       SCREAMING_SNAKE_CASE							( lowercase__   )			->						List[Any]:
 lowerCAmelCase__   :				Optional[Any] 			=				{}
 for article in dataset:
  for p in article["paragraphs"]:
   for qa in p["qas"]:
    lowerCAmelCase__   :				Tuple 			=				bool(qa["answers"]["text"]   )
 return qid_to_has_ans



def       SCREAMING_SNAKE_CASE							( lowercase__   )			->						Tuple:

 def remove_articles(lowercase__   ):
  return ARTICLES_REGEX.sub(" "					,					lowercase__   )

 def white_space_fix(lowercase__   ):
  return " ".join(text.split()   )

 def remove_punc(lowercase__   ):
  lowerCAmelCase__   :				Dict 			=				set(string.punctuation   )
  return "".join(ch for ch in text if ch not in exclude   )

 def lower(lowercase__   ):
  return text.lower()

 return white_space_fix(remove_articles(remove_punc(lower(lowercase__   )   )   )   )



def       SCREAMING_SNAKE_CASE							( lowercase__   )			->						Tuple:
 if not s:
  return []
 return normalize_answer(lowercase__   ).split()



def       SCREAMING_SNAKE_CASE							( lowercase__					,					lowercase__   )			->						int:
 return int(normalize_answer(lowercase__   ) == normalize_answer(lowercase__   )   )



def       SCREAMING_SNAKE_CASE							( lowercase__					,					lowercase__   )			->						Optional[Any]:
 lowerCAmelCase__   :				List[str] 			=				get_tokens(lowercase__   )
 lowerCAmelCase__   :				List[Any] 			=				get_tokens(lowercase__   )
 lowerCAmelCase__   :				Any 			=				collections.Counter(lowercase__   ) & collections.Counter(lowercase__   )
 lowerCAmelCase__   :				Optional[Any] 			=				sum(common.values()   )
 if len(lowercase__   ) == 0 or len(lowercase__   ) == 0:
  # If either is no-answer, then F1 is 1 if they agree, 0 otherwise
  return int(gold_toks == pred_toks   )
 if num_same == 0:
  return 0
 lowerCAmelCase__   :				int 			=				1.0 * num_same / len(lowercase__   )
 lowerCAmelCase__   :				Any 			=				1.0 * num_same / len(lowercase__   )
 lowerCAmelCase__   :				Tuple 			=				(2 * precision * recall) / (precision + recall)
 return fa



def       SCREAMING_SNAKE_CASE							( lowercase__					,					lowercase__   )			->						int:
 lowerCAmelCase__   :				Dict 			=				{}
 lowerCAmelCase__   :				Union[str, Any] 			=				{}
 for article in dataset:
  for p in article["paragraphs"]:
   for qa in p["qas"]:
    lowerCAmelCase__   :				Any 			=				qa["id"]
    lowerCAmelCase__   :				Optional[int] 			=				[t for t in qa["answers"]["text"] if normalize_answer(lowercase__   )]
    if not gold_answers:
     # For unanswerable questions, only correct answer is empty string
     lowerCAmelCase__   :				List[Any] 			=				[""]
    if qid not in preds:
     print(F"""Missing prediction for {qid}"""   )
     continue
    lowerCAmelCase__   :				Tuple 			=				preds[qid]
    # Take max over all gold answers
    lowerCAmelCase__   :				Any 			=				max(compute_exact(lowercase__					,					lowercase__   ) for a in gold_answers   )
    lowerCAmelCase__   :				Optional[Any] 			=				max(compute_fa(lowercase__					,					lowercase__   ) for a in gold_answers   )
 return exact_scores, fa_scores



def       SCREAMING_SNAKE_CASE							( lowercase__					,					lowercase__					,					lowercase__					,					lowercase__   )			->						Optional[Any]:
 lowerCAmelCase__   :				Optional[Any] 			=				{}
 for qid, s in scores.items():
  lowerCAmelCase__   :				str 			=				na_probs[qid] > na_prob_thresh
  if pred_na:
   lowerCAmelCase__   :				Dict 			=				float(not qid_to_has_ans[qid]   )
  else:
   lowerCAmelCase__   :				str 			=				s
 return new_scores



def       SCREAMING_SNAKE_CASE							( lowercase__					,					lowercase__					,					lowercase__=None   )			->						Dict:
 if not qid_list:
  lowerCAmelCase__   :				Optional[int] 			=				len(lowercase__   )
  return collections.OrderedDict(
      [
          ("exact", 100.0 * sum(exact_scores.values()   ) / total),
          ("f1", 100.0 * sum(fa_scores.values()   ) / total),
          ("total", total),
      ]   )
 else:
  lowerCAmelCase__   :				List[Any] 			=				len(lowercase__   )
  return collections.OrderedDict(
      [
          ("exact", 100.0 * sum(exact_scores[k] for k in qid_list   ) / total),
          ("f1", 100.0 * sum(fa_scores[k] for k in qid_list   ) / total),
          ("total", total),
      ]   )



def       SCREAMING_SNAKE_CASE							( lowercase__					,					lowercase__					,					lowercase__   )			->						int:
 for k in new_eval:
  lowerCAmelCase__   :				Optional[int] 			=				new_eval[k]



def       SCREAMING_SNAKE_CASE							( lowercase__					,					lowercase__					,					lowercase__					,					lowercase__   )			->						List[Any]:
 plt.step(lowercase__					,					lowercase__					,					color="b"					,					alpha=0.2					,					where="post"   )
 plt.fill_between(lowercase__					,					lowercase__					,					step="post"					,					alpha=0.2					,					color="b"   )
 plt.xlabel("Recall"   )
 plt.ylabel("Precision"   )
 plt.xlim([0.0, 1.05]   )
 plt.ylim([0.0, 1.05]   )
 plt.title(lowercase__   )
 plt.savefig(lowercase__   )
 plt.clf()



def       SCREAMING_SNAKE_CASE							( lowercase__					,					lowercase__					,					lowercase__					,					lowercase__					,					lowercase__=None					,					lowercase__=None   )			->						Optional[Any]:
 lowerCAmelCase__   :				str 			=				sorted(lowercase__					,					key=lambda lowercase__   : na_probs[k]   )
 lowerCAmelCase__   :				List[str] 			=				0.0
 lowerCAmelCase__   :				Union[str, Any] 			=				1.0
 lowerCAmelCase__   :				Optional[Any] 			=				0.0
 lowerCAmelCase__   :				List[Any] 			=				[1.0]
 lowerCAmelCase__   :				str 			=				[0.0]
 lowerCAmelCase__   :				List[Any] 			=				0.0
 for i, qid in enumerate(lowercase__   ):
  if qid_to_has_ans[qid]:
   true_pos += scores[qid]
  lowerCAmelCase__   :				Tuple 			=				true_pos / float(i + 1   )
  lowerCAmelCase__   :				Any 			=				true_pos / float(lowercase__   )
  if i == len(lowercase__   ) - 1 or na_probs[qid] != na_probs[qid_list[i + 1]]:
   # i.e., if we can put a threshold after this point
   avg_prec += cur_p * (cur_r - recalls[-1])
   precisions.append(lowercase__   )
   recalls.append(lowercase__   )
 if out_image:
  plot_pr_curve(lowercase__					,					lowercase__					,					lowercase__					,					lowercase__   )
 return {"ap": 100.0 * avg_prec}



def       SCREAMING_SNAKE_CASE							( lowercase__					,					lowercase__					,					lowercase__					,					lowercase__					,					lowercase__					,					lowercase__   )			->						str:
 if out_image_dir and not os.path.exists(lowercase__   ):
  os.makedirs(lowercase__   )
 lowerCAmelCase__   :				Dict 			=				sum(1 for v in qid_to_has_ans.values() if v   )
 if num_true_pos == 0:
  return
 lowerCAmelCase__   :				str 			=				make_precision_recall_eval(
     lowercase__					,					lowercase__					,					lowercase__					,					lowercase__					,					out_image=os.path.join(lowercase__					,					"pr_exact.png"   )					,					title="Precision-Recall curve for Exact Match score"					,					)
 lowerCAmelCase__   :				int 			=				make_precision_recall_eval(
     lowercase__					,					lowercase__					,					lowercase__					,					lowercase__					,					out_image=os.path.join(lowercase__					,					"pr_f1.png"   )					,					title="Precision-Recall curve for F1 score"					,					)
 lowerCAmelCase__   :				int 			=				{k: float(lowercase__   ) for k, v in qid_to_has_ans.items()}
 lowerCAmelCase__   :				List[str] 			=				make_precision_recall_eval(
     lowercase__					,					lowercase__					,					lowercase__					,					lowercase__					,					out_image=os.path.join(lowercase__					,					"pr_oracle.png"   )					,					title="Oracle Precision-Recall curve (binary task of HasAns vs. NoAns)"					,					)
 merge_eval(lowercase__					,					lowercase__					,					"pr_exact"   )
 merge_eval(lowercase__					,					lowercase__					,					"pr_f1"   )
 merge_eval(lowercase__					,					lowercase__					,					"pr_oracle"   )



def       SCREAMING_SNAKE_CASE							( lowercase__					,					lowercase__					,					lowercase__					,					lowercase__   )			->						Dict:
 if not qid_list:
  return
 lowerCAmelCase__   :				Dict 			=				[na_probs[k] for k in qid_list]
 lowerCAmelCase__   :				Any 			=				np.ones_like(lowercase__   ) / float(len(lowercase__   )   )
 plt.hist(lowercase__					,					weights=lowercase__					,					bins=2_0					,					range=(0.0, 1.0)   )
 plt.xlabel("Model probability of no-answer"   )
 plt.ylabel("Proportion of dataset"   )
 plt.title(F"""Histogram of no-answer probability: {name}"""   )
 plt.savefig(os.path.join(lowercase__					,					F"""na_prob_hist_{name}.png"""   )   )
 plt.clf()



def       SCREAMING_SNAKE_CASE							( lowercase__					,					lowercase__					,					lowercase__					,					lowercase__   )			->						Optional[Any]:
 lowerCAmelCase__   :				Optional[Any] 			=				sum(1 for k in qid_to_has_ans if not qid_to_has_ans[k]   )
 lowerCAmelCase__   :				str 			=				num_no_ans
 lowerCAmelCase__   :				Dict 			=				cur_score
 lowerCAmelCase__   :				Union[str, Any] 			=				0.0
 lowerCAmelCase__   :				Tuple 			=				sorted(lowercase__					,					key=lambda lowercase__   : na_probs[k]   )
 for i, qid in enumerate(lowercase__   ):
  if qid not in scores:
   continue
  if qid_to_has_ans[qid]:
   lowerCAmelCase__   :				Tuple 			=				scores[qid]
  else:
   if preds[qid]:
    lowerCAmelCase__   :				Tuple 			=				-1
   else:
    lowerCAmelCase__   :				Optional[Any] 			=				0
  cur_score += diff
  if cur_score > best_score:
   lowerCAmelCase__   :				Optional[Any] 			=				cur_score
   lowerCAmelCase__   :				Union[str, Any] 			=				na_probs[qid]
 return 100.0 * best_score / len(lowercase__   ), best_thresh



def       SCREAMING_SNAKE_CASE							( lowercase__					,					lowercase__					,					lowercase__					,					lowercase__					,					lowercase__					,					lowercase__   )			->						Any:
 lowerCAmelCase__			,						lowerCAmelCase__   :				int 			=				find_best_thresh(lowercase__					,					lowercase__					,					lowercase__					,					lowercase__   )
 lowerCAmelCase__			,						lowerCAmelCase__   :				Tuple 			=				find_best_thresh(lowercase__					,					lowercase__					,					lowercase__					,					lowercase__   )
 lowerCAmelCase__   :				int 			=				best_exact
 lowerCAmelCase__   :				List[str] 			=				exact_thresh
 lowerCAmelCase__   :				List[str] 			=				best_fa
 lowerCAmelCase__   :				int 			=				fa_thresh



def       SCREAMING_SNAKE_CASE							( )			->						List[str]:
 with open(OPTS.data_file   ) as f:
  lowerCAmelCase__   :				str 			=				json.load(lowercase__   )
  lowerCAmelCase__   :				str 			=				dataset_json["data"]
 with open(OPTS.pred_file   ) as f:
  lowerCAmelCase__   :				str 			=				json.load(lowercase__   )
 if OPTS.na_prob_file:
  with open(OPTS.na_prob_file   ) as f:
   lowerCAmelCase__   :				List[Any] 			=				json.load(lowercase__   )
 else:
  lowerCAmelCase__   :				List[str] 			=				{k: 0.0 for k in preds}
 lowerCAmelCase__   :				Optional[Any] 			=				make_qid_to_has_ans(lowercase__   )  # maps qid to True/False
 lowerCAmelCase__   :				Optional[Any] 			=				[k for k, v in qid_to_has_ans.items() if v]
 lowerCAmelCase__   :				List[Any] 			=				[k for k, v in qid_to_has_ans.items() if not v]
 lowerCAmelCase__			,						lowerCAmelCase__   :				str 			=				get_raw_scores(lowercase__					,					lowercase__   )
 lowerCAmelCase__   :				str 			=				apply_no_ans_threshold(lowercase__					,					lowercase__					,					lowercase__					,					OPTS.na_prob_thresh   )
 lowerCAmelCase__   :				int 			=				apply_no_ans_threshold(lowercase__					,					lowercase__					,					lowercase__					,					OPTS.na_prob_thresh   )
 lowerCAmelCase__   :				str 			=				make_eval_dict(lowercase__					,					lowercase__   )
 if has_ans_qids:
  lowerCAmelCase__   :				Optional[int] 			=				make_eval_dict(lowercase__					,					lowercase__					,					qid_list=lowercase__   )
  merge_eval(lowercase__					,					lowercase__					,					"HasAns"   )
 if no_ans_qids:
  lowerCAmelCase__   :				Optional[Any] 			=				make_eval_dict(lowercase__					,					lowercase__					,					qid_list=lowercase__   )
  merge_eval(lowercase__					,					lowercase__					,					"NoAns"   )
 if OPTS.na_prob_file:
  find_all_best_thresh(lowercase__					,					lowercase__					,					lowercase__					,					lowercase__					,					lowercase__					,					lowercase__   )
 if OPTS.na_prob_file and OPTS.out_image_dir:
  run_precision_recall_analysis(lowercase__					,					lowercase__					,					lowercase__					,					lowercase__					,					lowercase__					,					OPTS.out_image_dir   )
  histogram_na_prob(lowercase__					,					lowercase__					,					OPTS.out_image_dir					,					"hasAns"   )
  histogram_na_prob(lowercase__					,					lowercase__					,					OPTS.out_image_dir					,					"noAns"   )
 if OPTS.out_file:
  with open(OPTS.out_file					,					"w"   ) as f:
   json.dump(lowercase__					,					lowercase__   )
 else:
  print(json.dumps(lowercase__					,					indent=2   )   )


if __name__ == "__main__":
    _UpperCamelCase		     =    parse_args()
    if OPTS.out_image_dir:
        import matplotlib

        matplotlib.use("""Agg""")
        import matplotlib.pyplot as plt
    main()









"""simple docstring"""

from typing import Union

from ..utils import add_end_docstrings, is_torch_available, is_vision_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
    from PIL import Image

    from ..image_utils import load_image

if is_torch_available():
    from ..models.auto.modeling_auto import MODEL_FOR_VISUAL_QUESTION_ANSWERING_MAPPING

_UpperCamelCase		     =    logging.get_logger(__name__)






@add_end_docstrings(__magic_name__       )
class 			__a					(						__magic_name__       ):





       """simple docstring"""


       def __init__(   self			,					*snake_case			,					**snake_case   ):





        """simple docstring"""

        super().__init__(*snake_case			,					**snake_case   )
        self.check_model_type(snake_case   )






       def 			SCREAMING_SNAKE_CASE_	(   self			,					snake_case=None			,					snake_case=None			,					snake_case=None			,					**snake_case   ):





        """simple docstring"""

        lowerCAmelCase__			,						lowerCAmelCase__   :				List[Any] 			=				{}, {}
        if padding is not None:
         lowerCAmelCase__   :				Dict 			=				padding
        if truncation is not None:
         lowerCAmelCase__   :				Tuple 			=				truncation
        if top_k is not None:
         lowerCAmelCase__   :				str 			=				top_k
        return preprocess_params, {}, postprocess_params






       def __call__(   self			,					snake_case			,					snake_case = None			,					**snake_case   ):





        """simple docstring"""

        if isinstance(snake_case			,					(Image.Image, str)   ) and isinstance(snake_case			,					snake_case   ):
         lowerCAmelCase__   :				int 			=				{"image": image, "question": question}
        else:
         lowerCAmelCase__   :				Optional[int] 			=				image
        lowerCAmelCase__   :				Optional[Any] 			=				super().__call__(snake_case			,					**snake_case   )
        return results






       def 			SCREAMING_SNAKE_CASE_	(   self			,					snake_case			,					snake_case=False			,					snake_case=False   ):





        """simple docstring"""

        lowerCAmelCase__   :				Union[str, Any] 			=				load_image(inputs["image"]   )
        lowerCAmelCase__   :				str 			=				self.tokenizer(
            inputs["question"]			,					return_tensors=self.framework			,					padding=snake_case			,					truncation=snake_case   )
        lowerCAmelCase__   :				int 			=				self.image_processor(images=snake_case			,					return_tensors=self.framework   )
        model_inputs.update(snake_case   )
        return model_inputs






       def 			SCREAMING_SNAKE_CASE_	(   self			,					snake_case   ):





        """simple docstring"""

        lowerCAmelCase__   :				Optional[int] 			=				self.model(**snake_case   )
        return model_outputs






       def 			SCREAMING_SNAKE_CASE_	(   self			,					snake_case			,					snake_case=5   ):





        """simple docstring"""

        if top_k > self.model.config.num_labels:
         lowerCAmelCase__   :				List[Any] 			=				self.model.config.num_labels

        if self.framework == "pt":
         lowerCAmelCase__   :				Optional[int] 			=				model_outputs.logits.sigmoid()[0]
         lowerCAmelCase__			,						lowerCAmelCase__   :				Optional[int] 			=				probs.topk(snake_case   )
        else:
         raise ValueError(F"""Unsupported framework: {self.framework}"""   )

        lowerCAmelCase__   :				int 			=				scores.tolist()
        lowerCAmelCase__   :				str 			=				ids.tolist()
        return [{"score": score, "answer": self.model.config.idalabel[_id]} for score, _id in zip(snake_case			,					snake_case   )]






