






"""simple docstring"""


import argparse
import logging
import os
from pathlib import Path
from typing import Any, Dict

import pytorch_lightning as pl
from pytorch_lightning.utilities import rank_zero_info

from transformers import (
    AdamW,
    AutoConfig,
    AutoModel,
    AutoModelForPreTraining,
    AutoModelForQuestionAnswering,
    AutoModelForSeqaSeqLM,
    AutoModelForSequenceClassification,
    AutoModelForTokenClassification,
    AutoModelWithLMHead,
    AutoTokenizer,
    PretrainedConfig,
    PreTrainedTokenizer,
)
from transformers.optimization import (
    Adafactor,
    get_cosine_schedule_with_warmup,
    get_cosine_with_hard_restarts_schedule_with_warmup,
    get_linear_schedule_with_warmup,
    get_polynomial_decay_schedule_with_warmup,
)
from transformers.utils.versions import require_version


_UpperCamelCase    :   Union[str, Any]   			=						logging.getLogger(__name__)

require_version('pytorch_lightning>=1.0.4')

_UpperCamelCase    :   int   			=						{
    'base': AutoModel,
    'sequence-classification': AutoModelForSequenceClassification,
    'question-answering': AutoModelForQuestionAnswering,
    'pretraining': AutoModelForPreTraining,
    'token-classification': AutoModelForTokenClassification,
    'language-modeling': AutoModelWithLMHead,
    'summarization': AutoModelForSeqaSeqLM,
    'translation': AutoModelForSeqaSeqLM,
}


# update this and the import above to support new schedulers from transformers.optimization
_UpperCamelCase    :   Any   			=						{
    'linear': get_linear_schedule_with_warmup,
    'cosine': get_cosine_schedule_with_warmup,
    'cosine_w_restarts': get_cosine_with_hard_restarts_schedule_with_warmup,
    'polynomial': get_polynomial_decay_schedule_with_warmup,
    # '': get_constant_schedule,             # not supported for now
    # '': get_constant_schedule_with_warmup, # not supported for now
}
_UpperCamelCase    :   Optional[Any]   			=						sorted(arg_to_scheduler.keys())
_UpperCamelCase    :   Tuple   			=						'{' + ', '.join(arg_to_scheduler_choices) + '}'






class 						snake_case  (  pl.LightningModule	):





    def __init__(							self     : Optional[Any]    ,		A     : argparse.Namespace    ,		A     : List[str]=None    ,		A     : Optional[Any]="base"    ,		A     : int=None    ,		A     : Any=None    ,		A     : str=None    ,		**A     : str    ,		):


      '''simple docstring'''




      super().__init__()
      # TODO: move to self.save_hyperparameters()
      # self.save_hyperparameters()
      # can also expand arguments into trainer signature for easier reading

      self.save_hyperparameters(A      )
      a					:      Any    =       0
      a					:      Optional[int]    =       Path(self.hparams.output_dir      )
      a					:      List[Any]    =       self.hparams.cache_dir if self.hparams.cache_dir else None
      if config is None:
        a					:      Any    =       AutoConfig.from_pretrained(
            self.hparams.config_name if self.hparams.config_name else self.hparams.model_name_or_path    ,		**({'num_labels': num_labels} if num_labels is not None else {})    ,		cache_dir=A    ,		**A    ,		)
      else:
        a					:      PretrainedConfig    =       config

      a					:      Optional[Any]    =       ('encoder_layerdrop', 'decoder_layerdrop', 'dropout', 'attention_dropout')
      for p in extra_model_params:
        if getattr(self.hparams    ,		A    ,		A      ):
          assert hasattr(self.config    ,		A      ), F'''model config doesn\'t have a `{p}` attribute'''
          setattr(self.config    ,		A    ,		getattr(self.hparams    ,		A      )      )

      if tokenizer is None:
        a					:      int    =       AutoTokenizer.from_pretrained(
            self.hparams.tokenizer_name if self.hparams.tokenizer_name else self.hparams.model_name_or_path    ,		cache_dir=A    ,		)
      else:
        a					:      PreTrainedTokenizer    =       tokenizer
      a					:      Optional[int]    =       MODEL_MODES[mode]
      if model is None:
        a					:      Union[str, Any]    =       self.model_type.from_pretrained(
            self.hparams.model_name_or_path    ,		from_tf=bool('.ckpt' in self.hparams.model_name_or_path      )    ,		config=self.config    ,		cache_dir=A    ,		)
      else:
        a					:      Optional[int]    =       model





    def     lowerCamelCase__				(							self     : Dict    ,		*A     : str    ,		**A     : Any      ):


      '''simple docstring'''




      a					:      Union[str, Any]    =       self.model_type.from_pretrained(*A    ,		**A      )





    def     lowerCamelCase__				(							self     : str      ):


      '''simple docstring'''




      a					:      Union[str, Any]    =       arg_to_scheduler[self.hparams.lr_scheduler]
      a					:      Any    =       get_schedule_func(
          self.opt    ,		num_warmup_steps=self.hparams.warmup_steps    ,		num_training_steps=self.total_steps()      )
      a					:      Optional[Any]    =       {'scheduler': scheduler, 'interval': 'step', 'frequency': 1}
      return scheduler





    def     lowerCamelCase__				(							self     : Tuple      ):


      '''simple docstring'''




      a					:      int    =       self.model
      a					:      str    =       ['bias', 'LayerNorm.weight']
      a					:      Union[str, Any]    =       [
          {
              'params': [
                  p for n, p in model.named_parameters() if not any(nd in n for nd in no_decay      )
              ],  # check this named paramters
              'weight_decay': self.hparams.weight_decay,
          },
          {
              'params': [p for n, p in model.named_parameters() if any(nd in n for nd in no_decay      )],
              'weight_decay': 0.0,
          },
      ]
      if self.hparams.adafactor:
        a					:      Any    =       Adafactor(
            A    ,		lr=self.hparams.learning_rate    ,		scale_parameter=A    ,		relative_step=A      )

      else:
        a					:      List[Any]    =       AdamW(
            A    ,		lr=self.hparams.learning_rate    ,		eps=self.hparams.adam_epsilon      )
      a					:      List[Any]    =       optimizer

      a					:      int    =       self.get_lr_scheduler()

      return [optimizer], [scheduler]





    def     lowerCamelCase__				(							self     : Dict    ,		A     : Optional[Any]    ,		A     : Dict      ):


      '''simple docstring'''




      return self.validation_step(A    ,		A      )





    def     lowerCamelCase__				(							self     : Union[str, Any]    ,		A     : str      ):


      '''simple docstring'''




      return self.validation_end(A      )





    def     lowerCamelCase__				(							self     : List[Any]      ):


      '''simple docstring'''




      a					:      int    =       max(1    ,		self.hparams.gpus      )  # TODO: consider num_tpu_cores
      a					:      Optional[Any]    =       self.hparams.train_batch_size * self.hparams.accumulate_grad_batches * num_devices
      return (self.dataset_size / effective_batch_size) * self.hparams.max_epochs





    def     lowerCamelCase__				(							self     : int    ,		A     : Any      ):


      '''simple docstring'''




      if stage == "test":
        a					:      Optional[Any]    =       len(self.test_dataloader().dataset      )
      else:
        a					:      Optional[int]    =       self.get_dataloader('train'    ,		self.hparams.train_batch_size    ,		shuffle=A      )
        a					:      Any    =       len(self.train_dataloader().dataset      )





    def     lowerCamelCase__				(							self     : str    ,		A     : str    ,		A     : int    ,		A     : bool = False      ):


      '''simple docstring'''




      raise NotImplementedError('You must implement this for your task'      )





    def     lowerCamelCase__				(							self     : List[str]      ):


      '''simple docstring'''




      return self.train_loader





    def     lowerCamelCase__				(							self     : Dict      ):


      '''simple docstring'''




      return self.get_dataloader('dev'    ,		self.hparams.eval_batch_size    ,		shuffle=A      )





    def     lowerCamelCase__				(							self     : int      ):


      '''simple docstring'''




      return self.get_dataloader('test'    ,		self.hparams.eval_batch_size    ,		shuffle=A      )





    def     lowerCamelCase__				(							self     : Tuple    ,		A     : List[str]      ):


      '''simple docstring'''




      return os.path.join(
          self.hparams.data_dir    ,		'cached_{}_{}_{}'.format(
              A    ,		list(filter(A    ,		self.hparams.model_name_or_path.split('/'      )      )      ).pop()    ,		str(self.hparams.max_seq_length      )    ,		)    ,		)





    @pl.utilities.rank_zero_only
    def     lowerCamelCase__				(							self     : Any    ,		A     : Dict[str, Any]      ):


      '''simple docstring'''




      a					:      int    =       self.output_dir.joinpath('best_tfmr'      )
      a					:      Union[str, Any]    =       self.step_count
      self.model.save_pretrained(A      )
      self.tokenizer.save_pretrained(A      )





    @staticmethod
    def     lowerCamelCase__				(							A     : int    ,		A     : str      ):


      '''simple docstring'''




      parser.add_argument(
          '--model_name_or_path'    ,		default=A    ,		type=A    ,		required=A    ,		help='Path to pretrained model or model identifier from huggingface.co/models'    ,		)
      parser.add_argument(
          '--config_name'    ,		default=''    ,		type=A    ,		help='Pretrained config name or path if not the same as model_name'      )
      parser.add_argument(
          '--tokenizer_name'    ,		default=A    ,		type=A    ,		help='Pretrained tokenizer name or path if not the same as model_name'    ,		)
      parser.add_argument(
          '--cache_dir'    ,		default=str(Path(A      ).parent / 'test_run' / 'cache'      )    ,		type=A    ,		help='Where do you want to store the pre-trained models downloaded from huggingface.co'    ,		)
      parser.add_argument(
          '--encoder_layerdrop'    ,		type=A    ,		help='Encoder layer dropout probability (Optional). Goes into model.config'    ,		)
      parser.add_argument(
          '--decoder_layerdrop'    ,		type=A    ,		help='Decoder layer dropout probability (Optional). Goes into model.config'    ,		)
      parser.add_argument(
          '--dropout'    ,		type=A    ,		help='Dropout probability (Optional). Goes into model.config'    ,		)
      parser.add_argument(
          '--attention_dropout'    ,		type=A    ,		help='Attention dropout probability (Optional). Goes into model.config'    ,		)
      parser.add_argument('--learning_rate'    ,		default=5E-5    ,		type=A    ,		help='The initial learning rate for Adam.'      )
      parser.add_argument(
          '--lr_scheduler'    ,		default='linear'    ,		choices=A    ,		metavar=A    ,		type=A    ,		help='Learning rate scheduler'    ,		)
      parser.add_argument('--weight_decay'    ,		default=0.0    ,		type=A    ,		help='Weight decay if we apply some.'      )
      parser.add_argument('--adam_epsilon'    ,		default=1E-8    ,		type=A    ,		help='Epsilon for Adam optimizer.'      )
      parser.add_argument('--warmup_steps'    ,		default=0    ,		type=A    ,		help='Linear warmup over warmup_steps.'      )
      parser.add_argument('--num_workers'    ,		default=4    ,		type=A    ,		help='kwarg passed to DataLoader'      )
      parser.add_argument('--num_train_epochs'    ,		dest='max_epochs'    ,		default=3    ,		type=A      )
      parser.add_argument('--train_batch_size'    ,		default=3_2    ,		type=A      )
      parser.add_argument('--eval_batch_size'    ,		default=3_2    ,		type=A      )
      parser.add_argument('--adafactor'    ,		action='store_true'      )






class 						snake_case  (  pl.Callback	):





    def     lowerCamelCase__				(							self     : List[Any]    ,		A     : str    ,		A     : Tuple      ):


      '''simple docstring'''




      if (
          trainer.is_global_zero and trainer.global_rank == 0
      ):  # we initialize the retriever only on master worker with RAY. In new pytorch-lightning accelorators are removed.
        pl_module.model.rag.retriever.init_retrieval()  # better to use hook functions.






class 						snake_case  (  pl.Callback	):





    def     lowerCamelCase__				(							self     : Tuple    ,		A     : Tuple    ,		A     : List[str]      ):


      '''simple docstring'''




      for name, param in pl_module.model.rag.named_parameters():
        if param.grad is None:
          print(A      )






class 						snake_case  (  pl.Callback	):





    def     lowerCamelCase__				(							self     : Optional[int]    ,		A     : Tuple    ,		A     : Optional[Any]      ):


      '''simple docstring'''




      a					:      str    =       trainer.lr_schedulers[0]['scheduler']
      a					:      Dict    =       {F'''lr_group_{i}''': lr for i, lr in enumerate(lr_scheduler.get_lr()      )}
      pl_module.logger.log_metrics(A      )





    def     lowerCamelCase__				(							self     : Tuple    ,		A     : pl.Trainer    ,		A     : pl.LightningModule      ):


      '''simple docstring'''




      rank_zero_info('***** Validation results *****'      )
      a					:      int    =       trainer.callback_metrics
      # Log results
      for key in sorted(A      ):
        if key not in ["log", "progress_bar"]:
          rank_zero_info('{} = {}\n'.format(A    ,		str(metrics[key]      )      )      )





    def     lowerCamelCase__				(							self     : int    ,		A     : pl.Trainer    ,		A     : pl.LightningModule      ):


      '''simple docstring'''




      rank_zero_info('***** Test results *****'      )
      a					:      List[Any]    =       trainer.callback_metrics
      # Log and save results to file
      a					:      Tuple    =       os.path.join(pl_module.hparams.output_dir    ,		'test_results.txt'      )
      with open(A    ,		'w'      ) as writer:
        for key in sorted(A      ):
          if key not in ["log", "progress_bar"]:
            rank_zero_info('{} = {}\n'.format(A    ,		str(metrics[key]      )      )      )
            writer.write('{} = {}\n'.format(A    ,		str(metrics[key]      )      )      )




def       snake_case  (A_				:List[Any]     ,      A_				:List[str]				):







  '''simple docstring'''

  parser.add_argument(
      '--output_dir'     ,      default=str(Path(A_				).parent / 'test_run' / 'model_checkpoints'				)     ,      type=A_     ,      help='The output directory where the model predictions and checkpoints will be written.'     ,      )
  parser.add_argument(
      '--fp16'     ,      action='store_true'     ,      help='Whether to use 16-bit (mixed) precision (through NVIDIA apex) instead of 32-bit'     ,      )

  parser.add_argument(
      '--fp16_opt_level'     ,      type=A_     ,      default='O2'     ,      help=(
          'For fp16: Apex AMP optimization level selected in [\'O0\', \'O1\', \'O2\', and \'O3\'].'
          'See details at https://nvidia.github.io/apex/amp.html'
      )     ,      )
  parser.add_argument('--n_tpu_cores'     ,      dest='tpu_cores'     ,      type=A_				)
  parser.add_argument('--max_grad_norm'     ,      dest='gradient_clip_val'     ,      default=1.0     ,      type=A_     ,      help='Max gradient norm'				)
  parser.add_argument('--do_train'     ,      action='store_true'     ,      help='Whether to run training.'				)
  parser.add_argument('--do_predict'     ,      action='store_true'     ,      help='Whether to run predictions on the test set.'				)
  parser.add_argument(
      '--gradient_accumulation_steps'     ,      dest='accumulate_grad_batches'     ,      type=A_     ,      default=1     ,      help='Number of updates steps to accumulate before performing a backward/update pass.'     ,      )
  parser.add_argument('--seed'     ,      type=A_     ,      default=4_2     ,      help='random seed for initialization'				)
  parser.add_argument(
      '--data_dir'     ,      default=str(Path(A_				).parent / 'test_run' / 'dummy-train-data'				)     ,      type=A_     ,      help='The input data dir. Should contain the training files for the CoNLL-2003 NER task.'     ,      )






def       snake_case  (A_				:BaseTransformer     ,      A_				:argparse.Namespace     ,      A_				:Tuple=None     ,      A_				:Optional[int]=True     ,      A_				:Dict=[]     ,      A_				:Optional[int]=None     ,      A_				:Optional[Any]=None     ,      **A_				:Union[str, Any]     ,      ):







  '''simple docstring'''

  pl.seed_everything(args.seed				)

  # init model
  a					:      Union[str, Any]    =       Path(model.hparams.output_dir				)
  odir.mkdir(exist_ok=A_				)

  # add custom checkpoints
  if checkpoint_callback is None:
    a					:      List[Any]    =       pl.callbacks.ModelCheckpoint(
        filepath=args.output_dir     ,      prefix='checkpoint'     ,      monitor='val_loss'     ,      mode='min'     ,      save_top_k=1				)
  if early_stopping_callback:
    extra_callbacks.append(A_				)
  if logging_callback is None:
    a					:      Union[str, Any]    =       LoggingCallback()

  a					:      List[Any]    =       {}

  if args.fpaa:
    a					:      Union[str, Any]    =       1_6

  if args.gpus > 1:
    a					:      List[Any]    =       'auto'
    a					:      Dict    =       'ddp'

  a					:      Tuple    =       args.accumulate_grad_batches
  a					:      int    =       None
  a					:      List[str]    =       'auto'

  a					:      Tuple    =       pl.Trainer.from_argparse_args(
      A_     ,      weights_summary=A_     ,      callbacks=[logging_callback] + extra_callbacks + [InitCallback()] + [checkpoint_callback]     ,      logger=A_     ,      val_check_interval=1     ,      num_sanity_val_steps=2     ,      **A_     ,      )

  if args.do_train:
    trainer.fit(A_				)

  else:
    print('RAG modeling tests with new set functions successfuly executed!'				)
  return trainer












"""simple docstring"""


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

import numpy as np


_UpperCamelCase    :   Optional[Any]   			=						re.compile(r'\b(a|an|the)\b', re.UNICODE)

_UpperCamelCase    :   str   			=						None




def       snake_case  ():







  '''simple docstring'''

  a					:      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=A_     ,      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=A_     ,      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       snake_case  (A_				:Optional[int]				):







  '''simple docstring'''

  a					:      str    =       {}
  for article in dataset:
    for p in article["paragraphs"]:
      for qa in p["qas"]:
        a					:      Optional[int]    =       bool(qa['answers']['text']				)
  return qid_to_has_ans




def       snake_case  (A_				:List[Any]				):







  '''simple docstring'''

  def remove_articles(A_				:str				):
    return ARTICLES_REGEX.sub(' '     ,      A_				)

  def white_space_fix(A_				:str				):
    return " ".join(text.split()				)

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

  def lower(A_				:Union[str, Any]				):
    return text.lower()

  return white_space_fix(remove_articles(remove_punc(lower(A_				)				)				)				)




def       snake_case  (A_				:str				):







  '''simple docstring'''

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




def       snake_case  (A_				:int     ,      A_				:Union[str, Any]				):







  '''simple docstring'''

  return int(normalize_answer(A_				) == normalize_answer(A_				)				)




def       snake_case  (A_				:Optional[int]     ,      A_				:str				):







  '''simple docstring'''

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




def       snake_case  (A_				:Any     ,      A_				:Dict				):







  '''simple docstring'''

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




def       snake_case  (A_				:Union[str, Any]     ,      A_				:List[Any]     ,      A_				:List[Any]     ,      A_				:Dict				):







  '''simple docstring'''

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




def       snake_case  (A_				:Tuple     ,      A_				:int     ,      A_				:Tuple=None				):







  '''simple docstring'''

  if not qid_list:
    a					:      Optional[int]    =       len(A_				)
    return collections.OrderedDict(
        [
            ('exact', 100.0 * sum(exact_scores.values()				) / total),
            ('f1', 100.0 * sum(fa_scores.values()				) / total),
            ('total', total),
        ]				)
  else:
    a					:      List[Any]    =       len(A_				)
    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       snake_case  (A_				:str     ,      A_				:Dict     ,      A_				:List[Any]				):







  '''simple docstring'''

  for k in new_eval:
    a					:      Union[str, Any]    =       new_eval[k]




def       snake_case  (A_				:Optional[Any]     ,      A_				:Any     ,      A_				:Dict     ,      A_				:Optional[int]				):







  '''simple docstring'''

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




def       snake_case  (A_				:List[str]     ,      A_				:str     ,      A_				:Any     ,      A_				:Any     ,      A_				:List[Any]=None     ,      A_				:Union[str, Any]=None				):







  '''simple docstring'''

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




def       snake_case  (A_				:Optional[int]     ,      A_				:Any     ,      A_				:List[Any]     ,      A_				:int     ,      A_				:int     ,      A_				:List[str]				):







  '''simple docstring'''

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




def       snake_case  (A_				:List[str]     ,      A_				:List[str]     ,      A_				:List[Any]     ,      A_				:str				):







  '''simple docstring'''

  if not qid_list:
    return
  a					:      List[Any]    =       [na_probs[k] for k in qid_list]
  a					:      List[str]    =       np.ones_like(A_				) / float(len(A_				)				)
  plt.hist(A_     ,      weights=A_     ,      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(A_     ,      f'''na_prob_hist_{name}.png'''				)				)
  plt.clf()




def       snake_case  (A_				:Tuple     ,      A_				:Tuple     ,      A_				:List[str]     ,      A_				:List[str]				):







  '''simple docstring'''

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




def       snake_case  (A_				:List[Any]     ,      A_				:List[Any]     ,      A_				:str     ,      A_				:int     ,      A_				:Optional[Any]     ,      A_				:Union[str, Any]				):







  '''simple docstring'''

  a,					a					:      Any    =       find_best_thresh(A_     ,      A_     ,      A_     ,      A_				)
  a,					a					:      List[Any]    =       find_best_thresh(A_     ,      A_     ,      A_     ,      A_				)
  a					:      Union[str, Any]    =       best_exact
  a					:      List[Any]    =       exact_thresh
  a					:      List[str]    =       best_fa
  a					:      Any    =       fa_thresh




def       snake_case  ():







  '''simple docstring'''

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


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

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




