




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


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







@dataclass
class      lowerCamelCase__      :
			a__	: str        =       field(
			    metadata={"""help""": """The name of the task to train selected in the list: """ + """, """.join(hans_processors.keys()     )}     )
			a__	: str        =       field(
			    metadata={"""help""": """The input data dir. Should contain the .tsv files (or other data files) for the task."""}     )
			a__	: int        =       field(
			    default=1_28			,					metadata={
			        """help""": (
			            """The maximum total input sequence length after tokenization. Sequences longer """
			            """than this will be truncated, sequences shorter will be padded."""
			        )
			    }			,					)
			a__	: bool        =       field(
			    default=lowercase__			,					metadata={"""help""": """Overwrite the cached training and evaluation sets"""}     )







def 					UpperCamelCase__       (						lowercase__					:       Dict				):
		snake_case						:  int        =			default_data_collator(lowercase_				)
		snake_case						:  List[str]        =			batch.pop("pairID"    ,       lowercase_				)
		return batch



def 					UpperCamelCase__       (						):
		snake_case						:  Any        =			HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)				)
		snake_case     ,			snake_case     ,			snake_case						:  List[str]        =			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"    ,       lowercase_				)

		# Set seed
		set_seed(training_args.seed				)

		try:
				snake_case						:  List[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						:  Tuple        =			AutoConfig.from_pretrained(
		    model_args.config_name if model_args.config_name else model_args.model_name_or_path    ,       num_labels=lowercase_    ,       finetuning_task=data_args.task_name    ,       cache_dir=model_args.cache_dir    ,       )
		snake_case						:  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						:  List[str]        =			AutoModelForSequenceClassification.from_pretrained(
		    model_args.model_name_or_path    ,       from_tf=bool(".ckpt" in model_args.model_name_or_path				)    ,       config=lowercase_    ,       cache_dir=model_args.cache_dir    ,       )

		# Get datasets
		snake_case						:  str        =			(
		    HansDataset(
		        data_dir=data_args.data_dir    ,       tokenizer=lowercase_    ,       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=lowercase_    ,       task=data_args.task_name    ,       max_seq_length=data_args.max_seq_length    ,       overwrite_cache=data_args.overwrite_cache    ,       evaluate=lowercase_    ,       )
		    if training_args.do_eval
		    else None
		)

		# Initialize our Trainer
		snake_case						:  Tuple        =			Trainer(
		    model=lowercase_    ,       args=lowercase_    ,       train_dataset=lowercase_    ,       eval_dataset=lowercase_    ,       data_collator=lowercase_    ,       )

		# 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						:  Tuple        =			trainer.predict(lowercase_				)
				snake_case						:  int        =			output.predictions
				snake_case						:  Tuple        =			np.argmax(lowercase_    ,       axis=1				)

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

				trainer._log(output.metrics				)



def 					UpperCamelCase__       (						lowercase__					:       List[str]				):
		main()


if __name__ == "__main__":
							main()







"""simple docstring"""

def 					UpperCamelCase__       (						lowercase__					:       int				):
		if not isinstance(lowercase__    ,       lowercase__				):
				raise ValueError("double_factorial() only accepts integral values"				)
		if n < 0:
				raise ValueError("double_factorial() not defined for negative values"				)
		return 1 if n <= 1 else n * double_factorial(n - 2				)


if __name__ == "__main__":
							import doctest

							doctest.testmod()

