
from __future__ import annotations

import copy
import os
import tempfile
from importlib import import_module
from math import isnan

from transformers import is_tf_available
from transformers.models.auto import get_values
from transformers.testing_utils import _tf_gpu_memory_limit, require_tf, slow

from ..test_modeling_tf_common import ids_tensor


if is_tf_available():
					import numpy as np
					import tensorflow as tf

					from transformers import (
					    TF_MODEL_FOR_CAUSAL_LM_MAPPING,
					    TF_MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING,
					    TF_MODEL_FOR_MASKED_LM_MAPPING,
					    TF_MODEL_FOR_MULTIPLE_CHOICE_MAPPING,
					    TF_MODEL_FOR_NEXT_SENTENCE_PREDICTION_MAPPING,
					    TF_MODEL_FOR_PRETRAINING_MAPPING,
					    TF_MODEL_FOR_QUESTION_ANSWERING_MAPPING,
					    TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING,
					    TF_MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING,
					    TF_MODEL_FOR_TOKEN_CLASSIFICATION_MAPPING,
					    TFSharedEmbeddings,
					)

					if _tf_gpu_memory_limit is not None:
										SCREAMING_SNAKE_CASE							:int											=       tf.config.list_physical_devices('GPU')
										for gpu in gpus:
															# Restrict TensorFlow to only allocate x GB of memory on the GPUs
															try:
																				tf.config.set_logical_device_configuration(
																				    gpu, [tf.config.LogicalDeviceConfiguration(memory_limit=_tf_gpu_memory_limit)]
																				)
																				SCREAMING_SNAKE_CASE							:Tuple											=       tf.config.list_logical_devices('GPU')
																				print('Logical GPUs', logical_gpus)
															except RuntimeError as e:
																				# Virtual devices must be set before GPUs have been initialized
																				print(e)





@require_tf
class 							UpperCAmelCase   :




	'''simple docstring'''





	snake_case_      					=  None
	snake_case_      					=  ()
	snake_case_      					=  ()
	snake_case_      					=  True
	snake_case_      					=  True
	snake_case_      					=  True
	snake_case_      					=  False






	def        UpperCamelCase_						( self      :				Optional[Any]							,A      :				List[str]							,A      :				List[Any]							,A      :				Tuple=False				):
					__A         =				copy.deepcopy(A				)

					if model_class in get_values(A				):
									__A         =				{
									    k: tf.tile(tf.expand_dims(A							,1				)							,(1, self.model_tester.num_choices) + (1,) * (v.ndim - 1)				)
									    if isinstance(A							,tf.Tensor				) and v.ndim > 0
									    else v
									    for k, v in inputs_dict.items()
									}

					if return_labels:
									if model_class in get_values(A				):
													__A         =				tf.ones(self.model_tester.batch_size							,dtype=tf.intaa				)
									elif model_class in get_values(A				):
													__A         =				tf.zeros(self.model_tester.batch_size							,dtype=tf.intaa				)
													__A         =				tf.zeros(self.model_tester.batch_size							,dtype=tf.intaa				)
									elif model_class in [
									    *get_values(A				),
									    *get_values(A				),
									]:
													__A         =				tf.zeros(self.model_tester.batch_size							,dtype=tf.intaa				)
									elif model_class in get_values(A				):
													__A         =				tf.zeros(self.model_tester.batch_size							,dtype=tf.intaa				)
									elif model_class in [
									    *get_values(A				),
									    *get_values(A				),
									    *get_values(A				),
									    *get_values(A				),
									    *get_values(A				),
									]:
													__A         =				tf.zeros(
													    (self.model_tester.batch_size, self.model_tester.seq_length)							,dtype=tf.intaa				)
					return inputs_dict






	@slow
	def        UpperCamelCase_						( self      :				List[str]				):
					__A    ,							__A         =				self.model_tester.prepare_config_and_inputs_for_common()
					for model_class in self.all_model_classes[:2]:
									__A         =				self._prepare_for_class(A							,A				)
									__A         =				model_class(A				)

									@tf.function
									def run_in_graph_mode():
													return model(A				)

									__A         =				run_in_graph_mode()
									self.assertIsNotNone(A				)






	@slow
	def        UpperCamelCase_						( self      :				Optional[int]				):
					__A    ,							__A         =				self.model_tester.prepare_config_and_inputs_for_common()
					for model_class in self.all_model_classes[:2]:
									__A         =				self._prepare_for_class(A							,A				)
									__A         =				model_class(A				)

									@tf.function(experimental_compile=A				)
									def run_in_graph_mode():
													return model(A				)

									__A         =				run_in_graph_mode()
									self.assertIsNotNone(A				)






	@slow
	def        UpperCamelCase_						( self      :				Union[str, Any]				):
					# This is a copy of the test_keras_fit method, but we use XLA compilation instead of eager
					__A    ,							__A         =				self.model_tester.prepare_config_and_inputs_for_common()
					for model_class in self.all_model_classes[:2]:
									__A         =				model_class(A				)
									if getattr(A							,"hf_compute_loss"							,A				):
													# Test that model correctly compute the loss with kwargs
													__A         =				self._prepare_for_class(inputs_dict.copy()							,A							,return_labels=A				)
													# Is there a better way to remove these decoder inputs?
													__A         =				{
													    key: val
													    for key, val in prepared_for_class.items()
													    if key not in ("head_mask", "decoder_head_mask", "cross_attn_head_mask", "decoder_input_ids")
													}

													__A         =				{
													    "labels",
													    "label",
													    "label_ids",
													    "start_positions",
													    "start_position",
													    "end_positions",
													    "end_position",
													    "next_sentence_label",
													}
													__A         =				possible_label_cols.intersection(set(A				)				)
													self.assertGreater(len(A				)							,0							,msg="No matching label names found!"				)
													__A         =				{key: val for key, val in prepared_for_class.items() if key in label_names}
													__A         =				{key: val for key, val in prepared_for_class.items() if key not in label_names}
													self.assertGreater(len(A				)							,0				)

													# Make sure it works with XLA!
													model.compile(optimizer=tf.keras.optimizers.SGD(0.0				)							,jit_compile=A				)
													# Make sure the model fits without crashing regardless of where we pass the labels
													__A         =				model.fit(
													    A							,validation_data=A							,steps_per_epoch=1							,validation_steps=1							,shuffle=A							,verbose=0							,)
													__A         =				history.history["loss"][0]
													self.assertTrue(not isnan(A				)				)
													__A         =				history.history["val_loss"][0]
													self.assertTrue(not isnan(A				)				)

													# Now test it with separate labels, to make sure that path works in XLA too.
													__A         =				model_class(A				)
													model.compile(optimizer=tf.keras.optimizers.SGD(0.0				)							,jit_compile=A				)
													__A         =				model.fit(
													    A							,A							,validation_data=(inputs_minus_labels, labels)							,steps_per_epoch=1							,validation_steps=1							,shuffle=A							,verbose=0							,)

													__A         =				history.history["loss"][0]
													self.assertTrue(not isnan(A				)				)
													__A         =				history.history["val_loss"][0]
													self.assertTrue(not isnan(A				)				)






	@slow
	def        UpperCamelCase_						( self      :				Any				):
					__A    ,							__A         =				self.model_tester.prepare_config_and_inputs_for_common()
					__A         =				True
					__A         =				True

					if hasattr(A							,"use_cache"				):
									__A         =				True

					__A         =				getattr(self.model_tester							,"encoder_seq_length"							,self.model_tester.seq_length				)
					__A         =				getattr(self.model_tester							,"key_length"							,A				)

					for model_class in self.all_model_classes[:2]:
									__A         =				self._prepare_for_class(A							,A				)
									__A         =				model_class(A				)
									model.build()
									__A         =				len(model(A				)				)

									for key in list(class_inputs_dict.keys()				):
													# Remove keys not in the serving signature, as the SavedModel will not be compiled to deal with them
													if key not in model.input_signature:
																	del class_inputs_dict[key]
													# Check it's a tensor, in case the inputs dict has some bools in it too
													elif isinstance(class_inputs_dict[key]							,tf.Tensor				) and class_inputs_dict[key].dtype.is_integer:
																	__A         =				tf.cast(class_inputs_dict[key]							,tf.intaa				)

									if set(class_inputs_dict.keys()				) != set(model.input_signature.keys()				):
													continue  # Some models have inputs that the preparation functions don't create, we skip those

									with tempfile.TemporaryDirectory() as tmpdirname:
													model.save_pretrained(A							,saved_model=A				)
													__A         =				os.path.join(A							,"saved_model"							,"1"				)
													__A         =				tf.keras.models.load_model(A				)
													__A         =				model(A				)

													if self.is_encoder_decoder:
																	__A         =				outputs["encoder_hidden_states"]
																	__A         =				outputs["encoder_attentions"]
													else:
																	__A         =				outputs["hidden_states"]
																	__A         =				outputs["attentions"]

													self.assertEqual(len(A				)							,A				)

													__A         =				getattr(
													    self.model_tester							,"expected_num_hidden_layers"							,self.model_tester.num_hidden_layers + 1				)

													self.assertEqual(len(A				)							,A				)
													self.assertListEqual(
													    list(output_hidden_states[0].shape[-2:]				)							,[self.model_tester.seq_length, self.model_tester.hidden_size]							,)

													self.assertEqual(len(A				)							,self.model_tester.num_hidden_layers				)
													self.assertListEqual(
													    list(output_attentions[0].shape[-3:]				)							,[self.model_tester.num_attention_heads, encoder_seq_length, encoder_key_length]							,)






	@slow
	def        UpperCamelCase_						( self      :				Any				):
					tf.keras.mixed_precision.set_global_policy("mixed_float16"				)

					# try/finally block to ensure subsequent tests run in float32
					try:
									__A    ,							__A         =				self.model_tester.prepare_config_and_inputs_for_common()
									for model_class in self.all_model_classes[:2]:
													__A         =				self._prepare_for_class(A							,A				)
													__A         =				model_class(A				)
													__A         =				model(A				)

													self.assertIsNotNone(A				)
					finally:
									tf.keras.mixed_precision.set_global_policy("float32"				)






	@slow
	def        UpperCamelCase_						( self      :				Dict				):
					__A    ,							__A         =				self.model_tester.prepare_config_and_inputs_for_common()
					# head_mask and decoder_head_mask has different shapes than other input args
					if "head_mask" in inputs_dict:
									del inputs_dict["head_mask"]
					if "decoder_head_mask" in inputs_dict:
									del inputs_dict["decoder_head_mask"]
					if "cross_attn_head_mask" in inputs_dict:
									del inputs_dict["cross_attn_head_mask"]
					__A         =				{
					    module_member
					    for model_class in self.all_model_classes
					    for module in (import_module(model_class.__module__				),)
					    for module_member_name in dir(A				)
					    if module_member_name.endswith("MainLayer"				)
					    for module_member in (getattr(A							,A				),)
					    if isinstance(A							,A				)
					    and tf.keras.layers.Layer in module_member.__bases__
					    and getattr(A							,"_keras_serializable"							,A				)
					}

					for main_layer_class in tf_main_layer_classes:
									# T5MainLayer needs an embed_tokens parameter when called without the inputs_embeds parameter
									if "T5" in main_layer_class.__name__:
													# Take the same values than in TFT5ModelTester for this shared layer
													__A         =				TFSharedEmbeddings(self.model_tester.vocab_size							,self.model_tester.hidden_size							,name="shared"				)
													__A         =				False
													__A         =				main_layer_class(A							,embed_tokens=A				)
									else:
													__A         =				main_layer_class(A				)

									__A         =				{
									    name: tf.keras.Input(tensor.shape[1:]							,dtype=tensor.dtype				) for name, tensor in inputs_dict.items()
									}

									if hasattr(self.model_tester							,"num_labels"				):
													__A         =				self.model_tester.num_labels
									else:
													__A         =				2

									__A         =				tf.data.Dataset.from_tensor_slices(
									    (inputs_dict, np.ones((self.model_tester.batch_size, self.model_tester.seq_length, num_labels, 1)				))				).batch(1				)

									__A         =				main_layer(A				)[0]
									__A         =				tf.keras.layers.Dense(A							,activation="softmax"							,name="outputs"				)(A				)
									__A         =				tf.keras.models.Model(inputs=A							,outputs=[outputs]				)

									model.compile(loss="binary_crossentropy"							,optimizer="adam"							,metrics=["binary_accuracy"]				)
									model.fit(A							,epochs=1				)

									with tempfile.TemporaryDirectory() as tmpdirname:
													__A         =				os.path.join(A							,"keras_model.h5"				)
													model.save(A				)
													if "T5" in main_layer_class.__name__:
																	__A         =				tf.keras.models.load_model(
																	    A							,custom_objects={
																	        main_layer_class.__name__: main_layer_class,
																	        "TFSharedEmbeddings": TFSharedEmbeddings,
																	    }							,)
													else:
																	__A         =				tf.keras.models.load_model(
																	    A							,custom_objects={main_layer_class.__name__: main_layer_class}				)
													assert isinstance(A							,tf.keras.Model				)
													model(A				)






	@slow
	def        UpperCamelCase_						( self      :				Any				):
					__A    ,							__A         =				self.model_tester.prepare_config_and_inputs_for_common()

					for model_class in self.all_model_classes[:2]:
									__A         =				model_class(A				)

									__A         =				copy.deepcopy(A				)

									if not self.is_encoder_decoder:
													__A         =				inputs["input_ids"]
													del inputs["input_ids"]
									else:
													__A         =				inputs["input_ids"]
													__A         =				inputs.get("decoder_input_ids"							,A				)
													del inputs["input_ids"]
													inputs.pop("decoder_input_ids"							,A				)

									if not self.is_encoder_decoder:
													__A         =				model.get_input_embeddings()(A				)
									else:
													__A         =				model.get_input_embeddings()(A				)
													__A         =				model.get_input_embeddings()(A				)

									__A         =				self._prepare_for_class(A							,A				)

									@tf.function
									def run_in_graph_mode():
													return model(A				)

									__A         =				run_in_graph_mode()
									self.assertIsNotNone(A				)






	def        UpperCamelCase_						( self      :				Tuple							,A      :				str							,A      :				Tuple				):
					# special tokens cannot be bad tokens
					__A         =				[]
					if model.config.bos_token_id is not None:
									special_tokens.append(model.config.bos_token_id				)
					if model.config.pad_token_id is not None:
									special_tokens.append(model.config.pad_token_id				)
					if model.config.eos_token_id is not None:
									special_tokens.append(model.config.eos_token_id				)

					# create random bad tokens that are not special tokens
					__A         =				[]
					while len(A				) < num_bad_tokens:
									__A         =				tf.squeeze(ids_tensor((1, 1)							,self.model_tester.vocab_size				)							,0				).numpy()[0]
									if token not in special_tokens:
													bad_tokens.append(A				)
					return bad_tokens






	def        UpperCamelCase_						( self      :				Dict							,A      :				str				):
					for token_id in output_ids[0].numpy().tolist():
									self.assertGreaterEqual(A							,0				)
									self.assertLess(A							,self.model_tester.vocab_size				)







	def        UpperCamelCase_						( self      :				List[str]							,A      :				Optional[int]							,A      :				Optional[Any]				):
					# for all bad word tokens
					for bad_word_ids in bad_words_ids:
									# for all slices in batch
									for generated_ids_slice in generated_ids:
													# for all word idx
													for i in range(len(A				)							,len(A				)				):
																	# if tokens match
																	if generated_ids_slice[i - len(A				) : i] == bad_word_ids:
																					return True
					return False


import torch

from transformers import PreTrainedModel

from .custom_configuration import CustomConfig, NoSuperInitConfig





class 							UpperCAmelCase   (						__SCREAMING_SNAKE_CASE						):




	'''simple docstring'''





	snake_case_      					=  CustomConfig






	def __init__( self      :				Optional[int]							,A      :				List[str]				):
					super().__init__(A				)
					__A         =				torch.nn.Linear(config.hidden_size							,config.hidden_size				)






	def        UpperCamelCase_						( self      :				Union[str, Any]							,A      :				Optional[int]				):
					return self.linear(A				)







	def        UpperCamelCase_						( self      :				Optional[int]							,A      :				int				):
					pass





class 							UpperCAmelCase   (						__SCREAMING_SNAKE_CASE						):




	'''simple docstring'''





	snake_case_      					=  NoSuperInitConfig






	def __init__( self      :				Optional[int]							,A      :				Union[str, Any]				):
					super().__init__(A				)
					__A         =				torch.nn.Linear(config.attribute							,config.attribute				)






	def        UpperCamelCase_						( self      :				str							,A      :				Any				):
					return self.linear(A				)







	def        UpperCamelCase_						( self      :				Tuple							,A      :				int				):
					pass
