






"""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:
												_snake_case							:   Union[str, Any]	  =       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)]
																								)
																								_snake_case							:   Optional[int]	  =       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 							__lowercase:
						lowerCAmelCase_		:				List[Any]=							None
						lowerCAmelCase_		:				Dict=							()
						lowerCAmelCase_		:				List[Any]=							()
						lowerCAmelCase_		:				Union[str, Any]=							True
						lowerCAmelCase_		:				str=							True
						lowerCAmelCase_		:				List[Any]=							True
						lowerCAmelCase_		:				int=							False






						def 		lowerCAmelCase (       self     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE=False)      ->	dict:




											"""simple docstring"""





											lowerCamelCase__					=copy.deepcopy(_SCREAMING_SNAKE_CASE)

											if model_class in get_values(_SCREAMING_SNAKE_CASE):
																lowerCamelCase__					={
																    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):
																					lowerCamelCase__					=tf.ones(self.model_tester.batch_size     ,    dtype=tf.intaa)
																elif model_class in get_values(_SCREAMING_SNAKE_CASE):
																					lowerCamelCase__					=tf.zeros(self.model_tester.batch_size     ,    dtype=tf.intaa)
																					lowerCamelCase__					=tf.zeros(self.model_tester.batch_size     ,    dtype=tf.intaa)
																elif model_class in [
																    *get_values(_SCREAMING_SNAKE_CASE),
																    *get_values(_SCREAMING_SNAKE_CASE),
																]:
																					lowerCamelCase__					=tf.zeros(self.model_tester.batch_size     ,    dtype=tf.intaa)
																elif model_class in get_values(_SCREAMING_SNAKE_CASE):
																					lowerCamelCase__					=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),
																]:
																					lowerCamelCase__					=tf.zeros(
																					    (self.model_tester.batch_size, self.model_tester.seq_length)     ,    dtype=tf.intaa)
											return inputs_dict






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




											"""simple docstring"""





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

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

																lowerCamelCase__					=run_in_graph_mode()
																self.assertIsNotNone(_SCREAMING_SNAKE_CASE)






						@slow
						def 		lowerCAmelCase (       self)      ->	int:




											"""simple docstring"""





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

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

																lowerCamelCase__					=run_in_graph_mode()
																self.assertIsNotNone(_SCREAMING_SNAKE_CASE)






						@slow
						def 		lowerCAmelCase (       self)      ->	int:




											"""simple docstring"""





											lowerCamelCase__					=self.model_tester.prepare_config_and_inputs_for_common()
											for model_class in self.all_model_classes[:2]:
																lowerCamelCase__					=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
																					lowerCamelCase__					=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?
																					lowerCamelCase__					={
																					    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")
																					}

																					lowerCamelCase__					={
																					    "labels",
																					    "label",
																					    "label_ids",
																					    "start_positions",
																					    "start_position",
																					    "end_positions",
																					    "end_position",
																					    "next_sentence_label",
																					}
																					lowerCamelCase__					=possible_label_cols.intersection(set(_SCREAMING_SNAKE_CASE))
																					self.assertGreater(len(_SCREAMING_SNAKE_CASE)     ,    0     ,    msg='No matching label names found!')
																					lowerCamelCase__					={key: val for key, val in prepared_for_class.items() if key in label_names}
																					lowerCamelCase__					={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
																					lowerCamelCase__					=model.fit(
																					    _SCREAMING_SNAKE_CASE     ,    validation_data=_SCREAMING_SNAKE_CASE     ,    steps_per_epoch=1     ,    validation_steps=1     ,    shuffle=_SCREAMING_SNAKE_CASE     ,    verbose=0     ,    )
																					lowerCamelCase__					=history.history["loss"][0]
																					self.assertTrue(not isnan(_SCREAMING_SNAKE_CASE))
																					lowerCamelCase__					=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.
																					lowerCamelCase__					=model_class(_SCREAMING_SNAKE_CASE)
																					model.compile(optimizer=tf.keras.optimizers.SGD(0.0)     ,    jit_compile=_SCREAMING_SNAKE_CASE)
																					lowerCamelCase__					=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     ,    )

																					lowerCamelCase__					=history.history["loss"][0]
																					self.assertTrue(not isnan(_SCREAMING_SNAKE_CASE))
																					lowerCamelCase__					=history.history["val_loss"][0]
																					self.assertTrue(not isnan(_SCREAMING_SNAKE_CASE))






						@slow
						def 		lowerCAmelCase (       self)      ->	List[str]:




											"""simple docstring"""





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

											if hasattr(_SCREAMING_SNAKE_CASE     ,    'use_cache'):
																lowerCamelCase__					=True

											lowerCamelCase__					=getattr(self.model_tester     ,    'encoder_seq_length'     ,    self.model_tester.seq_length)
											lowerCamelCase__					=getattr(self.model_tester     ,    'key_length'     ,    _SCREAMING_SNAKE_CASE)

											for model_class in self.all_model_classes[:2]:
																lowerCamelCase__					=self._prepare_for_class(_SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE)
																lowerCamelCase__					=model_class(_SCREAMING_SNAKE_CASE)
																model.build()
																lowerCamelCase__					=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:
																										lowerCamelCase__					=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)
																					lowerCamelCase__					=os.path.join(_SCREAMING_SNAKE_CASE     ,    'saved_model'     ,    '1')
																					lowerCamelCase__					=tf.keras.models.load_model(_SCREAMING_SNAKE_CASE)
																					lowerCamelCase__					=model(_SCREAMING_SNAKE_CASE)

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

																					self.assertEqual(len(_SCREAMING_SNAKE_CASE)     ,    _SCREAMING_SNAKE_CASE)

																					lowerCamelCase__					=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)      ->	int:




											"""simple docstring"""





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

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

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






						@slow
						def 		lowerCAmelCase (       self)      ->	Optional[int]:




											"""simple docstring"""





											lowerCamelCase__					=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"]
											lowerCamelCase__					={
											    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
																					lowerCamelCase__					=TFSharedEmbeddings(self.model_tester.vocab_size     ,    self.model_tester.hidden_size     ,    name='shared')
																					lowerCamelCase__					=False
																					lowerCamelCase__					=main_layer_class(_SCREAMING_SNAKE_CASE     ,    embed_tokens=_SCREAMING_SNAKE_CASE)
																else:
																					lowerCamelCase__					=main_layer_class(_SCREAMING_SNAKE_CASE)

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

																if hasattr(self.model_tester     ,    'num_labels'):
																					lowerCamelCase__					=self.model_tester.num_labels
																else:
																					lowerCamelCase__					=2

																lowerCamelCase__					=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)

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




											"""simple docstring"""





											lowerCamelCase__					=self.model_tester.prepare_config_and_inputs_for_common()

											for model_class in self.all_model_classes[:2]:
																lowerCamelCase__					=model_class(_SCREAMING_SNAKE_CASE)

																lowerCamelCase__					=copy.deepcopy(_SCREAMING_SNAKE_CASE)

																if not self.is_encoder_decoder:
																					lowerCamelCase__					=inputs["input_ids"]
																					del inputs["input_ids"]
																else:
																					lowerCamelCase__					=inputs["input_ids"]
																					lowerCamelCase__					=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:
																					lowerCamelCase__					=model.get_input_embeddings()(_SCREAMING_SNAKE_CASE)
																else:
																					lowerCamelCase__					=model.get_input_embeddings()(_SCREAMING_SNAKE_CASE)
																					lowerCamelCase__					=model.get_input_embeddings()(_SCREAMING_SNAKE_CASE)

																lowerCamelCase__					=self._prepare_for_class(_SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE)

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

																lowerCamelCase__					=run_in_graph_mode()
																self.assertIsNotNone(_SCREAMING_SNAKE_CASE)






						def 		lowerCAmelCase (       self     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE)      ->	Any:




											"""simple docstring"""





											lowerCamelCase__					=[]
											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
											lowerCamelCase__					=[]
											while len(_SCREAMING_SNAKE_CASE) < num_bad_tokens:
																lowerCamelCase__					=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)      ->	Any:




											"""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)      ->	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(_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






from typing import List, Optional, Tuple, Union

import torch

from ...utils import randn_tensor
from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput



class  _a  (					lowercase__				):






							"""simple docstring"""

							def __init__(   self					:  Dict				,						a					:  str				,						a					:  Optional[Any]     )						->Any:
													super().__init__()
													self.register_modules(unet=a				,						scheduler=a     )






							@torch.no_grad()
							def __call__(   self					:  int				,						a					:  int = 1				,						a					:  Optional[Union[torch.Generator, List[torch.Generator]]] = None				,						a					:  int = 10_00				,						a					:  Optional[str] = "pil"				,						a					:  bool = True				,						)						->Union[ImagePipelineOutput, Tuple]:
													# Sample gaussian noise to begin loop
													if isinstance(self.unet.config.sample_size				,						a     ):
																			SCREAMING_SNAKE_CASE__  :					int =       (
																			    batch_size,
																			    self.unet.config.in_channels,
																			    self.unet.config.sample_size,
																			    self.unet.config.sample_size,
																			)
													else:
																			SCREAMING_SNAKE_CASE__  :					Dict =       (batch_size, self.unet.config.in_channels, *self.unet.config.sample_size)

													if self.device.type == "mps":
																			# randn does not work reproducibly on mps
																			SCREAMING_SNAKE_CASE__  :					List[str] =       randn_tensor(a				,						generator=a     )
																			SCREAMING_SNAKE_CASE__  :					Optional[int] =       image.to(self.device     )
													else:
																			SCREAMING_SNAKE_CASE__  :					Union[str, Any] =       randn_tensor(a				,						generator=a				,						device=self.device     )

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

													for t in self.progress_bar(self.scheduler.timesteps     ):
																			# 1. predict noise model_output
																			SCREAMING_SNAKE_CASE__  :					Optional[Any] =       self.unet(a				,						a     ).sample

																			# 2. compute previous image: x_t -> x_t-1
																			SCREAMING_SNAKE_CASE__  :					Tuple =       self.scheduler.step(a				,						a				,						a				,						generator=a     ).prev_sample

													SCREAMING_SNAKE_CASE__  :					List[str] =       (image / 2 + 0.5).clamp(0				,						1     )
													SCREAMING_SNAKE_CASE__  :					Tuple =       image.cpu().permute(0				,						2				,						3				,						1     ).numpy()
													if output_type == "pil":
																			SCREAMING_SNAKE_CASE__  :					Dict =       self.numpy_to_pil(a     )

													if not return_dict:
																			return (image,)

													return ImagePipelineOutput(images=a     )