






import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMParallelScheduler,
    DDPMParallelScheduler,
    StableDiffusionParadigmsPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    require_torch_gpu,
)

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()

class      snake_case__       (A__      ,			A__      ,			unittest.TestCase	):







						"""simple docstring"""




						__lowerCAmelCase  :List[Any]			      =  StableDiffusionParadigmsPipeline
						__lowerCAmelCase  :Any			      =  TEXT_TO_IMAGE_PARAMS
						__lowerCAmelCase  :Dict			      =  TEXT_TO_IMAGE_BATCH_PARAMS
						__lowerCAmelCase  :Optional[int]			      =  TEXT_TO_IMAGE_IMAGE_PARAMS
						__lowerCAmelCase  :Optional[Any]			      =  TEXT_TO_IMAGE_IMAGE_PARAMS

						def 							SCREAMING_SNAKE_CASE__(     self						)      ->   str:
									"""simple docstring"""




									torch.manual_seed(0						)
									a__     :		Tuple		 =						UNetaDConditionModel(
									    block_out_channels=(3_2, 6_4)      ,     layers_per_block=2      ,     sample_size=3_2      ,     in_channels=4      ,     out_channels=4      ,     down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")      ,     up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")      ,     cross_attention_dim=3_2      ,     attention_head_dim=(2, 4)      ,     use_linear_projection=__lowercase      ,     )
									a__     :		Union[str, Any]		 =						DDIMParallelScheduler(
									    beta_start=0.0_0_0_8_5      ,     beta_end=0.0_1_2      ,     beta_schedule="""scaled_linear"""      ,     clip_sample=__lowercase      ,     set_alpha_to_one=__lowercase      ,     )
									torch.manual_seed(0						)
									a__     :		int		 =						AutoencoderKL(
									    block_out_channels=[3_2, 6_4]      ,     in_channels=3      ,     out_channels=3      ,     down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""]      ,     up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""]      ,     latent_channels=4      ,     sample_size=1_2_8      ,     )
									torch.manual_seed(0						)
									a__     :		Optional[Any]		 =						CLIPTextConfig(
									    bos_token_id=0      ,     eos_token_id=2      ,     hidden_size=3_2      ,     intermediate_size=3_7      ,     layer_norm_eps=1E-05      ,     num_attention_heads=4      ,     num_hidden_layers=5      ,     pad_token_id=1      ,     vocab_size=1_0_0_0      ,     hidden_act="""gelu"""      ,     projection_dim=5_1_2      ,     )
									a__     :		Union[str, Any]		 =						CLIPTextModel(__lowercase						)
									a__     :		Optional[int]		 =						CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""						)

									a__     :		Optional[int]		 =						{
									    """unet""": unet,
									    """scheduler""": scheduler,
									    """vae""": vae,
									    """text_encoder""": text_encoder,
									    """tokenizer""": tokenizer,
									    """safety_checker""": None,
									    """feature_extractor""": None,
									}
									return components

						def 							SCREAMING_SNAKE_CASE__(     self      ,     __lowercase      ,     __lowercase=0						)      ->   Tuple:
									"""simple docstring"""




									if str(__lowercase						).startswith("""mps"""						):
												a__     :		Union[str, Any]		 =						torch.manual_seed(__lowercase						)
									else:
												a__     :		Any		 =						torch.Generator(device=__lowercase						).manual_seed(__lowercase						)
									a__     :		Any		 =						{
									    """prompt""": """a photograph of an astronaut riding a horse""",
									    """generator""": generator,
									    """num_inference_steps""": 1_0,
									    """guidance_scale""": 6.0,
									    """output_type""": """numpy""",
									    """parallel""": 3,
									    """debug""": True,
									}
									return inputs

						def 							SCREAMING_SNAKE_CASE__(     self						)      ->   Optional[Any]:
									"""simple docstring"""




									a__     :		Optional[Any]		 =						"""cpu"""  # ensure determinism for the device-dependent torch.Generator
									a__     :		Optional[int]		 =						self.get_dummy_components()
									a__     :		Tuple		 =						StableDiffusionParadigmsPipeline(**__lowercase						)
									a__     :		List[Any]		 =						sd_pipe.to(__lowercase						)
									sd_pipe.set_progress_bar_config(disable=__lowercase						)

									a__     :		List[Any]		 =						self.get_dummy_inputs(__lowercase						)
									a__     :		str		 =						sd_pipe(**__lowercase						).images
									a__     :		List[Any]		 =						image[0, -3:, -3:, -1]
									assert image.shape == (1, 6_4, 6_4, 3)

									a__     :		Union[str, Any]		 =						np.array([0.4_7_7_3, 0.5_4_1_7, 0.4_7_2_3, 0.4_9_2_5, 0.5_6_3_1, 0.4_7_5_2, 0.5_2_4_0, 0.4_9_3_5, 0.5_0_2_3]						)

									assert np.abs(image_slice.flatten() - expected_slice						).max() < 1E-2

						def 							SCREAMING_SNAKE_CASE__(     self						)      ->   int:
									"""simple docstring"""




									a__     :		Tuple		 =						"""cpu"""  # ensure determinism for the device-dependent torch.Generator
									a__     :		str		 =						self.get_dummy_components()
									torch.manual_seed(0						)
									a__     :		Tuple		 =						DDPMParallelScheduler()
									torch.manual_seed(0						)
									a__     :		Dict		 =						StableDiffusionParadigmsPipeline(**__lowercase						)
									a__     :		int		 =						sd_pipe.to(__lowercase						)
									sd_pipe.set_progress_bar_config(disable=__lowercase						)

									a__     :		Optional[Any]		 =						self.get_dummy_inputs(__lowercase						)
									a__     :		Any		 =						sd_pipe(**__lowercase						).images
									a__     :		Dict		 =						image[0, -3:, -3:, -1]
									assert image.shape == (1, 6_4, 6_4, 3)

									a__     :		Any		 =						np.array([0.3_5_7_3, 0.4_4_2_0, 0.4_9_6_0, 0.4_7_9_9, 0.3_7_9_6, 0.3_8_7_9, 0.4_8_1_9, 0.4_3_6_5, 0.4_4_6_8]						)

									assert np.abs(image_slice.flatten() - expected_slice						).max() < 1E-2

						def 							SCREAMING_SNAKE_CASE__(     self						)      ->   Dict:
									"""simple docstring"""




									super().test_inference_batch_consistent(batch_sizes=[1, 2]						)

						def 							SCREAMING_SNAKE_CASE__(     self						)      ->   Optional[int]:
									"""simple docstring"""




									super().test_inference_batch_single_identical(batch_size=2      ,     expected_max_diff=3E-3						)

						def 							SCREAMING_SNAKE_CASE__(     self						)      ->   Any:
									"""simple docstring"""




									a__     :		str		 =						"""cpu"""  # ensure determinism for the device-dependent torch.Generator
									a__     :		Dict		 =						self.get_dummy_components()
									a__     :		Dict		 =						StableDiffusionParadigmsPipeline(**__lowercase						)
									a__     :		Dict		 =						sd_pipe.to(__lowercase						)
									sd_pipe.set_progress_bar_config(disable=__lowercase						)

									a__     :		Union[str, Any]		 =						self.get_dummy_inputs(__lowercase						)
									a__     :		Dict		 =						"""french fries"""
									a__     :		str		 =						sd_pipe(**__lowercase      ,     negative_prompt=__lowercase						)
									a__     :		int		 =						output.images
									a__     :		Optional[int]		 =						image[0, -3:, -3:, -1]

									assert image.shape == (1, 6_4, 6_4, 3)

									a__     :		Dict		 =						np.array([0.4_7_7_1, 0.5_4_2_0, 0.4_6_8_3, 0.4_9_1_8, 0.5_6_3_6, 0.4_7_2_5, 0.5_2_3_0, 0.4_9_2_3, 0.5_0_1_5]						)

									assert np.abs(image_slice.flatten() - expected_slice						).max() < 1E-2




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







						"""simple docstring"""




						def 							SCREAMING_SNAKE_CASE__(     self						)      ->   int:
									"""simple docstring"""




									super().tearDown()
									gc.collect()
									torch.cuda.empty_cache()

						def 							SCREAMING_SNAKE_CASE__(     self      ,     __lowercase=0						)      ->   List[Any]:
									"""simple docstring"""




									a__     :		List[str]		 =						torch.Generator(device=__lowercase						).manual_seed(__lowercase						)
									a__     :		int		 =						{
									    """prompt""": """a photograph of an astronaut riding a horse""",
									    """generator""": generator,
									    """num_inference_steps""": 1_0,
									    """guidance_scale""": 7.5,
									    """output_type""": """numpy""",
									    """parallel""": 3,
									    """debug""": True,
									}
									return inputs

						def 							SCREAMING_SNAKE_CASE__(     self						)      ->   List[str]:
									"""simple docstring"""




									a__     :		int		 =						"""stabilityai/stable-diffusion-2-base"""
									a__     :		Union[str, Any]		 =						DDIMParallelScheduler.from_pretrained(__lowercase      ,     subfolder="""scheduler"""						)
									a__     :		Any		 =						StableDiffusionParadigmsPipeline.from_pretrained(__lowercase      ,     scheduler=__lowercase      ,     safety_checker=__lowercase						)
									pipe.to(__lowercase						)
									pipe.set_progress_bar_config(disable=__lowercase						)
									pipe.enable_attention_slicing()

									a__     :		Any		 =						self.get_inputs()
									a__     :		Dict		 =						pipe(**__lowercase						).images
									a__     :		Tuple		 =						image[0, -3:, -3:, -1].flatten()

									assert image.shape == (1, 5_1_2, 5_1_2, 3)

									a__     :		Union[str, Any]		 =						np.array([0.9_6_2_2, 0.9_6_0_2, 0.9_7_4_8, 0.9_5_9_1, 0.9_6_3_0, 0.9_6_9_1, 0.9_6_6_1, 0.9_6_3_1, 0.9_7_4_1]						)

									assert np.abs(expected_slice - image_slice						).max() < 1E-2









from __future__ import annotations

import unittest

from transformers import OpenAIGPTConfig, is_tf_available
from transformers.testing_utils import require_tf, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
		import tensorflow as tf

		from transformers.models.openai.modeling_tf_openai import (
		    TF_OPENAI_GPT_PRETRAINED_MODEL_ARCHIVE_LIST,
		    TFOpenAIGPTDoubleHeadsModel,
		    TFOpenAIGPTForSequenceClassification,
		    TFOpenAIGPTLMHeadModel,
		    TFOpenAIGPTModel,
		)

class      snake_case__       :







						"""simple docstring"""




						def __init__(     self      ,     __lowercase      ,     )      ->   str:
									"""simple docstring"""




									a__     :		Union[str, Any]		 =						parent
									a__     :		Dict		 =						1_3
									a__     :		List[Any]		 =						7
									a__     :		Tuple		 =						True
									a__     :		List[str]		 =						True
									a__     :		List[str]		 =						True
									a__     :		Optional[int]		 =						True
									a__     :		int		 =						True
									a__     :		Optional[int]		 =						9_9
									a__     :		Optional[int]		 =						3_2
									a__     :		Tuple		 =						2
									a__     :		Any		 =						4
									a__     :		Optional[Any]		 =						3_7
									a__     :		Union[str, Any]		 =						"""gelu"""
									a__     :		Dict		 =						0.1
									a__     :		Optional[int]		 =						0.1
									a__     :		int		 =						5_1_2
									a__     :		Any		 =						1_6
									a__     :		List[Any]		 =						2
									a__     :		Optional[Any]		 =						0.0_2
									a__     :		List[Any]		 =						3
									a__     :		Dict		 =						4
									a__     :		List[str]		 =						None
									a__     :		Tuple		 =						self.vocab_size - 1

						def 							SCREAMING_SNAKE_CASE__(     self						)      ->   Dict:
									"""simple docstring"""




									a__     :		Tuple		 =						ids_tensor([self.batch_size, self.seq_length]      ,     self.vocab_size						)

									a__     :		int		 =						None
									if self.use_input_mask:
												a__     :		Optional[int]		 =						random_attention_mask([self.batch_size, self.seq_length]						)

									a__     :		List[Any]		 =						None
									if self.use_token_type_ids:
												a__     :		Any		 =						ids_tensor([self.batch_size, self.seq_length]      ,     self.type_vocab_size						)

									a__     :		int		 =						None
									if self.use_mc_token_ids:
												a__     :		int		 =						ids_tensor([self.batch_size, self.num_choices]      ,     self.seq_length						)

									a__     :		str		 =						None
									a__     :		Dict		 =						None
									a__     :		Optional[int]		 =						None
									if self.use_labels:
												a__     :		List[Any]		 =						ids_tensor([self.batch_size]      ,     self.type_sequence_label_size						)
												a__     :		Dict		 =						ids_tensor([self.batch_size, self.seq_length]      ,     self.num_labels						)
												a__     :		str		 =						ids_tensor([self.batch_size]      ,     self.num_choices						)

									a__     :		Union[str, Any]		 =						OpenAIGPTConfig(
									    vocab_size=self.vocab_size      ,     n_embd=self.hidden_size      ,     n_layer=self.num_hidden_layers      ,     n_head=self.num_attention_heads      ,     n_positions=self.max_position_embeddings      ,     pad_token_id=self.pad_token_id      ,     )

									a__     :		Optional[int]		 =						ids_tensor([self.num_hidden_layers, self.num_attention_heads]      ,     2						)

									return (
									    config,
									    input_ids,
									    input_mask,
									    head_mask,
									    token_type_ids,
									    mc_token_ids,
									    sequence_labels,
									    token_labels,
									    choice_labels,
									)

						def 							SCREAMING_SNAKE_CASE__(     self      ,     __lowercase      ,     __lowercase      ,     __lowercase      ,     __lowercase      ,     __lowercase      ,     *__lowercase						)      ->   str:
									"""simple docstring"""




									a__     :		Dict		 =						TFOpenAIGPTModel(config=__lowercase						)
									a__     :		Tuple		 =						{"""input_ids""": input_ids, """attention_mask""": input_mask, """token_type_ids""": token_type_ids}
									a__     :		int		 =						model(__lowercase						)

									a__     :		List[Any]		 =						[input_ids, input_mask]
									a__     :		int		 =						model(__lowercase						)

									a__     :		Optional[int]		 =						model(__lowercase						)

									self.parent.assertEqual(result.last_hidden_state.shape      ,     (self.batch_size, self.seq_length, self.hidden_size)						)

						def 							SCREAMING_SNAKE_CASE__(     self      ,     __lowercase      ,     __lowercase      ,     __lowercase      ,     __lowercase      ,     __lowercase      ,     *__lowercase						)      ->   int:
									"""simple docstring"""




									a__     :		List[str]		 =						TFOpenAIGPTLMHeadModel(config=__lowercase						)
									a__     :		List[str]		 =						{"""input_ids""": input_ids, """attention_mask""": input_mask, """token_type_ids""": token_type_ids}
									a__     :		Tuple		 =						model(__lowercase						)
									self.parent.assertEqual(result.logits.shape      ,     (self.batch_size, self.seq_length, self.vocab_size)						)

						def 							SCREAMING_SNAKE_CASE__(     self      ,     __lowercase      ,     __lowercase      ,     __lowercase      ,     __lowercase      ,     __lowercase      ,     __lowercase      ,     *__lowercase						)      ->   Tuple:
									"""simple docstring"""




									a__     :		List[Any]		 =						TFOpenAIGPTDoubleHeadsModel(config=__lowercase						)

									a__     :		int		 =						tf.tile(tf.expand_dims(__lowercase      ,     1						)      ,     (1, self.num_choices, 1)						)
									a__     :		List[str]		 =						tf.tile(tf.expand_dims(__lowercase      ,     1						)      ,     (1, self.num_choices, 1)						)
									a__     :		Dict		 =						tf.tile(tf.expand_dims(__lowercase      ,     1						)      ,     (1, self.num_choices, 1)						)

									a__     :		Optional[Any]		 =						{
									    """input_ids""": multiple_choice_inputs_ids,
									    """mc_token_ids""": mc_token_ids,
									    """attention_mask""": multiple_choice_input_mask,
									    """token_type_ids""": multiple_choice_token_type_ids,
									}
									a__     :		Optional[Any]		 =						model(__lowercase						)
									self.parent.assertEqual(
									    result.logits.shape      ,     (self.batch_size, self.num_choices, self.seq_length, self.vocab_size)						)
									self.parent.assertEqual(result.mc_logits.shape      ,     (self.batch_size, self.num_choices)						)

						def 							SCREAMING_SNAKE_CASE__(     self      ,     __lowercase      ,     __lowercase      ,     __lowercase      ,     __lowercase      ,     __lowercase      ,     *__lowercase						)      ->   Dict:
									"""simple docstring"""




									a__     :		Tuple		 =						self.num_labels
									a__     :		Optional[int]		 =						ids_tensor([self.batch_size]      ,     self.type_sequence_label_size						)
									a__     :		Tuple		 =						{
									    """input_ids""": input_ids,
									    """attention_mask""": input_mask,
									    """token_type_ids""": token_type_ids,
									    """labels""": sequence_labels,
									}
									a__     :		Optional[Any]		 =						TFOpenAIGPTForSequenceClassification(__lowercase						)
									a__     :		Union[str, Any]		 =						model(__lowercase						)
									self.parent.assertEqual(result.logits.shape      ,     (self.batch_size, self.num_labels)						)

						def 							SCREAMING_SNAKE_CASE__(     self						)      ->   int:
									"""simple docstring"""




									a__     :		Optional[Any]		 =						self.prepare_config_and_inputs()

									(
									    (
									    a__
									) ,		(
									    a__
									) ,		(
									    a__
									) ,		(
									    a__
									) ,		(
									    a__
									) ,		(
									    a__
									) ,		(
									    a__
									) ,		(
									    a__
									) ,		(
									    a__
									) ,		
									)     :		Dict		 =						config_and_inputs

									a__     :		Union[str, Any]		 =						{"""input_ids""": input_ids, """token_type_ids""": token_type_ids, """attention_mask""": input_mask}
									return config, inputs_dict




@require_tf
class      snake_case__       (A__      ,			A__      ,			unittest.TestCase	):







						"""simple docstring"""




						__lowerCAmelCase  :Tuple			      =  (
						    (TFOpenAIGPTModel, TFOpenAIGPTLMHeadModel, TFOpenAIGPTDoubleHeadsModel, TFOpenAIGPTForSequenceClassification)
						    if is_tf_available()
						    else ()
						)
						__lowerCAmelCase  :str			      =  (
						    (TFOpenAIGPTLMHeadModel,) if is_tf_available() else ()
						)  # TODO (PVP): Add Double HeadsModel when generate() function is changed accordingly
						__lowerCAmelCase  :Union[str, Any]			      =  (
						    {
						        "feature-extraction": TFOpenAIGPTModel,
						        "text-classification": TFOpenAIGPTForSequenceClassification,
						        "text-generation": TFOpenAIGPTLMHeadModel,
						        "zero-shot": TFOpenAIGPTForSequenceClassification,
						    }
						    if is_tf_available()
						    else {}
						)
						__lowerCAmelCase  :Dict			      =  False
						__lowerCAmelCase  :Tuple			      =  False

						def 							SCREAMING_SNAKE_CASE__(     self      ,     __lowercase      ,     __lowercase      ,     __lowercase      ,     __lowercase      ,     __lowercase						)      ->   Optional[Any]:
									"""simple docstring"""




									if pipeline_test_casse_name == "ZeroShotClassificationPipelineTests":
												# Get `tokenizer does not have a padding token` error for both fast/slow tokenizers.
												# `OpenAIGPTConfig` was never used in pipeline tests, either because of a missing checkpoint or because a
												# tiny config could not be created.
												return True

									return False

						def 							SCREAMING_SNAKE_CASE__(     self						)      ->   int:
									"""simple docstring"""




									a__     :		Optional[Any]		 =						TFOpenAIGPTModelTester(self						)
									a__     :		List[Any]		 =						ConfigTester(self      ,     config_class=__lowercase      ,     n_embd=3_7						)

						def 							SCREAMING_SNAKE_CASE__(     self						)      ->   Optional[Any]:
									"""simple docstring"""




									self.config_tester.run_common_tests()

						def 							SCREAMING_SNAKE_CASE__(     self						)      ->   Union[str, Any]:
									"""simple docstring"""




									a__     :		Tuple		 =						self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_openai_gpt_model(*__lowercase						)

						def 							SCREAMING_SNAKE_CASE__(     self						)      ->   Optional[int]:
									"""simple docstring"""




									a__     :		Optional[Any]		 =						self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_openai_gpt_lm_head(*__lowercase						)

						def 							SCREAMING_SNAKE_CASE__(     self						)      ->   List[str]:
									"""simple docstring"""




									a__     :		int		 =						self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_openai_gpt_double_head(*__lowercase						)

						def 							SCREAMING_SNAKE_CASE__(     self						)      ->   Dict:
									"""simple docstring"""




									a__     :		Optional[Any]		 =						self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_openai_gpt_for_sequence_classification(*__lowercase						)

						@slow
						def 							SCREAMING_SNAKE_CASE__(     self						)      ->   Optional[Any]:
									"""simple docstring"""




									for model_name in TF_OPENAI_GPT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
												a__     :		int		 =						TFOpenAIGPTModel.from_pretrained(__lowercase						)
												self.assertIsNotNone(__lowercase						)




@require_tf
class      snake_case__       (unittest.TestCase	):







						"""simple docstring"""




						@slow
						def 							SCREAMING_SNAKE_CASE__(     self						)      ->   Dict:
									"""simple docstring"""




									a__     :		int		 =						TFOpenAIGPTLMHeadModel.from_pretrained("""openai-gpt"""						)
									a__     :		Tuple		 =						tf.convert_to_tensor([[4_8_1, 4_7_3_5, 5_4_4]]      ,     dtype=tf.intaa						)  # the president is
									a__     :		List[str]		 =						[
									    4_8_1,
									    4_7_3_5,
									    5_4_4,
									    2_4_6,
									    9_6_3,
									    8_7_0,
									    7_6_2,
									    2_3_9,
									    2_4_4,
									    4_0_4_7_7,
									    2_4_4,
									    2_4_9,
									    7_1_9,
									    8_8_1,
									    4_8_7,
									    5_4_4,
									    2_4_0,
									    2_4_4,
									    6_0_3,
									    4_8_1,
									]  # the president is a very good man. " \n " i\'m sure he is, " said the

									a__     :		Union[str, Any]		 =						model.generate(__lowercase      ,     do_sample=__lowercase						)
									self.assertListEqual(output_ids[0].numpy().tolist()      ,     __lowercase						)




