



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       ):
				__lowerCamelCase =				StableUnCLIPImgaImgPipeline
				__lowerCamelCase =				TEXT_GUIDED_IMAGE_VARIATION_PARAMS
				__lowerCamelCase =				TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
				__lowerCamelCase =				frozenset(
				    []       )  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
				__lowerCamelCase =				frozenset([]       )




				def 							snake_case   (							self		):



									"""simple docstring"""





									_lowerCAmelCase						=    32
									_lowerCAmelCase						=    embedder_hidden_size

									# image encoding components

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

									torch.manual_seed(0		)
									_lowerCAmelCase						=    CLIPVisionModelWithProjection(
									    CLIPVisionConfig(
									        hidden_size=_snake_case						,				projection_dim=_snake_case						,				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						=    StableUnCLIPImageNormalizer(embedding_dim=_snake_case		)
									_lowerCAmelCase						=    DDPMScheduler(beta_schedule="""squaredcos_cap_v2"""		)

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

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

									torch.manual_seed(0		)
									_lowerCAmelCase						=    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		)
									_lowerCAmelCase						=    DDIMScheduler(
									    beta_schedule="""scaled_linear"""						,				beta_start=0.0_0085						,				beta_end=0.012						,				prediction_type="""v_prediction"""						,				set_alpha_to_one=_snake_case						,				steps_offset=1						,				)

									torch.manual_seed(0		)
									_lowerCAmelCase						=    AutoencoderKL()

									_lowerCAmelCase						=    {
									    # 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 							snake_case   (							self						,				_snake_case						,				_snake_case=0						,				_snake_case=True		):



									"""simple docstring"""





									if str(_snake_case		).startswith("""mps"""		):
														_lowerCAmelCase						=    torch.manual_seed(_snake_case		)
									else:
														_lowerCAmelCase						=    torch.Generator(device=_snake_case		).manual_seed(_snake_case		)

									_lowerCAmelCase						=    floats_tensor((1, 3, 32, 32)						,				rng=random.Random(_snake_case		)		).to(_snake_case		)

									if pil_image:
														_lowerCAmelCase						=    input_image * 0.5 + 0.5
														_lowerCAmelCase						=    input_image.clamp(0						,				1		)
														_lowerCAmelCase						=    input_image.cpu().permute(0						,				2						,				3						,				1		).float().numpy()
														_lowerCAmelCase						=    DiffusionPipeline.numpy_to_pil(_snake_case		)[0]

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




				@skip_mps
				def 							snake_case   (							self		):



									"""simple docstring"""





									_lowerCAmelCase						=    """cpu"""  # ensure determinism for the device-dependent torch.Generator
									_lowerCAmelCase						=    self.get_dummy_components()
									_lowerCAmelCase						=    StableUnCLIPImgaImgPipeline(**_snake_case		)
									_lowerCAmelCase						=    sd_pipe.to(_snake_case		)
									sd_pipe.set_progress_bar_config(disable=_snake_case		)

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

									assert image.shape == (1, 32, 32, 3)
									_lowerCAmelCase						=    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 							snake_case   (							self		):



									"""simple docstring"""





									_lowerCAmelCase						=    torch_device in ["""cpu""", """mps"""]

									self._test_attention_slicing_forward_pass(test_max_difference=_snake_case		)




				def 							snake_case   (							self		):



									"""simple docstring"""





									_lowerCAmelCase						=    torch_device in ["""cpu""", """mps"""]

									self._test_inference_batch_single_identical(test_max_difference=_snake_case		)




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



									"""simple docstring"""





									self._test_xformers_attention_forwardGenerator_pass(test_max_difference=_snake_case		)





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




				def 							snake_case   (							self		):



									"""simple docstring"""





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




				def 							snake_case   (							self		):



									"""simple docstring"""





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

									_lowerCAmelCase						=    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						=    StableUnCLIPImgaImgPipeline.from_pretrained(
									    """fusing/stable-unclip-2-1-l-img2img"""						,				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()

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

									_lowerCAmelCase						=    output.images[0]

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

									assert_mean_pixel_difference(_snake_case						,				_snake_case		)




				def 							snake_case   (							self		):



									"""simple docstring"""





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

									_lowerCAmelCase						=    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						=    StableUnCLIPImgaImgPipeline.from_pretrained(
									    """fusing/stable-unclip-2-1-h-img2img"""						,				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()

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

									_lowerCAmelCase						=    output.images[0]

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

									assert_mean_pixel_difference(_snake_case						,				_snake_case		)




				def 							snake_case   (							self		):



									"""simple docstring"""





									_lowerCAmelCase						=    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						=    StableUnCLIPImgaImgPipeline.from_pretrained(
									    """fusing/stable-unclip-2-1-h-img2img"""						,				torch_dtype=torch.floataa		)
									_lowerCAmelCase						=    pipe.to(_snake_case		)
									pipe.set_progress_bar_config(disable=_snake_case		)
									pipe.enable_attention_slicing()
									pipe.enable_sequential_cpu_offload()

									_lowerCAmelCase						=    pipe(
									    _snake_case						,				"""anime turtle"""						,				num_inference_steps=2						,				output_type="""np"""						,				)

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











"""simple docstring"""



def 						lowercase    (				__snake_case				: int	):
		if not isinstance(__snake_case		,					__snake_case	):
				raise ValueError('''Input must be an integer'''	)
		if input_num <= 0:
				raise ValueError('''Input must be positive'''	)
		return sum(
		    divisor for divisor in range(1		,					input_num // 2 + 1	) if input_num % divisor == 0	)


if __name__ == "__main__":
						import doctest

						doctest.testmod()



