


"""simple docstring"""


import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_image_inputs


if is_torch_available():
		import torch

if is_vision_available():
		from PIL import Image

		from transformers import ConvNextImageProcessor
class snake_case					(     unittest.TestCase):







		def __init__(    self					:				str			,      a__					:				Dict			,      a__					:				List[Any]=7			,      a__					:				Union[str, Any]=3			,      a__					:				Dict=18			,      a__					:				Tuple=30			,      a__					:				List[Any]=4_00			,      a__					:				Optional[int]=True			,      a__					:				Any=None			,      a__					:				List[str]=0.8_7_5			,      a__					:				int=True			,      a__					:				List[str]=[0.5, 0.5, 0.5]			,      a__					:				List[str]=[0.5, 0.5, 0.5]			,      )				->  str:





						'''simple docstring'''







						_A         =	size if size is not None else {"shortest_edge": 20}
						_A         =	parent
						_A         =	batch_size
						_A         =	num_channels
						_A         =	image_size
						_A         =	min_resolution
						_A         =	max_resolution
						_A         =	do_resize
						_A         =	size
						_A         =	crop_pct
						_A         =	do_normalize
						_A         =	image_mean
						_A         =	image_std







		def        a_       (    self					:				Dict						)				->  List[Any]:





						'''simple docstring'''







						return {
						    "image_mean": self.image_mean,
						    "image_std": self.image_std,
						    "do_normalize": self.do_normalize,
						    "do_resize": self.do_resize,
						    "size": self.size,
						    "crop_pct": self.crop_pct,
						}





@require_torch
@require_vision
class snake_case					(     _UpperCamelCase   , unittest.TestCase):
		__UpperCamelCase							=	ConvNextImageProcessor if is_vision_available() else None







		def        a_       (    self					:				Optional[int]						)				->  Optional[Any]:





						'''simple docstring'''







						_A         =	ConvNextImageProcessingTester(self						)







		@property
		def        a_       (    self					:				int						)				->  List[Any]:





						'''simple docstring'''







						return self.image_processor_tester.prepare_image_processor_dict()







		def        a_       (    self					:				Union[str, Any]						)				->  List[Any]:





						'''simple docstring'''







						_A         =	self.image_processing_class(**self.image_processor_dict						)
						self.assertTrue(hasattr(a__			,      "do_resize"						)						)
						self.assertTrue(hasattr(a__			,      "size"						)						)
						self.assertTrue(hasattr(a__			,      "crop_pct"						)						)
						self.assertTrue(hasattr(a__			,      "do_normalize"						)						)
						self.assertTrue(hasattr(a__			,      "image_mean"						)						)
						self.assertTrue(hasattr(a__			,      "image_std"						)						)







		def        a_       (    self					:				Dict						)				->  Any:





						'''simple docstring'''







						_A         =	self.image_processing_class.from_dict(self.image_processor_dict						)
						self.assertEqual(image_processor.size			,      {"shortest_edge": 20}						)

						_A         =	self.image_processing_class.from_dict(self.image_processor_dict			,      size=42						)
						self.assertEqual(image_processor.size			,      {"shortest_edge": 42}						)







		def        a_       (    self					:				Any						)				->  Any:





						'''simple docstring'''







						pass







		def        a_       (    self					:				Optional[Any]						)				->  Any:





						'''simple docstring'''







						_A         =	self.image_processing_class(**self.image_processor_dict						)
						# create random PIL images
						_A         =	prepare_image_inputs(self.image_processor_tester			,      equal_resolution=a__						)
						for image in image_inputs:
										self.assertIsInstance(a__			,      Image.Image						)

						# Test not batched input
						_A         =	image_processing(image_inputs[0]			,      return_tensors="pt"						).pixel_values
						self.assertEqual(
						    encoded_images.shape			,      (
						        1,
						        self.image_processor_tester.num_channels,
						        self.image_processor_tester.size["shortest_edge"],
						        self.image_processor_tester.size["shortest_edge"],
						    )			,      )

						# Test batched
						_A         =	image_processing(a__			,      return_tensors="pt"						).pixel_values
						self.assertEqual(
						    encoded_images.shape			,      (
						        self.image_processor_tester.batch_size,
						        self.image_processor_tester.num_channels,
						        self.image_processor_tester.size["shortest_edge"],
						        self.image_processor_tester.size["shortest_edge"],
						    )			,      )







		def        a_       (    self					:				Union[str, Any]						)				->  Optional[int]:





						'''simple docstring'''







						_A         =	self.image_processing_class(**self.image_processor_dict						)
						# create random numpy tensors
						_A         =	prepare_image_inputs(self.image_processor_tester			,      equal_resolution=a__			,      numpify=a__						)
						for image in image_inputs:
										self.assertIsInstance(a__			,      np.ndarray						)

						# Test not batched input
						_A         =	image_processing(image_inputs[0]			,      return_tensors="pt"						).pixel_values
						self.assertEqual(
						    encoded_images.shape			,      (
						        1,
						        self.image_processor_tester.num_channels,
						        self.image_processor_tester.size["shortest_edge"],
						        self.image_processor_tester.size["shortest_edge"],
						    )			,      )

						# Test batched
						_A         =	image_processing(a__			,      return_tensors="pt"						).pixel_values
						self.assertEqual(
						    encoded_images.shape			,      (
						        self.image_processor_tester.batch_size,
						        self.image_processor_tester.num_channels,
						        self.image_processor_tester.size["shortest_edge"],
						        self.image_processor_tester.size["shortest_edge"],
						    )			,      )







		def        a_       (    self					:				Union[str, Any]						)				->  Dict:





						'''simple docstring'''







						_A         =	self.image_processing_class(**self.image_processor_dict						)
						# create random PyTorch tensors
						_A         =	prepare_image_inputs(self.image_processor_tester			,      equal_resolution=a__			,      torchify=a__						)
						for image in image_inputs:
										self.assertIsInstance(a__			,      torch.Tensor						)

						# Test not batched input
						_A         =	image_processing(image_inputs[0]			,      return_tensors="pt"						).pixel_values
						self.assertEqual(
						    encoded_images.shape			,      (
						        1,
						        self.image_processor_tester.num_channels,
						        self.image_processor_tester.size["shortest_edge"],
						        self.image_processor_tester.size["shortest_edge"],
						    )			,      )

						# Test batched
						_A         =	image_processing(a__			,      return_tensors="pt"						).pixel_values
						self.assertEqual(
						    encoded_images.shape			,      (
						        self.image_processor_tester.batch_size,
						        self.image_processor_tester.num_channels,
						        self.image_processor_tester.size["shortest_edge"],
						        self.image_processor_tester.size["shortest_edge"],
						    )			,      )



"""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


a_											=			logging.getLogger(__name__)
@dataclass
class snake_case					:
		__UpperCamelCase							=	field(
		    metadata={'help': 'Path to pretrained model or model identifier from huggingface.co/models'})
		__UpperCamelCase							=	field(
		    default=_UpperCamelCase   , metadata={'help': 'Pretrained config name or path if not the same as model_name'})
		__UpperCamelCase							=	field(
		    default=_UpperCamelCase   , metadata={'help': 'Pretrained tokenizer name or path if not the same as model_name'})
		__UpperCamelCase							=	field(
		    default=_UpperCamelCase   , metadata={'help': 'Where do you want to store the pretrained models downloaded from huggingface.co'}   , )





@dataclass
class snake_case					:
		__UpperCamelCase							=	field(
		    metadata={'help': 'The name of the task to train selected in the list: ' + ', '.join(hans_processors.keys())})
		__UpperCamelCase							=	field(
		    metadata={'help': 'The input data dir. Should contain the .tsv files (or other data files) for the task.'})
		__UpperCamelCase							=	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.'
		        )
		    }   , )
		__UpperCamelCase							=	field(
		    default=_UpperCamelCase   , metadata={'help': 'Overwrite the cached training and evaluation sets'})






def    a__       ( __lowercase      )		->		Dict[str, torch.Tensor]:
				_A         =	default_data_collator(__lowercase      )
				_A         =	batch.pop("pairID"   ,	__lowercase      )
				return batch






def    a__       ( )		->		Optional[Any]:
				# 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         =	HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)      )
				_A     ,						_A     ,						_A         =	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:
								_A         =	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.

				_A         =	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   ,	)
				_A         =	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         =	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
				_A         =	(
				    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
				)
				_A         =	(
				    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
				_A         =	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 ***"      )

								_A         =	trainer.predict(__lowercase      )
								_A         =	output.predictions
								_A         =	np.argmax(__lowercase   ,	axis=1      )

								_A         =	[ex.pairID for ex in eval_dataset]
								_A         =	os.path.join(training_args.output_dir   ,	"hans_predictions.txt"      )
								_A         =	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    a__       ( __lowercase      )		->		Dict:
				# For xla_spawn (TPUs)
				main()


if __name__ == "__main__":
		main()