


'''simple docstring'''


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


__magic_name__			=      logging.getLogger(__name__)




def      lowerCamelCase					(      lowerCamelCase  :      List[str]							,						lowerCamelCase  :      Tuple):
      return (preds == labels).mean()



@dataclass
class 	__lowerCAmelCase		:


  '''simple docstring'''







  a_  					=							field(
      metadata={"""help""": """Path to pretrained model or model identifier from huggingface.co/models"""}    )
  a_  					=							field(
      default=__SCREAMING_SNAKE_CASE  , metadata={"""help""": """Pretrained config name or path if not the same as model_name"""}    )
  a_  					=							field(
      default=__SCREAMING_SNAKE_CASE  , metadata={"""help""": """Pretrained tokenizer name or path if not the same as model_name"""}    )
  a_  					=							field(
      default=__SCREAMING_SNAKE_CASE  , metadata={"""help""": """Where do you want to store the pretrained models downloaded from huggingface.co"""}  , )







@dataclass
class 	__lowerCAmelCase		:


  '''simple docstring'''







  a_  					=							field(metadata={"""help""": """The name of the task to train on: """ + """, """.join(processors.keys()    )}    )
  a_  					=							field(metadata={"""help""": """Should contain the data files for the task."""}    )
  a_  					=							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."""
          )
      }  , )
  a_  					=							field(
      default=__SCREAMING_SNAKE_CASE  , metadata={"""help""": """Overwrite the cached training and evaluation sets"""}    )







def      lowerCamelCase					(      ):
      # See all possible arguments in src/transformers/training_args.py
      # or by passing the --help flag to this script.
      # We now keep distinct sets of args, for a cleaner separation of concerns.

      A_						:  Any          =				HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments))
      A_   ,					A_   ,					A_						:  Any          =				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"""							,						lowerCamelCase)

      # Set seed
      set_seed(training_args.seed)

      try:
            A_						:  List[str]          =				processors[data_args.task_name]()
            A_						:  Optional[int]          =				processor.get_labels()
            A_						:  List[str]          =				len(lowerCamelCase)
      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_						:  Union[str, Any]          =				AutoConfig.from_pretrained(
          model_args.config_name if model_args.config_name else model_args.model_name_or_path							,						num_labels=lowerCamelCase							,						finetuning_task=data_args.task_name							,						cache_dir=model_args.cache_dir							,						)
      A_						:  Tuple          =				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_						:  Tuple          =				AutoModelForMultipleChoice.from_pretrained(
          model_args.model_name_or_path							,						from_tf=bool(""".ckpt""" in model_args.model_name_or_path)							,						config=lowerCamelCase							,						cache_dir=model_args.cache_dir							,						)

      # Get datasets
      A_						:  Any          =				(
          MultipleChoiceDataset(
              data_dir=data_args.data_dir							,						tokenizer=lowerCamelCase							,						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_						:  List[Any]          =				(
          MultipleChoiceDataset(
              data_dir=data_args.data_dir							,						tokenizer=lowerCamelCase							,						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(lowerCamelCase  :      EvalPrediction) -> Dict:
            A_						:  Dict          =				np.argmax(p.predictions							,						axis=1)
            return {"acc": simple_accuracy(lowerCamelCase							,						p.label_ids)}

      # Data collator
      A_						:  str          =				DataCollatorWithPadding(lowerCamelCase							,						pad_to_multiple_of=8) if training_args.fpaa else None

      # Initialize our Trainer
      A_						:  Optional[Any]          =				Trainer(
          model=lowerCamelCase							,						args=lowerCamelCase							,						train_dataset=lowerCamelCase							,						eval_dataset=lowerCamelCase							,						compute_metrics=lowerCamelCase							,						data_collator=lowerCamelCase							,						)

      # 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_						:  List[Any]          =				{}
      if training_args.do_eval:
            logger.info("""*** Evaluate ***""")

            A_						:  List[str]          =				trainer.evaluate()

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

                        results.update(lowerCamelCase)

      return results




def      lowerCamelCase					(      lowerCamelCase  :      Any):
      # For xla_spawn (TPUs)
      main()


if __name__ == "__main__":
     main()




'''simple docstring'''


import warnings
from typing import List, Optional, Union

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding, PaddingStrategy, PreTokenizedInput, TextInput, TruncationStrategy
from ...utils import TensorType



class 	__lowerCAmelCase		(							__SCREAMING_SNAKE_CASE    ):


  '''simple docstring'''







  a_  					=							["""image_processor""", """tokenizer"""]
  a_  					=							"""ViltImageProcessor"""
  a_  					=							("""BertTokenizer""", """BertTokenizerFast""")






  def __init__(							self :       List[Any]	,_a :       Optional[Any]=None	,_a :       List[str]=None	,**_a :       Any ):

        '''simple docstring'''




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

        A_						:  List[Any]          =				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`.""" )

        super().__init__(_a	,_a )
        A_						:  Optional[Any]          =				self.image_processor






  def __call__(							self :       Any	,_a :       Tuple	,_a :       Union[TextInput, PreTokenizedInput, List[TextInput], List[PreTokenizedInput]] = None	,_a :       bool = True	,_a :       Union[bool, str, PaddingStrategy] = False	,_a :       Union[bool, str, TruncationStrategy] = None	,_a :       Optional[int] = None	,_a :       int = 0	,_a :       Optional[int] = None	,_a :       Optional[bool] = None	,_a :       Optional[bool] = None	,_a :       bool = False	,_a :       bool = False	,_a :       bool = False	,_a :       bool = False	,_a :       bool = True	,_a :       Optional[Union[str, TensorType]] = None	,**_a :       Tuple	,):

        '''simple docstring'''




        A_						:  int          =				self.tokenizer(
            text=_a	,add_special_tokens=_a	,padding=_a	,truncation=_a	,max_length=_a	,stride=_a	,pad_to_multiple_of=_a	,return_token_type_ids=_a	,return_attention_mask=_a	,return_overflowing_tokens=_a	,return_special_tokens_mask=_a	,return_offsets_mapping=_a	,return_length=_a	,verbose=_a	,return_tensors=_a	,**_a	,)
        # add pixel_values + pixel_mask
        A_						:  Optional[int]          =				self.image_processor(_a	,return_tensors=_a )
        encoding.update(_a )

        return encoding






  def        _a							(							self :       List[Any]	,*_a :       Any	,**_a :       Any ):

        '''simple docstring'''




        return self.tokenizer.batch_decode(*_a	,**_a )






  def        _a							(							self :       int	,*_a :       int	,**_a :       Optional[int] ):

        '''simple docstring'''




        return self.tokenizer.decode(*_a	,**_a )






  @property
  def        _a							(							self :       List[Any] ):

        '''simple docstring'''




        A_						:  Optional[int]          =				self.tokenizer.model_input_names
        A_						:  str          =				self.image_processor.model_input_names
        return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names ) )






  @property
  def        _a							(							self :       str ):

        '''simple docstring'''




        warnings.warn(
            """`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead."""	,_a	,)
        return self.image_processor_class






  @property
  def        _a							(							self :       int ):

        '''simple docstring'''




        warnings.warn(
            """`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead."""	,_a	,)
        return self.image_processor
