




'''simple docstring'''


import gc
import random
import unittest

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

from diffusers import (
    AutoencoderKL,
    DiffusionPipeline,
    EulerDiscreteScheduler,
    StableDiffusionXLImgaImgPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import floats_tensor, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import (
    IMAGE_TO_IMAGE_IMAGE_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()
class 						a_				(			UpperCAmelCase__     ,						UpperCAmelCase__     ,						unittest.TestCase						):
	lowercase_       :     Union[str, Any]			  =      StableDiffusionXLImgaImgPipeline
	lowercase_       :     Tuple			  =      TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'''height''', '''width'''}
	lowercase_       :     Tuple			  =      PipelineTesterMixin.required_optional_params - {'''latents'''}
	lowercase_       :     str			  =      TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
	lowercase_       :     int			  =      IMAGE_TO_IMAGE_IMAGE_PARAMS
	lowercase_       :     Tuple			  =      IMAGE_TO_IMAGE_IMAGE_PARAMS




	def   lowercase__   (       self     :  List[str]      ):
							torch.manual_seed(0      )
							__snake_case						 =   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')       ,		attention_head_dim=(2, 4)       ,		use_linear_projection=__lowerCAmelCase       ,		addition_embed_type='text_time'       ,		addition_time_embed_dim=8       ,		transformer_layers_per_block=(1, 2)       ,		projection_class_embeddings_input_dim=8_0       ,		cross_attention_dim=6_4       ,		)
							__snake_case						 =   EulerDiscreteScheduler(
							    beta_start=0.00085       ,		beta_end=0.012       ,		steps_offset=1       ,		beta_schedule='scaled_linear'       ,		timestep_spacing='leading'       ,		)
							torch.manual_seed(0      )
							__snake_case						 =   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      )
							__snake_case						 =   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=3_2       ,		)
							__snake_case						 =   CLIPTextModel(__lowerCAmelCase      )
							__snake_case						 =   CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'       ,		local_files_only=__lowerCAmelCase      )

							__snake_case						 =   CLIPTextModelWithProjection(__lowerCAmelCase      )
							__snake_case						 =   CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'       ,		local_files_only=__lowerCAmelCase      )

							__snake_case						 =   {
							    'unet': unet,
							    'scheduler': scheduler,
							    'vae': vae,
							    'text_encoder': text_encoder,
							    'tokenizer': tokenizer,
							    'text_encoder_2': text_encoder_a,
							    'tokenizer_2': tokenizer_a,
							    # "safety_checker": None,
							    # "feature_extractor": None,
							}
							return components




	def   lowercase__   (       self     :  Optional[int]       ,		__lowerCAmelCase     :  List[str]       ,		__lowerCAmelCase     :  str=0      ):
							__snake_case						 =   floats_tensor((1, 3, 3_2, 3_2)       ,		rng=random.Random(__lowerCAmelCase      )      ).to(__lowerCAmelCase      )
							__snake_case						 =   image / 2 + 0.5
							if str(__lowerCAmelCase      ).startswith('mps'      ):
													__snake_case						 =   torch.manual_seed(__lowerCAmelCase      )
							else:
													__snake_case						 =   torch.Generator(device=__lowerCAmelCase      ).manual_seed(__lowerCAmelCase      )
							__snake_case						 =   {
							    'prompt': 'A painting of a squirrel eating a burger',
							    'image': image,
							    'generator': generator,
							    'num_inference_steps': 2,
							    'guidance_scale': 5.0,
							    'output_type': 'numpy',
							    'strength': 0.75,
							}
							return inputs




	def   lowercase__   (       self     :  Optional[Any]      ):
							__snake_case						 =   'cpu'  # ensure determinism for the device-dependent torch.Generator
							__snake_case						 =   self.get_dummy_components()
							__snake_case						 =   StableDiffusionXLImgaImgPipeline(**__lowerCAmelCase      )
							__snake_case						 =   sd_pipe.to(__lowerCAmelCase      )
							sd_pipe.set_progress_bar_config(disable=__lowerCAmelCase      )

							__snake_case						 =   self.get_dummy_inputs(__lowerCAmelCase      )
							__snake_case						 =   sd_pipe(**__lowerCAmelCase      ).images
							__snake_case						 =   image[0, -3:, -3:, -1]

							assert image.shape == (1, 3_2, 3_2, 3)

							__snake_case						 =   np.array([0.4656, 0.4840, 0.4439, 0.6698, 0.5574, 0.4524, 0.5799, 0.5943, 0.5165]      )

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




	def   lowercase__   (       self     :  Dict      ):
							super().test_attention_slicing_forward_pass(expected_max_diff=3E-3      )




	def   lowercase__   (       self     :  Dict      ):
							super().test_inference_batch_single_identical(expected_max_diff=3E-3      )




	def   lowercase__   (       self     :  str      ):
							pass




	def   lowercase__   (       self     :  Optional[Any]      ):
							__snake_case						 =   self.get_dummy_components()
							__snake_case						 =   StableDiffusionXLImgaImgPipeline(**__lowerCAmelCase      )
							__snake_case						 =   sd_pipe.to(__lowerCAmelCase      )
							__snake_case						 =   sd_pipe.to(__lowerCAmelCase      )
							sd_pipe.set_progress_bar_config(disable=__lowerCAmelCase      )

							# forward without prompt embeds
							__snake_case						 =   self.get_dummy_inputs(__lowerCAmelCase      )
							__snake_case						 =   3 * ['this is a negative prompt']
							__snake_case						 =   negative_prompt
							__snake_case						 =   3 * [inputs['prompt']]

							__snake_case						 =   sd_pipe(**__lowerCAmelCase      )
							__snake_case						 =   output.images[0, -3:, -3:, -1]

							# forward with prompt embeds
							__snake_case						 =   self.get_dummy_inputs(__lowerCAmelCase      )
							__snake_case						 =   3 * ['this is a negative prompt']
							__snake_case						 =   3 * [inputs.pop('prompt'      )]

							(
							    (
							    __snake_case
							) ,       (
							    __snake_case
							) ,       (
							    __snake_case
							) ,       (
							    __snake_case
							) ,       
							)						 =   sd_pipe.encode_prompt(__lowerCAmelCase       ,		negative_prompt=__lowerCAmelCase      )

							__snake_case						 =   sd_pipe(
							    **__lowerCAmelCase       ,		prompt_embeds=__lowerCAmelCase       ,		negative_prompt_embeds=__lowerCAmelCase       ,		pooled_prompt_embeds=__lowerCAmelCase       ,		negative_pooled_prompt_embeds=__lowerCAmelCase       ,		)
							__snake_case						 =   output.images[0, -3:, -3:, -1]

							# make sure that it's equal
							assert np.abs(image_slice_a.flatten() - image_slice_a.flatten()      ).max() < 1E-4





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




	def   lowercase__   (       self     :  Optional[int]      ):
							super().tearDown()
							gc.collect()
							torch.cuda.empty_cache()




	def   lowercase__   (       self     :  Union[str, Any]       ,		__lowerCAmelCase     :  Tuple       ,		__lowerCAmelCase     :  Optional[int]="cpu"       ,		__lowerCAmelCase     :  Union[str, Any]=torch.floataa       ,		__lowerCAmelCase     :  Any=0      ):
							__snake_case						 =   torch.Generator(device=__lowerCAmelCase      ).manual_seed(__lowerCAmelCase      )
							__snake_case						 =   np.random.RandomState(__lowerCAmelCase      ).standard_normal((1, 4, 6_4, 6_4)      )
							__snake_case						 =   torch.from_numpy(__lowerCAmelCase      ).to(device=__lowerCAmelCase       ,		dtype=__lowerCAmelCase      )
							__snake_case						 =   {
							    'prompt': 'a photograph of an astronaut riding a horse',
							    'latents': latents,
							    'generator': generator,
							    'num_inference_steps': 3,
							    'guidance_scale': 7.5,
							    'output_type': 'numpy',
							}
							return inputs




	def   lowercase__   (       self     :  Dict      ):
							__snake_case						 =   DiffusionPipeline.from_pretrained('stabilityai/stable-diffusion-2-base'      )
							pipe.to(__lowerCAmelCase      )
							pipe.set_progress_bar_config(disable=__lowerCAmelCase      )

							__snake_case						 =   self.get_inputs(__lowerCAmelCase      )
							__snake_case						 =   pipe(**__lowerCAmelCase      ).images
							__snake_case						 =   image[0, -3:, -3:, -1].flatten()

							assert image.shape == (1, 5_1_2, 5_1_2, 3)
							__snake_case						 =   np.array([0.49493, 0.47896, 0.40798, 0.54214, 0.53212, 0.48202, 0.47656, 0.46329, 0.48506]      )
							assert np.abs(image_slice - expected_slice      ).max() < 7E-3







'''simple docstring'''


import unittest

from transformers import is_torch_available
from transformers.testing_utils import require_torch, slow, torch_device

from ...generation.test_utils import GenerationTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
	import torch

	from transformers import (
	    OPENAI_GPT_PRETRAINED_MODEL_ARCHIVE_LIST,
	    OpenAIGPTConfig,
	    OpenAIGPTDoubleHeadsModel,
	    OpenAIGPTForSequenceClassification,
	    OpenAIGPTLMHeadModel,
	    OpenAIGPTModel,
	)
class 						a_				:




	def __init__(       self     :  Optional[Any]       ,		__lowerCAmelCase     :  Optional[Any]       ,		__lowerCAmelCase     :  Optional[int]=1_3       ,		__lowerCAmelCase     :  Tuple=7       ,		__lowerCAmelCase     :  Optional[int]=True       ,		__lowerCAmelCase     :  Any=True       ,		__lowerCAmelCase     :  List[Any]=True       ,		__lowerCAmelCase     :  Union[str, Any]=9_9       ,		__lowerCAmelCase     :  Any=3_2       ,		__lowerCAmelCase     :  Tuple=5       ,		__lowerCAmelCase     :  int=4       ,		__lowerCAmelCase     :  List[str]=3_7       ,		__lowerCAmelCase     :  List[Any]="gelu"       ,		__lowerCAmelCase     :  Union[str, Any]=0.1       ,		__lowerCAmelCase     :  int=0.1       ,		__lowerCAmelCase     :  Any=5_1_2       ,		__lowerCAmelCase     :  Dict=1_6       ,		__lowerCAmelCase     :  Dict=2       ,		__lowerCAmelCase     :  Union[str, Any]=0.02       ,		__lowerCAmelCase     :  Optional[Any]=3       ,		__lowerCAmelCase     :  str=4       ,		__lowerCAmelCase     :  List[str]=None       ,		):
							__snake_case						 =   parent
							__snake_case						 =   batch_size
							__snake_case						 =   seq_length
							__snake_case						 =   is_training
							__snake_case						 =   use_token_type_ids
							__snake_case						 =   use_labels
							__snake_case						 =   vocab_size
							__snake_case						 =   hidden_size
							__snake_case						 =   num_hidden_layers
							__snake_case						 =   num_attention_heads
							__snake_case						 =   intermediate_size
							__snake_case						 =   hidden_act
							__snake_case						 =   hidden_dropout_prob
							__snake_case						 =   attention_probs_dropout_prob
							__snake_case						 =   max_position_embeddings
							__snake_case						 =   type_vocab_size
							__snake_case						 =   type_sequence_label_size
							__snake_case						 =   initializer_range
							__snake_case						 =   num_labels
							__snake_case						 =   num_choices
							__snake_case						 =   scope
							__snake_case						 =   self.vocab_size - 1




	def   lowercase__   (       self     :  Union[str, Any]      ):
							__snake_case						 =   ids_tensor([self.batch_size, self.seq_length]       ,		self.vocab_size      )

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

							__snake_case						 =   None
							__snake_case						 =   None
							__snake_case						 =   None
							if self.use_labels:
													__snake_case						 =   ids_tensor([self.batch_size]       ,		self.type_sequence_label_size      )
													__snake_case						 =   ids_tensor([self.batch_size, self.seq_length]       ,		self.num_labels      )
													__snake_case						 =   ids_tensor([self.batch_size]       ,		self.num_choices      )

							__snake_case						 =   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       ,		)

							__snake_case						 =   ids_tensor([self.num_hidden_layers, self.num_attention_heads]       ,		2      )

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




	def   lowercase__   (       self     :  List[str]       ,		__lowerCAmelCase     :  Tuple       ,		__lowerCAmelCase     :  Optional[Any]       ,		__lowerCAmelCase     :  Any       ,		__lowerCAmelCase     :  Union[str, Any]       ,		*__lowerCAmelCase     :  List[str]      ):
							__snake_case						 =   OpenAIGPTModel(config=__lowerCAmelCase      )
							model.to(__lowerCAmelCase      )
							model.eval()

							__snake_case						 =   model(__lowerCAmelCase       ,		token_type_ids=__lowerCAmelCase       ,		head_mask=__lowerCAmelCase      )
							__snake_case						 =   model(__lowerCAmelCase       ,		token_type_ids=__lowerCAmelCase      )
							__snake_case						 =   model(__lowerCAmelCase      )

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




	def   lowercase__   (       self     :  Union[str, Any]       ,		__lowerCAmelCase     :  int       ,		__lowerCAmelCase     :  Dict       ,		__lowerCAmelCase     :  Dict       ,		__lowerCAmelCase     :  Dict       ,		*__lowerCAmelCase     :  str      ):
							__snake_case						 =   OpenAIGPTLMHeadModel(__lowerCAmelCase      )
							model.to(__lowerCAmelCase      )
							model.eval()

							__snake_case						 =   model(__lowerCAmelCase       ,		token_type_ids=__lowerCAmelCase       ,		labels=__lowerCAmelCase      )
							self.parent.assertEqual(result.loss.shape       ,		()      )
							self.parent.assertEqual(result.logits.shape       ,		(self.batch_size, self.seq_length, self.vocab_size)      )




	def   lowercase__   (       self     :  Optional[Any]       ,		__lowerCAmelCase     :  str       ,		__lowerCAmelCase     :  Any       ,		__lowerCAmelCase     :  int       ,		__lowerCAmelCase     :  List[Any]       ,		*__lowerCAmelCase     :  Tuple      ):
							__snake_case						 =   OpenAIGPTDoubleHeadsModel(__lowerCAmelCase      )
							model.to(__lowerCAmelCase      )
							model.eval()

							__snake_case						 =   model(__lowerCAmelCase       ,		token_type_ids=__lowerCAmelCase       ,		labels=__lowerCAmelCase      )
							self.parent.assertEqual(result.loss.shape       ,		()      )
							self.parent.assertEqual(result.logits.shape       ,		(self.batch_size, self.seq_length, self.vocab_size)      )




	def   lowercase__   (       self     :  Optional[int]       ,		__lowerCAmelCase     :  Union[str, Any]       ,		__lowerCAmelCase     :  int       ,		__lowerCAmelCase     :  Tuple       ,		__lowerCAmelCase     :  int       ,		*__lowerCAmelCase     :  int      ):
							__snake_case						 =   self.num_labels
							__snake_case						 =   OpenAIGPTForSequenceClassification(__lowerCAmelCase      )
							model.to(__lowerCAmelCase      )
							model.eval()

							__snake_case						 =   ids_tensor([self.batch_size]       ,		self.type_sequence_label_size      )
							__snake_case						 =   model(__lowerCAmelCase       ,		token_type_ids=__lowerCAmelCase       ,		labels=__lowerCAmelCase      )
							self.parent.assertEqual(result.logits.shape       ,		(self.batch_size, self.num_labels)      )




	def   lowercase__   (       self     :  List[str]      ):
							__snake_case						 =   self.prepare_config_and_inputs()
							(
							    (
							    __snake_case
							) ,       (
							    __snake_case
							) ,       (
							    __snake_case
							) ,       (
							    __snake_case
							) ,       (
							    __snake_case
							) ,       (
							    __snake_case
							) ,       (
							    __snake_case
							) ,       
							)						 =   config_and_inputs
							__snake_case						 =   {
							    'input_ids': input_ids,
							    'token_type_ids': token_type_ids,
							    'head_mask': head_mask,
							}

							return config, inputs_dict





@require_torch
class 						a_				(			UpperCAmelCase__     ,						UpperCAmelCase__     ,						UpperCAmelCase__     ,						unittest.TestCase						):
	lowercase_       :     List[str]			  =      (
	    (OpenAIGPTModel, OpenAIGPTLMHeadModel, OpenAIGPTDoubleHeadsModel, OpenAIGPTForSequenceClassification)
	    if is_torch_available()
	    else ()
	)
	lowercase_       :     Union[str, Any]			  =      (
	    (OpenAIGPTLMHeadModel,) if is_torch_available() else ()
	)  # TODO (PVP): Add Double HeadsModel when generate() function is changed accordingly
	lowercase_       :     str			  =      (
	    {
	        '''feature-extraction''': OpenAIGPTModel,
	        '''text-classification''': OpenAIGPTForSequenceClassification,
	        '''text-generation''': OpenAIGPTLMHeadModel,
	        '''zero-shot''': OpenAIGPTForSequenceClassification,
	    }
	    if is_torch_available()
	    else {}
	)




	def   lowercase__   (       self     :  Optional[Any]       ,		__lowerCAmelCase     :  str       ,		__lowerCAmelCase     :  int       ,		__lowerCAmelCase     :  str       ,		__lowerCAmelCase     :  int       ,		__lowerCAmelCase     :  Any      ):
							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   lowercase__   (       self     :  Tuple       ,		__lowerCAmelCase     :  Union[str, Any]       ,		__lowerCAmelCase     :  List[str]       ,		__lowerCAmelCase     :  List[str]=False      ):
							__snake_case						 =   super()._prepare_for_class(__lowerCAmelCase       ,		__lowerCAmelCase       ,		return_labels=__lowerCAmelCase      )

							if return_labels:
													if model_class.__name__ == "OpenAIGPTDoubleHeadsModel":
																			__snake_case						 =   torch.zeros(
																			    (self.model_tester.batch_size, self.model_tester.num_choices, self.model_tester.seq_length)       ,		dtype=torch.long       ,		device=__lowerCAmelCase       ,		)
																			__snake_case						 =   inputs_dict['labels']
																			__snake_case						 =   inputs_dict['labels']
																			__snake_case						 =   torch.zeros(
																			    (self.model_tester.batch_size, self.model_tester.num_choices)       ,		dtype=torch.long       ,		device=__lowerCAmelCase       ,		)
																			__snake_case						 =   torch.zeros(
																			    self.model_tester.batch_size       ,		dtype=torch.long       ,		device=__lowerCAmelCase      )
							return inputs_dict




	def   lowercase__   (       self     :  Tuple      ):
							__snake_case						 =   OpenAIGPTModelTester(self      )
							__snake_case						 =   ConfigTester(self       ,		config_class=__lowerCAmelCase       ,		n_embd=3_7      )




	def   lowercase__   (       self     :  int      ):
							self.config_tester.run_common_tests()




	def   lowercase__   (       self     :  Optional[Any]      ):
							__snake_case						 =   self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_openai_gpt_model(*__lowerCAmelCase      )




	def   lowercase__   (       self     :  int      ):
							__snake_case						 =   self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_lm_head_model(*__lowerCAmelCase      )




	def   lowercase__   (       self     :  Any      ):
							__snake_case						 =   self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_double_lm_head_model(*__lowerCAmelCase      )




	def   lowercase__   (       self     :  str      ):
							__snake_case						 =   self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_openai_gpt_for_sequence_classification(*__lowerCAmelCase      )




	@slow
	def   lowercase__   (       self     :  str      ):
							for model_name in OPENAI_GPT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
													__snake_case						 =   OpenAIGPTModel.from_pretrained(__lowerCAmelCase      )
													self.assertIsNotNone(__lowerCAmelCase      )





@require_torch
class 						a_				(			unittest.TestCase						):




	@slow
	def   lowercase__   (       self     :  int      ):
							__snake_case						 =   OpenAIGPTLMHeadModel.from_pretrained('openai-gpt'      )
							model.to(__lowerCAmelCase      )
							__snake_case						 =   torch.tensor([[4_8_1, 4_7_3_5, 5_4_4]]       ,		dtype=torch.long       ,		device=__lowerCAmelCase      )  # the president is
							__snake_case						 =   [
							    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

							__snake_case						 =   model.generate(__lowerCAmelCase       ,		do_sample=__lowerCAmelCase      )
							self.assertListEqual(output_ids[0].tolist()       ,		__lowerCAmelCase      )

