






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


_snake_case							:   List[str]	  =       logging.getLogger(__name__)




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




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




def     lowerCamelCase_					(	__A	:		List[InputFeatures]      ):
					lowerCamelCase__					=default_data_collator(__A      )
					lowerCamelCase__					=batch.pop('pairID'      ,						__A      )
					return batch




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.

					lowerCamelCase__					=HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)      )
					lowerCamelCase__					,  lowerCamelCase__					,  lowerCamelCase__					=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'      ,						__A      )

					# Set seed
					set_seed(training_args.seed      )

					try:
										lowerCamelCase__					=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.

					lowerCamelCase__					=AutoConfig.from_pretrained(
					    model_args.config_name if model_args.config_name else model_args.model_name_or_path      ,						num_labels=__A      ,						finetuning_task=data_args.task_name      ,						cache_dir=model_args.cache_dir      ,						)
					lowerCamelCase__					=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      ,						)
					lowerCamelCase__					=AutoModelForSequenceClassification.from_pretrained(
					    model_args.model_name_or_path      ,						from_tf=bool('.ckpt' in model_args.model_name_or_path      )      ,						config=__A      ,						cache_dir=model_args.cache_dir      ,						)

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

					# Initialize our Trainer
					lowerCamelCase__					=Trainer(
					    model=__A      ,						args=__A      ,						train_dataset=__A      ,						eval_dataset=__A      ,						data_collator=__A      ,						)

					# 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 ***'      )

										lowerCamelCase__					=trainer.predict(__A      )
										lowerCamelCase__					=output.predictions
										lowerCamelCase__					=np.argmax(__A      ,						axis=1      )

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

										trainer._log(output.metrics      )




def     lowerCamelCase_					(	__A	:		List[Any]      ):
					# For xla_spawn (TPUs)
					main()


if __name__ == "__main__":
						main()







"""simple docstring"""



import inspect
import unittest

from transformers import ResNetConfig
from transformers.testing_utils import require_torch, require_vision, slow, torch_device
from transformers.utils import cached_property, is_torch_available, is_vision_available

from ...test_backbone_common import BackboneTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
						import torch
						from torch import nn

						from transformers import ResNetBackbone, ResNetForImageClassification, ResNetModel
						from transformers.models.resnet.modeling_resnet import RESNET_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
						from PIL import Image

						from transformers import AutoImageProcessor




class 							__lowercase:






						def __init__(       self     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE=3     ,    _SCREAMING_SNAKE_CASE=32     ,    _SCREAMING_SNAKE_CASE=3     ,    _SCREAMING_SNAKE_CASE=10     ,    _SCREAMING_SNAKE_CASE=[10, 20, 30, 40]     ,    _SCREAMING_SNAKE_CASE=[1, 1, 2, 1]     ,    _SCREAMING_SNAKE_CASE=True     ,    _SCREAMING_SNAKE_CASE=True     ,    _SCREAMING_SNAKE_CASE="relu"     ,    _SCREAMING_SNAKE_CASE=3     ,    _SCREAMING_SNAKE_CASE=None     ,    _SCREAMING_SNAKE_CASE=["stage2", "stage3", "stage4"]     ,    _SCREAMING_SNAKE_CASE=[2, 3, 4]     ,    )      ->	Union[str, Any]:




											"""simple docstring"""





											lowerCamelCase__					=parent
											lowerCamelCase__					=batch_size
											lowerCamelCase__					=image_size
											lowerCamelCase__					=num_channels
											lowerCamelCase__					=embeddings_size
											lowerCamelCase__					=hidden_sizes
											lowerCamelCase__					=depths
											lowerCamelCase__					=is_training
											lowerCamelCase__					=use_labels
											lowerCamelCase__					=hidden_act
											lowerCamelCase__					=num_labels
											lowerCamelCase__					=scope
											lowerCamelCase__					=len(_SCREAMING_SNAKE_CASE)
											lowerCamelCase__					=out_features
											lowerCamelCase__					=out_indices






						def 		lowerCAmelCase (       self)      ->	Tuple:




											"""simple docstring"""





											lowerCamelCase__					=floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size])

											lowerCamelCase__					=None
											if self.use_labels:
																lowerCamelCase__					=ids_tensor([self.batch_size]     ,    self.num_labels)

											lowerCamelCase__					=self.get_config()

											return config, pixel_values, labels






						def 		lowerCAmelCase (       self)      ->	int:




											"""simple docstring"""





											return ResNetConfig(
											    num_channels=self.num_channels     ,    embeddings_size=self.embeddings_size     ,    hidden_sizes=self.hidden_sizes     ,    depths=self.depths     ,    hidden_act=self.hidden_act     ,    num_labels=self.num_labels     ,    out_features=self.out_features     ,    out_indices=self.out_indices     ,    )






						def 		lowerCAmelCase (       self     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE)      ->	Union[str, Any]:




											"""simple docstring"""





											lowerCamelCase__					=ResNetModel(config=_SCREAMING_SNAKE_CASE)
											model.to(_SCREAMING_SNAKE_CASE)
											model.eval()
											lowerCamelCase__					=model(_SCREAMING_SNAKE_CASE)
											# expected last hidden states: B, C, H // 32, W // 32
											self.parent.assertEqual(
											    result.last_hidden_state.shape     ,    (self.batch_size, self.hidden_sizes[-1], self.image_size // 32, self.image_size // 32)     ,    )






						def 		lowerCAmelCase (       self     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE)      ->	List[Any]:




											"""simple docstring"""





											lowerCamelCase__					=self.num_labels
											lowerCamelCase__					=ResNetForImageClassification(_SCREAMING_SNAKE_CASE)
											model.to(_SCREAMING_SNAKE_CASE)
											model.eval()
											lowerCamelCase__					=model(_SCREAMING_SNAKE_CASE     ,    labels=_SCREAMING_SNAKE_CASE)
											self.parent.assertEqual(result.logits.shape     ,    (self.batch_size, self.num_labels))






						def 		lowerCAmelCase (       self     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE)      ->	Union[str, Any]:




											"""simple docstring"""





											lowerCamelCase__					=ResNetBackbone(config=_SCREAMING_SNAKE_CASE)
											model.to(_SCREAMING_SNAKE_CASE)
											model.eval()
											lowerCamelCase__					=model(_SCREAMING_SNAKE_CASE)

											# verify feature maps
											self.parent.assertEqual(len(result.feature_maps)     ,    len(config.out_features))
											self.parent.assertListEqual(list(result.feature_maps[0].shape)     ,    [self.batch_size, self.hidden_sizes[1], 4, 4])

											# verify channels
											self.parent.assertEqual(len(model.channels)     ,    len(config.out_features))
											self.parent.assertListEqual(model.channels     ,    config.hidden_sizes[1:])

											# verify backbone works with out_features=None
											lowerCamelCase__					=None
											lowerCamelCase__					=ResNetBackbone(config=_SCREAMING_SNAKE_CASE)
											model.to(_SCREAMING_SNAKE_CASE)
											model.eval()
											lowerCamelCase__					=model(_SCREAMING_SNAKE_CASE)

											# verify feature maps
											self.parent.assertEqual(len(result.feature_maps)     ,    1)
											self.parent.assertListEqual(list(result.feature_maps[0].shape)     ,    [self.batch_size, self.hidden_sizes[-1], 1, 1])

											# verify channels
											self.parent.assertEqual(len(model.channels)     ,    1)
											self.parent.assertListEqual(model.channels     ,    [config.hidden_sizes[-1]])






						def 		lowerCAmelCase (       self)      ->	Tuple:




											"""simple docstring"""





											lowerCamelCase__					=self.prepare_config_and_inputs()
											lowerCamelCase__					,  lowerCamelCase__					,  lowerCamelCase__					=config_and_inputs
											lowerCamelCase__					={'pixel_values': pixel_values}
											return config, inputs_dict




@require_torch
class 							__lowercase(							_UpperCamelCase    ,				_UpperCamelCase    ,				unittest.TestCase	):
						lowerCAmelCase_		:				List[Any]=							(
						    (
						        ResNetModel,
						        ResNetForImageClassification,
						        ResNetBackbone,
						    )
						    if is_torch_available()
						    else ()
						)
						lowerCAmelCase_		:				Optional[Any]=							(
						    {"""feature-extraction""": ResNetModel, """image-classification""": ResNetForImageClassification}
						    if is_torch_available()
						    else {}
						)

						lowerCAmelCase_		:				Optional[int]=							True
						lowerCAmelCase_		:				Tuple=							False
						lowerCAmelCase_		:				Union[str, Any]=							False
						lowerCAmelCase_		:				Tuple=							False
						lowerCAmelCase_		:				Optional[int]=							False






						def 		lowerCAmelCase (       self)      ->	List[str]:




											"""simple docstring"""





											lowerCamelCase__					=ResNetModelTester(self)
											lowerCamelCase__					=ConfigTester(self     ,    config_class=_SCREAMING_SNAKE_CASE     ,    has_text_modality=_SCREAMING_SNAKE_CASE)






						def 		lowerCAmelCase (       self)      ->	Any:




											"""simple docstring"""





											self.create_and_test_config_common_properties()
											self.config_tester.create_and_test_config_to_json_string()
											self.config_tester.create_and_test_config_to_json_file()
											self.config_tester.create_and_test_config_from_and_save_pretrained()
											self.config_tester.create_and_test_config_with_num_labels()
											self.config_tester.check_config_can_be_init_without_params()
											self.config_tester.check_config_arguments_init()






						def 		lowerCAmelCase (       self)      ->	int:




											"""simple docstring"""





											return






						@unittest.skip(reason='ResNet does not use inputs_embeds')
						def 		lowerCAmelCase (       self)      ->	int:




											"""simple docstring"""





											pass






						@unittest.skip(reason='ResNet does not support input and output embeddings')
						def 		lowerCAmelCase (       self)      ->	List[str]:




											"""simple docstring"""





											pass






						def 		lowerCAmelCase (       self)      ->	str:




											"""simple docstring"""





											lowerCamelCase__					,  lowerCamelCase__					=self.model_tester.prepare_config_and_inputs_for_common()

											for model_class in self.all_model_classes:
																lowerCamelCase__					=model_class(_SCREAMING_SNAKE_CASE)
																lowerCamelCase__					=inspect.signature(model.forward)
																# signature.parameters is an OrderedDict => so arg_names order is deterministic
																lowerCamelCase__					=[*signature.parameters.keys()]

																lowerCamelCase__					=['pixel_values']
																self.assertListEqual(arg_names[:1]     ,    _SCREAMING_SNAKE_CASE)






						def 		lowerCAmelCase (       self)      ->	Any:




											"""simple docstring"""





											lowerCamelCase__					=self.model_tester.prepare_config_and_inputs()
											self.model_tester.create_and_check_model(*_SCREAMING_SNAKE_CASE)






						def 		lowerCAmelCase (       self)      ->	List[Any]:




											"""simple docstring"""





											lowerCamelCase__					=self.model_tester.prepare_config_and_inputs()
											self.model_tester.create_and_check_backbone(*_SCREAMING_SNAKE_CASE)






						def 		lowerCAmelCase (       self)      ->	List[str]:




											"""simple docstring"""





											lowerCamelCase__					,  lowerCamelCase__					=self.model_tester.prepare_config_and_inputs_for_common()

											for model_class in self.all_model_classes:
																lowerCamelCase__					=model_class(config=_SCREAMING_SNAKE_CASE)
																for name, module in model.named_modules():
																					if isinstance(_SCREAMING_SNAKE_CASE     ,    (nn.BatchNormad, nn.GroupNorm)):
																										self.assertTrue(
																										    torch.all(module.weight == 1)     ,    msg=F"Parameter {name} of model {model_class} seems not properly initialized"     ,    )
																										self.assertTrue(
																										    torch.all(module.bias == 0)     ,    msg=F"Parameter {name} of model {model_class} seems not properly initialized"     ,    )






						def 		lowerCAmelCase (       self)      ->	List[str]:




											"""simple docstring"""





											def check_hidden_states_output(_SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE):
																lowerCamelCase__					=model_class(_SCREAMING_SNAKE_CASE)
																model.to(_SCREAMING_SNAKE_CASE)
																model.eval()

																with torch.no_grad():
																					lowerCamelCase__					=model(**self._prepare_for_class(_SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE))

																lowerCamelCase__					=outputs.encoder_hidden_states if config.is_encoder_decoder else outputs.hidden_states

																lowerCamelCase__					=self.model_tester.num_stages
																self.assertEqual(len(_SCREAMING_SNAKE_CASE)     ,    expected_num_stages + 1)

																# ResNet's feature maps are of shape (batch_size, num_channels, height, width)
																self.assertListEqual(
																    list(hidden_states[0].shape[-2:])     ,    [self.model_tester.image_size // 4, self.model_tester.image_size // 4]     ,    )

											lowerCamelCase__					,  lowerCamelCase__					=self.model_tester.prepare_config_and_inputs_for_common()
											lowerCamelCase__					=['basic', 'bottleneck']
											for model_class in self.all_model_classes:
																for layer_type in layers_type:
																					lowerCamelCase__					=layer_type
																					lowerCamelCase__					=True
																					check_hidden_states_output(_SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE)

																					# check that output_hidden_states also work using config
																					del inputs_dict["output_hidden_states"]
																					lowerCamelCase__					=True

																					check_hidden_states_output(_SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE)






						@unittest.skip(reason='ResNet does not use feedforward chunking')
						def 		lowerCAmelCase (       self)      ->	Any:




											"""simple docstring"""





											pass






						def 		lowerCAmelCase (       self)      ->	Dict:




											"""simple docstring"""





											lowerCamelCase__					=self.model_tester.prepare_config_and_inputs()
											self.model_tester.create_and_check_for_image_classification(*_SCREAMING_SNAKE_CASE)






						@slow
						def 		lowerCAmelCase (       self)      ->	int:




											"""simple docstring"""





											for model_name in RESNET_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
																lowerCamelCase__					=ResNetModel.from_pretrained(_SCREAMING_SNAKE_CASE)
																self.assertIsNotNone(_SCREAMING_SNAKE_CASE)




def     lowerCamelCase_					(	):
					lowerCamelCase__					=Image.open('./tests/fixtures/tests_samples/COCO/000000039769.png'      )
					return image




@require_torch
@require_vision
class 							__lowercase(							unittest.TestCase	):






						@cached_property
						def 		lowerCAmelCase (       self)      ->	Union[str, Any]:




											"""simple docstring"""





											return (
											    AutoImageProcessor.from_pretrained(RESNET_PRETRAINED_MODEL_ARCHIVE_LIST[0])
											    if is_vision_available()
											    else None
											)






						@slow
						def 		lowerCAmelCase (       self)      ->	Any:




											"""simple docstring"""





											lowerCamelCase__					=ResNetForImageClassification.from_pretrained(RESNET_PRETRAINED_MODEL_ARCHIVE_LIST[0]).to(_SCREAMING_SNAKE_CASE)

											lowerCamelCase__					=self.default_image_processor
											lowerCamelCase__					=prepare_img()
											lowerCamelCase__					=image_processor(images=_SCREAMING_SNAKE_CASE     ,    return_tensors='pt').to(_SCREAMING_SNAKE_CASE)

											# forward pass
											with torch.no_grad():
																lowerCamelCase__					=model(**_SCREAMING_SNAKE_CASE)

											# verify the logits
											lowerCamelCase__					=torch.Size((1, 10_00))
											self.assertEqual(outputs.logits.shape     ,    _SCREAMING_SNAKE_CASE)

											lowerCamelCase__					=torch.tensor([-11.1069, -9.7877, -8.3777]).to(_SCREAMING_SNAKE_CASE)

											self.assertTrue(torch.allclose(outputs.logits[0, :3]     ,    _SCREAMING_SNAKE_CASE     ,    atol=1E-4))




@require_torch
class 							__lowercase(							_UpperCamelCase    ,				unittest.TestCase	):
						lowerCAmelCase_		:				Optional[Any]=							(ResNetBackbone,) if is_torch_available() else ()
						lowerCAmelCase_		:				Dict=							False
						lowerCAmelCase_		:				int=							ResNetConfig






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




											"""simple docstring"""





											lowerCamelCase__					=ResNetModelTester(self)