



import gc
import random
import unittest

import numpy as np
import torch
from transformers import (
    CLIPImageProcessor,
    CLIPTextConfig,
    CLIPTextModel,
    CLIPTokenizer,
    CLIPVisionConfig,
    CLIPVisionModelWithProjection,
)

from diffusers import AutoencoderKL, DDIMScheduler, DDPMScheduler, StableUnCLIPImgaImgPipeline, UNetaDConditionModel
from diffusers.pipelines.pipeline_utils import DiffusionPipeline
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    floats_tensor,
    load_image,
    load_numpy,
    require_torch_gpu,
    skip_mps,
    slow,
    torch_device,
)

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


enable_full_determinism()



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




		'''simple docstring'''



		snake_case_     =StableUnCLIPImgaImgPipeline
		snake_case_     =TEXT_GUIDED_IMAGE_VARIATION_PARAMS
		snake_case_     =TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
		snake_case_     =frozenset(
		    [])  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
		snake_case_     =frozenset([])



		def 		lowerCAmelCase__				(self			)					->      Any:

						"""simple docstring"""






						lowerCAmelCase__					:    List[str]												=					32
						lowerCAmelCase__					:    List[str]												=					embedder_hidden_size

						# image encoding components

						lowerCAmelCase__					:    Any												=					CLIPImageProcessor(crop_size=32		,size=32			)

						torch.manual_seed(0			)
						lowerCAmelCase__					:    Dict												=					CLIPVisionModelWithProjection(
						    CLIPVisionConfig(
						        hidden_size=__lowerCamelCase		,projection_dim=__lowerCamelCase		,num_hidden_layers=5		,num_attention_heads=4		,image_size=32		,intermediate_size=37		,patch_size=1		,)			)

						# regular denoising components

						torch.manual_seed(0			)
						lowerCAmelCase__					:    Dict												=					StableUnCLIPImageNormalizer(embedding_dim=__lowerCamelCase			)
						lowerCAmelCase__					:    Tuple												=					DDPMScheduler(beta_schedule='''squaredcos_cap_v2'''			)

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

						torch.manual_seed(0			)
						lowerCAmelCase__					:    Optional[Any]												=					CLIPTextModel(
						    CLIPTextConfig(
						        bos_token_id=0		,eos_token_id=2		,hidden_size=__lowerCamelCase		,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			)
						lowerCAmelCase__					:    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=__lowerCamelCase		,layers_per_block=1		,upcast_attention=__lowerCamelCase		,use_linear_projection=__lowerCamelCase		,)

						torch.manual_seed(0			)
						lowerCAmelCase__					:    Dict												=					DDIMScheduler(
						    beta_schedule='''scaled_linear'''		,beta_start=0.0_0085		,beta_end=0.012		,prediction_type='''v_prediction'''		,set_alpha_to_one=__lowerCamelCase		,steps_offset=1		,)

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

						lowerCAmelCase__					:    Optional[Any]												=					{
						    # image encoding components
						    '''feature_extractor''': feature_extractor,
						    '''image_encoder''': image_encoder.eval(),
						    # image noising components
						    '''image_normalizer''': image_normalizer.eval(),
						    '''image_noising_scheduler''': image_noising_scheduler,
						    # regular denoising components
						    '''tokenizer''': tokenizer,
						    '''text_encoder''': text_encoder.eval(),
						    '''unet''': unet.eval(),
						    '''scheduler''': scheduler,
						    '''vae''': vae.eval(),
						}

						return components



		def 		lowerCAmelCase__				(self		,__lowerCamelCase		,__lowerCamelCase=0		,__lowerCamelCase=True			)					->      Union[str, Any]:

						"""simple docstring"""






						if str(__lowerCamelCase			).startswith('''mps'''			):
										lowerCAmelCase__					:    Dict												=					torch.manual_seed(__lowerCamelCase			)
						else:
										lowerCAmelCase__					:    Optional[Any]												=					torch.Generator(device=__lowerCamelCase			).manual_seed(__lowerCamelCase			)

						lowerCAmelCase__					:    str												=					floats_tensor((1, 3, 32, 32)		,rng=random.Random(__lowerCamelCase			)			).to(__lowerCamelCase			)

						if pil_image:
										lowerCAmelCase__					:    Union[str, Any]												=					input_image * 0.5 + 0.5
										lowerCAmelCase__					:    List[str]												=					input_image.clamp(0		,1			)
										lowerCAmelCase__					:    Union[str, Any]												=					input_image.cpu().permute(0		,2		,3		,1			).float().numpy()
										lowerCAmelCase__					:    List[Any]												=					DiffusionPipeline.numpy_to_pil(__lowerCamelCase			)[0]

						return {
						    "prompt": "An anime racoon running a marathon",
						    "image": input_image,
						    "generator": generator,
						    "num_inference_steps": 2,
						    "output_type": "np",
						}



		@skip_mps
		def 		lowerCAmelCase__				(self			)					->      Any:

						"""simple docstring"""






						lowerCAmelCase__					:    Union[str, Any]												=					'''cpu'''  # ensure determinism for the device-dependent torch.Generator
						lowerCAmelCase__					:    Any												=					self.get_dummy_components()
						lowerCAmelCase__					:    List[str]												=					StableUnCLIPImgaImgPipeline(**__lowerCamelCase			)
						lowerCAmelCase__					:    Tuple												=					sd_pipe.to(__lowerCamelCase			)
						sd_pipe.set_progress_bar_config(disable=__lowerCamelCase			)

						lowerCAmelCase__					:    Optional[int]												=					self.get_dummy_inputs(__lowerCamelCase			)
						inputs.update({'''image_embeds''': None}			)
						lowerCAmelCase__					:    List[Any]												=					sd_pipe(**__lowerCamelCase			).images
						lowerCAmelCase__					:    List[str]												=					image[0, -3:, -3:, -1]

						assert image.shape == (1, 32, 32, 3)
						lowerCAmelCase__					:    Any												=					np.array([0.3872, 0.7224, 0.5601, 0.4741, 0.6872, 0.5814, 0.4636, 0.3867, 0.5078]			)

						assert np.abs(image_slice.flatten() - expected_slice			).max() < 1e-3



		def 		lowerCAmelCase__				(self			)					->      Union[str, Any]:

						"""simple docstring"""






						lowerCAmelCase__					:    Dict												=					torch_device in ['''cpu''', '''mps''']

						self._test_attention_slicing_forward_pass(test_max_difference=__lowerCamelCase			)



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

						"""simple docstring"""






						lowerCAmelCase__					:    Any												=					torch_device in ['''cpu''', '''mps''']

						self._test_inference_batch_single_identical(test_max_difference=__lowerCamelCase			)



		@unittest.skipIf(
		    torch_device != '''cuda''' or not is_xformers_available()		,reason='''XFormers attention is only available with CUDA and `xformers` installed'''		,)
		def 		lowerCAmelCase__				(self			)					->      Dict:

						"""simple docstring"""






						self._test_xformers_attention_forwardGenerator_pass(test_max_difference=__lowerCamelCase			)



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




		'''simple docstring'''



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

						"""simple docstring"""






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



		def 		lowerCAmelCase__				(self			)					->      Dict:

						"""simple docstring"""






						lowerCAmelCase__					:    List[Any]												=					load_image(
						    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png'''			)

						lowerCAmelCase__					:    Dict												=					load_numpy(
						    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_img2img_anime_turtle_fp16.npy'''			)

						lowerCAmelCase__					:    Optional[int]												=					StableUnCLIPImgaImgPipeline.from_pretrained(
						    '''fusing/stable-unclip-2-1-l-img2img'''		,torch_dtype=torch.floataa			)
						pipe.to(__lowerCamelCase			)
						pipe.set_progress_bar_config(disable=__lowerCamelCase			)
						# 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()

						lowerCAmelCase__					:    Any												=					torch.Generator(device='''cpu'''			).manual_seed(0			)
						lowerCAmelCase__					:    Any												=					pipe(__lowerCamelCase		,'''anime turle'''		,generator=__lowerCamelCase		,output_type='''np'''			)

						lowerCAmelCase__					:    Tuple												=					output.images[0]

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

						assert_mean_pixel_difference(__lowerCamelCase		,__lowerCamelCase			)



		def 		lowerCAmelCase__				(self			)					->      Any:

						"""simple docstring"""






						lowerCAmelCase__					:    Dict												=					load_image(
						    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png'''			)

						lowerCAmelCase__					:    List[str]												=					load_numpy(
						    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_h_img2img_anime_turtle_fp16.npy'''			)

						lowerCAmelCase__					:    int												=					StableUnCLIPImgaImgPipeline.from_pretrained(
						    '''fusing/stable-unclip-2-1-h-img2img'''		,torch_dtype=torch.floataa			)
						pipe.to(__lowerCamelCase			)
						pipe.set_progress_bar_config(disable=__lowerCamelCase			)
						# 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()

						lowerCAmelCase__					:    Optional[int]												=					torch.Generator(device='''cpu'''			).manual_seed(0			)
						lowerCAmelCase__					:    str												=					pipe(__lowerCamelCase		,'''anime turle'''		,generator=__lowerCamelCase		,output_type='''np'''			)

						lowerCAmelCase__					:    Any												=					output.images[0]

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

						assert_mean_pixel_difference(__lowerCamelCase		,__lowerCamelCase			)



		def 		lowerCAmelCase__				(self			)					->      Tuple:

						"""simple docstring"""






						lowerCAmelCase__					:    Any												=					load_image(
						    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png'''			)

						torch.cuda.empty_cache()
						torch.cuda.reset_max_memory_allocated()
						torch.cuda.reset_peak_memory_stats()

						lowerCAmelCase__					:    str												=					StableUnCLIPImgaImgPipeline.from_pretrained(
						    '''fusing/stable-unclip-2-1-h-img2img'''		,torch_dtype=torch.floataa			)
						lowerCAmelCase__					:    List[Any]												=					pipe.to(__lowerCamelCase			)
						pipe.set_progress_bar_config(disable=__lowerCamelCase			)
						pipe.enable_attention_slicing()
						pipe.enable_sequential_cpu_offload()

						lowerCAmelCase__					:    Tuple												=					pipe(
						    __lowerCamelCase		,'''anime turtle'''		,num_inference_steps=2		,output_type='''np'''		,)

						lowerCAmelCase__					:    Any												=					torch.cuda.max_memory_allocated()
						# make sure that less than 7 GB is allocated
						assert mem_bytes < 7 * 10**9






from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_flax_available,
    is_tf_available,
    is_torch_available,
)


__snake_case    :		Optional[int]  ={
    'configuration_vision_text_dual_encoder': ['VisionTextDualEncoderConfig'],
    'processing_vision_text_dual_encoder': ['VisionTextDualEncoderProcessor'],
}


try:
							if not is_torch_available():
														raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
							pass
else:
							__snake_case    :		int  =['VisionTextDualEncoderModel']


try:
							if not is_flax_available():
														raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
							pass
else:
							__snake_case    :		List[str]  =['FlaxVisionTextDualEncoderModel']

try:
							if not is_tf_available():
														raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
							pass
else:
							__snake_case    :		Tuple  =['TFVisionTextDualEncoderModel']


if TYPE_CHECKING:
							from .configuration_vision_text_dual_encoder import VisionTextDualEncoderConfig
							from .processing_vision_text_dual_encoder import VisionTextDualEncoderProcessor

							try:
														if not is_torch_available():
																					raise OptionalDependencyNotAvailable()
							except OptionalDependencyNotAvailable:
														pass
							else:
														from .modeling_vision_text_dual_encoder import VisionTextDualEncoderModel

							try:
														if not is_flax_available():
																					raise OptionalDependencyNotAvailable()
							except OptionalDependencyNotAvailable:
														pass
							else:
														from .modeling_flax_vision_text_dual_encoder import FlaxVisionTextDualEncoderModel

							try:
														if not is_tf_available():
																					raise OptionalDependencyNotAvailable()
							except OptionalDependencyNotAvailable:
														pass
							else:
														from .modeling_tf_vision_text_dual_encoder import TFVisionTextDualEncoderModel


else:
							import sys

							__snake_case    :		str  =_LazyModule(__name__, globals()['__file__'], _import_structure)

