





import unittest
from pathlib import Path
from tempfile import TemporaryDirectory

from transformers import AutoConfig, TFGPTaLMHeadModel, is_keras_nlp_available, is_tf_available
from transformers.models.gpta.tokenization_gpta import GPTaTokenizer
from transformers.testing_utils import require_keras_nlp, require_tf, slow


if is_tf_available():
							import tensorflow as tf

if is_keras_nlp_available():
							from transformers.models.gpta import TFGPTaTokenizer


UpperCAmelCase_							=				["""gpt2"""]
UpperCAmelCase_							=				"""gpt2"""

if is_tf_available():
							class     UpperCamelCase__					(     tf.Module				):






									'''simple docstring'''






									def __init__(							self,   snake_case__  )       -> Optional[Any]:






																"""simple docstring"""
																super().__init__()
																lowercase_	:   int  				=     tokenizer
																lowercase_	:   str  				=     AutoConfig.from_pretrained(snake_case__  )
																lowercase_	:   Any  				=     TFGPTaLMHeadModel.from_config(snake_case__  )






									@tf.function(input_signature=(tf.TensorSpec((None,),   tf.string,   name="""text"""  ),)  )
									def 						snake_case__				(							self,   snake_case__  )       -> Dict:






																"""simple docstring"""
																lowercase_	:   Tuple  				=     self.tokenizer(snake_case__  )
																lowercase_	:   Optional[Any]  				=     tokenized["""input_ids"""].to_tensor()

																lowercase_	:   Optional[Any]  				=     tf.cast(input_ids_dense > 0,   tf.intaa  )
																# input_mask = tf.reshape(input_mask, [-1, MAX_SEQ_LEN])

																lowercase_	:   Dict  				=     self.model(input_ids=snake_case__,   attention_mask=snake_case__  )["""logits"""]

																return outputs
@require_tf
@require_keras_nlp
class     UpperCamelCase__					(     unittest.TestCase				):






		'''simple docstring'''






		def 						snake_case__				(							self  )       -> int:






									"""simple docstring"""
									super().setUp()

									lowercase_	:   Any  				=     [GPTaTokenizer.from_pretrained(snake_case__  ) for checkpoint in (TOKENIZER_CHECKPOINTS)]
									lowercase_	:   Dict  				=     [TFGPTaTokenizer.from_pretrained(snake_case__  ) for checkpoint in TOKENIZER_CHECKPOINTS]
									assert len(self.tokenizers  ) == len(self.tf_tokenizers  )

									lowercase_	:   Dict  				=     [
									    """This is a straightforward English test sentence.""",
									    """This one has some weird characters\rto\nsee\r\nif  those\u00E9break things.""",
									    """Now we're going to add some Chinese: 一 二 三 一二三""",
									    """And some much more rare Chinese: 齉 堃 齉堃""",
									    """Je vais aussi écrire en français pour tester les accents""",
									    """Classical Irish also has some unusual characters, so in they go: Gaelaċ, ꝼ""",
									]
									lowercase_	:   int  				=     list(zip(self.test_sentences,   self.test_sentences[::-1]  )  )






		def 						snake_case__				(							self  )       -> Optional[int]:






									"""simple docstring"""
									for tokenizer, tf_tokenizer in zip(self.tokenizers,   self.tf_tokenizers  ):
																for test_inputs in self.test_sentences:
																							lowercase_	:   int  				=     tokenizer([test_inputs],   return_tensors="""tf"""  )
																							lowercase_	:   Tuple  				=     tf_tokenizer([test_inputs]  )

																							for key in python_outputs.keys():
																														# convert them to numpy to avoid messing with ragged tensors
																														lowercase_	:   Dict  				=     python_outputs[key].numpy()
																														lowercase_	:   int  				=     tf_outputs[key].numpy()

																														self.assertTrue(tf.reduce_all(python_outputs_values.shape == tf_outputs_values.shape  )  )
																														self.assertTrue(tf.reduce_all(tf.cast(snake_case__,   tf.intaa  ) == tf_outputs_values  )  )






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






									"""simple docstring"""
									for tf_tokenizer in self.tf_tokenizers:
																lowercase_	:   Union[str, Any]  				=     tf.function(snake_case__  )
																for test_inputs in self.test_sentences:
																							lowercase_	:   List[str]  				=     tf.constant(snake_case__  )
																							lowercase_	:   Optional[Any]  				=     compiled_tokenizer(snake_case__  )
																							lowercase_	:   Union[str, Any]  				=     tf_tokenizer(snake_case__  )

																							for key in eager_outputs.keys():
																														self.assertTrue(tf.reduce_all(eager_outputs[key] == compiled_outputs[key]  )  )






		@slow
		def 						snake_case__				(							self  )       -> int:






									"""simple docstring"""
									for tf_tokenizer in self.tf_tokenizers:
																lowercase_	:   Dict  				=     ModelToSave(tokenizer=snake_case__  )
																lowercase_	:   List[Any]  				=     tf.convert_to_tensor([self.test_sentences[0]]  )
																lowercase_	:   Union[str, Any]  				=     model.serving(snake_case__  )  # Build model with some sample inputs
																with TemporaryDirectory() as tempdir:
																							lowercase_	:   str  				=     Path(snake_case__  ) / """saved.model"""
																							tf.saved_model.save(snake_case__,   snake_case__,   signatures={"""serving_default""": model.serving}  )
																							lowercase_	:   Union[str, Any]  				=     tf.saved_model.load(snake_case__  )
																lowercase_	:   List[str]  				=     loaded_model.signatures["""serving_default"""](snake_case__  )["""output_0"""]
																# We may see small differences because the loaded model is compiled, so we need an epsilon for the test
																self.assertTrue(tf.reduce_all(out == loaded_output  )  )






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






									"""simple docstring"""
									for tf_tokenizer in self.tf_tokenizers:
																lowercase_	:   int  				=     tf.convert_to_tensor([self.test_sentences[0]]  )
																lowercase_	:   List[Any]  				=     tf_tokenizer(snake_case__  )  # Build model with some sample inputs

																lowercase_	:   int  				=     tf_tokenizer.get_config()
																lowercase_	:   int  				=     TFGPTaTokenizer.from_config(snake_case__  )
																lowercase_	:   Union[str, Any]  				=     model_from_config(snake_case__  )

																for key in from_config_output.keys():
																							self.assertTrue(tf.reduce_all(from_config_output[key] == out[key]  )  )






		@slow
		def 						snake_case__				(							self  )       -> Dict:






									"""simple docstring"""
									for tf_tokenizer in self.tf_tokenizers:
																# for the test to run
																lowercase_	:   List[Any]  				=     12_31_23

																for max_length in [3, 5, 10_24]:
																							lowercase_	:   Optional[int]  				=     tf.convert_to_tensor([self.test_sentences[0]]  )
																							lowercase_	:   Union[str, Any]  				=     tf_tokenizer(snake_case__,   max_length=snake_case__  )

																							lowercase_	:   List[Any]  				=     out["""input_ids"""].numpy().shape[1]

																							assert out_length == max_length






import gc
import unittest

import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DDPMScheduler,
    PriorTransformer,
    StableUnCLIPPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.testing_utils import enable_full_determinism, load_numpy, require_torch_gpu, slow, torch_device

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()
class     UpperCamelCase__					(     lowerCamelCase__				, lowerCamelCase__				, lowerCamelCase__				, unittest.TestCase				):






		'''simple docstring'''






		__a       :       Optional[int]       =	StableUnCLIPPipeline
		__a       :       int       =	TEXT_TO_IMAGE_PARAMS
		__a       :       Optional[Any]       =	TEXT_TO_IMAGE_BATCH_PARAMS
		__a       :       Tuple       =	TEXT_TO_IMAGE_IMAGE_PARAMS
		__a       :       Optional[int]       =	TEXT_TO_IMAGE_IMAGE_PARAMS

		# TODO(will) Expected attn_bias.stride(1) == 0 to be true, but got false
		__a       :       Tuple       =	False






		def 						snake_case__				(							self  )       -> Optional[Any]:






									"""simple docstring"""
									lowercase_	:   int  				=     32
									lowercase_	:   Tuple  				=     embedder_hidden_size

									# prior components

									torch.manual_seed(0  )
									lowercase_	:   Optional[Any]  				=     CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""  )

									torch.manual_seed(0  )
									lowercase_	:   Dict  				=     CLIPTextModelWithProjection(
									    CLIPTextConfig(
									        bos_token_id=0,   eos_token_id=2,   hidden_size=snake_case__,   projection_dim=snake_case__,   intermediate_size=37,   layer_norm_eps=1E-05,   num_attention_heads=4,   num_hidden_layers=5,   pad_token_id=1,   vocab_size=10_00,   )  )

									torch.manual_seed(0  )
									lowercase_	:   Tuple  				=     PriorTransformer(
									    num_attention_heads=2,   attention_head_dim=12,   embedding_dim=snake_case__,   num_layers=1,   )

									torch.manual_seed(0  )
									lowercase_	:   Optional[Any]  				=     DDPMScheduler(
									    variance_type="""fixed_small_log""",   prediction_type="""sample""",   num_train_timesteps=10_00,   clip_sample=snake_case__,   clip_sample_range=5.0,   beta_schedule="""squaredcos_cap_v2""",   )

									# regular denoising components

									torch.manual_seed(0  )
									lowercase_	:   Union[str, Any]  				=     StableUnCLIPImageNormalizer(embedding_dim=snake_case__  )
									lowercase_	:   str  				=     DDPMScheduler(beta_schedule="""squaredcos_cap_v2"""  )

									torch.manual_seed(0  )
									lowercase_	:   str  				=     CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""  )

									torch.manual_seed(0  )
									lowercase_	:   List[Any]  				=     CLIPTextModel(
									    CLIPTextConfig(
									        bos_token_id=0,   eos_token_id=2,   hidden_size=snake_case__,   projection_dim=32,   intermediate_size=37,   layer_norm_eps=1E-05,   num_attention_heads=4,   num_hidden_layers=5,   pad_token_id=1,   vocab_size=10_00,   )  )

									torch.manual_seed(0  )
									lowercase_	:   Optional[int]  				=     UNetaDConditionModel(
									    sample_size=32,   in_channels=4,   out_channels=4,   down_block_types=("""CrossAttnDownBlock2D""", """DownBlock2D"""),   up_block_types=("""UpBlock2D""", """CrossAttnUpBlock2D"""),   block_out_channels=(32, 64),   attention_head_dim=(2, 4),   class_embed_type="""projection""",   projection_class_embeddings_input_dim=embedder_projection_dim * 2,   cross_attention_dim=snake_case__,   layers_per_block=1,   upcast_attention=snake_case__,   use_linear_projection=snake_case__,   )

									torch.manual_seed(0  )
									lowercase_	:   Dict  				=     DDIMScheduler(
									    beta_schedule="""scaled_linear""",   beta_start=0.00085,   beta_end=0.012,   prediction_type="""v_prediction""",   set_alpha_to_one=snake_case__,   steps_offset=1,   )

									torch.manual_seed(0  )
									lowercase_	:   str  				=     AutoencoderKL()

									lowercase_	:   str  				=     {
									    # prior components
									    """prior_tokenizer""": prior_tokenizer,
									    """prior_text_encoder""": prior_text_encoder,
									    """prior""": prior,
									    """prior_scheduler""": prior_scheduler,
									    # image noising components
									    """image_normalizer""": image_normalizer,
									    """image_noising_scheduler""": image_noising_scheduler,
									    # regular denoising components
									    """tokenizer""": tokenizer,
									    """text_encoder""": text_encoder,
									    """unet""": unet,
									    """scheduler""": scheduler,
									    """vae""": vae,
									}

									return components






		def 						snake_case__				(							self,   snake_case__,   snake_case__=0  )       -> str:






									"""simple docstring"""
									if str(snake_case__  ).startswith("""mps"""  ):
																lowercase_	:   Tuple  				=     torch.manual_seed(snake_case__  )
									else:
																lowercase_	:   int  				=     torch.Generator(device=snake_case__  ).manual_seed(snake_case__  )
									lowercase_	:   Dict  				=     {
									    """prompt""": """A painting of a squirrel eating a burger""",
									    """generator""": generator,
									    """num_inference_steps""": 2,
									    """prior_num_inference_steps""": 2,
									    """output_type""": """numpy""",
									}
									return inputs






		def 						snake_case__				(							self  )       -> List[Any]:






									"""simple docstring"""
									lowercase_	:   int  				=     torch_device == """cpu"""

									self._test_attention_slicing_forward_pass(test_max_difference=snake_case__  )






		def 						snake_case__				(							self  )       -> Any:






									"""simple docstring"""
									lowercase_	:   List[str]  				=     torch_device in ["""cpu""", """mps"""]

									self._test_inference_batch_single_identical(test_max_difference=snake_case__  )




@slow
@require_torch_gpu
class     UpperCamelCase__					(     unittest.TestCase				):






		'''simple docstring'''






		def 						snake_case__				(							self  )       -> Any:






									"""simple docstring"""
									# clean up the VRAM after each test
									super().tearDown()
									gc.collect()
									torch.cuda.empty_cache()






		def 						snake_case__				(							self  )       -> List[str]:






									"""simple docstring"""
									lowercase_	:   Union[str, Any]  				=     load_numpy(
									    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_anime_turtle_fp16.npy"""  )

									lowercase_	:   List[Any]  				=     StableUnCLIPPipeline.from_pretrained("""fusing/stable-unclip-2-1-l""",   torch_dtype=torch.floataa  )
									pipe.to(snake_case__  )
									pipe.set_progress_bar_config(disable=snake_case__  )
									# stable unclip will oom when integration tests are run on a V100,
									# so turn on memory savings
									pipe.enable_attention_slicing()
									pipe.enable_sequential_cpu_offload()

									lowercase_	:   Any  				=     torch.Generator(device="""cpu"""  ).manual_seed(0  )
									lowercase_	:   Dict  				=     pipe("""anime turle""",   generator=snake_case__,   output_type="""np"""  )

									lowercase_	:   str  				=     output.images[0]

									assert image.shape == (7_68, 7_68, 3)

									assert_mean_pixel_difference(snake_case__,   snake_case__  )






		def 						snake_case__				(							self  )       -> Dict:






									"""simple docstring"""
									torch.cuda.empty_cache()
									torch.cuda.reset_max_memory_allocated()
									torch.cuda.reset_peak_memory_stats()

									lowercase_	:   str  				=     StableUnCLIPPipeline.from_pretrained("""fusing/stable-unclip-2-1-l""",   torch_dtype=torch.floataa  )
									lowercase_	:   Optional[int]  				=     pipe.to(snake_case__  )
									pipe.set_progress_bar_config(disable=snake_case__  )
									pipe.enable_attention_slicing()
									pipe.enable_sequential_cpu_offload()

									lowercase_	:   List[Any]  				=     pipe(
									    """anime turtle""",   prior_num_inference_steps=2,   num_inference_steps=2,   output_type="""np""",   )

									lowercase_	:   Optional[int]  				=     torch.cuda.max_memory_allocated()
									# make sure that less than 7 GB is allocated
									assert mem_bytes < 7 * 10**9