






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        __snake_case:
 lowercase :       Dict		      =		None
 lowercase :       Dict		      =		()
 lowercase :       str		      =		()
 lowercase :       List[str]		      =		True
 lowercase :       List[str]		      =		True
 lowercase :       Optional[Any]		      =		True
 lowercase :       Optional[Any]		      =		False



 def 						UpperCAmelCase    (	self   :   Any ,		UpperCAmelCase__   :   str ,		UpperCAmelCase__   :   List[str] ,		UpperCAmelCase__   :   List[str]=False	):

    """simple docstring"""






    lowercase__						:	List[str]         = copy.deepcopy(UpperCAmelCase__	)

    if model_class in get_values(UpperCAmelCase__	):
       lowercase__						:	Union[str, Any]         = {
           k: tf.tile(tf.expand_dims(UpperCAmelCase__ ,		1	) ,		(1, self.model_tester.num_choices) + (1,) * (v.ndim - 1)	)
           if isinstance(UpperCAmelCase__ ,		tf.Tensor	) and v.ndim > 0
           else v
           for k, v in inputs_dict.items()
       }

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



 @slow
 def 						UpperCAmelCase    (	self   :   Tuple	):

    """simple docstring"""






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

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

       lowercase__						:	Optional[int]         = run_in_graph_mode()
       self.assertIsNotNone(UpperCAmelCase__	)



 @slow
 def 						UpperCAmelCase    (	self   :   Any	):

    """simple docstring"""






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

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

       lowercase__						:	List[Any]         = run_in_graph_mode()
       self.assertIsNotNone(UpperCAmelCase__	)



 @slow
 def 						UpperCAmelCase    (	self   :   List[str]	):

    """simple docstring"""






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

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

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

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

          lowercase__						:	List[str]         = history.history["""loss"""][0]
          self.assertTrue(not isnan(UpperCAmelCase__	)	)
          lowercase__						:	List[str]         = history.history["""val_loss"""][0]
          self.assertTrue(not isnan(UpperCAmelCase__	)	)



 @slow
 def 						UpperCAmelCase    (	self   :   Optional[int]	):

    """simple docstring"""






    lowercase__     ,					lowercase__						:	Tuple         = self.model_tester.prepare_config_and_inputs_for_common()
    lowercase__						:	List[str]         = True
    lowercase__						:	List[str]         = True

    if hasattr(UpperCAmelCase__ ,		"""use_cache"""	):
       lowercase__						:	Tuple         = True

    lowercase__						:	Union[str, Any]         = getattr(self.model_tester ,		"""encoder_seq_length""" ,		self.model_tester.seq_length	)
    lowercase__						:	str         = getattr(self.model_tester ,		"""key_length""" ,		UpperCAmelCase__	)

    for model_class in self.all_model_classes[:2]:
       lowercase__						:	List[str]         = self._prepare_for_class(UpperCAmelCase__ ,		UpperCAmelCase__	)
       lowercase__						:	Dict         = model_class(UpperCAmelCase__	)
       model.build()
       lowercase__						:	Dict         = len(model(UpperCAmelCase__	)	)

       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:
             lowercase__						:	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(UpperCAmelCase__ ,		saved_model=UpperCAmelCase__	)
          lowercase__						:	Union[str, Any]         = os.path.join(UpperCAmelCase__ ,		"""saved_model""" ,		"""1"""	)
          lowercase__						:	Dict         = tf.keras.models.load_model(UpperCAmelCase__	)
          lowercase__						:	Optional[Any]         = model(UpperCAmelCase__	)

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

          self.assertEqual(len(UpperCAmelCase__	) ,		UpperCAmelCase__	)

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

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

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



 @slow
 def 						UpperCAmelCase    (	self   :   Any	):

    """simple docstring"""






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

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

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



 @slow
 def 						UpperCAmelCase    (	self   :   int	):

    """simple docstring"""






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

    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
          lowercase__						:	List[str]         = TFSharedEmbeddings(self.model_tester.vocab_size ,		self.model_tester.hidden_size ,		name="""shared"""	)
          lowercase__						:	Dict         = False
          lowercase__						:	List[str]         = main_layer_class(UpperCAmelCase__ ,		embed_tokens=UpperCAmelCase__	)
       else:
          lowercase__						:	Dict         = main_layer_class(UpperCAmelCase__	)

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

       if hasattr(self.model_tester ,		"""num_labels"""	):
          lowercase__						:	Union[str, Any]         = self.model_tester.num_labels
       else:
          lowercase__						:	Optional[int]         = 2

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

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

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

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



 @slow
 def 						UpperCAmelCase    (	self   :   List[Any]	):

    """simple docstring"""






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

    for model_class in self.all_model_classes[:2]:
       lowercase__						:	Dict         = model_class(UpperCAmelCase__	)

       lowercase__						:	Optional[Any]         = copy.deepcopy(UpperCAmelCase__	)

       if not self.is_encoder_decoder:
          lowercase__						:	Tuple         = inputs["""input_ids"""]
          del inputs["input_ids"]
       else:
          lowercase__						:	List[Any]         = inputs["""input_ids"""]
          lowercase__						:	Dict         = inputs.get("""decoder_input_ids""" ,		UpperCAmelCase__	)
          del inputs["input_ids"]
          inputs.pop("""decoder_input_ids""" ,		UpperCAmelCase__	)

       if not self.is_encoder_decoder:
          lowercase__						:	Optional[int]         = model.get_input_embeddings()(UpperCAmelCase__	)
       else:
          lowercase__						:	List[str]         = model.get_input_embeddings()(UpperCAmelCase__	)
          lowercase__						:	Union[str, Any]         = model.get_input_embeddings()(UpperCAmelCase__	)

       lowercase__						:	List[Any]         = self._prepare_for_class(UpperCAmelCase__ ,		UpperCAmelCase__	)

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

       lowercase__						:	Dict         = run_in_graph_mode()
       self.assertIsNotNone(UpperCAmelCase__	)



 def 						UpperCAmelCase    (	self   :   Dict ,		UpperCAmelCase__   :   Union[str, Any] ,		UpperCAmelCase__   :   List[Any]	):

    """simple docstring"""






    lowercase__						:	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
    lowercase__						:	Dict         = []
    while len(UpperCAmelCase__	) < num_bad_tokens:
       lowercase__						:	Union[str, Any]         = tf.squeeze(ids_tensor((1, 1) ,		self.model_tester.vocab_size	) ,		0	).numpy()[0]
       if token not in special_tokens:
          bad_tokens.append(UpperCAmelCase__	)
    return bad_tokens



 def 						UpperCAmelCase    (	self   :   Optional[Any] ,		UpperCAmelCase__   :   List[str]	):

    """simple docstring"""






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







 def 						UpperCAmelCase    (	self   :   str ,		UpperCAmelCase__   :   Dict ,		UpperCAmelCase__   :   Any	):

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










from dataclasses import dataclass
from typing import Callable, Optional

import datasets
@dataclass
class        __snake_case(					datasets.BuilderConfig      ):
 lowercase :       Optional[Callable]		      =		None
 lowercase :       Optional[dict]		      =		None
 lowercase :       Optional[datasets.Features]		      =		None



 def 						UpperCAmelCase    (	self   :   int	):

    """simple docstring"""






    assert self.generator is not None, "generator must be specified"

    if self.gen_kwargs is None:
       lowercase__						:	List[Any]         = {}
class        __snake_case(					datasets.GeneratorBasedBuilder      ):
 lowercase :       Dict		      =		GeneratorConfig



 def 						UpperCAmelCase    (	self   :   Any	):

    """simple docstring"""






    return datasets.DatasetInfo(features=self.config.features	)



 def 						UpperCAmelCase    (	self   :   List[str] ,		UpperCAmelCase__   :   Optional[int]	):

    """simple docstring"""






    return [datasets.SplitGenerator(name=datasets.Split.TRAIN ,		gen_kwargs=self.config.gen_kwargs	)]







 def 						UpperCAmelCase    (	self   :   Dict ,		**UpperCAmelCase__   :   List[str]	):

    """simple docstring"""






    for idx, ex in enumerate(self.config.generator(**UpperCAmelCase__	)	):
       yield idx, ex


