



"""simple docstring"""







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

import numpy as np


lowercase__		:				Union[str, Any]          =	re.compile(R"""\b(a|an|the)\b""", re.UNICODE)

lowercase__		:				Tuple          =	None

def 				UpperCamelCase_    (    )						->   Tuple:

  """simple docstring"""




  lowerCAmelCase_   :				Optional[int]								= 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=lowerCAmelCase__    ,   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=lowerCAmelCase__    ,   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 				UpperCamelCase_    (    lowerCAmelCase__      :				Optional[Any]						)						->   Dict:

  """simple docstring"""




  lowerCAmelCase_   :				Dict								= {}
  for article in dataset:
    for p in article["paragraphs"]:
      for qa in p["qas"]:
        lowerCAmelCase_   :				Dict								= bool(qa['answers']['text']						)
  return qid_to_has_ans

def 				UpperCamelCase_    (    lowerCAmelCase__      :				List[str]						)						->   Tuple:

  """simple docstring"""




  def remove_articles(lowerCAmelCase__      :				Any						):
    return ARTICLES_REGEX.sub(' '    ,   lowerCAmelCase__						)

  def white_space_fix(lowerCAmelCase__      :				Union[str, Any]						):
    return " ".join(text.split()						)

  def remove_punc(lowerCAmelCase__      :				Union[str, Any]						):
    lowerCAmelCase_   :				Dict								= set(string.punctuation						)
    return "".join(ch for ch in text if ch not in exclude						)

  def lower(lowerCAmelCase__      :				Tuple						):
    return text.lower()

  return white_space_fix(remove_articles(remove_punc(lower(lowerCAmelCase__						)						)						)						)

def 				UpperCamelCase_    (    lowerCAmelCase__      :				Dict						)						->   List[Any]:

  """simple docstring"""




  if not s:
    return []
  return normalize_answer(lowerCAmelCase__						).split()

def 				UpperCamelCase_    (    lowerCAmelCase__      :				List[Any]    ,   lowerCAmelCase__      :				str						)						->   Dict:

  """simple docstring"""




  return int(normalize_answer(lowerCAmelCase__						) == normalize_answer(lowerCAmelCase__						)						)

def 				UpperCamelCase_    (    lowerCAmelCase__      :				List[Any]    ,   lowerCAmelCase__      :				Dict						)						->   List[str]:

  """simple docstring"""




  lowerCAmelCase_   :				List[str]								= get_tokens(lowerCAmelCase__						)
  lowerCAmelCase_   :				List[Any]								= get_tokens(lowerCAmelCase__						)
  lowerCAmelCase_   :				str								= collections.Counter(lowerCAmelCase__						) & collections.Counter(lowerCAmelCase__						)
  lowerCAmelCase_   :				Optional[int]								= sum(common.values()						)
  if len(lowerCAmelCase__						) == 0 or len(lowerCAmelCase__						) == 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_   :				Optional[int]								= 1.0 * num_same / len(lowerCAmelCase__						)
  lowerCAmelCase_   :				Optional[Any]								= 1.0 * num_same / len(lowerCAmelCase__						)
  lowerCAmelCase_   :				Union[str, Any]								= (2 * precision * recall) / (precision + recall)
  return fa

def 				UpperCamelCase_    (    lowerCAmelCase__      :				str    ,   lowerCAmelCase__      :				Dict						)						->   List[str]:

  """simple docstring"""




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

def 				UpperCamelCase_    (    lowerCAmelCase__      :				Any    ,   lowerCAmelCase__      :				str    ,   lowerCAmelCase__      :				Tuple    ,   lowerCAmelCase__      :				List[Any]						)						->   List[str]:

  """simple docstring"""




  lowerCAmelCase_   :				Any								= {}
  for qid, s in scores.items():
    lowerCAmelCase_   :				Tuple								= na_probs[qid] > na_prob_thresh
    if pred_na:
      lowerCAmelCase_   :				Optional[Any]								= float(not qid_to_has_ans[qid]						)
    else:
      lowerCAmelCase_   :				List[Any]								= s
  return new_scores

def 				UpperCamelCase_    (    lowerCAmelCase__      :				Any    ,   lowerCAmelCase__      :				List[str]    ,   lowerCAmelCase__      :				Union[str, Any]=None						)						->   List[str]:

  """simple docstring"""




  if not qid_list:
    lowerCAmelCase_   :				List[Any]								= len(lowerCAmelCase__						)
    return collections.OrderedDict(
        [
            ('exact', 100.0 * sum(exact_scores.values()						) / total),
            ('f1', 100.0 * sum(fa_scores.values()						) / total),
            ('total', total),
        ]						)
  else:
    lowerCAmelCase_   :				str								= len(lowerCAmelCase__						)
    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 				UpperCamelCase_    (    lowerCAmelCase__      :				Tuple    ,   lowerCAmelCase__      :				Any    ,   lowerCAmelCase__      :				int						)						->   Union[str, Any]:

  """simple docstring"""




  for k in new_eval:
    lowerCAmelCase_   :				Optional[Any]								= new_eval[k]

def 				UpperCamelCase_    (    lowerCAmelCase__      :				Union[str, Any]    ,   lowerCAmelCase__      :				Optional[Any]    ,   lowerCAmelCase__      :				List[str]    ,   lowerCAmelCase__      :				str						)						->   Tuple:

  """simple docstring"""




  plt.step(lowerCAmelCase__    ,   lowerCAmelCase__    ,   color='b'    ,   alpha=0.2    ,   where='post'						)
  plt.fill_between(lowerCAmelCase__    ,   lowerCAmelCase__    ,   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(lowerCAmelCase__						)
  plt.savefig(lowerCAmelCase__						)
  plt.clf()

def 				UpperCamelCase_    (    lowerCAmelCase__      :				str    ,   lowerCAmelCase__      :				Optional[Any]    ,   lowerCAmelCase__      :				List[Any]    ,   lowerCAmelCase__      :				Any    ,   lowerCAmelCase__      :				Optional[Any]=None    ,   lowerCAmelCase__      :				str=None						)						->   Optional[int]:

  """simple docstring"""




  lowerCAmelCase_   :				Optional[int]								= sorted(lowerCAmelCase__    ,   key=lambda lowerCAmelCase__						: na_probs[k]						)
  lowerCAmelCase_   :				Optional[int]								= 0.0
  lowerCAmelCase_   :				Tuple								= 1.0
  lowerCAmelCase_   :				Union[str, Any]								= 0.0
  lowerCAmelCase_   :				Optional[Any]								= [1.0]
  lowerCAmelCase_   :				Any								= [0.0]
  lowerCAmelCase_   :				Any								= 0.0
  for i, qid in enumerate(lowerCAmelCase__						):
    if qid_to_has_ans[qid]:
      true_pos += scores[qid]
    lowerCAmelCase_   :				Optional[int]								= true_pos / float(i + 1						)
    lowerCAmelCase_   :				List[Any]								= true_pos / float(lowerCAmelCase__						)
    if i == len(lowerCAmelCase__						) - 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(lowerCAmelCase__						)
      recalls.append(lowerCAmelCase__						)
  if out_image:
    plot_pr_curve(lowerCAmelCase__    ,   lowerCAmelCase__    ,   lowerCAmelCase__    ,   lowerCAmelCase__						)
  return {"ap": 100.0 * avg_prec}

def 				UpperCamelCase_    (    lowerCAmelCase__      :				List[Any]    ,   lowerCAmelCase__      :				str    ,   lowerCAmelCase__      :				int    ,   lowerCAmelCase__      :				int    ,   lowerCAmelCase__      :				List[Any]    ,   lowerCAmelCase__      :				Any						)						->   Union[str, Any]:

  """simple docstring"""




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

def 				UpperCamelCase_    (    lowerCAmelCase__      :				List[Any]    ,   lowerCAmelCase__      :				Union[str, Any]    ,   lowerCAmelCase__      :				Tuple    ,   lowerCAmelCase__      :				Dict						)						->   Dict:

  """simple docstring"""




  if not qid_list:
    return
  lowerCAmelCase_   :				int								= [na_probs[k] for k in qid_list]
  lowerCAmelCase_   :				Dict								= np.ones_like(lowerCAmelCase__						) / float(len(lowerCAmelCase__						)						)
  plt.hist(lowerCAmelCase__    ,   weights=lowerCAmelCase__    ,   bins=20    ,   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(lowerCAmelCase__    ,   f"na_prob_hist_{name}.png"						)						)
  plt.clf()

def 				UpperCamelCase_    (    lowerCAmelCase__      :				int    ,   lowerCAmelCase__      :				Tuple    ,   lowerCAmelCase__      :				Union[str, Any]    ,   lowerCAmelCase__      :				Optional[int]						)						->   List[str]:

  """simple docstring"""




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

def 				UpperCamelCase_    (    lowerCAmelCase__      :				Dict    ,   lowerCAmelCase__      :				int    ,   lowerCAmelCase__      :				List[str]    ,   lowerCAmelCase__      :				Optional[Any]    ,   lowerCAmelCase__      :				Dict    ,   lowerCAmelCase__      :				str						)						->   Union[str, Any]:

  """simple docstring"""




  lowerCAmelCase_						,lowerCAmelCase_   :				Tuple								= find_best_thresh(lowerCAmelCase__    ,   lowerCAmelCase__    ,   lowerCAmelCase__    ,   lowerCAmelCase__						)
  lowerCAmelCase_						,lowerCAmelCase_   :				Optional[int]								= find_best_thresh(lowerCAmelCase__    ,   lowerCAmelCase__    ,   lowerCAmelCase__    ,   lowerCAmelCase__						)
  lowerCAmelCase_   :				Tuple								= best_exact
  lowerCAmelCase_   :				Any								= exact_thresh
  lowerCAmelCase_   :				Tuple								= best_fa
  lowerCAmelCase_   :				Union[str, Any]								= fa_thresh

def 				UpperCamelCase_    (    )						->   Dict:

  """simple docstring"""




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


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

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










"""simple docstring"""







import inspect
import os
import unittest

import torch

import accelerate
from accelerate import Accelerator
from accelerate.test_utils import execute_subprocess_async, require_multi_gpu
from accelerate.utils import patch_environment







class       UpperCamelCase__   (	unittest.TestCase  ):




      """simple docstring"""




      def      SCREAMING_SNAKE_CASE__ (      self				:      Optional[int]       ):
        lowerCAmelCase_   :				Union[str, Any]								= inspect.getfile(accelerate.test_utils       )
        lowerCAmelCase_   :				Dict								= os.path.sep.join(mod_file.split(os.path.sep       )[:-1] + ['scripts', 'test_script.py']       )
        lowerCAmelCase_   :				int								= os.path.sep.join(
            mod_file.split(os.path.sep       )[:-1] + ['scripts', 'test_distributed_data_loop.py']       )
        lowerCAmelCase_   :				Union[str, Any]								= os.path.sep.join(mod_file.split(os.path.sep       )[:-1] + ['scripts', 'test_ops.py']       )



      @require_multi_gpu
      def      SCREAMING_SNAKE_CASE__ (      self				:      Any       ):
        print(F"Found {torch.cuda.device_count()} devices."       )
        lowerCAmelCase_   :				Optional[int]								= ['torchrun', F"--nproc_per_node={torch.cuda.device_count()}", self.test_file_path]
        with patch_environment(omp_num_threads=1       ):
          execute_subprocess_async(SCREAMING_SNAKE_CASE_    ,			env=os.environ.copy()       )



      @require_multi_gpu
      def      SCREAMING_SNAKE_CASE__ (      self				:      Optional[Any]       ):
        print(F"Found {torch.cuda.device_count()} devices."       )
        lowerCAmelCase_   :				int								= ['torchrun', F"--nproc_per_node={torch.cuda.device_count()}", self.operation_file_path]
        print(F"Command: {cmd}"       )
        with patch_environment(omp_num_threads=1       ):
          execute_subprocess_async(SCREAMING_SNAKE_CASE_    ,			env=os.environ.copy()       )



      @require_multi_gpu
      def      SCREAMING_SNAKE_CASE__ (      self				:      Optional[int]       ):
        lowerCAmelCase_   :				List[Any]								= ['torchrun', F"--nproc_per_node={torch.cuda.device_count()}", inspect.getfile(self.__class__       )]
        with patch_environment(omp_num_threads=1       ):
          execute_subprocess_async(SCREAMING_SNAKE_CASE_    ,			env=os.environ.copy()       )





      @require_multi_gpu
      def      SCREAMING_SNAKE_CASE__ (      self				:      Union[str, Any]       ):
        print(F"Found {torch.cuda.device_count()} devices, using 2 devices only"       )
        lowerCAmelCase_   :				Any								= ['torchrun', F"--nproc_per_node={torch.cuda.device_count()}", self.data_loop_file_path]
        with patch_environment(omp_num_threads=1    ,			cuda_visible_devices='0,1'       ):
          execute_subprocess_async(SCREAMING_SNAKE_CASE_    ,			env=os.environ.copy()       )


if __name__ == "__main__":
     lowercase__		:				Dict          =	Accelerator()
     lowercase__		:				List[Any]          =	(accelerator.state.process_index + 2, 1_0)
     lowercase__		:				Any          =	torch.randint(0, 1_0, shape).to(accelerator.device)

     lowercase__		:				List[Any]          =	""""""

     lowercase__		:				Union[str, Any]          =	accelerator.pad_across_processes(tensor)
     if tensora.shape[0] != accelerator.state.num_processes + 1:
          error_msg += f"Found shape {tensora.shape} but should have {accelerator.state.num_processes + 1} at dim 0."
     if not torch.equal(tensora[: accelerator.state.process_index + 2], tensor):
          error_msg += "Tensors have different values."
     if not torch.all(tensora[accelerator.state.process_index + 2 :] == 0):
          error_msg += "Padding was not done with the right value (0)."

     lowercase__		:				Optional[Any]          =	accelerator.pad_across_processes(tensor, pad_first=True)
     if tensora.shape[0] != accelerator.state.num_processes + 1:
          error_msg += f"Found shape {tensora.shape} but should have {accelerator.state.num_processes + 1} at dim 0."
     lowercase__		:				Any          =	accelerator.state.num_processes - accelerator.state.process_index - 1
     if not torch.equal(tensora[index:], tensor):
          error_msg += "Tensors have different values."
     if not torch.all(tensora[:index] == 0):
          error_msg += "Padding was not done with the right value (0)."

     # Raise error at the end to make sure we don't stop at the first failure.
     if len(error_msg) > 0:
          raise ValueError(error_msg)





