




'''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:
      lowercase__   :    Optional[int] 					=      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)]
            )
            lowercase__   :    Union[str, Any] 					=      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     a__ :
     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						,	A						,	A						,	A=False				)						->					Dict:



          '''simple docstring'''



          a											=   copy.deepcopy(A				)

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

          if return_labels:
               if model_class in get_values(A				):
                    a											=   tf.ones(self.model_tester.batch_size						,	dtype=tf.intaa				)
               elif model_class in get_values(A				):
                    a											=   tf.zeros(self.model_tester.batch_size						,	dtype=tf.intaa				)
                    a											=   tf.zeros(self.model_tester.batch_size						,	dtype=tf.intaa				)
               elif model_class in [
                   *get_values(A				),
                   *get_values(A				),
               ]:
                    a											=   tf.zeros(self.model_tester.batch_size						,	dtype=tf.intaa				)
               elif model_class in get_values(A				):
                    a											=   tf.zeros(self.model_tester.batch_size						,	dtype=tf.intaa				)
               elif model_class in [
                   *get_values(A				),
                   *get_values(A				),
                   *get_values(A				),
                   *get_values(A				),
                   *get_values(A				),
               ]:
                    a											=   tf.zeros(
                        (self.model_tester.batch_size, self.model_tester.seq_length)						,	dtype=tf.intaa				)
          return inputs_dict







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



          '''simple docstring'''



          a											=   self.model_tester.prepare_config_and_inputs_for_common()
          for model_class in self.all_model_classes[:2]:
               a											=   self._prepare_for_class(A						,	A				)
               a											=   model_class(A				)

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

               a											=   run_in_graph_mode()
               self.assertIsNotNone(A				)







     @slow
     def     lowerCAmelCase_       (			self				)						->					List[Any]:



          '''simple docstring'''



          a											=   self.model_tester.prepare_config_and_inputs_for_common()
          for model_class in self.all_model_classes[:2]:
               a											=   self._prepare_for_class(A						,	A				)
               a											=   model_class(A				)

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

               a											=   run_in_graph_mode()
               self.assertIsNotNone(A				)







     @slow
     def     lowerCAmelCase_       (			self				)						->					Optional[Any]:



          '''simple docstring'''



          a											=   self.model_tester.prepare_config_and_inputs_for_common()
          for model_class in self.all_model_classes[:2]:
               a											=   model_class(A				)
               if getattr(A						,	"hf_compute_loss"						,	A				):
                    # Test that model correctly compute the loss with kwargs
                    a											=   self._prepare_for_class(inputs_dict.copy()						,	A						,	return_labels=A				)
                    # Is there a better way to remove these decoder inputs?
                    a											=   {
                        key: val
                        for key, val in prepared_for_class.items()
                        if key not in ("head_mask", "decoder_head_mask", "cross_attn_head_mask", "decoder_input_ids")
                    }

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

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

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

                    a											=   history.history["loss"][0]
                    self.assertTrue(not isnan(A				)				)
                    a											=   history.history["val_loss"][0]
                    self.assertTrue(not isnan(A				)				)







     @slow
     def     lowerCAmelCase_       (			self				)						->					List[Any]:



          '''simple docstring'''



          a											=   self.model_tester.prepare_config_and_inputs_for_common()
          a											=   True
          a											=   True

          if hasattr(A						,	"use_cache"				):
               a											=   True

          a											=   getattr(self.model_tester						,	"encoder_seq_length"						,	self.model_tester.seq_length				)
          a											=   getattr(self.model_tester						,	"key_length"						,	A				)

          for model_class in self.all_model_classes[:2]:
               a											=   self._prepare_for_class(A						,	A				)
               a											=   model_class(A				)
               model.build()
               a											=   len(model(A				)				)

               for key in list(class_inputs_dict.keys()				):
                    # Remove keys not in the serving signature, as the SavedModel will not be compiled to deal with them
                    if key not in model.input_signature:
                         del class_inputs_dict[key]
                    # Check it's a tensor, in case the inputs dict has some bools in it too
                    elif isinstance(class_inputs_dict[key]						,	tf.Tensor				) and class_inputs_dict[key].dtype.is_integer:
                         a											=   tf.cast(class_inputs_dict[key]						,	tf.intaa				)

               if set(class_inputs_dict.keys()				) != set(model.input_signature.keys()				):
                    continue  # Some models have inputs that the preparation functions don't create, we skip those

               with tempfile.TemporaryDirectory() as tmpdirname:
                    model.save_pretrained(A						,	saved_model=A				)
                    a											=   os.path.join(A						,	"saved_model"						,	"1"				)
                    a											=   tf.keras.models.load_model(A				)
                    a											=   model(A				)

                    if self.is_encoder_decoder:
                         a											=   outputs["encoder_hidden_states"]
                         a											=   outputs["encoder_attentions"]
                    else:
                         a											=   outputs["hidden_states"]
                         a											=   outputs["attentions"]

                    self.assertEqual(len(A				)						,	A				)

                    a											=   getattr(
                        self.model_tester						,	"expected_num_hidden_layers"						,	self.model_tester.num_hidden_layers + 1				)

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

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







     @slow
     def     lowerCAmelCase_       (			self				)						->					Any:



          '''simple docstring'''



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

          # try/finally block to ensure subsequent tests run in float32
          try:
               a											=   self.model_tester.prepare_config_and_inputs_for_common()
               for model_class in self.all_model_classes[:2]:
                    a											=   self._prepare_for_class(A						,	A				)
                    a											=   model_class(A				)
                    a											=   model(A				)

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







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



          '''simple docstring'''



          a											=   self.model_tester.prepare_config_and_inputs_for_common()
          # head_mask and decoder_head_mask has different shapes than other input args
          if "head_mask" in inputs_dict:
               del inputs_dict["head_mask"]
          if "decoder_head_mask" in inputs_dict:
               del inputs_dict["decoder_head_mask"]
          if "cross_attn_head_mask" in inputs_dict:
               del inputs_dict["cross_attn_head_mask"]
          a											=   {
              module_member
              for model_class in self.all_model_classes
              for module in (import_module(model_class.__module__				),)
              for module_member_name in dir(A				)
              if module_member_name.endswith("MainLayer"				)
              for module_member in (getattr(A						,	A				),)
              if isinstance(A						,	A				)
              and tf.keras.layers.Layer in module_member.__bases__
              and getattr(A						,	"_keras_serializable"						,	A				)
          }

          for main_layer_class in tf_main_layer_classes:
               # T5MainLayer needs an embed_tokens parameter when called without the inputs_embeds parameter
               if "T5" in main_layer_class.__name__:
                    # Take the same values than in TFT5ModelTester for this shared layer
                    a											=   TFSharedEmbeddings(self.model_tester.vocab_size						,	self.model_tester.hidden_size						,	name="shared"				)
                    a											=   False
                    a											=   main_layer_class(A						,	embed_tokens=A				)
               else:
                    a											=   main_layer_class(A				)

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

               if hasattr(self.model_tester						,	"num_labels"				):
                    a											=   self.model_tester.num_labels
               else:
                    a											=   2

               a											=   tf.data.Dataset.from_tensor_slices(
                   (inputs_dict, np.ones((self.model_tester.batch_size, self.model_tester.seq_length, num_labels, 1)				))				).batch(1				)

               a											=   main_layer(A				)[0]
               a											=   tf.keras.layers.Dense(A						,	activation="softmax"						,	name="outputs"				)(A				)
               a											=   tf.keras.models.Model(inputs=A						,	outputs=[outputs]				)

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

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







     @slow
     def     lowerCAmelCase_       (			self				)						->					str:



          '''simple docstring'''



          a											=   self.model_tester.prepare_config_and_inputs_for_common()

          for model_class in self.all_model_classes[:2]:
               a											=   model_class(A				)

               a											=   copy.deepcopy(A				)

               if not self.is_encoder_decoder:
                    a											=   inputs["input_ids"]
                    del inputs["input_ids"]
               else:
                    a											=   inputs["input_ids"]
                    a											=   inputs.get("decoder_input_ids"						,	A				)
                    del inputs["input_ids"]
                    inputs.pop("decoder_input_ids"						,	A				)

               if not self.is_encoder_decoder:
                    a											=   model.get_input_embeddings()(A				)
               else:
                    a											=   model.get_input_embeddings()(A				)
                    a											=   model.get_input_embeddings()(A				)

               a											=   self._prepare_for_class(A						,	A				)

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

               a											=   run_in_graph_mode()
               self.assertIsNotNone(A				)







     def     lowerCAmelCase_       (			self						,	A						,	A				)						->					int:



          '''simple docstring'''



          a											=   []
          if model.config.bos_token_id is not None:
               special_tokens.append(model.config.bos_token_id				)
          if model.config.pad_token_id is not None:
               special_tokens.append(model.config.pad_token_id				)
          if model.config.eos_token_id is not None:
               special_tokens.append(model.config.eos_token_id				)

          # create random bad tokens that are not special tokens
          a											=   []
          while len(A				) < num_bad_tokens:
               a											=   tf.squeeze(ids_tensor((1, 1)						,	self.model_tester.vocab_size				)						,	0				).numpy()[0]
               if token not in special_tokens:
                    bad_tokens.append(A				)
          return bad_tokens







     def     lowerCAmelCase_       (			self						,	A				)						->					Optional[Any]:



          '''simple docstring'''



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







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







from dataclasses import dataclass
from typing import List, Optional, Union

import numpy as np
import PIL

from ...utils import BaseOutput, is_invisible_watermark_available, is_torch_available, is_transformers_available




@dataclass
class     a__ (      UpperCamelCase__   ):
     a	:    Union[List[PIL.Image.Image], np.ndarray]


if is_transformers_available() and is_torch_available() and is_invisible_watermark_available():
   from .pipeline_stable_diffusion_xl import StableDiffusionXLPipeline
   from .pipeline_stable_diffusion_xl_imgaimg import StableDiffusionXLImgaImgPipeline

