



"""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    :






					'''simple docstring'''





					_A						:     str		=	None
					_A						:     Dict		=	()
					_A						:     Optional[Any]		=	()
					_A						:     List[Any]		=	True
					_A						:     Optional[Any]		=	True
					_A						:     int		=	True
					_A						:     Any		=	False






					def   lowerCAmelCase_							(			self:				List[str]		,	snake_case:				List[Any]		,	snake_case:				List[str]		,	snake_case:				Dict=False		)  ->      dict:
											snake_case_					:Tuple		=     copy.deepcopy(snake_case		)

											if model_class in get_values(snake_case		):
																	snake_case_					:List[str]		=     {
																	    k: tf.tile(tf.expand_dims(snake_case		,	1		)		,	(1, self.model_tester.num_choices) + (1,) * (v.ndim - 1)		)
																	    if isinstance(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(snake_case		):
																							snake_case_					:Dict		=     tf.ones(self.model_tester.batch_size		,	dtype=tf.intaa		)
																	elif model_class in get_values(snake_case		):
																							snake_case_					:List[str]		=     tf.zeros(self.model_tester.batch_size		,	dtype=tf.intaa		)
																							snake_case_					:Tuple		=     tf.zeros(self.model_tester.batch_size		,	dtype=tf.intaa		)
																	elif model_class in [
																	    *get_values(snake_case		),
																	    *get_values(snake_case		),
																	]:
																							snake_case_					:List[Any]		=     tf.zeros(self.model_tester.batch_size		,	dtype=tf.intaa		)
																	elif model_class in get_values(snake_case		):
																							snake_case_					:List[Any]		=     tf.zeros(self.model_tester.batch_size		,	dtype=tf.intaa		)
																	elif model_class in [
																	    *get_values(snake_case		),
																	    *get_values(snake_case		),
																	    *get_values(snake_case		),
																	    *get_values(snake_case		),
																	    *get_values(snake_case		),
																	]:
																							snake_case_					:Optional[int]		=     tf.zeros(
																							    (self.model_tester.batch_size, self.model_tester.seq_length)		,	dtype=tf.intaa		)
											return inputs_dict






					@slow
					def   lowerCAmelCase_							(			self:				str		)  ->      Any:
											snake_case_,			snake_case_					:Dict		=     self.model_tester.prepare_config_and_inputs_for_common()
											for model_class in self.all_model_classes[:2]:
																	snake_case_					:int		=     self._prepare_for_class(snake_case		,	snake_case		)
																	snake_case_					:List[Any]		=     model_class(snake_case		)

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

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






					@slow
					def   lowerCAmelCase_							(			self:				str		)  ->      List[str]:
											snake_case_,			snake_case_					:List[str]		=     self.model_tester.prepare_config_and_inputs_for_common()
											for model_class in self.all_model_classes[:2]:
																	snake_case_					:int		=     self._prepare_for_class(snake_case		,	snake_case		)
																	snake_case_					:Any		=     model_class(snake_case		)

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

																	snake_case_					:List[Any]		=     run_in_graph_mode()
																	self.assertIsNotNone(snake_case		)






					@slow
					def   lowerCAmelCase_							(			self:				int		)  ->      List[Any]:
											# This is a copy of the test_keras_fit method, but we use XLA compilation instead of eager
											snake_case_,			snake_case_					:str		=     self.model_tester.prepare_config_and_inputs_for_common()
											for model_class in self.all_model_classes[:2]:
																	snake_case_					:int		=     model_class(snake_case		)
																	if getattr(snake_case		,	"""hf_compute_loss"""		,	snake_case		):
																							# Test that model correctly compute the loss with kwargs
																							snake_case_					:Optional[Any]		=     self._prepare_for_class(inputs_dict.copy()		,	snake_case		,	return_labels=snake_case		)
																							# Is there a better way to remove these decoder inputs?
																							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""")
																							}

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

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

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

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






					@slow
					def   lowerCAmelCase_							(			self:				Dict		)  ->      int:
											snake_case_,			snake_case_					:Union[str, Any]		=     self.model_tester.prepare_config_and_inputs_for_common()
											snake_case_					:Dict		=     True
											snake_case_					:Tuple		=     True

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

											snake_case_					:List[str]		=     getattr(self.model_tester		,	"""encoder_seq_length"""		,	self.model_tester.seq_length		)
											snake_case_					:Union[str, Any]		=     getattr(self.model_tester		,	"""key_length"""		,	snake_case		)

											for model_class in self.all_model_classes[:2]:
																	snake_case_					:List[Any]		=     self._prepare_for_class(snake_case		,	snake_case		)
																	snake_case_					:Union[str, Any]		=     model_class(snake_case		)
																	model.build()
																	snake_case_					:Dict		=     len(model(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_					:str		=     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(snake_case		,	saved_model=snake_case		)
																							snake_case_					:List[Any]		=     os.path.join(snake_case		,	"""saved_model"""		,	"""1"""		)
																							snake_case_					:Any		=     tf.keras.models.load_model(snake_case		)
																							snake_case_					:Union[str, Any]		=     model(snake_case		)

																							if self.is_encoder_decoder:
																													snake_case_					:Any		=     outputs["""encoder_hidden_states"""]
																													snake_case_					:List[Any]		=     outputs["""encoder_attentions"""]
																							else:
																													snake_case_					:List[str]		=     outputs["""hidden_states"""]
																													snake_case_					:str		=     outputs["""attentions"""]

																							self.assertEqual(len(snake_case		)		,	snake_case		)

																							snake_case_					:Tuple		=     getattr(
																							    self.model_tester		,	"""expected_num_hidden_layers"""		,	self.model_tester.num_hidden_layers + 1		)

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

																							self.assertEqual(len(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:				Optional[Any]		)  ->      List[Any]:
											tf.keras.mixed_precision.set_global_policy("""mixed_float16"""		)

											# try/finally block to ensure subsequent tests run in float32
											try:
																	snake_case_,			snake_case_					: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(snake_case		,	snake_case		)
																							snake_case_					:List[Any]		=     model_class(snake_case		)
																							snake_case_					:Tuple		=     model(snake_case		)

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






					@slow
					def   lowerCAmelCase_							(			self:				int		)  ->      List[str]:
											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_					:Tuple		=     {
											    module_member
											    for model_class in self.all_model_classes
											    for module in (import_module(model_class.__module__		),)
											    for module_member_name in dir(snake_case		)
											    if module_member_name.endswith("""MainLayer"""		)
											    for module_member in (getattr(snake_case		,	snake_case		),)
											    if isinstance(snake_case		,	snake_case		)
											    and tf.keras.layers.Layer in module_member.__bases__
											    and getattr(snake_case		,	"""_keras_serializable"""		,	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_					:Dict		=     TFSharedEmbeddings(self.model_tester.vocab_size		,	self.model_tester.hidden_size		,	name="""shared"""		)
																							snake_case_					:int		=     False
																							snake_case_					:Dict		=     main_layer_class(snake_case		,	embed_tokens=snake_case		)
																	else:
																							snake_case_					:str		=     main_layer_class(snake_case		)

																	snake_case_					:Union[str, 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_					:Tuple		=     self.model_tester.num_labels
																	else:
																							snake_case_					:Any		=     2

																	snake_case_					:Dict		=     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_					:str		=     main_layer(snake_case		)[0]
																	snake_case_					:str		=     tf.keras.layers.Dense(snake_case		,	activation="""softmax"""		,	name="""outputs"""		)(snake_case		)
																	snake_case_					:int		=     tf.keras.models.Model(inputs=snake_case		,	outputs=[outputs]		)

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

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






					@slow
					def   lowerCAmelCase_							(			self:				List[str]		)  ->      str:
											snake_case_,			snake_case_					:int		=     self.model_tester.prepare_config_and_inputs_for_common()

											for model_class in self.all_model_classes[:2]:
																	snake_case_					:str		=     model_class(snake_case		)

																	snake_case_					:Optional[Any]		=     copy.deepcopy(snake_case		)

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

																	if not self.is_encoder_decoder:
																							snake_case_					:Union[str, Any]		=     model.get_input_embeddings()(snake_case		)
																	else:
																							snake_case_					:Dict		=     model.get_input_embeddings()(snake_case		)
																							snake_case_					:Union[str, Any]		=     model.get_input_embeddings()(snake_case		)

																	snake_case_					:Tuple		=     self._prepare_for_class(snake_case		,	snake_case		)

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

																	snake_case_					:Union[str, Any]		=     run_in_graph_mode()
																	self.assertIsNotNone(snake_case		)






					def   lowerCAmelCase_							(			self:				Any		,	snake_case:				List[str]		,	snake_case:				Optional[Any]		)  ->      str:
											# special tokens cannot be bad tokens
											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_					:Optional[int]		=     []
											while len(snake_case		) < num_bad_tokens:
																	snake_case_					:str		=     tf.squeeze(ids_tensor((1, 1)		,	self.model_tester.vocab_size		)		,	0		).numpy()[0]
																	if token not in special_tokens:
																							bad_tokens.append(snake_case		)
											return bad_tokens






					def   lowerCAmelCase_							(			self:				str		,	snake_case:				Tuple		)  ->      Union[str, Any]:
											for token_id in output_ids[0].numpy().tolist():
																	self.assertGreaterEqual(snake_case		,	0		)
																	self.assertLess(snake_case		,	self.model_tester.vocab_size		)






					def   lowerCAmelCase_							(			self:				str		,	snake_case:				Union[str, Any]		,	snake_case:				str		)  ->      Tuple:
											# 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(snake_case		)		,	len(snake_case		)		):
																													# if tokens match
																													if generated_ids_slice[i - len(snake_case		) : i] == bad_word_ids:
																																			return True
											return False




import enum
import inspect
import warnings
from functools import wraps
from typing import Callable, Optional

from .logging import get_logger


_A							     =       set()
_A							     =       get_logger(__name__)

def 			__SCREAMING_SNAKE_CASE		(		UpperCamelCase		:      Optional[str] = None	)     ->	Optional[int]:



	"""simple docstring"""






	def decorator(UpperCamelCase		:      Callable	):
		global _emitted_deprecation_warnings

		if inspect.isclass(UpperCamelCase	):
			a_     			=   deprecated_class_or_function.__init__
			a_     			=   deprecated_class_or_function.__name__
		else:
			a_     			=   deprecated_class_or_function
			a_     			=   deprecated_function.__name__
			# Support deprecating __init__ class method: class name instead
			a_     			=   name if name != """__init__""" else deprecated_function.__qualname__.split("""."""	)[-2]

		a_     			=   (
		    F"""{name} is deprecated and will be removed in the next major version of datasets.""" + F""" {help_message}"""
		    if help_message
		    else """"""
		)

		@wraps(UpperCamelCase	)
		def wrapper(*UpperCamelCase		:      List[str]     ,					**UpperCamelCase		:      Union[str, Any]	):
			a_     			=   hash(UpperCamelCase	)
			if func_hash not in _emitted_deprecation_warnings:
				warnings.warn(UpperCamelCase     ,					category=UpperCamelCase     ,					stacklevel=2	)
				_emitted_deprecation_warnings.add(UpperCamelCase	)
			return deprecated_function(*UpperCamelCase     ,					**UpperCamelCase	)

		a_     			=   """deprecated"""

		if inspect.isclass(UpperCamelCase	):
			a_     			=   wrapper
			return deprecated_class_or_function
		else:
			return wrapper

	return decorator

class 			lowerCamelCase_					(    enum.EnumMeta    ):





							def __getattribute__(							cls			,       _SCREAMING_SNAKE_CASE    ):
								a_     			=   super().__getattribute__(_SCREAMING_SNAKE_CASE    )
								if isinstance(_SCREAMING_SNAKE_CASE			,       enum.Enum    ) and obj._on_access:
									obj._on_access()
								return obj





							def __getitem__(							cls			,       _SCREAMING_SNAKE_CASE    ):
								a_     			=   super().__getitem__(_SCREAMING_SNAKE_CASE    )
								if member._on_access:
									member._on_access()
								return member





							def __call__(							cls			,       _SCREAMING_SNAKE_CASE			,       _SCREAMING_SNAKE_CASE=None			,       *, _SCREAMING_SNAKE_CASE=None			,       _SCREAMING_SNAKE_CASE=None			,       _SCREAMING_SNAKE_CASE=None			,       _SCREAMING_SNAKE_CASE=1    ):
								a_     			=   super().__call__(_SCREAMING_SNAKE_CASE			,       _SCREAMING_SNAKE_CASE			,       module=_SCREAMING_SNAKE_CASE			,       qualname=_SCREAMING_SNAKE_CASE			,       type=_SCREAMING_SNAKE_CASE			,       start=_SCREAMING_SNAKE_CASE    )
								if isinstance(_SCREAMING_SNAKE_CASE			,       enum.Enum    ) and obj._on_access:
									obj._on_access()
								return obj







class 			lowerCamelCase_					(    enum.Enum   ,		metaclass=_SCREAMING_SNAKE_CASE    ):





							def __new__(							cls			,       _SCREAMING_SNAKE_CASE    ):
								a_     			=   object.__new__(cls    )
								a_     			=   value
								a_     			=   member.deprecate
								return member





							@property
							def   __magic_name__      (							self    ):
								return ""





							def   __magic_name__      (							self    ):
								a_     			=   f""" {self.help_message}""" if self.help_message else """"""
								warnings.warn(
								    f"""'{self.__objclass__.__name__}' is deprecated and will be removed in the next major version of datasets."""
								    + help_message			,       _SCREAMING_SNAKE_CASE			,       stacklevel=3			,       )