





'''simple docstring'''

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

import numpy as np
import torch
from utils_hans import HansDataset, InputFeatures, hans_processors, hans_tasks_num_labels

import transformers
from transformers import (
    AutoConfig,
    AutoModelForSequenceClassification,
    AutoTokenizer,
    HfArgumentParser,
    Trainer,
    TrainingArguments,
    default_data_collator,
    set_seed,
)
from transformers.trainer_utils import is_main_process


UpperCAmelCase       :		List[Any]		=				logging.getLogger(__name__)





@dataclass
class lowerCamelCase    :
  _lowercase	:						str							       =					field(
      metadata={"""help""": """Path to pretrained model or model identifier from huggingface.co/models"""}       )
  _lowercase	:						Optional[str]							       =					field(
      default=a__    ,   metadata={"""help""": """Pretrained config name or path if not the same as model_name"""}       )
  _lowercase	:						Optional[str]							       =					field(
      default=a__    ,   metadata={"""help""": """Pretrained tokenizer name or path if not the same as model_name"""}       )
  _lowercase	:						Optional[str]							       =					field(
      default=a__    ,   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 selected in the list: """ + """, """.join(hans_processors.keys()       )}       )
  _lowercase	:						str							       =					field(
      metadata={"""help""": """The input data dir. Should contain the .tsv files (or other 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=a__    ,   metadata={"""help""": """Overwrite the cached training and evaluation sets"""}       )




def  _a     (				lowerCAmelCase_						):





      """simple docstring"""



      _snake_case		:						Optional[int]         =  default_data_collator(lowerCAmelCase_						)
      _snake_case		:						List[Any]         =  batch.pop('''pairID'''      ,   lowerCAmelCase_						)
      return batch




def  _a     (				):





      """simple docstring"""



      _snake_case		:						int         =  HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)						)
      _snake_case  ,				_snake_case  ,				_snake_case		:						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:
            _snake_case		:						Union[str, Any]         =  hans_tasks_num_labels[data_args.task_name]
      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.

      _snake_case		:						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      ,   )
      _snake_case		:						Union[str, Any]         =  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      ,   )
      _snake_case		:						Any         =  AutoModelForSequenceClassification.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
      _snake_case		:						Any         =  (
          HansDataset(
              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      ,   )
          if training_args.do_train
          else None
      )
      _snake_case		:						Tuple         =  (
          HansDataset(
              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      ,   evaluate=lowerCAmelCase_      ,   )
          if training_args.do_eval
          else None
      )

      # Initialize our Trainer
      _snake_case		:						Optional[Any]         =  Trainer(
          model=lowerCAmelCase_      ,   args=lowerCAmelCase_      ,   train_dataset=lowerCAmelCase_      ,   eval_dataset=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
      if training_args.do_eval:
            logger.info('''*** Evaluate ***'''						)

            _snake_case		:						List[Any]         =  trainer.predict(lowerCAmelCase_						)
            _snake_case		:						Tuple         =  output.predictions
            _snake_case		:						Tuple         =  np.argmax(lowerCAmelCase_      ,   axis=1						)

            _snake_case		:						Optional[Any]         =  [ex.pairID for ex in eval_dataset]
            _snake_case		:						int         =  os.path.join(training_args.output_dir      ,   '''hans_predictions.txt'''						)
            _snake_case		:						Dict         =  eval_dataset.get_labels()
            if trainer.is_world_master():
                  with open(lowerCAmelCase_      ,   '''w'''						) as writer:
                        writer.write('''pairID,gold_label\n'''						)
                        for pid, pred in zip(lowerCAmelCase_      ,   lowerCAmelCase_						):
                              writer.write('''ex''' + str(lowerCAmelCase_						) + ''',''' + label_list[int(lowerCAmelCase_						)] + '''\n'''						)

            trainer._log(output.metrics						)




def  _a     (				lowerCAmelCase_						):





      """simple docstring"""



      main()


if __name__ == "__main__":
   main()









'''simple docstring'''

from __future__ import annotations




def  _a     (				lowerCAmelCase_						):





      """simple docstring"""



      _snake_case		:						List[str]         =  sorted(lowerCAmelCase_						)
      _snake_case		:						Any         =  len(lowerCAmelCase_						)
      _snake_case		:						List[str]         =  length >> 1
      return (
          (sorted_list[mid_index] + sorted_list[mid_index - 1]) / 2
          if length % 2 == 0
          else sorted_list[mid_index]
      )




def  _a     (				):





      """simple docstring"""



      import doctest

      doctest.testmod()


if __name__ == "__main__":
   main()


