



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   UpperCamelCase_       (     __A				,  __A				,  __A				,  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    :Dict		)   ->   Any:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__			=      32
									SCREAMING_SNAKE_CASE__			=      embedder_hidden_size

									# image encoding components

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

									torch.manual_seed(0		)
									SCREAMING_SNAKE_CASE__			=      CLIPVisionModelWithProjection(
									    CLIPVisionConfig(
									        hidden_size=__A							,				projection_dim=__A							,				num_hidden_layers=5							,				num_attention_heads=4							,				image_size=32							,				intermediate_size=37							,				patch_size=1							,				)		)

									# regular denoising components

									torch.manual_seed(0		)
									SCREAMING_SNAKE_CASE__			=      StableUnCLIPImageNormalizer(embedding_dim=__A		)
									SCREAMING_SNAKE_CASE__			=      DDPMScheduler(beta_schedule="""squaredcos_cap_v2"""		)

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

									torch.manual_seed(0		)
									SCREAMING_SNAKE_CASE__			=      CLIPTextModel(
									    CLIPTextConfig(
									        bos_token_id=0							,				eos_token_id=2							,				hidden_size=__A							,				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		)
									SCREAMING_SNAKE_CASE__			=      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=__A							,				layers_per_block=1							,				upcast_attention=__A							,				use_linear_projection=__A							,				)

									torch.manual_seed(0		)
									SCREAMING_SNAKE_CASE__			=      DDIMScheduler(
									    beta_schedule="""scaled_linear"""							,				beta_start=0.0_0_0_8_5							,				beta_end=0.0_1_2							,				prediction_type="""v_prediction"""							,				set_alpha_to_one=__A							,				steps_offset=1							,				)

									torch.manual_seed(0		)
									SCREAMING_SNAKE_CASE__			=      AutoencoderKL()

									SCREAMING_SNAKE_CASE__			=      {
									    # 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    :List[Any]							,				__A    :str							,				__A    :int=0							,				__A    :Dict=True		)   ->   Dict:



									"""simple docstring"""



									if str(__A		).startswith("""mps"""		):
											SCREAMING_SNAKE_CASE__			=      torch.manual_seed(__A		)
									else:
											SCREAMING_SNAKE_CASE__			=      torch.Generator(device=__A		).manual_seed(__A		)

									SCREAMING_SNAKE_CASE__			=      floats_tensor((1, 3, 32, 32)							,				rng=random.Random(__A		)		).to(__A		)

									if pil_image:
											SCREAMING_SNAKE_CASE__			=      input_image * 0.5 + 0.5
											SCREAMING_SNAKE_CASE__			=      input_image.clamp(0							,				1		)
											SCREAMING_SNAKE_CASE__			=      input_image.cpu().permute(0							,				2							,				3							,				1		).float().numpy()
											SCREAMING_SNAKE_CASE__			=      DiffusionPipeline.numpy_to_pil(__A		)[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    :List[Any]		)   ->   int:



									"""simple docstring"""



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

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

									assert image.shape == (1, 32, 32, 3)
									SCREAMING_SNAKE_CASE__			=      np.array([0.3_8_7_2, 0.7_2_2_4, 0.5_6_0_1, 0.4_7_4_1, 0.6_8_7_2, 0.5_8_1_4, 0.4_6_3_6, 0.3_8_6_7, 0.5_0_7_8]		)

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


							def      _snake_case  (							self    :Tuple		)   ->   str:



									"""simple docstring"""



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

									self._test_attention_slicing_forward_pass(test_max_difference=__A		)


							def      _snake_case  (							self    :Tuple		)   ->   Any:



									"""simple docstring"""



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

									self._test_inference_batch_single_identical(test_max_difference=__A		)


							@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    :List[str]		)   ->   int:



									"""simple docstring"""



									self._test_xformers_attention_forwardGenerator_pass(test_max_difference=__A		)




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


							def      _snake_case  (							self    :str		)   ->   List[Any]:



									"""simple docstring"""



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


							def      _snake_case  (							self    :str		)   ->   Optional[Any]:



									"""simple docstring"""



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

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

									SCREAMING_SNAKE_CASE__			=      StableUnCLIPImgaImgPipeline.from_pretrained(
									    """fusing/stable-unclip-2-1-l-img2img"""							,				torch_dtype=torch.floataa		)
									pipe.to(__A		)
									pipe.set_progress_bar_config(disable=__A		)
									# 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()

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

									SCREAMING_SNAKE_CASE__			=      output.images[0]

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

									assert_mean_pixel_difference(__A							,				__A		)


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



									"""simple docstring"""



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

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

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

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

									SCREAMING_SNAKE_CASE__			=      output.images[0]

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

									assert_mean_pixel_difference(__A							,				__A		)


							def      _snake_case  (							self    :int		)   ->   Tuple:



									"""simple docstring"""



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

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

									SCREAMING_SNAKE_CASE__			=      pipe(
									    __A							,				"""anime turtle"""							,				num_inference_steps=2							,				output_type="""np"""							,				)

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






import warnings

from ...utils import logging
from .image_processing_layoutlmva import LayoutLMvaImageProcessor


_lowerCamelCase   				=			logging.get_logger(__name__)




class   UpperCamelCase_       (     UpperCamelCase__    ):


							def __init__(							self    :List[Any]							,				*__A    :Tuple							,				**__A    :Dict		)   ->   None:



									"""simple docstring"""



									warnings.warn(
									    """The class LayoutLMv2FeatureExtractor is deprecated and will be removed in version 5 of Transformers."""
									    """ Please use LayoutLMv2ImageProcessor instead."""							,				__A							,				)
									super().__init__(*__A							,				**__A		)