



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        lowercase_   (					__snake_case						,			__snake_case						,			__snake_case						,			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 	UpperCamelCase	(						self    ):
													_snake_case						:					Any    =   32
													_snake_case						:					Any    =   embedder_hidden_size

													# image encoding components

													_snake_case						:					Union[str, Any]    =   CLIPImageProcessor(crop_size=32				,    size=32    )

													torch.manual_seed(0    )
													_snake_case						:					Tuple    =   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    )
													_snake_case						:					int    =   StableUnCLIPImageNormalizer(embedding_dim=lowercase_    )
													_snake_case						:					List[Any]    =   DDPMScheduler(beta_schedule="squaredcos_cap_v2"    )

													torch.manual_seed(0    )
													_snake_case						:					List[str]    =   CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"    )

													torch.manual_seed(0    )
													_snake_case						:					Union[str, Any]    =   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    )
													_snake_case						:					List[Any]    =   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    )
													_snake_case						:					Optional[Any]    =   DDIMScheduler(
													    beta_schedule="scaled_linear"				,    beta_start=0.00_085				,    beta_end=0.012				,    prediction_type="v_prediction"				,    set_alpha_to_one=lowercase_				,    steps_offset=1				,    )

													torch.manual_seed(0    )
													_snake_case						:					Dict    =   AutoencoderKL()

													_snake_case						:					Dict    =   {
													    # 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 	UpperCamelCase	(						self				,    lowercase_				,    lowercase_=0				,    lowercase_=True    ):
													if str(lowercase_    ).startswith("mps"    ):
																			_snake_case						:					List[Any]    =   torch.manual_seed(lowercase_    )
													else:
																			_snake_case						:					Optional[int]    =   torch.Generator(device=lowercase_    ).manual_seed(lowercase_    )

													_snake_case						:					List[Any]    =   floats_tensor((1, 3, 32, 32)				,    rng=random.Random(lowercase_    )    ).to(lowercase_    )

													if pil_image:
																			_snake_case						:					Union[str, Any]    =   input_image * 0.5 + 0.5
																			_snake_case						:					List[Any]    =   input_image.clamp(0				,    1    )
																			_snake_case						:					str    =   input_image.cpu().permute(0				,    2				,    3				,    1    ).float().numpy()
																			_snake_case						:					int    =   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 	UpperCamelCase	(						self    ):
													_snake_case						:					Any    =   "cpu"  # ensure determinism for the device-dependent torch.Generator
													_snake_case						:					List[str]    =   self.get_dummy_components()
													_snake_case						:					Dict    =   StableUnCLIPImgaImgPipeline(**lowercase_    )
													_snake_case						:					Dict    =   sd_pipe.to(lowercase_    )
													sd_pipe.set_progress_bar_config(disable=lowercase_    )

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

													assert image.shape == (1, 32, 32, 3)
													_snake_case						:					Dict    =   np.array([0.3_872, 0.7_224, 0.5_601, 0.4_741, 0.6_872, 0.5_814, 0.4_636, 0.3_867, 0.5_078]    )

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



							def 	UpperCamelCase	(						self    ):
													_snake_case						:					Dict    =   torch_device in ["cpu", "mps"]

													self._test_attention_slicing_forward_pass(test_max_difference=lowercase_    )



							def 	UpperCamelCase	(						self    ):
													_snake_case						:					Any    =   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 	UpperCamelCase	(						self    ):
													self._test_xformers_attention_forwardGenerator_pass(test_max_difference=lowercase_    )




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



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



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

													_snake_case						:					str    =   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"    )

													_snake_case						:					Optional[int]    =   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()

													_snake_case						:					Tuple    =   torch.Generator(device="cpu"    ).manual_seed(0    )
													_snake_case						:					int    =   pipe(lowercase_				,    "anime turle"				,    generator=lowercase_				,    output_type="np"    )

													_snake_case						:					Tuple    =   output.images[0]

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

													assert_mean_pixel_difference(lowercase_				,    lowercase_    )



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

													_snake_case						:					List[Any]    =   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"    )

													_snake_case						:					Union[str, Any]    =   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()

													_snake_case						:					Optional[int]    =   torch.Generator(device="cpu"    ).manual_seed(0    )
													_snake_case						:					Tuple    =   pipe(lowercase_				,    "anime turle"				,    generator=lowercase_				,    output_type="np"    )

													_snake_case						:					Optional[Any]    =   output.images[0]

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

													assert_mean_pixel_difference(lowercase_				,    lowercase_    )



							def 	UpperCamelCase	(						self    ):
													_snake_case						:					Union[str, 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()

													_snake_case						:					Tuple    =   StableUnCLIPImgaImgPipeline.from_pretrained(
													    "fusing/stable-unclip-2-1-h-img2img"				,    torch_dtype=torch.floataa    )
													_snake_case						:					Optional[Any]    =   pipe.to(lowercase_    )
													pipe.set_progress_bar_config(disable=lowercase_    )
													pipe.enable_attention_slicing()
													pipe.enable_sequential_cpu_offload()

													_snake_case						:					Dict    =   pipe(
													    lowercase_				,    "anime turtle"				,    num_inference_steps=2				,    output_type="np"				,    )

													_snake_case						:					Any    =   torch.cuda.max_memory_allocated()
													# make sure that less than 7 GB is allocated
													assert mem_bytes < 7 * 10**9
import torch
from transformers import PreTrainedModel, XLMRobertaConfig, XLMRobertaModel



class        lowercase_   (					__snake_case		):
							_lowerCamelCase						=		'M-CLIP'



							def __init__(						self				,    lowercase_=1_024				,    lowercase_=768				,    **lowercase_    ):
													_snake_case						:					str    =   transformerDimSize
													_snake_case						:					Union[str, Any]    =   imageDimSize
													super().__init__(**lowercase_    )




class        lowercase_   (					__snake_case		):
							_lowerCamelCase						=		MCLIPConfig



							def __init__(						self				,    lowercase_				,    *lowercase_				,    **lowercase_    ):
													super().__init__(lowercase_				,    *lowercase_				,    **lowercase_    )
													_snake_case						:					List[Any]    =   XLMRobertaModel(lowercase_    )
													_snake_case						:					int    =   torch.nn.Linear(
													    in_features=config.transformerDimensions				,    out_features=config.numDims    )



							def 	UpperCamelCase	(						self				,    lowercase_				,    lowercase_    ):
													_snake_case						:					Tuple    =   self.transformer(input_ids=lowercase_				,    attention_mask=lowercase_    )[0]
													_snake_case						:					Tuple    =   (embs * attention_mask.unsqueeze(2    )).sum(dim=1    ) / attention_mask.sum(dim=1    )[:, None]
													return self.LinearTransformation(lowercase_    ), embs