


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:
      UpperCamelCase__				       =			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)]
            )
            UpperCamelCase__				       =			tf.config.list_logical_devices('GPU')
            print('Logical GPUs', logical_gpus)
         except RuntimeError as e:
            # Virtual devices must be set before GPUs have been initialized
            print(e)

@require_tf
class 					UpperCAmelCase__   :



     '''simple docstring'''
     UpperCAmelCase_         =  None
     UpperCAmelCase_         =  ()
     UpperCAmelCase_         =  ()
     UpperCAmelCase_         =  True
     UpperCAmelCase_         =  True
     UpperCAmelCase_         =  True
     UpperCAmelCase_         =  False




     def   lowerCAmelCase_   ( self       :       Any	,      UpperCamelCase       :       List[str]	,      UpperCamelCase       :       int	,      UpperCamelCase       :       int=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					:				str           =			tf.ones(self.model_tester.batch_size	,      dtype=tf.intaa		)
                   elif model_class in get_values(UpperCamelCase		):
                          _lowercase					:				str           =			tf.zeros(self.model_tester.batch_size	,      dtype=tf.intaa		)
                          _lowercase					:				Union[str, Any]           =			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					:				List[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					:				Optional[int]           =			tf.zeros(
                              (self.model_tester.batch_size, self.model_tester.seq_length)	,      dtype=tf.intaa		)
            return inputs_dict




     @slow
     def   lowerCAmelCase_   ( self       :       int		):





            """simple docstring"""



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

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

                   _lowercase					:				int           =			run_in_graph_mode()
                   self.assertIsNotNone(UpperCamelCase		)




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





            """simple docstring"""



            _lowercase					:				Dict           =			self.model_tester.prepare_config_and_inputs_for_common()
            for model_class in self.all_model_classes[:2]:
                   _lowercase					:				List[Any]           =			self._prepare_for_class(UpperCamelCase	,      UpperCamelCase		)
                   _lowercase					:				int           =			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   lowerCAmelCase_   ( self       :       Union[str, Any]		):





            """simple docstring"""



            _lowercase					:				Optional[int]           =			self.model_tester.prepare_config_and_inputs_for_common()
            for model_class in self.all_model_classes[:2]:
                   _lowercase					:				Dict           =			model_class(UpperCamelCase		)
                   if getattr(UpperCamelCase	,      '''hf_compute_loss'''	,      UpperCamelCase		):
                          # Test that model correctly compute the loss with kwargs
                          _lowercase					:				Dict           =			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					:				int           =			{
                              '''labels''',
                              '''label''',
                              '''label_ids''',
                              '''start_positions''',
                              '''start_position''',
                              '''end_positions''',
                              '''end_position''',
                              '''next_sentence_label''',
                          }
                          _lowercase					:				List[Any]           =			possible_label_cols.intersection(set(UpperCamelCase		)		)
                          self.assertGreater(len(UpperCamelCase		)	,      0	,      msg='''No matching label names found!'''		)
                          _lowercase					:				Union[str, Any]           =			{key: val for key, val in prepared_for_class.items() if key in label_names}
                          _lowercase					:				List[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					:				Tuple           =			model.fit(
                              UpperCamelCase	,      validation_data=UpperCamelCase	,      steps_per_epoch=1	,      validation_steps=1	,      shuffle=UpperCamelCase	,      verbose=0	,      )
                          _lowercase					:				int           =			history.history['''loss'''][0]
                          self.assertTrue(not isnan(UpperCamelCase		)		)
                          _lowercase					:				List[Any]           =			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					:				Any           =			model.fit(
                              UpperCamelCase	,      UpperCamelCase	,      validation_data=(inputs_minus_labels, labels)	,      steps_per_epoch=1	,      validation_steps=1	,      shuffle=UpperCamelCase	,      verbose=0	,      )

                          _lowercase					:				int           =			history.history['''loss'''][0]
                          self.assertTrue(not isnan(UpperCamelCase		)		)
                          _lowercase					:				Any           =			history.history['''val_loss'''][0]
                          self.assertTrue(not isnan(UpperCamelCase		)		)




     @slow
     def   lowerCAmelCase_   ( self       :       Tuple		):





            """simple docstring"""



            _lowercase					:				Dict           =			self.model_tester.prepare_config_and_inputs_for_common()
            _lowercase					:				Any           =			True
            _lowercase					:				Dict           =			True

            if hasattr(UpperCamelCase	,      '''use_cache'''		):
                   _lowercase					:				Optional[Any]           =			True

            _lowercase					:				Union[str, Any]           =			getattr(self.model_tester	,      '''encoder_seq_length'''	,      self.model_tester.seq_length		)
            _lowercase					:				List[Any]           =			getattr(self.model_tester	,      '''key_length'''	,      UpperCamelCase		)

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

                          if self.is_encoder_decoder:
                                 _lowercase					:				str           =			outputs['''encoder_hidden_states''']
                                 _lowercase					:				str           =			outputs['''encoder_attentions''']
                          else:
                                 _lowercase					:				Optional[int]           =			outputs['''hidden_states''']
                                 _lowercase					:				Tuple           =			outputs['''attentions''']

                          self.assertEqual(len(UpperCamelCase		)	,      UpperCamelCase		)

                          _lowercase					:				Dict           =			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   lowerCAmelCase_   ( self       :       Optional[Any]		):





            """simple docstring"""



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

            # try/finally block to ensure subsequent tests run in float32
            try:
                   _lowercase					:				str           =			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					:				Tuple           =			model_class(UpperCamelCase		)
                          _lowercase					:				int           =			model(UpperCamelCase		)

                          self.assertIsNotNone(UpperCamelCase		)
            finally:
                   tf.keras.mixed_precision.set_global_policy('''float32'''		)




     @slow
     def   lowerCAmelCase_   ( self       :       Union[str, Any]		):





            """simple docstring"""



            _lowercase					:				Dict           =			self.model_tester.prepare_config_and_inputs_for_common()
            # head_mask and decoder_head_mask has different shapes than other input args
            if "head_mask" in inputs_dict:
                   del inputs_dict["head_mask"]
            if "decoder_head_mask" in inputs_dict:
                   del inputs_dict["decoder_head_mask"]
            if "cross_attn_head_mask" in inputs_dict:
                   del inputs_dict["cross_attn_head_mask"]
            _lowercase					:				str           =			{
                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					:				Tuple           =			TFSharedEmbeddings(self.model_tester.vocab_size	,      self.model_tester.hidden_size	,      name='''shared'''		)
                          _lowercase					:				List[Any]           =			False
                          _lowercase					:				Any           =			main_layer_class(UpperCamelCase	,      embed_tokens=UpperCamelCase		)
                   else:
                          _lowercase					:				Optional[Any]           =			main_layer_class(UpperCamelCase		)

                   _lowercase					:				int           =			{
                       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					:				Optional[Any]           =			self.model_tester.num_labels
                   else:
                          _lowercase					:				Tuple           =			2

                   _lowercase					:				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		)

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





            """simple docstring"""



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

            for model_class in self.all_model_classes[:2]:
                   _lowercase					:				int           =			model_class(UpperCamelCase		)

                   _lowercase					:				Tuple           =			copy.deepcopy(UpperCamelCase		)

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

                   if not self.is_encoder_decoder:
                          _lowercase					:				str           =			model.get_input_embeddings()(UpperCamelCase		)
                   else:
                          _lowercase					:				Dict           =			model.get_input_embeddings()(UpperCamelCase		)
                          _lowercase					:				str           =			model.get_input_embeddings()(UpperCamelCase		)

                   _lowercase					:				List[str]           =			self._prepare_for_class(UpperCamelCase	,      UpperCamelCase		)

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

                   _lowercase					:				int           =			run_in_graph_mode()
                   self.assertIsNotNone(UpperCamelCase		)




     def   lowerCAmelCase_   ( self       :       Any	,      UpperCamelCase       :       str	,      UpperCamelCase       :       str		):





            """simple docstring"""



            _lowercase					:				Dict           =			[]
            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					:				Optional[Any]           =			[]
            while len(UpperCamelCase		) < num_bad_tokens:
                   _lowercase					:				Dict           =			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   lowerCAmelCase_   ( self       :       str	,      UpperCamelCase       :       Tuple		):





            """simple docstring"""



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




     def   lowerCAmelCase_   ( self       :       Optional[int]	,      UpperCamelCase       :       List[str]	,      UpperCamelCase       :       Optional[int]		):





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



# limitations under the License.


from typing import Optional, Tuple, Union

import torch

from diffusers.pipeline_utils import DiffusionPipeline, ImagePipelineOutput

class 						UpperCAmelCase__   (      A_		):



     '''simple docstring'''




     def __init__( self       :       List[str]	,      UpperCamelCase       :       Union[str, Any]	,      UpperCamelCase       :       int		):





            """simple docstring"""



            super().__init__()
            self.register_modules(unet=UpperCamelCase	,      scheduler=UpperCamelCase		)




     @torch.no_grad()
     def __call__( self       :       Any	,      UpperCamelCase       :       int = 1	,      UpperCamelCase       :       Optional[torch.Generator] = None	,      UpperCamelCase       :       int = 50	,      UpperCamelCase       :       Optional[str] = "pil"	,      UpperCamelCase       :       bool = True	,      **UpperCamelCase       :       int	,      ):





            """simple docstring"""



            _lowercase					:				Optional[int]           =			torch.randn(
                (batch_size, self.unet.config.in_channels, self.unet.config.sample_size, self.unet.config.sample_size)	,      generator=UpperCamelCase	,      )
            _lowercase					:				Dict           =			image.to(self.device		)

            # set step values
            self.scheduler.set_timesteps(UpperCamelCase		)

            for t in self.progress_bar(self.scheduler.timesteps		):
                   # 1. predict noise model_output
                   _lowercase					:				Dict           =			self.unet(UpperCamelCase	,      UpperCamelCase		).sample

                   # 2. predict previous mean of image x_t-1 and add variance depending on eta
                   # eta corresponds to η in paper and should be between [0, 1]
                   # do x_t -> x_t-1
                   _lowercase					:				Tuple           =			self.scheduler.step(UpperCamelCase	,      UpperCamelCase	,      UpperCamelCase		).prev_sample

            _lowercase					:				Optional[int]           =			(image / 2 + 0.5).clamp(0	,      1		)
            _lowercase					:				int           =			image.cpu().permute(0	,      2	,      3	,      1		).numpy()
            if output_type == "pil":
                   _lowercase					:				List[Any]           =			self.numpy_to_pil(UpperCamelCase		)

            if not return_dict:
                   return (image,), "This is a local test"

            return ImagePipelineOutput(images=UpperCamelCase		), "This is a local test"