



import unittest

from transformers import ErnieMConfig, is_torch_available
from transformers.testing_utils import require_torch, slow, torch_device

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
		import torch

		from transformers import (
		    ErnieMForInformationExtraction,
		    ErnieMForMultipleChoice,
		    ErnieMForQuestionAnswering,
		    ErnieMForSequenceClassification,
		    ErnieMForTokenClassification,
		    ErnieMModel,
		)
		from transformers.models.ernie_m.modeling_ernie_m import ERNIE_M_PRETRAINED_MODEL_ARCHIVE_LIST


class 	lowerCamelCase__ :


						def __init__( self						:  Union[str, Any]      ,       _a						:  Optional[Any]      ,       _a						:  List[str]=1_3      ,       _a						:  str=7      ,       _a						:  Optional[int]=True      ,       _a						:  Union[str, Any]=True      ,       _a						:  int=True      ,       _a						:  Optional[int]=9_9      ,       _a						:  Any=3_2      ,       _a						:  List[str]=5      ,       _a						:  Any=4      ,       _a						:  List[Any]=3_7      ,       _a						:  str="gelu"      ,       _a						:  Dict=0.1      ,       _a						:  Tuple=0.1      ,       _a						:  str=5_1_2      ,       _a						:  Any=1_6      ,       _a						:  Optional[int]=2      ,       _a						:  List[Any]=0.0_2      ,       _a						:  List[Any]=3      ,       _a						:  List[Any]=4      ,       _a						:  Optional[Any]=None      ,       ):
													a__:       int											=parent
													a__:       Optional[int]											=batch_size
													a__:       Any											=seq_length
													a__:       Any											=is_training
													a__:       int											=use_input_mask
													a__:       Optional[Any]											=use_labels
													a__:       Dict											=vocab_size
													a__:       List[Any]											=hidden_size
													a__:       Dict											=num_hidden_layers
													a__:       int											=num_attention_heads
													a__:       List[str]											=intermediate_size
													a__:       int											=hidden_act
													a__:       Tuple											=hidden_dropout_prob
													a__:       List[Any]											=attention_probs_dropout_prob
													a__:       int											=max_position_embeddings
													a__:       List[Any]											=type_vocab_size
													a__:       Tuple											=type_sequence_label_size
													a__:       Any											=initializer_range
													a__:       Optional[Any]											=num_labels
													a__:       Optional[Any]											=num_choices
													a__:       List[str]											=scope


						def   _lowerCamelCase			( self						:  List[Any]			):
													a__:       Optional[Any]											=ids_tensor([self.batch_size, self.seq_length]      ,       self.vocab_size			)

													a__:       Optional[int]											=None
													if self.use_input_mask:
																				a__:       str											=random_attention_mask([self.batch_size, self.seq_length]			)

													a__:       Optional[Any]											=None
													a__:       Any											=None
													a__:       Union[str, Any]											=None
													if self.use_labels:
																				a__:       List[str]											=ids_tensor([self.batch_size]      ,       self.type_sequence_label_size			)
																				a__:       Any											=ids_tensor([self.batch_size, self.seq_length]      ,       self.num_labels			)
																				a__:       Any											=ids_tensor([self.batch_size]      ,       self.num_choices			)

													a__:       str											=self.get_config()

													return config, input_ids, input_mask, sequence_labels, token_labels, choice_labels


						def   _lowerCamelCase			( self						:  Tuple			):
													a__:       int											=ids_tensor([self.batch_size, self.seq_length]      ,       self.vocab_size			)

													a__:       Dict											=None
													if self.use_input_mask:
																				a__:       List[Any]											=random_attention_mask([self.batch_size, self.seq_length]			)
													a__:       Optional[Any]											=self.get_config()

													return config, input_ids, input_mask


						def   _lowerCamelCase			( self						:  Tuple			):
													return ErnieMConfig(
													    vocab_size=self.vocab_size      ,       hidden_size=self.hidden_size      ,       num_hidden_layers=self.num_hidden_layers      ,       num_attention_heads=self.num_attention_heads      ,       intermediate_size=self.intermediate_size      ,       hidden_act=self.hidden_act      ,       hidden_dropout_prob=self.hidden_dropout_prob      ,       attention_probs_dropout_prob=self.attention_probs_dropout_prob      ,       max_position_embeddings=self.max_position_embeddings      ,       type_vocab_size=self.type_vocab_size      ,       initializer_range=self.initializer_range      ,       )


						def   _lowerCamelCase			( self						:  Optional[Any]      ,       _a						:  str      ,       _a						:  int      ,       _a						:  List[str]      ,       _a						:  str      ,       _a						:  Union[str, Any]      ,       _a						:  Any			):
													a__:       Dict											=ErnieMModel(config=_a			)
													model.to(_a			)
													model.eval()
													a__:       Optional[Any]											=model(_a      ,       return_dict=_a			)
													self.parent.assertEqual(result.last_hidden_state.shape      ,       (self.batch_size, self.seq_length, self.hidden_size)			)


						def   _lowerCamelCase			( self						:  List[str]      ,       _a						:  List[Any]      ,       _a						:  Optional[int]      ,       _a						:  Any      ,       _a						:  str      ,       _a						:  Dict      ,       _a						:  Dict			):
													a__:       Optional[Any]											=ErnieMForQuestionAnswering(config=_a			)
													model.to(_a			)
													model.eval()
													a__:       Dict											=model(
													    _a      ,       attention_mask=_a      ,       start_positions=_a      ,       end_positions=_a      ,       )
													self.parent.assertEqual(result.start_logits.shape      ,       (self.batch_size, self.seq_length)			)
													self.parent.assertEqual(result.end_logits.shape      ,       (self.batch_size, self.seq_length)			)


						def   _lowerCamelCase			( self						:  Tuple      ,       _a						:  Tuple      ,       _a						:  Tuple      ,       _a						:  Tuple      ,       _a						:  str      ,       _a						:  List[str]      ,       _a						:  Dict			):
													a__:       Any											=ErnieMForInformationExtraction(config=_a			)
													model.to(_a			)
													model.eval()
													a__:       List[str]											=torch.ones_like(_a      ,       dtype=torch.floataa			)
													a__:       Union[str, Any]											=model(
													    _a      ,       attention_mask=_a      ,       start_positions=_a      ,       end_positions=_a      ,       )
													self.parent.assertEqual(result.start_logits.shape      ,       (self.batch_size, self.seq_length)			)
													self.parent.assertEqual(result.end_logits.shape      ,       (self.batch_size, self.seq_length)			)


						def   _lowerCamelCase			( self						:  Dict      ,       _a						:  str      ,       _a						:  Union[str, Any]      ,       _a						:  Tuple      ,       _a						:  Any      ,       _a						:  Any      ,       _a						:  Optional[Any]			):
													a__:       Optional[int]											=self.num_labels
													a__:       Tuple											=ErnieMForSequenceClassification(_a			)
													model.to(_a			)
													model.eval()
													a__:       Optional[int]											=model(_a      ,       attention_mask=_a      ,       labels=_a			)
													self.parent.assertEqual(result.logits.shape      ,       (self.batch_size, self.num_labels)			)


						def   _lowerCamelCase			( self						:  List[str]      ,       _a						:  Any      ,       _a						:  List[Any]      ,       _a						:  List[str]      ,       _a						:  Any      ,       _a						:  Optional[int]      ,       _a						:  Optional[int]			):
													a__:       Optional[Any]											=self.num_labels
													a__:       Union[str, Any]											=ErnieMForTokenClassification(config=_a			)
													model.to(_a			)
													model.eval()
													input_ids.to(_a			)
													input_mask.to(_a			)
													token_labels.to(_a			)

													a__:       Union[str, Any]											=model(_a      ,       attention_mask=_a      ,       labels=_a			)

													self.parent.assertEqual(result.logits.shape      ,       (self.batch_size, self.seq_length, self.num_labels)			)


						def   _lowerCamelCase			( self						:  Tuple      ,       _a						:  int      ,       _a						:  List[str]      ,       _a						:  str      ,       _a						:  List[str]      ,       _a						:  Any      ,       _a						:  Dict			):
													a__:       Optional[Any]											=self.num_choices
													a__:       Tuple											=ErnieMForMultipleChoice(config=_a			)
													model.to(_a			)
													model.eval()
													a__:       List[str]											=input_ids.unsqueeze(1			).expand(-1      ,       self.num_choices      ,       -1			).contiguous()
													a__:       Dict											=input_mask.unsqueeze(1			).expand(-1      ,       self.num_choices      ,       -1			).contiguous()
													a__:       int											=model(
													    _a      ,       attention_mask=_a      ,       labels=_a      ,       )
													self.parent.assertEqual(result.logits.shape      ,       (self.batch_size, self.num_choices)			)







						def   _lowerCamelCase			( self						:  Tuple			):
													a__:       List[str]											=self.prepare_config_and_inputs()
													(
													    (
													    a__
													)							,			(
													    a__
													)							,			(
													    a__
													)							,			(
													    a__
													)							,			(
													    a__
													)							,			(
													    a__
													)							,			
													):       Dict											=config_and_inputs
													a__:       Optional[int]											={"input_ids": input_ids, "attention_mask": input_mask}
													return config, inputs_dict


@require_torch
class 	lowerCamelCase__ (		_a				,					_a				,					unittest.TestCase   ):
						_lowerCAmelCase								= (
						    (
						        ErnieMModel,
						        ErnieMForMultipleChoice,
						        ErnieMForQuestionAnswering,
						        ErnieMForSequenceClassification,
						        ErnieMForTokenClassification,
						    )
						    if is_torch_available()
						    else ()
						)
						_lowerCAmelCase								= ()
						_lowerCAmelCase								= (
						    {
						        '''feature-extraction''': ErnieMModel,
						        '''question-answering''': ErnieMForQuestionAnswering,
						        '''text-classification''': ErnieMForSequenceClassification,
						        '''token-classification''': ErnieMForTokenClassification,
						        '''zero-shot''': ErnieMForSequenceClassification,
						    }
						    if is_torch_available()
						    else {}
						)
						_lowerCAmelCase								= False


						def   _lowerCamelCase			( self						:  int      ,       _a						:  Dict      ,       _a						:  int      ,       _a						:  Dict      ,       _a						:  List[Any]      ,       _a						:  Any			):
													if pipeline_test_casse_name == "QAPipelineTests":
																				return True

													return False


						def   _lowerCamelCase			( self						:  Optional[Any]			):
													a__:       str											=ErnieMModelTester(self			)
													a__:       Tuple											=ConfigTester(self      ,       config_class=_a      ,       hidden_size=3_7			)


						def   _lowerCamelCase			( self						:  Optional[int]			):
													self.config_tester.run_common_tests()


						def   _lowerCamelCase			( self						:  Optional[Any]			):
													a__:       Optional[Any]											=self.model_tester.prepare_config_and_inputs()
													self.model_tester.create_and_check_model(*_a			)


						def   _lowerCamelCase			( self						:  Tuple			):
													a__:       Optional[int]											=self.model_tester.prepare_config_and_inputs()
													for type in ["absolute", "relative_key", "relative_key_query"]:
																				a__:       Dict											=type
																				self.model_tester.create_and_check_model(*_a			)


						def   _lowerCamelCase			( self						:  int			):
													a__:       Union[str, Any]											=self.model_tester.prepare_config_and_inputs()
													self.model_tester.create_and_check_for_multiple_choice(*_a			)


						def   _lowerCamelCase			( self						:  Any			):
													a__:       List[Any]											=self.model_tester.prepare_config_and_inputs()
													self.model_tester.create_and_check_for_question_answering(*_a			)


						def   _lowerCamelCase			( self						:  List[Any]			):
													a__:       Dict											=self.model_tester.prepare_config_and_inputs()
													self.model_tester.create_and_check_for_sequence_classification(*_a			)


						def   _lowerCamelCase			( self						:  Optional[Any]			):
													a__:       List[Any]											=self.model_tester.prepare_config_and_inputs()
													self.model_tester.create_and_check_for_information_extraction(*_a			)


						def   _lowerCamelCase			( self						:  int			):
													a__:       List[Any]											=self.model_tester.prepare_config_and_inputs()
													self.model_tester.create_and_check_for_token_classification(*_a			)







						@slow
						def   _lowerCamelCase			( self						:  List[str]			):
													for model_name in ERNIE_M_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
																				a__:       Optional[Any]											=ErnieMModel.from_pretrained(_a			)
																				self.assertIsNotNone(_a			)


@require_torch
class 	lowerCamelCase__ (		unittest.TestCase   ):


						@slow
						def   _lowerCamelCase			( self						:  Optional[int]			):
													a__:       Tuple											=ErnieMModel.from_pretrained("susnato/ernie-m-base_pytorch"			)
													model.eval()
													a__:       Any											=torch.tensor([[0, 1, 2, 3, 4, 5]]			)
													a__:       Any											=model(_a			)[0]

													# TODO Replace vocab size
													a__:       Optional[int]											=7_6_8

													a__:       int											=torch.Size((1, 6, hidden_size)			)
													self.assertEqual(output.shape      ,       _a			)

													a__:       List[Any]											=torch.tensor(
													    [[[-0.0_0_1_2, 0.1_2_4_5, -0.0_2_1_4], [-0.0_7_4_2, 0.0_2_4_4, -0.0_7_7_1], [-0.0_3_3_3, 0.1_1_6_4, -0.1_5_5_4]]]			)

													self.assertTrue(torch.allclose(output[:, :3, :3]      ,       _a      ,       atol=1e-3			)			)










# This test is meant to be run in on an instance with TPUs like this:
#
#   python examples/pytorch/xla_spawn.py --num_cores=8 tests/test_trainer_tpu.py
#
# Replace 8 with the number of TPU cores you have.
#

import sys
from typing import Dict

from transformers import EvalPrediction, HfArgumentParser, TrainingArguments, is_torch_available
from transformers.utils import logging


__UpperCAmelCase						  =     logging.get_logger(__name__)


if is_torch_available():
		import torch
		from torch import nn
		from torch.utils.data import Dataset

		from transformers import Trainer


		class 	lowerCamelCase__ (		_a   ):


								def __init__( self						:  Any      ,       _a						:  int = 1_0_1			):
															a__:       Dict											=length


								def __len__( self						:  Optional[Any]			):
															return self.length







								def __getitem__( self						:  Dict      ,       _a						:  int			):
															return i


		class 	lowerCamelCase__ :


								def __call__( self						:  Any      ,       _a						:  List[str]			):
															return {"input_ids": torch.tensor(_a			), "labels": torch.tensor(_a			)}


		class 	lowerCamelCase__ (		nn.Module   ):


								def __init__( self						:  Tuple			):
															super().__init__()
															# Add some (unused) params otherwise DDP will complain.
															a__:       int											=nn.Linear(1_2_0      ,       8_0			)







								def   _lowerCamelCase			( self						:  str      ,       _a						:  List[Any]      ,       _a						:  List[str]=None			):
															if labels is not None:
																						return torch.tensor(0.0      ,       device=input_ids.device			), input_ids
															else:
																						return input_ids





def 						__lowerCamelCase       (      ):
							a__:       List[str]											=HfArgumentParser((TrainingArguments,)						)
							sys.argv += ["--output_dir", "./examples"]
							a__:       int											=parser.parse_args_into_dataclasses()[0]

							logger.warning(
							    F"Process rank: {training_args.local_rank}, device: {training_args.device}, "
							    F"tpu_num_cores: {training_args.tpu_num_cores}"     , )

							# Essentially, what we want to verify in the distributed case is
							# that we get all samples back, in the right order.
							# (this is crucial for prediction for instance)
							for dataset_length in [1_001, 256, 15]:
														a__:       Optional[Any]											=DummyDataset(__magic_name__						)

														def compute_metrics(__magic_name__	:  EvalPrediction						) -> Dict:
																					a__:       List[str]											=list(range(len(__magic_name__						)						)						)
																					a__:       Union[str, Any]											=p.predictions.tolist() == sequential and p.label_ids.tolist() == sequential
																					return {"success": success}

														a__:       Union[str, Any]											=Trainer(
														    model=DummyModel()     , args=__magic_name__     , data_collator=DummyDataCollator()     , eval_dataset=__magic_name__     , compute_metrics=__magic_name__     , )
														a__:       str											=trainer.evaluate()
														logger.info(__magic_name__						)
														if metrics["eval_success"] is not True:
																					logger.error(__magic_name__						)
																					exit(1						)

														a__:       List[str]											=trainer.predict(__magic_name__						)
														logger.info(p.metrics						)
														if p.metrics["test_success"] is not True:
																					logger.error(p.metrics						)
																					exit(1						)

														a__:       Tuple											=2

														a__:       Optional[int]											=trainer.evaluate()
														logger.info(__magic_name__						)
														if metrics["eval_success"] is not True:
																					logger.error(__magic_name__						)
																					exit(1						)

														a__:       List[str]											=trainer.predict(__magic_name__						)
														logger.info(p.metrics						)
														if p.metrics["test_success"] is not True:
																					logger.error(p.metrics						)
																					exit(1						)

														a__:       List[str]											=None

							logger.info("🔥 All distributed tests successful"						)





def 						__lowerCamelCase       (      __magic_name__	:  Optional[Any]						):
							# For xla_spawn (TPUs)
							main()


if __name__ == "__main__":
		main()





