

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:
						a_         =      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)]
												)
												a_         =      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'''







			lowerCamelCase__          =None
			lowerCamelCase__          =()
			lowerCamelCase__          =()
			lowerCamelCase__          =True
			lowerCamelCase__          =True
			lowerCamelCase__          =True
			lowerCamelCase__          =False
			def 							__UpperCamelCase	(     self							:	Dict , a							:	int , a							:	int , a							:	str=False		)							->    dict:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE				: Optional[Any]			       =    copy.deepcopy(a		)

				if model_class in get_values(a		):
					SCREAMING_SNAKE_CASE				: Tuple			       =    {
					    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		):
						SCREAMING_SNAKE_CASE				: Optional[int]			       =    tf.ones(self.model_tester.batch_size , dtype=tf.intaa		)
					elif model_class in get_values(a		):
						SCREAMING_SNAKE_CASE				: Tuple			       =    tf.zeros(self.model_tester.batch_size , dtype=tf.intaa		)
						SCREAMING_SNAKE_CASE				: str			       =    tf.zeros(self.model_tester.batch_size , dtype=tf.intaa		)
					elif model_class in [
					    *get_values(a		),
					    *get_values(a		),
					]:
						SCREAMING_SNAKE_CASE				: List[str]			       =    tf.zeros(self.model_tester.batch_size , dtype=tf.intaa		)
					elif model_class in get_values(a		):
						SCREAMING_SNAKE_CASE				: List[str]			       =    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		),
					]:
						SCREAMING_SNAKE_CASE				: List[str]			       =    tf.zeros(
						    (self.model_tester.batch_size, self.model_tester.seq_length) , dtype=tf.intaa		)
				return inputs_dict
			@slow
			def 							__UpperCamelCase	(     self							:	int		)							->    str:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE       ,SCREAMING_SNAKE_CASE				: List[str]			       =    self.model_tester.prepare_config_and_inputs_for_common()
				for model_class in self.all_model_classes[:2]:
					SCREAMING_SNAKE_CASE				: str			       =    self._prepare_for_class(a , a		)
					SCREAMING_SNAKE_CASE				: Optional[Any]			       =    model_class(a		)

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

					SCREAMING_SNAKE_CASE				: Dict			       =    run_in_graph_mode()
					self.assertIsNotNone(a		)
			@slow
			def 							__UpperCamelCase	(     self							:	Tuple		)							->    Any:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE       ,SCREAMING_SNAKE_CASE				: Any			       =    self.model_tester.prepare_config_and_inputs_for_common()
				for model_class in self.all_model_classes[:2]:
					SCREAMING_SNAKE_CASE				: Dict			       =    self._prepare_for_class(a , a		)
					SCREAMING_SNAKE_CASE				: Union[str, Any]			       =    model_class(a		)

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

					SCREAMING_SNAKE_CASE				: Optional[int]			       =    run_in_graph_mode()
					self.assertIsNotNone(a		)
			@slow
			def 							__UpperCamelCase	(     self							:	Dict		)							->    Optional[int]:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE       ,SCREAMING_SNAKE_CASE				: Optional[int]			       =    self.model_tester.prepare_config_and_inputs_for_common()
				for model_class in self.all_model_classes[:2]:
					SCREAMING_SNAKE_CASE				: str			       =    model_class(a		)
					if getattr(a , "hf_compute_loss" , a		):
						# Test that model correctly compute the loss with kwargs
						SCREAMING_SNAKE_CASE				: Dict			       =    self._prepare_for_class(inputs_dict.copy() , a , return_labels=a		)
						# Is there a better way to remove these decoder inputs?
						SCREAMING_SNAKE_CASE				: Optional[int]			       =    {
						    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")
						}

						SCREAMING_SNAKE_CASE				: Any			       =    {
						    "labels",
						    "label",
						    "label_ids",
						    "start_positions",
						    "start_position",
						    "end_positions",
						    "end_position",
						    "next_sentence_label",
						}
						SCREAMING_SNAKE_CASE				: Optional[int]			       =    possible_label_cols.intersection(set(a		)		)
						self.assertGreater(len(a		) , 0 , msg="No matching label names found!"		)
						SCREAMING_SNAKE_CASE				: Optional[Any]			       =    {key: val for key, val in prepared_for_class.items() if key in label_names}
						SCREAMING_SNAKE_CASE				: Optional[int]			       =    {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
						SCREAMING_SNAKE_CASE				: Any			       =    model.fit(
						    a , validation_data=a , steps_per_epoch=1 , validation_steps=1 , shuffle=a , verbose=0 , )
						SCREAMING_SNAKE_CASE				: List[str]			       =    history.history["loss"][0]
						self.assertTrue(not isnan(a		)		)
						SCREAMING_SNAKE_CASE				: int			       =    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.
						SCREAMING_SNAKE_CASE				: Any			       =    model_class(a		)
						model.compile(optimizer=tf.keras.optimizers.SGD(0.0		) , jit_compile=a		)
						SCREAMING_SNAKE_CASE				: List[str]			       =    model.fit(
						    a , a , validation_data=(inputs_minus_labels, labels) , steps_per_epoch=1 , validation_steps=1 , shuffle=a , verbose=0 , )

						SCREAMING_SNAKE_CASE				: Optional[Any]			       =    history.history["loss"][0]
						self.assertTrue(not isnan(a		)		)
						SCREAMING_SNAKE_CASE				: List[str]			       =    history.history["val_loss"][0]
						self.assertTrue(not isnan(a		)		)
			@slow
			def 							__UpperCamelCase	(     self							:	List[str]		)							->    int:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE       ,SCREAMING_SNAKE_CASE				: str			       =    self.model_tester.prepare_config_and_inputs_for_common()
				SCREAMING_SNAKE_CASE				: Any			       =    True
				SCREAMING_SNAKE_CASE				: str			       =    True

				if hasattr(a , "use_cache"		):
					SCREAMING_SNAKE_CASE				: Dict			       =    True

				SCREAMING_SNAKE_CASE				: int			       =    getattr(self.model_tester , "encoder_seq_length" , self.model_tester.seq_length		)
				SCREAMING_SNAKE_CASE				: Union[str, Any]			       =    getattr(self.model_tester , "key_length" , a		)

				for model_class in self.all_model_classes[:2]:
					SCREAMING_SNAKE_CASE				: Tuple			       =    self._prepare_for_class(a , a		)
					SCREAMING_SNAKE_CASE				: int			       =    model_class(a		)
					model.build()
					SCREAMING_SNAKE_CASE				: Optional[Any]			       =    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:
							SCREAMING_SNAKE_CASE				: Optional[int]			       =    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		)
						SCREAMING_SNAKE_CASE				: Tuple			       =    os.path.join(a , "saved_model" , "1"		)
						SCREAMING_SNAKE_CASE				: Union[str, Any]			       =    tf.keras.models.load_model(a		)
						SCREAMING_SNAKE_CASE				: List[str]			       =    model(a		)

						if self.is_encoder_decoder:
							SCREAMING_SNAKE_CASE				: Dict			       =    outputs["encoder_hidden_states"]
							SCREAMING_SNAKE_CASE				: str			       =    outputs["encoder_attentions"]
						else:
							SCREAMING_SNAKE_CASE				: Union[str, Any]			       =    outputs["hidden_states"]
							SCREAMING_SNAKE_CASE				: int			       =    outputs["attentions"]

						self.assertEqual(len(a		) , a		)

						SCREAMING_SNAKE_CASE				: Dict			       =    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							:	Tuple		)							->    Union[str, Any]:

				"""simple docstring"""

				tf.keras.mixed_precision.set_global_policy("mixed_float16"		)

				# try/finally block to ensure subsequent tests run in float32
				try:
					SCREAMING_SNAKE_CASE       ,SCREAMING_SNAKE_CASE				: Dict			       =    self.model_tester.prepare_config_and_inputs_for_common()
					for model_class in self.all_model_classes[:2]:
						SCREAMING_SNAKE_CASE				: List[str]			       =    self._prepare_for_class(a , a		)
						SCREAMING_SNAKE_CASE				: int			       =    model_class(a		)
						SCREAMING_SNAKE_CASE				: Tuple			       =    model(a		)

						self.assertIsNotNone(a		)
				finally:
					tf.keras.mixed_precision.set_global_policy("float32"		)
			@slow
			def 							__UpperCamelCase	(     self							:	Any		)							->    int:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE       ,SCREAMING_SNAKE_CASE				: Dict			       =    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"]
				SCREAMING_SNAKE_CASE				: Any			       =    {
				    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
						SCREAMING_SNAKE_CASE				: int			       =    TFSharedEmbeddings(self.model_tester.vocab_size , self.model_tester.hidden_size , name="shared"		)
						SCREAMING_SNAKE_CASE				: Optional[Any]			       =    False
						SCREAMING_SNAKE_CASE				: Union[str, Any]			       =    main_layer_class(a , embed_tokens=a		)
					else:
						SCREAMING_SNAKE_CASE				: List[Any]			       =    main_layer_class(a		)

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

					if hasattr(self.model_tester , "num_labels"		):
						SCREAMING_SNAKE_CASE				: Tuple			       =    self.model_tester.num_labels
					else:
						SCREAMING_SNAKE_CASE				: Optional[Any]			       =    2

					SCREAMING_SNAKE_CASE				: Optional[int]			       =    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		)

					SCREAMING_SNAKE_CASE				: Optional[Any]			       =    main_layer(a		)[0]
					SCREAMING_SNAKE_CASE				: Optional[int]			       =    tf.keras.layers.Dense(a , activation="softmax" , name="outputs"		)(a		)
					SCREAMING_SNAKE_CASE				: int			       =    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:
						SCREAMING_SNAKE_CASE				: List[Any]			       =    os.path.join(a , "keras_model.h5"		)
						model.save(a		)
						if "T5" in main_layer_class.__name__:
							SCREAMING_SNAKE_CASE				: Any			       =    tf.keras.models.load_model(
							    a , custom_objects={
							        main_layer_class.__name__: main_layer_class,
							        "TFSharedEmbeddings": TFSharedEmbeddings,
							    } , )
						else:
							SCREAMING_SNAKE_CASE				: List[Any]			       =    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							:	Union[str, Any]		)							->    List[str]:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE       ,SCREAMING_SNAKE_CASE				: List[str]			       =    self.model_tester.prepare_config_and_inputs_for_common()

				for model_class in self.all_model_classes[:2]:
					SCREAMING_SNAKE_CASE				: Optional[int]			       =    model_class(a		)

					SCREAMING_SNAKE_CASE				: List[Any]			       =    copy.deepcopy(a		)

					if not self.is_encoder_decoder:
						SCREAMING_SNAKE_CASE				: Any			       =    inputs["input_ids"]
						del inputs["input_ids"]
					else:
						SCREAMING_SNAKE_CASE				: Union[str, Any]			       =    inputs["input_ids"]
						SCREAMING_SNAKE_CASE				: Dict			       =    inputs.get("decoder_input_ids" , a		)
						del inputs["input_ids"]
						inputs.pop("decoder_input_ids" , a		)

					if not self.is_encoder_decoder:
						SCREAMING_SNAKE_CASE				: Optional[int]			       =    model.get_input_embeddings()(a		)
					else:
						SCREAMING_SNAKE_CASE				: List[str]			       =    model.get_input_embeddings()(a		)
						SCREAMING_SNAKE_CASE				: Optional[int]			       =    model.get_input_embeddings()(a		)

					SCREAMING_SNAKE_CASE				: List[Any]			       =    self._prepare_for_class(a , a		)

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

					SCREAMING_SNAKE_CASE				: str			       =    run_in_graph_mode()
					self.assertIsNotNone(a		)
			def 							__UpperCamelCase	(     self							:	Tuple , a							:	List[str] , a							:	Union[str, Any]		)							->    Optional[Any]:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE				: int			       =    []
				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
				SCREAMING_SNAKE_CASE				: Any			       =    []
				while len(a		) < num_bad_tokens:
					SCREAMING_SNAKE_CASE				: List[Any]			       =    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							:	Any , a							:	Union[str, Any]		)							->    Optional[int]:

				"""simple docstring"""

				for token_id in output_ids[0].numpy().tolist():
					self.assertGreaterEqual(a , 0		)
					self.assertLess(a , self.model_tester.vocab_size		)






			def 							__UpperCamelCase	(     self							:	Any , a							:	int , a							:	List[Any]		)							->    List[str]:

				"""simple docstring"""

				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


from __future__ import annotations

from typing import Any






class _UpperCamelCase							:
			'''simple docstring'''







			def __init__(     self							:	Dict , a							:	Any , a							:	Any		)							->    None:  # noqa: A002

				"""simple docstring"""

				SCREAMING_SNAKE_CASE				: str			       =    item
				SCREAMING_SNAKE_CASE				: List[Any]			       =    next






class _UpperCamelCase							:
			'''simple docstring'''







			def __init__(     self							:	Optional[int]		)							->    None:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE				: Node | None			       =    None
				SCREAMING_SNAKE_CASE				: List[Any]			       =    0
			def 							__UpperCamelCase	(     self							:	Union[str, Any] , a							:	Any		)							->    None:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE				: Any			       =    Node(a , self.head		)
				self.size += 1
			def 							__UpperCamelCase	(     self							:	Dict		)							->    Any:

				"""simple docstring"""

				if self.head is None:
					return None
				else:
					SCREAMING_SNAKE_CASE				: Optional[int]			       =    self.head.item
					SCREAMING_SNAKE_CASE				: Dict			       =    self.head.next
					self.size -= 1
					return item
			def 							__UpperCamelCase	(     self							:	Any		)							->    bool:

				"""simple docstring"""

				return self.head is None
			def __str__(     self							:	Union[str, Any]		)							->    str:

				"""simple docstring"""

				if self.is_empty():
					return ""
				else:
					SCREAMING_SNAKE_CASE				: Union[str, Any]			       =    self.head
					SCREAMING_SNAKE_CASE				: str			       =    ""
					SCREAMING_SNAKE_CASE				: list[str]			       =    []
					while iterate:
						item_list.append(str(iterate.item		)		)
						SCREAMING_SNAKE_CASE				: str			       =    iterate.next

					SCREAMING_SNAKE_CASE				: List[str]			       =    " --> ".join(a		)

					return item_str






			def __len__(     self							:	List[Any]		)							->    int:

				"""simple docstring"""

				return self.size