





import logging
import os
from dataclasses import dataclass, field
from typing import Dict, Optional

import numpy as np
from utils_multiple_choice import MultipleChoiceDataset, Split, processors

import transformers
from transformers import (
    AutoConfig,
    AutoModelForMultipleChoice,
    AutoTokenizer,
    DataCollatorWithPadding,
    EvalPrediction,
    HfArgumentParser,
    Trainer,
    TrainingArguments,
    set_seed,
)
from transformers.trainer_utils import is_main_process


UpperCAmelCase     :     Tuple							=	logging.getLogger(__name__)
def 					_A	(		SCREAMING_SNAKE_CASE					:       Tuple	,	SCREAMING_SNAKE_CASE					:       Dict						):





  """simple docstring"""



  return (preds == labels).mean()







@dataclass
class     __lowerCAmelCase	:
      _lowercase    : str            =      field(
          metadata={"""help""": """Path to pretrained model or model identifier from huggingface.co/models"""})
      _lowercase    : Optional[str]            =      field(
          default=UpperCamelCase__						,    metadata={"""help""": """Pretrained config name or path if not the same as model_name"""})
      _lowercase    : Optional[str]            =      field(
          default=UpperCamelCase__						,    metadata={"""help""": """Pretrained tokenizer name or path if not the same as model_name"""})
      _lowercase    : Optional[str]            =      field(
          default=UpperCamelCase__						,    metadata={"""help""": """Where do you want to store the pretrained models downloaded from huggingface.co"""}						,    )







@dataclass
class     __lowerCAmelCase	:
      _lowercase    : str            =      field(metadata={"""help""": """The name of the task to train on: """ + """, """.join(processors.keys())})
      _lowercase    : str            =      field(metadata={"""help""": """Should contain the data files for the task."""})
      _lowercase    : int            =      field(
          default=128						,    metadata={
              """help""": (
                  """The maximum total input sequence length after tokenization. Sequences longer """
                  """than this will be truncated, sequences shorter will be padded."""
              )
          }						,    )
      _lowercase    : bool            =      field(
          default=UpperCamelCase__						,    metadata={"""help""": """Overwrite the cached training and evaluation sets"""})
def 					_A	(		):





  """simple docstring"""



  a__     : List[Any]  =HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)						)
  a__ ,							a__ ,							a__     : int  =parser.parse_args_into_dataclasses()

  if (
      os.path.exists(training_args.output_dir						)
      and os.listdir(training_args.output_dir						)
      and training_args.do_train
      and not training_args.overwrite_output_dir
  ):
    raise ValueError(
        f'''Output directory ({training_args.output_dir}) already exists and is not empty. Use'''
        " --overwrite_output_dir to overcome."						)

  # Setup logging
  logging.basicConfig(
      format="%(asctime)s - %(levelname)s - %(name)s - %(message)s"	,	datefmt="%m/%d/%Y %H:%M:%S"	,	level=logging.INFO if training_args.local_rank in [-1, 0] else logging.WARN	,	)
  logger.warning(
      "Process rank: %s, device: %s, n_gpu: %s, distributed training: %s, 16-bits training: %s"	,	training_args.local_rank	,	training_args.device	,	training_args.n_gpu	,	bool(training_args.local_rank != -1						)	,	training_args.fpaa	,	)
  # Set the verbosity to info of the Transformers logger (on main process only):
  if is_main_process(training_args.local_rank						):
    transformers.utils.logging.set_verbosity_info()
    transformers.utils.logging.enable_default_handler()
    transformers.utils.logging.enable_explicit_format()
  logger.info("Training/evaluation parameters %s"	,	SCREAMING_SNAKE_CASE						)

  # Set seed
  set_seed(training_args.seed						)

  try:
    a__     : Tuple  =processors[data_args.task_name]()
    a__     : Optional[Any]  =processor.get_labels()
    a__     : Optional[int]  =len(SCREAMING_SNAKE_CASE						)
  except KeyError:
    raise ValueError("Task not found: %s" % (data_args.task_name)						)

  # Load pretrained model and tokenizer
  #
  # Distributed training:
  # The .from_pretrained methods guarantee that only one local process can concurrently
  # download model & vocab.

  a__     : List[Any]  =AutoConfig.from_pretrained(
      model_args.config_name if model_args.config_name else model_args.model_name_or_path	,	num_labels=SCREAMING_SNAKE_CASE	,	finetuning_task=data_args.task_name	,	cache_dir=model_args.cache_dir	,	)
  a__     : int  =AutoTokenizer.from_pretrained(
      model_args.tokenizer_name if model_args.tokenizer_name else model_args.model_name_or_path	,	cache_dir=model_args.cache_dir	,	)
  a__     : List[str]  =AutoModelForMultipleChoice.from_pretrained(
      model_args.model_name_or_path	,	from_tf=bool(".ckpt" in model_args.model_name_or_path						)	,	config=SCREAMING_SNAKE_CASE	,	cache_dir=model_args.cache_dir	,	)

  # Get datasets
  a__     : Dict  =(
      MultipleChoiceDataset(
          data_dir=data_args.data_dir	,	tokenizer=SCREAMING_SNAKE_CASE	,	task=data_args.task_name	,	max_seq_length=data_args.max_seq_length	,	overwrite_cache=data_args.overwrite_cache	,	mode=Split.train	,	)
      if training_args.do_train
      else None
  )
  a__     : Tuple  =(
      MultipleChoiceDataset(
          data_dir=data_args.data_dir	,	tokenizer=SCREAMING_SNAKE_CASE	,	task=data_args.task_name	,	max_seq_length=data_args.max_seq_length	,	overwrite_cache=data_args.overwrite_cache	,	mode=Split.dev	,	)
      if training_args.do_eval
      else None
  )

  def compute_metrics(SCREAMING_SNAKE_CASE					:       EvalPrediction						) -> Dict:
    a__     : List[str]  =np.argmax(p.predictions	,	axis=1						)
    return {"acc": simple_accuracy(SCREAMING_SNAKE_CASE	,	p.label_ids						)}

  # Data collator
  a__     : Any  =DataCollatorWithPadding(SCREAMING_SNAKE_CASE	,	pad_to_multiple_of=8						) if training_args.fpaa else None

  # Initialize our Trainer
  a__     : Any  =Trainer(
      model=SCREAMING_SNAKE_CASE	,	args=SCREAMING_SNAKE_CASE	,	train_dataset=SCREAMING_SNAKE_CASE	,	eval_dataset=SCREAMING_SNAKE_CASE	,	compute_metrics=SCREAMING_SNAKE_CASE	,	data_collator=SCREAMING_SNAKE_CASE	,	)

  # Training
  if training_args.do_train:
    trainer.train(
        model_path=model_args.model_name_or_path if os.path.isdir(model_args.model_name_or_path						) else None						)
    trainer.save_model()
    # For convenience, we also re-save the tokenizer to the same directory,
    # so that you can share your model easily on huggingface.co/models =)
    if trainer.is_world_master():
      tokenizer.save_pretrained(training_args.output_dir						)

    # Evaluation
  a__     : Optional[Any]  ={}
  if training_args.do_eval:
    logger.info("*** Evaluate ***"						)

    a__     : Union[str, Any]  =trainer.evaluate()

    a__     : Any  =os.path.join(training_args.output_dir	,	"eval_results.txt"						)
    if trainer.is_world_master():
      with open(SCREAMING_SNAKE_CASE	,	"w"						) as writer:
        logger.info("***** Eval results *****"						)
        for key, value in result.items():
          logger.info("  %s = %s"	,	SCREAMING_SNAKE_CASE	,	SCREAMING_SNAKE_CASE						)
          writer.write("%s = %s\n" % (key, value)						)

        results.update(SCREAMING_SNAKE_CASE						)

  return results
def 					_A	(		SCREAMING_SNAKE_CASE					:       Tuple						):





  """simple docstring"""



  main()


if __name__ == "__main__":
     main()













import warnings

from transformers import AutoTokenizer
from transformers.utils import is_torch_available
from transformers.utils.generic import ExplicitEnum

from ...processing_utils import ProcessorMixin


if is_torch_available():
     import torch







class     __lowerCAmelCase	(   UpperCamelCase__):
      _lowercase    : Any            =      """char"""
      _lowercase    : str            =      """bpe"""
      _lowercase    : List[Any]            =      """wp"""
UpperCAmelCase     :     List[Any]							=	(DecodeType.CHARACTER, DecodeType.BPE, DecodeType.WORDPIECE)







class     __lowerCAmelCase	(   UpperCamelCase__):
      _lowercase    : Tuple            =      ["""image_processor""", """char_tokenizer"""]
      _lowercase    : Any            =      """ViTImageProcessor"""
      _lowercase    : Optional[Any]            =      """MgpstrTokenizer"""







      def __init__(   self		,				lowerCAmelCase__=None		,				lowerCAmelCase__=None		,				**lowerCAmelCase__							)       ->						Optional[Any]:



        '''simple docstring'''




        a__     : Optional[int]  =None
        if "feature_extractor" in kwargs:
          warnings.warn(
              "The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`"
              " instead."		,				lowerCAmelCase__		,				)
          a__     : List[str]  =kwargs.pop("feature_extractor"							)

        a__     : List[str]  =image_processor if image_processor is not None else feature_extractor
        if image_processor is None:
          raise ValueError("You need to specify an `image_processor`."							)
        if tokenizer is None:
          raise ValueError("You need to specify a `tokenizer`."							)

        a__     : str  =tokenizer
        a__     : List[str]  =AutoTokenizer.from_pretrained("gpt2"							)
        a__     : Optional[int]  =AutoTokenizer.from_pretrained("bert-base-uncased"							)

        super().__init__(lowerCAmelCase__		,				lowerCAmelCase__							)







      def __call__(   self		,				lowerCAmelCase__=None		,				lowerCAmelCase__=None		,				lowerCAmelCase__=None		,				**lowerCAmelCase__							)       ->						Union[str, Any]:



        '''simple docstring'''




        if images is None and text is None:
          raise ValueError("You need to specify either an `images` or `text` input to process."							)

        if images is not None:
          a__     : Union[str, Any]  =self.image_processor(lowerCAmelCase__		,				return_tensors=lowerCAmelCase__		,				**lowerCAmelCase__							)
        if text is not None:
          a__     : int  =self.char_tokenizer(lowerCAmelCase__		,				return_tensors=lowerCAmelCase__		,				**lowerCAmelCase__							)

        if text is None:
          return inputs
        elif images is None:
          return encodings
        else:
          a__     : Tuple  =encodings["input_ids"]
          return inputs







      def        _lowercase  (   self		,				lowerCAmelCase__							)       ->						Optional[Any]:



        '''simple docstring'''




        a__ ,							a__ ,							a__     : Any  =sequences
        a__     : Union[str, Any]  =char_preds.size(0							)

        a__ ,							a__     : Dict  =self._decode_helper(lowerCAmelCase__		,				"char"							)
        a__ ,							a__     : List[Any]  =self._decode_helper(lowerCAmelCase__		,				"bpe"							)
        a__ ,							a__     : Optional[int]  =self._decode_helper(lowerCAmelCase__		,				"wp"							)

        a__     : List[Any]  =[]
        a__     : Dict  =[]
        for i in range(lowerCAmelCase__							):
          a__     : int  =[char_scores[i], bpe_scores[i], wp_scores[i]]
          a__     : Tuple  =[char_strs[i], bpe_strs[i], wp_strs[i]]
          a__     : Any  =scores.index(max(lowerCAmelCase__							)							)
          final_strs.append(strs[max_score_index]							)
          final_scores.append(scores[max_score_index]							)

        a__     : Dict  ={}
        a__     : str  =final_strs
        a__     : Optional[int]  =final_scores
        a__     : Union[str, Any]  =char_strs
        a__     : List[str]  =bpe_strs
        a__     : Union[str, Any]  =wp_strs
        return out







      def        _lowercase  (   self		,				lowerCAmelCase__		,				lowerCAmelCase__							)       ->						Tuple:



        '''simple docstring'''




        if format == DecodeType.CHARACTER:
          a__     : Optional[Any]  =self.char_decode
          a__     : Dict  =1
          a__     : Tuple  ="[s]"
        elif format == DecodeType.BPE:
          a__     : str  =self.bpe_decode
          a__     : Dict  =2
          a__     : Optional[int]  ="#"
        elif format == DecodeType.WORDPIECE:
          a__     : Union[str, Any]  =self.wp_decode
          a__     : List[Any]  =1_0_2
          a__     : Dict  ="[SEP]"
        else:
          raise ValueError(F'''Format {format} is not supported.'''							)

        a__ ,							a__     : Any  =[], []
        a__     : str  =pred_logits.size(0							)
        a__     : Optional[Any]  =pred_logits.size(1							)
        a__ ,							a__     : Optional[int]  =pred_logits.topk(1		,				dim=-1		,				largest=lowerCAmelCase__		,				sorted=lowerCAmelCase__							)
        a__     : Optional[Any]  =preds_index.view(-1		,				lowerCAmelCase__							)[:, 1:]
        a__     : Dict  =decoder(lowerCAmelCase__							)
        a__ ,							a__     : Any  =torch.nn.functional.softmax(lowerCAmelCase__		,				dim=2							).max(dim=2							)
        a__     : int  =preds_max_prob[:, 1:]

        for index in range(lowerCAmelCase__							):
          a__     : Optional[Any]  =preds_str[index].find(lowerCAmelCase__							)
          a__     : Optional[int]  =preds_str[index][:pred_eos]
          a__     : List[Any]  =preds_index[index].cpu().tolist()
          a__     : List[Any]  =pred_index.index(lowerCAmelCase__							) if eos_token in pred_index else -1
          a__     : Union[str, Any]  =preds_max_prob[index][: pred_eos_index + 1]
          a__     : List[Any]  =pred_max_prob.cumprod(dim=0							)[-1] if pred_max_prob.nelement() != 0 else 0.0
          dec_strs.append(lowerCAmelCase__							)
          conf_scores.append(lowerCAmelCase__							)

        return dec_strs, conf_scores







      def        _lowercase  (   self		,				lowerCAmelCase__							)       ->						str:



        '''simple docstring'''




        a__     : int  =[seq.replace(" "		,				""							) for seq in self.char_tokenizer.batch_decode(lowerCAmelCase__							)]
        return decode_strs







      def        _lowercase  (   self		,				lowerCAmelCase__							)       ->						Tuple:



        '''simple docstring'''




        return self.bpe_tokenizer.batch_decode(lowerCAmelCase__							)







      def        _lowercase  (   self		,				lowerCAmelCase__							)       ->						Optional[Any]:



        '''simple docstring'''




        a__     : List[Any]  =[seq.replace(" "		,				""							) for seq in self.wp_tokenizer.batch_decode(lowerCAmelCase__							)]
        return decode_strs






