






'''simple docstring'''





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 		SCREAMING_SNAKE_CASE__   :
		_UpperCAmelCase         =None
		_UpperCAmelCase         =()
		_UpperCAmelCase         =()
		_UpperCAmelCase         =True
		_UpperCAmelCase         =True
		_UpperCAmelCase         =True
		_UpperCAmelCase         =False



		def _lowerCAmelCase     (    self:    Dict				,   a:    int				,   a:    Optional[int]				,   a:    Any=False)  ->dict:






								'''simple docstring'''


								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 _lowerCAmelCase     (    self:    Optional[int])  ->List[Any]:






								'''simple docstring'''


								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 _lowerCAmelCase     (    self:    Any)  ->Dict:






								'''simple docstring'''


								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 _lowerCAmelCase     (    self:    Any)  ->List[Any]:






								'''simple docstring'''


								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 _lowerCAmelCase     (    self:    str)  ->List[str]:






								'''simple docstring'''


								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 _lowerCAmelCase     (    self:    List[str])  ->Optional[int]:






								'''simple docstring'''


								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 _lowerCAmelCase     (    self:    List[str])  ->Optional[int]:






								'''simple docstring'''


								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 _lowerCAmelCase     (    self:    Any)  ->str:






								'''simple docstring'''


								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 _lowerCAmelCase     (    self:    Optional[int]				,   a:    List[str]				,   a:    str)  ->int:






								'''simple docstring'''


								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 _lowerCAmelCase     (    self:    Tuple				,   a:    int)  ->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 _lowerCAmelCase     (    self:    List[str]				,   a:    Any				,   a:    Union[str, Any])  ->Tuple:






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











'''simple docstring'''





from functools import partial
from typing import Any, Callable, Dict, List, Type, TypeVar, Union, overload

import torch
import torch.nn as nn
import torch.types



def 			__UpperCAmelCase				(lowercase__	,lowercase__	,lowercase__				)		->				torch.Tensor:

						'''simple docstring'''


						if not inplace:
												a_				     =   ma + ma
						else:
												ma += ma

						return ma



def 			__UpperCAmelCase				(lowercase__	,lowercase__				)		->				torch.Tensor:

						'''simple docstring'''


						a_				     =   -1 * len(lowercase__				)
						a_				     =   list(range(len(tensor.shape[:zero_index]				)				)				)
						return tensor.permute(first_inds + [zero_index + i for i in inds]				)



def 			__UpperCAmelCase				(lowercase__	,lowercase__				)		->				torch.Tensor:

						'''simple docstring'''


						return t.reshape(t.shape[:-no_dims] + (-1,)				)



def 			__UpperCAmelCase				(lowercase__	,lowercase__	,lowercase__	,lowercase__ = 1e-4				)		->				torch.Tensor:

						'''simple docstring'''


						a_				     =   mask.expand(*value.shape				)
						return torch.sum(mask * value	,dim=lowercase__				) / (eps + torch.sum(lowercase__	,dim=lowercase__				))



def 			__UpperCAmelCase				(lowercase__	,lowercase__ = 2.3125	,lowercase__ = 21.6875	,lowercase__ = 64				)		->				torch.Tensor:

						'''simple docstring'''


						a_				     =   torch.linspace(lowercase__	,lowercase__	,no_bins - 1	,device=pts.device				)
						a_				     =   torch.sqrt(torch.sum((pts.unsqueeze(-2				) - pts.unsqueeze(-3				)) ** 2	,dim=-1				)				)
						return torch.bucketize(lowercase__	,lowercase__				)



def 			__UpperCAmelCase				(lowercase__	,lowercase__				)		->				dict:

						'''simple docstring'''


						a_				     =   dicts[0]
						a_				     =   {}
						for k, v in first.items():
												a_				     =   [d[k] for d in dicts]
												if isinstance(lowercase__	,lowercase__				):
																		a_				     =   dict_multimap(lowercase__	,lowercase__				)
												else:
																		a_				     =   fn(lowercase__				)

						return new_dict



def 			__UpperCAmelCase				(lowercase__	,lowercase__				)		->				torch.Tensor:

						'''simple docstring'''


						a_				     =   v_bins.view(((1,) * len(x.shape				)) + (len(lowercase__				),)				)
						a_				     =   x[..., None] - reshaped_bins
						a_				     =   torch.argmin(torch.abs(lowercase__				)	,dim=-1				)
						return nn.functional.one_hot(lowercase__	,num_classes=len(lowercase__				)				).float()



def 			__UpperCAmelCase				(lowercase__	,lowercase__	,lowercase__ = 0	,lowercase__ = 0				)		->				torch.Tensor:

						'''simple docstring'''


						a_				     =   []
						for i, s in enumerate(data.shape[:no_batch_dims]				):
												a_				     =   torch.arange(lowercase__				)
												a_				     =   r.view(*(*((1,) * i), -1, *((1,) * (len(inds.shape				) - i - 1)))				)
												ranges.append(lowercase__				)

						a_				     =   [slice(lowercase__				) for _ in range(len(data.shape				) - no_batch_dims				)]
						a_				     =   inds
						ranges.extend(lowercase__				)
						# Matt note: Editing this to get around the behaviour of using a list as an array index changing
						# in recent Numpy versions
						return data[tuple(lowercase__				)]


a_      = TypeVar('T')



def 			__UpperCAmelCase				(lowercase__	,lowercase__	,lowercase__				)		->				Dict[Any, Union[dict, list, tuple, Any]]:

						'''simple docstring'''


						a_				     =   {}
						for k, v in dic.items():
												if isinstance(lowercase__	,lowercase__				):
																		a_				     =   dict_map(lowercase__	,lowercase__	,lowercase__				)
												else:
																		a_				     =   tree_map(lowercase__	,lowercase__	,lowercase__				)

						return new_dict



@overload
def 			__UpperCAmelCase				(lowercase__	,lowercase__	,lowercase__				)		->				Any:

						'''simple docstring'''


						...



@overload
def 			__UpperCAmelCase				(lowercase__	,lowercase__	,lowercase__				)		->				dict:

						'''simple docstring'''


						...



@overload
def 			__UpperCAmelCase				(lowercase__	,lowercase__	,lowercase__				)		->				list:

						'''simple docstring'''


						...



@overload
def 			__UpperCAmelCase				(lowercase__	,lowercase__	,lowercase__				)		->				tuple:

						'''simple docstring'''


						...



def 			__UpperCAmelCase				(lowercase__	,lowercase__	,lowercase__				)		->				List[Any]:

						'''simple docstring'''


						if isinstance(lowercase__	,lowercase__				):
												return dict_map(lowercase__	,lowercase__	,lowercase__				)
						elif isinstance(lowercase__	,lowercase__				):
												return [tree_map(lowercase__	,lowercase__	,lowercase__				) for x in tree]
						elif isinstance(lowercase__	,lowercase__				):
												return tuple(tree_map(lowercase__	,lowercase__	,lowercase__				) for x in tree				)
						elif isinstance(lowercase__	,lowercase__				):
												return fn(lowercase__				)
						else:
												print(type(lowercase__				)				)
												raise ValueError("Not supported"				)
a_      = partial(tree_map, leaf_type=torch.Tensor)



