




"""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 			lowerCamelCase__    :
			a__ :      List[str]								=     None
			a__ :      Any								=     ()
			a__ :      Optional[int]								=     ()
			a__ :      List[Any]								=     True
			a__ :      List[str]								=     True
			a__ :      str								=     True
			a__ :      List[Any]								=     False


			def 							lowerCamelCase_				(     self				, SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE=False							):



					"""simple docstring"""







					snake_case		:						Tuple								=				copy.deepcopy(SCREAMING_SNAKE_CASE							)

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

					if return_labels:
							if model_class in get_values(SCREAMING_SNAKE_CASE							):
									snake_case		:						List[str]								=				tf.ones(self.model_tester.batch_size				, dtype=tf.intaa							)
							elif model_class in get_values(SCREAMING_SNAKE_CASE							):
									snake_case		:						Union[str, Any]								=				tf.zeros(self.model_tester.batch_size				, dtype=tf.intaa							)
									snake_case		:						List[Any]								=				tf.zeros(self.model_tester.batch_size				, dtype=tf.intaa							)
							elif model_class in [
							    *get_values(SCREAMING_SNAKE_CASE							),
							    *get_values(SCREAMING_SNAKE_CASE							),
							]:
									snake_case		:						Dict								=				tf.zeros(self.model_tester.batch_size				, dtype=tf.intaa							)
							elif model_class in get_values(SCREAMING_SNAKE_CASE							):
									snake_case		:						Any								=				tf.zeros(self.model_tester.batch_size				, dtype=tf.intaa							)
							elif model_class in [
							    *get_values(SCREAMING_SNAKE_CASE							),
							    *get_values(SCREAMING_SNAKE_CASE							),
							    *get_values(SCREAMING_SNAKE_CASE							),
							    *get_values(SCREAMING_SNAKE_CASE							),
							    *get_values(SCREAMING_SNAKE_CASE							),
							]:
									snake_case		:						Optional[Any]								=				tf.zeros(
									    (self.model_tester.batch_size, self.model_tester.seq_length)				, dtype=tf.intaa							)
					return inputs_dict


			@slow
			def 							lowerCamelCase_				(     self							):



					"""simple docstring"""







					snake_case			,				snake_case		:						Optional[int]								=				self.model_tester.prepare_config_and_inputs_for_common()
					for model_class in self.all_model_classes[:2]:
							snake_case		:						Union[str, Any]								=				self._prepare_for_class(SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE							)
							snake_case		:						Tuple								=				model_class(SCREAMING_SNAKE_CASE							)

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

							snake_case		:						str								=				run_in_graph_mode()
							self.assertIsNotNone(SCREAMING_SNAKE_CASE							)


			@slow
			def 							lowerCamelCase_				(     self							):



					"""simple docstring"""







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

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

							snake_case		:						Dict								=				run_in_graph_mode()
							self.assertIsNotNone(SCREAMING_SNAKE_CASE							)


			@slow
			def 							lowerCamelCase_				(     self							):



					"""simple docstring"""







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

									snake_case		:						Optional[Any]								=				{
									    "labels",
									    "label",
									    "label_ids",
									    "start_positions",
									    "start_position",
									    "end_positions",
									    "end_position",
									    "next_sentence_label",
									}
									snake_case		:						Tuple								=				possible_label_cols.intersection(set(SCREAMING_SNAKE_CASE							)							)
									self.assertGreater(len(SCREAMING_SNAKE_CASE							)				, 0				, msg="No matching label names found!"							)
									snake_case		:						List[str]								=				{key: val for key, val in prepared_for_class.items() if key in label_names}
									snake_case		:						Union[str, Any]								=				{key: val for key, val in prepared_for_class.items() if key not in label_names}
									self.assertGreater(len(SCREAMING_SNAKE_CASE							)				, 0							)

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

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

									snake_case		:						Union[str, Any]								=				history.history["loss"][0]
									self.assertTrue(not isnan(SCREAMING_SNAKE_CASE							)							)
									snake_case		:						Union[str, Any]								=				history.history["val_loss"][0]
									self.assertTrue(not isnan(SCREAMING_SNAKE_CASE							)							)


			@slow
			def 							lowerCamelCase_				(     self							):



					"""simple docstring"""







					snake_case			,				snake_case		:						Any								=				self.model_tester.prepare_config_and_inputs_for_common()
					snake_case		:						Any								=				True
					snake_case		:						Tuple								=				True

					if hasattr(SCREAMING_SNAKE_CASE				, "use_cache"							):
							snake_case		:						int								=				True

					snake_case		:						Optional[int]								=				getattr(self.model_tester				, "encoder_seq_length"				, self.model_tester.seq_length							)
					snake_case		:						Tuple								=				getattr(self.model_tester				, "key_length"				, SCREAMING_SNAKE_CASE							)

					for model_class in self.all_model_classes[:2]:
							snake_case		:						int								=				self._prepare_for_class(SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE							)
							snake_case		:						int								=				model_class(SCREAMING_SNAKE_CASE							)
							model.build()
							snake_case		:						List[str]								=				len(model(SCREAMING_SNAKE_CASE							)							)

							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:
											snake_case		:						List[Any]								=				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(SCREAMING_SNAKE_CASE				, saved_model=SCREAMING_SNAKE_CASE							)
									snake_case		:						Dict								=				os.path.join(SCREAMING_SNAKE_CASE				, "saved_model"				, "1"							)
									snake_case		:						Any								=				tf.keras.models.load_model(SCREAMING_SNAKE_CASE							)
									snake_case		:						int								=				model(SCREAMING_SNAKE_CASE							)

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

									self.assertEqual(len(SCREAMING_SNAKE_CASE							)				, SCREAMING_SNAKE_CASE							)

									snake_case		:						Union[str, Any]								=				getattr(
									    self.model_tester				, "expected_num_hidden_layers"				, self.model_tester.num_hidden_layers + 1							)

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

									self.assertEqual(len(SCREAMING_SNAKE_CASE							)				, 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							):



					"""simple docstring"""







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

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

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


			@slow
			def 							lowerCamelCase_				(     self							):



					"""simple docstring"""







					snake_case			,				snake_case		:						str								=				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"]
					snake_case		:						List[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(SCREAMING_SNAKE_CASE							)
					    if module_member_name.endswith("MainLayer"							)
					    for module_member in (getattr(SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE							),)
					    if isinstance(SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE							)
					    and tf.keras.layers.Layer in module_member.__bases__
					    and getattr(SCREAMING_SNAKE_CASE				, "_keras_serializable"				, SCREAMING_SNAKE_CASE							)
					}

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

							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"							):
									snake_case		:						Union[str, Any]								=				self.model_tester.num_labels
							else:
									snake_case		:						Optional[Any]								=				2

							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							)

							snake_case		:						int								=				main_layer(SCREAMING_SNAKE_CASE							)[0]
							snake_case		:						str								=				tf.keras.layers.Dense(SCREAMING_SNAKE_CASE				, activation="softmax"				, name="outputs"							)(SCREAMING_SNAKE_CASE							)
							snake_case		:						List[str]								=				tf.keras.models.Model(inputs=SCREAMING_SNAKE_CASE				, outputs=[outputs]							)

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

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


			@slow
			def 							lowerCamelCase_				(     self							):



					"""simple docstring"""







					snake_case			,				snake_case		:						Optional[Any]								=				self.model_tester.prepare_config_and_inputs_for_common()

					for model_class in self.all_model_classes[:2]:
							snake_case		:						List[Any]								=				model_class(SCREAMING_SNAKE_CASE							)

							snake_case		:						int								=				copy.deepcopy(SCREAMING_SNAKE_CASE							)

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

							if not self.is_encoder_decoder:
									snake_case		:						Tuple								=				model.get_input_embeddings()(SCREAMING_SNAKE_CASE							)
							else:
									snake_case		:						List[Any]								=				model.get_input_embeddings()(SCREAMING_SNAKE_CASE							)
									snake_case		:						Dict								=				model.get_input_embeddings()(SCREAMING_SNAKE_CASE							)

							snake_case		:						str								=				self._prepare_for_class(SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE							)

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

							snake_case		:						str								=				run_in_graph_mode()
							self.assertIsNotNone(SCREAMING_SNAKE_CASE							)


			def 							lowerCamelCase_				(     self				, SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE							):



					"""simple docstring"""







					snake_case		:						str								=				[]
					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
					snake_case		:						Dict								=				[]
					while len(SCREAMING_SNAKE_CASE							) < num_bad_tokens:
							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(SCREAMING_SNAKE_CASE							)
					return bad_tokens


			def 							lowerCamelCase_				(     self				, SCREAMING_SNAKE_CASE							):



					"""simple docstring"""







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



			def 							lowerCamelCase_				(     self				, SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE							):



					"""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(SCREAMING_SNAKE_CASE							)				, len(SCREAMING_SNAKE_CASE							)							):
											# if tokens match
											if generated_ids_slice[i - len(SCREAMING_SNAKE_CASE							) : i] == bad_word_ids:
													return True
					return False







"""simple docstring"""

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding
class 			lowerCamelCase__    (  lowerCamelCase_			):
			a__ :      Tuple								=     ["""image_processor""", """tokenizer"""]
			a__ :      List[Any]								=     """EfficientNetImageProcessor"""
			a__ :      str								=     ("""BertTokenizer""", """BertTokenizerFast""")


			def __init__(     self				, SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE							):



					"""simple docstring"""







					super().__init__(SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE							)


			def __call__(     self				, SCREAMING_SNAKE_CASE=None				, SCREAMING_SNAKE_CASE=None				, SCREAMING_SNAKE_CASE="max_length"				, SCREAMING_SNAKE_CASE=64				, SCREAMING_SNAKE_CASE=None				, **SCREAMING_SNAKE_CASE							):



					"""simple docstring"""







					if text is None and images is None:
							raise ValueError("You have to specify either text or images. Both cannot be none."							)

					if text is not None:
							snake_case		:						Dict								=				self.tokenizer(
							    SCREAMING_SNAKE_CASE				, padding=SCREAMING_SNAKE_CASE				, max_length=SCREAMING_SNAKE_CASE				, return_tensors=SCREAMING_SNAKE_CASE				, **SCREAMING_SNAKE_CASE							)

					if images is not None:
							snake_case		:						Tuple								=				self.image_processor(SCREAMING_SNAKE_CASE				, return_tensors=SCREAMING_SNAKE_CASE				, **SCREAMING_SNAKE_CASE							)

					if text is not None and images is not None:
							snake_case		:						Dict								=				image_features.pixel_values
							return encoding
					elif text is not None:
							return encoding
					else:
							return BatchEncoding(data=dict(**SCREAMING_SNAKE_CASE							)				, tensor_type=SCREAMING_SNAKE_CASE							)


			def 							lowerCamelCase_				(     self				, *SCREAMING_SNAKE_CASE				, **SCREAMING_SNAKE_CASE							):



					"""simple docstring"""







					return self.tokenizer.batch_decode(*SCREAMING_SNAKE_CASE				, **SCREAMING_SNAKE_CASE							)


			def 							lowerCamelCase_				(     self				, *SCREAMING_SNAKE_CASE				, **SCREAMING_SNAKE_CASE							):



					"""simple docstring"""







					return self.tokenizer.decode(*SCREAMING_SNAKE_CASE				, **SCREAMING_SNAKE_CASE							)



			@property
			def 							lowerCamelCase_				(     self							):



					"""simple docstring"""







					snake_case		:						Optional[Any]								=				self.tokenizer.model_input_names
					snake_case		:						Any								=				self.image_processor.model_input_names
					return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names							)							)

