




'''simple docstring'''







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     _a						(			__a							,							__a							,							__a							,							unittest.TestCase				):
       __a       :  Union[str, Any]            =      StableUnCLIPImgaImgPipeline
       __a       :  Optional[Any]            =      TEXT_GUIDED_IMAGE_VARIATION_PARAMS
       __a       :  List[str]            =      TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
       __a       :  Any            =      frozenset(
           []				)  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
       __a       :  Optional[int]            =      frozenset([]				)
       def 					A   (			self		:			Tuple       ):



           '''simple docstring'''

           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=lowercase     ,       projection_dim=lowercase     ,       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=lowercase       )
           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=lowercase     ,       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=1_000     ,       )       )

           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=lowercase     ,       layers_per_block=1     ,       upcast_attention=lowercase     ,       use_linear_projection=lowercase     ,       )

           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=lowercase     ,       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 					A   (			self		:			List[Any]     ,       lowercase		:			Optional[Any]     ,       lowercase		:			Dict=0     ,       lowercase		:			Any=True       ):



           '''simple docstring'''

           if str(lowercase       ).startswith('''mps'''       ):
               UpperCAmelCase     =	torch.manual_seed(lowercase       )
           else:
               UpperCAmelCase     =	torch.Generator(device=lowercase       ).manual_seed(lowercase       )

           UpperCAmelCase     =	floats_tensor((1, 3, 32, 32)     ,       rng=random.Random(lowercase       )       ).to(lowercase       )

           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(lowercase       )[0]

           return {
               "prompt": "An anime racoon running a marathon",
               "image": input_image,
               "generator": generator,
               "num_inference_steps": 2,
               "output_type": "np",
           }
       @skip_mps
       def 					A   (			self		:			Union[str, Any]       ):



           '''simple docstring'''

           UpperCAmelCase     =	'''cpu'''  # ensure determinism for the device-dependent torch.Generator
           UpperCAmelCase     =	self.get_dummy_components()
           UpperCAmelCase     =	StableUnCLIPImgaImgPipeline(**lowercase       )
           UpperCAmelCase     =	sd_pipe.to(lowercase       )
           sd_pipe.set_progress_bar_config(disable=lowercase       )

           UpperCAmelCase     =	self.get_dummy_inputs(lowercase       )
           inputs.update({'''image_embeds''': None}       )
           UpperCAmelCase     =	sd_pipe(**lowercase       ).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 					A   (			self		:			Any       ):



           '''simple docstring'''

           UpperCAmelCase     =	torch_device in ['''cpu''', '''mps''']

           self._test_attention_slicing_forward_pass(test_max_difference=lowercase       )
       def 					A   (			self		:			Optional[Any]       ):



           '''simple docstring'''

           UpperCAmelCase     =	torch_device in ['''cpu''', '''mps''']

           self._test_inference_batch_single_identical(test_max_difference=lowercase       )







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



           '''simple docstring'''

           self._test_xformers_attention_forwardGenerator_pass(test_max_difference=lowercase       )






@slow
@require_torch_gpu
class     _a						(			unittest.TestCase				):
       def 					A   (			self		:			List[str]       ):



           '''simple docstring'''

           super().tearDown()
           gc.collect()
           torch.cuda.empty_cache()
       def 					A   (			self		:			List[str]       ):



           '''simple docstring'''

           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(lowercase       )
           pipe.set_progress_bar_config(disable=lowercase       )
           # 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(lowercase     ,       '''anime turle'''     ,       generator=lowercase     ,       output_type='''np'''       )

           UpperCAmelCase     =	output.images[0]

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

           assert_mean_pixel_difference(lowercase     ,       lowercase       )
       def 					A   (			self		:			List[str]       ):



           '''simple docstring'''

           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(lowercase       )
           pipe.set_progress_bar_config(disable=lowercase       )
           # 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(lowercase     ,       '''anime turle'''     ,       generator=lowercase     ,       output_type='''np'''       )

           UpperCAmelCase     =	output.images[0]

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

           assert_mean_pixel_difference(lowercase     ,       lowercase       )







       def 					A   (			self		:			Optional[Any]       ):



           '''simple docstring'''

           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(lowercase       )
           pipe.set_progress_bar_config(disable=lowercase       )
           pipe.enable_attention_slicing()
           pipe.enable_sequential_cpu_offload()

           UpperCAmelCase     =	pipe(
               lowercase     ,       '''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










'''simple docstring'''







from datetime import datetime

import matplotlib.pyplot as plt
import torch





def 		snake_case_							(_a			:							List[str]  ):
    for param in module.parameters():
        UpperCAmelCase     =	False





def 		snake_case_							():
    UpperCAmelCase     =	'''cuda''' if torch.cuda.is_available() else '''cpu'''
    if torch.backends.mps.is_available() and torch.backends.mps.is_built():
        UpperCAmelCase     =	'''mps'''
    if device == "mps":
        print(
            '''WARNING: MPS currently doesn\'t seem to work, and messes up backpropagation without any visible torch'''
            ''' errors. I recommend using CUDA on a colab notebook or CPU instead if you\'re facing inexplicable issues'''
            ''' with generations.'''  )
    return device





def 		snake_case_							(_a			:							Optional[Any]  ):
    UpperCAmelCase     =	plt.imshow(_a  )
    fig.axes.get_xaxis().set_visible(_a  )
    fig.axes.get_yaxis().set_visible(_a  )
    plt.show()







def 		snake_case_							():
    UpperCAmelCase     =	datetime.now()
    UpperCAmelCase     =	current_time.strftime('''%H:%M:%S'''  )
    return timestamp




