






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 ):





				"""simple docstring"""







				UpperCamelCase__ :      Dict			 =StableUnCLIPImgaImgPipeline
				UpperCamelCase__ :      Optional[int]			 =TEXT_GUIDED_IMAGE_VARIATION_PARAMS
				UpperCamelCase__ :      int			 =TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
				UpperCamelCase__ :      Tuple			 =frozenset(
				    [] )  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
				UpperCamelCase__ :      Optional[Any]			 =frozenset([] )


				def       __lowercase							( self    ):




									"""simple docstring"""

									__UpperCamelCase      :					Optional[int]       		=32
									__UpperCamelCase      :					Optional[int]       		=embedder_hidden_size

									# image encoding components

									__UpperCamelCase      :					Optional[int]       		=CLIPImageProcessor(crop_size=32							,  size=32    )

									torch.manual_seed(0    )
									__UpperCamelCase      :					List[str]       		=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      :					Any       		=StableUnCLIPImageNormalizer(embedding_dim=lowerCamelCase__    )
									__UpperCamelCase      :					Any       		=DDPMScheduler(beta_schedule='squaredcos_cap_v2'    )

									torch.manual_seed(0    )
									__UpperCamelCase      :					Optional[Any]       		=CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'    )

									torch.manual_seed(0    )
									__UpperCamelCase      :					Union[str, Any]       		=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      :					int       		=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      :					int       		=DDIMScheduler(
									    beta_schedule='scaled_linear'							,  beta_start=0.00_085							,  beta_end=0.012							,  prediction_type='v_prediction'							,  set_alpha_to_one=lowerCamelCase__							,  steps_offset=1							,  )

									torch.manual_seed(0    )
									__UpperCamelCase      :					int       		=AutoencoderKL()

									__UpperCamelCase      :					int       		={
									    # 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       __lowercase							( self							,  lowerCamelCase__							,  lowerCamelCase__=0							,  lowerCamelCase__=True    ):




									"""simple docstring"""

									if str(lowerCamelCase__    ).startswith('mps'    ):
														__UpperCamelCase      :					Union[str, Any]       		=torch.manual_seed(lowerCamelCase__    )
									else:
														__UpperCamelCase      :					Optional[Any]       		=torch.Generator(device=lowerCamelCase__    ).manual_seed(lowerCamelCase__    )

									__UpperCamelCase      :					int       		=floats_tensor((1, 3, 32, 32)							,  rng=random.Random(lowerCamelCase__    )    ).to(lowerCamelCase__    )

									if pil_image:
														__UpperCamelCase      :					Union[str, Any]       		=input_image * 0.5 + 0.5
														__UpperCamelCase      :					List[Any]       		=input_image.clamp(0							,  1    )
														__UpperCamelCase      :					Optional[int]       		=input_image.cpu().permute(0							,  2							,  3							,  1    ).float().numpy()
														__UpperCamelCase      :					Dict       		=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       __lowercase							( self    ):




									"""simple docstring"""

									__UpperCamelCase      :					Optional[int]       		='cpu'  # ensure determinism for the device-dependent torch.Generator
									__UpperCamelCase      :					List[str]       		=self.get_dummy_components()
									__UpperCamelCase      :					Union[str, Any]       		=StableUnCLIPImgaImgPipeline(**lowerCamelCase__    )
									__UpperCamelCase      :					Tuple       		=sd_pipe.to(lowerCamelCase__    )
									sd_pipe.set_progress_bar_config(disable=lowerCamelCase__    )

									__UpperCamelCase      :					Tuple       		=self.get_dummy_inputs(lowerCamelCase__    )
									inputs.update({'image_embeds': None}    )
									__UpperCamelCase      :					Optional[int]       		=sd_pipe(**lowerCamelCase__    ).images
									__UpperCamelCase      :					int       		=image[0, -3:, -3:, -1]

									assert image.shape == (1, 32, 32, 3)
									__UpperCamelCase      :					Union[str, Any]       		=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       __lowercase							( self    ):




									"""simple docstring"""

									__UpperCamelCase      :					Optional[Any]       		=torch_device in ['cpu', 'mps']

									self._test_attention_slicing_forward_pass(test_max_difference=lowerCamelCase__    )


				def       __lowercase							( self    ):




									"""simple docstring"""

									__UpperCamelCase      :					Tuple       		=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       __lowercase							( self    ):




									"""simple docstring"""

									self._test_xformers_attention_forwardGenerator_pass(test_max_difference=lowerCamelCase__    )




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





				"""simple docstring"""







				def       __lowercase							( self    ):




									"""simple docstring"""

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


				def       __lowercase							( self    ):




									"""simple docstring"""

									__UpperCamelCase      :					Union[str, Any]       		=load_image(
									    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png'    )

									__UpperCamelCase      :					List[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'    )

									__UpperCamelCase      :					str       		=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      :					Optional[Any]       		=torch.Generator(device='cpu'    ).manual_seed(0    )
									__UpperCamelCase      :					Union[str, Any]       		=pipe(lowerCamelCase__							,  'anime turle'							,  generator=lowerCamelCase__							,  output_type='np'    )

									__UpperCamelCase      :					Union[str, Any]       		=output.images[0]

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

									assert_mean_pixel_difference(lowerCamelCase__							,  lowerCamelCase__    )


				def       __lowercase							( self    ):




									"""simple docstring"""

									__UpperCamelCase      :					Union[str, Any]       		=load_image(
									    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png'    )

									__UpperCamelCase      :					int       		=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      :					Union[str, Any]       		=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      :					int       		=torch.Generator(device='cpu'    ).manual_seed(0    )
									__UpperCamelCase      :					Union[str, Any]       		=pipe(lowerCamelCase__							,  'anime turle'							,  generator=lowerCamelCase__							,  output_type='np'    )

									__UpperCamelCase      :					Dict       		=output.images[0]

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

									assert_mean_pixel_difference(lowerCamelCase__							,  lowerCamelCase__    )







				def       __lowercase							( self    ):




									"""simple docstring"""

									__UpperCamelCase      :					Optional[int]       		=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      :					int       		=StableUnCLIPImgaImgPipeline.from_pretrained(
									    'fusing/stable-unclip-2-1-h-img2img'							,  torch_dtype=torch.floataa    )
									__UpperCamelCase      :					Dict       		=pipe.to(lowerCamelCase__    )
									pipe.set_progress_bar_config(disable=lowerCamelCase__    )
									pipe.enable_attention_slicing()
									pipe.enable_sequential_cpu_offload()

									__UpperCamelCase      :					Tuple       		=pipe(
									    lowerCamelCase__							,  'anime turtle'							,  num_inference_steps=2							,  output_type='np'							,  )

									__UpperCamelCase      :					List[Any]       		=torch.cuda.max_memory_allocated()
									# make sure that less than 7 GB is allocated
									assert mem_bytes < 7 * 10**9






'''simple docstring'''

from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_tf_available, is_torch_available


__lowerCAmelCase									=   {'''configuration_swin''': ['''SWIN_PRETRAINED_CONFIG_ARCHIVE_MAP''', '''SwinConfig''', '''SwinOnnxConfig''']}


try:
							if not is_torch_available():
														raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
							pass
else:
							__lowerCAmelCase									=   [
							    '''SWIN_PRETRAINED_MODEL_ARCHIVE_LIST''',
							    '''SwinForImageClassification''',
							    '''SwinForMaskedImageModeling''',
							    '''SwinModel''',
							    '''SwinPreTrainedModel''',
							    '''SwinBackbone''',
							]

try:
							if not is_tf_available():
														raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
							pass
else:
							__lowerCAmelCase									=   [
							    '''TF_SWIN_PRETRAINED_MODEL_ARCHIVE_LIST''',
							    '''TFSwinForImageClassification''',
							    '''TFSwinForMaskedImageModeling''',
							    '''TFSwinModel''',
							    '''TFSwinPreTrainedModel''',
							]

if TYPE_CHECKING:
							from .configuration_swin import SWIN_PRETRAINED_CONFIG_ARCHIVE_MAP, SwinConfig, SwinOnnxConfig

							try:
														if not is_torch_available():
																					raise OptionalDependencyNotAvailable()
							except OptionalDependencyNotAvailable:
														pass
							else:
														from .modeling_swin import (
														    SWIN_PRETRAINED_MODEL_ARCHIVE_LIST,
														    SwinBackbone,
														    SwinForImageClassification,
														    SwinForMaskedImageModeling,
														    SwinModel,
														    SwinPreTrainedModel,
														)

							try:
														if not is_tf_available():
																					raise OptionalDependencyNotAvailable()
							except OptionalDependencyNotAvailable:
														pass
							else:
														from .modeling_tf_swin import (
														    TF_SWIN_PRETRAINED_MODEL_ARCHIVE_LIST,
														    TFSwinForImageClassification,
														    TFSwinForMaskedImageModeling,
														    TFSwinModel,
														    TFSwinPreTrainedModel,
														)

else:
							import sys

							__lowerCAmelCase									=   _LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)




