





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        __SCREAMING_SNAKE_CASE					(     UpperCamelCase_   ,		UpperCamelCase_   ,		UpperCamelCase_   ,		unittest.TestCase       ):
							snake_case     :      Optional[int]						     =       StableUnCLIPImgaImgPipeline
							snake_case     :      Union[str, Any]						     =       TEXT_GUIDED_IMAGE_VARIATION_PARAMS
							snake_case     :      str						     =       TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
							snake_case     :      Dict						     =       frozenset(
							    []       )  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
							snake_case     :      Union[str, Any]						     =       frozenset([]       )






							def 				_lowerCamelCase		(						self  ):
														UpperCamelCase__				     =			32
														UpperCamelCase__				     =			embedder_hidden_size

														# image encoding components

														UpperCamelCase__				     =			CLIPImageProcessor(crop_size=32  , size=32  )

														torch.manual_seed(0  )
														UpperCamelCase__				     =			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  )
														UpperCamelCase__				     =			StableUnCLIPImageNormalizer(embedding_dim=__lowerCAmelCase  )
														UpperCamelCase__				     =			DDPMScheduler(beta_schedule="""squaredcos_cap_v2"""  )

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

														torch.manual_seed(0  )
														UpperCamelCase__				     =			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=1000  , )  )

														torch.manual_seed(0  )
														UpperCamelCase__				     =			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  )
														UpperCamelCase__				     =			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  )
														UpperCamelCase__				     =			AutoencoderKL()

														UpperCamelCase__				     =			{
														    # 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  ):
														if str(__lowerCAmelCase  ).startswith("""mps"""  ):
																					UpperCamelCase__				     =			torch.manual_seed(__lowerCAmelCase  )
														else:
																					UpperCamelCase__				     =			torch.Generator(device=__lowerCAmelCase  ).manual_seed(__lowerCAmelCase  )

														UpperCamelCase__				     =			floats_tensor((1, 3, 32, 32)  , rng=random.Random(__lowerCAmelCase  )  ).to(__lowerCAmelCase  )

														if pil_image:
																					UpperCamelCase__				     =			input_image * 0.5 + 0.5
																					UpperCamelCase__				     =			input_image.clamp(0  , 1  )
																					UpperCamelCase__				     =			input_image.cpu().permute(0  , 2  , 3  , 1  ).float().numpy()
																					UpperCamelCase__				     =			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  ):
														UpperCamelCase__				     =			'''cpu'''  # ensure determinism for the device-dependent torch.Generator
														UpperCamelCase__				     =			self.get_dummy_components()
														UpperCamelCase__				     =			StableUnCLIPImgaImgPipeline(**__lowerCAmelCase  )
														UpperCamelCase__				     =			sd_pipe.to(__lowerCAmelCase  )
														sd_pipe.set_progress_bar_config(disable=__lowerCAmelCase  )

														UpperCamelCase__				     =			self.get_dummy_inputs(__lowerCAmelCase  )
														inputs.update({"""image_embeds""": None}  )
														UpperCamelCase__				     =			sd_pipe(**__lowerCAmelCase  ).images
														UpperCamelCase__				     =			image[0, -3:, -3:, -1]

														assert image.shape == (1, 32, 32, 3)
														UpperCamelCase__				     =			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  ):
														UpperCamelCase__				     =			torch_device in ['''cpu''', '''mps''']

														self._test_attention_slicing_forward_pass(test_max_difference=__lowerCAmelCase  )






							def 				_lowerCamelCase		(						self  ):
														UpperCamelCase__				     =			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  ):
														self._test_xformers_attention_forwardGenerator_pass(test_max_difference=__lowerCAmelCase  )






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






							def 				_lowerCamelCase		(						self  ):
														# clean up the VRAM after each test
														super().tearDown()
														gc.collect()
														torch.cuda.empty_cache()






							def 				_lowerCamelCase		(						self  ):
														UpperCamelCase__				     =			load_image(
														    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png"""  )

														UpperCamelCase__				     =			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"""  )

														UpperCamelCase__				     =			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()

														UpperCamelCase__				     =			torch.Generator(device="""cpu"""  ).manual_seed(0  )
														UpperCamelCase__				     =			pipe(__lowerCAmelCase  , """anime turle"""  , generator=__lowerCAmelCase  , output_type="""np"""  )

														UpperCamelCase__				     =			output.images[0]

														assert image.shape == (768, 768, 3)

														assert_mean_pixel_difference(__lowerCAmelCase  , __lowerCAmelCase  )






							def 				_lowerCamelCase		(						self  ):
														UpperCamelCase__				     =			load_image(
														    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png"""  )

														UpperCamelCase__				     =			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"""  )

														UpperCamelCase__				     =			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()

														UpperCamelCase__				     =			torch.Generator(device="""cpu"""  ).manual_seed(0  )
														UpperCamelCase__				     =			pipe(__lowerCAmelCase  , """anime turle"""  , generator=__lowerCAmelCase  , output_type="""np"""  )

														UpperCamelCase__				     =			output.images[0]

														assert image.shape == (768, 768, 3)

														assert_mean_pixel_difference(__lowerCAmelCase  , __lowerCAmelCase  )







							def 				_lowerCamelCase		(						self  ):
														UpperCamelCase__				     =			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()

														UpperCamelCase__				     =			StableUnCLIPImgaImgPipeline.from_pretrained(
														    """fusing/stable-unclip-2-1-h-img2img"""  , torch_dtype=torch.floataa  )
														UpperCamelCase__				     =			pipe.to(__lowerCAmelCase  )
														pipe.set_progress_bar_config(disable=__lowerCAmelCase  )
														pipe.enable_attention_slicing()
														pipe.enable_sequential_cpu_offload()

														UpperCamelCase__				     =			pipe(
														    __lowerCAmelCase  , """anime turtle"""  , num_inference_steps=2  , output_type="""np"""  , )

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









import inspect
import unittest

class        __SCREAMING_SNAKE_CASE					(     unittest.TestCase       ):






							def 				_lowerCamelCase		(						self  ):
														try:
																					import diffusers  # noqa: F401
														except ImportError:
																					assert False







							def 				_lowerCamelCase		(						self  ):
														import diffusers
														from diffusers.dependency_versions_table import deps

														UpperCamelCase__				     =			inspect.getmembers(__lowerCAmelCase  , inspect.isclass  )

														for cls_name, cls_module in all_classes:
																					if "dummy_" in cls_module.__module__:
																												for backend in cls_module._backends:
																																			if backend == "k_diffusion":
																																										UpperCamelCase__				     =			"""k-diffusion"""
																																			elif backend == "invisible_watermark":
																																										UpperCamelCase__				     =			"""invisible-watermark"""
																																			assert backend in deps, f"""{backend} is not in the deps table!"""




