


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 			SCREAMING_SNAKE_CASE__  (    lowercase__						,   lowercase__						,   lowercase__						,   unittest.TestCase      ):
      snake_case__    :					Tuple					       =						StableUnCLIPImgaImgPipeline
      snake_case__    :					Dict					       =						TEXT_GUIDED_IMAGE_VARIATION_PARAMS
      snake_case__    :					List[str]					       =						TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
      snake_case__    :					str					       =						frozenset(
          []      )  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
      snake_case__    :					List[Any]					       =						frozenset([]      )
      def 	SCREAMING_SNAKE_CASE							(	self    :	List[str]					)		->      Any:
             a_ :		List[Any]							=  3_2
             a_ :		List[str]							=  embedder_hidden_size

             # image encoding components

             a_ :		Optional[int]							=  CLIPImageProcessor(crop_size=3_2					,							size=3_2					)

             torch.manual_seed(0					)
             a_ :		Optional[Any]							=  CLIPVisionModelWithProjection(
                 CLIPVisionConfig(
                     hidden_size=SCREAMING_SNAKE_CASE__					,							projection_dim=SCREAMING_SNAKE_CASE__					,							num_hidden_layers=5					,							num_attention_heads=4					,							image_size=3_2					,							intermediate_size=3_7					,							patch_size=1					,							)					)

             # regular denoising components

             torch.manual_seed(0					)
             a_ :		Any							=  StableUnCLIPImageNormalizer(embedding_dim=SCREAMING_SNAKE_CASE__					)
             a_ :		Optional[Any]							=  DDPMScheduler(beta_schedule='squaredcos_cap_v2'					)

             torch.manual_seed(0					)
             a_ :		str							=  CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'					)

             torch.manual_seed(0					)
             a_ :		Dict							=  CLIPTextModel(
                 CLIPTextConfig(
                     bos_token_id=0					,							eos_token_id=2					,							hidden_size=SCREAMING_SNAKE_CASE__					,							projection_dim=3_2					,							intermediate_size=3_7					,							layer_norm_eps=1E-05					,							num_attention_heads=4					,							num_hidden_layers=5					,							pad_token_id=1					,							vocab_size=1_0_0_0					,							)					)

             torch.manual_seed(0					)
             a_ :		List[Any]							=  UNetaDConditionModel(
                 sample_size=3_2					,							in_channels=4					,							out_channels=4					,							down_block_types=('CrossAttnDownBlock2D', 'DownBlock2D')					,							up_block_types=('UpBlock2D', 'CrossAttnUpBlock2D')					,							block_out_channels=(3_2, 6_4)					,							attention_head_dim=(2, 4)					,							class_embed_type='projection'					,							projection_class_embeddings_input_dim=embedder_projection_dim * 2					,							cross_attention_dim=SCREAMING_SNAKE_CASE__					,							layers_per_block=1					,							upcast_attention=SCREAMING_SNAKE_CASE__					,							use_linear_projection=SCREAMING_SNAKE_CASE__					,							)

             torch.manual_seed(0					)
             a_ :		Tuple							=  DDIMScheduler(
                 beta_schedule='scaled_linear'					,							beta_start=0.00085					,							beta_end=0.012					,							prediction_type='v_prediction'					,							set_alpha_to_one=SCREAMING_SNAKE_CASE__					,							steps_offset=1					,							)

             torch.manual_seed(0					)
             a_ :		Tuple							=  AutoencoderKL()

             a_ :		List[str]							=  {
                 # 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 	SCREAMING_SNAKE_CASE							(	self    :	Optional[int]					,							SCREAMING_SNAKE_CASE__    :	int					,							SCREAMING_SNAKE_CASE__    :	List[Any]=0					,							SCREAMING_SNAKE_CASE__    :	Union[str, Any]=True					)		->      Optional[int]:
             if str(SCREAMING_SNAKE_CASE__					).startswith('mps'					):
                    a_ :		List[Any]							=  torch.manual_seed(SCREAMING_SNAKE_CASE__					)
             else:
                    a_ :		List[Any]							=  torch.Generator(device=SCREAMING_SNAKE_CASE__					).manual_seed(SCREAMING_SNAKE_CASE__					)

             a_ :		List[Any]							=  floats_tensor((1, 3, 3_2, 3_2)					,							rng=random.Random(SCREAMING_SNAKE_CASE__					)					).to(SCREAMING_SNAKE_CASE__					)

             if pil_image:
                    a_ :		int							=  input_image * 0.5 + 0.5
                    a_ :		str							=  input_image.clamp(0					,							1					)
                    a_ :		Tuple							=  input_image.cpu().permute(0					,							2					,							3					,							1					).float().numpy()
                    a_ :		Any							=  DiffusionPipeline.numpy_to_pil(SCREAMING_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 	SCREAMING_SNAKE_CASE							(	self    :	Optional[Any]					)		->      Optional[int]:
             a_ :		str							=  'cpu'  # ensure determinism for the device-dependent torch.Generator
             a_ :		str							=  self.get_dummy_components()
             a_ :		Union[str, Any]							=  StableUnCLIPImgaImgPipeline(**SCREAMING_SNAKE_CASE__					)
             a_ :		int							=  sd_pipe.to(SCREAMING_SNAKE_CASE__					)
             sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__					)

             a_ :		List[str]							=  self.get_dummy_inputs(SCREAMING_SNAKE_CASE__					)
             inputs.update({'image_embeds': None}					)
             a_ :		int							=  sd_pipe(**SCREAMING_SNAKE_CASE__					).images
             a_ :		List[str]							=  image[0, -3:, -3:, -1]

             assert image.shape == (1, 3_2, 3_2, 3)
             a_ :		List[Any]							=  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 	SCREAMING_SNAKE_CASE							(	self    :	Union[str, Any]					)		->      str:
             a_ :		str							=  torch_device in ['cpu', 'mps']

             self._test_attention_slicing_forward_pass(test_max_difference=SCREAMING_SNAKE_CASE__					)
      def 	SCREAMING_SNAKE_CASE							(	self    :	int					)		->      str:
             a_ :		List[Any]							=  torch_device in ['cpu', 'mps']

             self._test_inference_batch_single_identical(test_max_difference=SCREAMING_SNAKE_CASE__					)

      @unittest.skipIf(
          torch_device != 'cuda' or not is_xformers_available()					,							reason='XFormers attention is only available with CUDA and `xformers` installed'					,							)
      def 	SCREAMING_SNAKE_CASE							(	self    :	Union[str, Any]					)		->      int:
             self._test_xformers_attention_forwardGenerator_pass(test_max_difference=SCREAMING_SNAKE_CASE__					)
@slow
@require_torch_gpu
class 			SCREAMING_SNAKE_CASE__  (    unittest.TestCase      ):
      def 	SCREAMING_SNAKE_CASE							(	self    :	List[Any]					)		->      int:
             # clean up the VRAM after each test
             super().tearDown()
             gc.collect()
             torch.cuda.empty_cache()
      def 	SCREAMING_SNAKE_CASE							(	self    :	Tuple					)		->      int:
             a_ :		Dict							=  load_image(
                 'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png'					)

             a_ :		Optional[int]							=  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'					)

             a_ :		Optional[int]							=  StableUnCLIPImgaImgPipeline.from_pretrained(
                 'fusing/stable-unclip-2-1-l-img2img'					,							torch_dtype=torch.floataa					)
             pipe.to(SCREAMING_SNAKE_CASE__					)
             pipe.set_progress_bar_config(disable=SCREAMING_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()

             a_ :		str							=  torch.Generator(device='cpu'					).manual_seed(0					)
             a_ :		int							=  pipe(SCREAMING_SNAKE_CASE__					,							'anime turle'					,							generator=SCREAMING_SNAKE_CASE__					,							output_type='np'					)

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

             assert image.shape == (7_6_8, 7_6_8, 3)

             assert_mean_pixel_difference(SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					)
      def 	SCREAMING_SNAKE_CASE							(	self    :	Tuple					)		->      Union[str, Any]:
             a_ :		List[Any]							=  load_image(
                 'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png'					)

             a_ :		Union[str, 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'					)

             a_ :		Optional[Any]							=  StableUnCLIPImgaImgPipeline.from_pretrained(
                 'fusing/stable-unclip-2-1-h-img2img'					,							torch_dtype=torch.floataa					)
             pipe.to(SCREAMING_SNAKE_CASE__					)
             pipe.set_progress_bar_config(disable=SCREAMING_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()

             a_ :		List[Any]							=  torch.Generator(device='cpu'					).manual_seed(0					)
             a_ :		Any							=  pipe(SCREAMING_SNAKE_CASE__					,							'anime turle'					,							generator=SCREAMING_SNAKE_CASE__					,							output_type='np'					)

             a_ :		List[Any]							=  output.images[0]

             assert image.shape == (7_6_8, 7_6_8, 3)

             assert_mean_pixel_difference(SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					)

      def 	SCREAMING_SNAKE_CASE							(	self    :	Tuple					)		->      int:
             a_ :		Dict							=  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()

             a_ :		str							=  StableUnCLIPImgaImgPipeline.from_pretrained(
                 'fusing/stable-unclip-2-1-h-img2img'					,							torch_dtype=torch.floataa					)
             a_ :		Tuple							=  pipe.to(SCREAMING_SNAKE_CASE__					)
             pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__					)
             pipe.enable_attention_slicing()
             pipe.enable_sequential_cpu_offload()

             a_ :		Optional[int]							=  pipe(
                 SCREAMING_SNAKE_CASE__					,							'anime turtle'					,							num_inference_steps=2					,							output_type='np'					,							)

             a_ :		List[str]							=  torch.cuda.max_memory_allocated()
             # make sure that less than 7 GB is allocated
             assert mem_bytes < 7 * 1_0**9






from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available


UpperCAmelCase_				:			Dict   =	{'configuration_vit_msn': ['VIT_MSN_PRETRAINED_CONFIG_ARCHIVE_MAP', 'ViTMSNConfig']}

try:
   if not is_torch_available():
      raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
   pass
else:
   UpperCAmelCase_				:			int   =	[
       'VIT_MSN_PRETRAINED_MODEL_ARCHIVE_LIST',
       'ViTMSNModel',
       'ViTMSNForImageClassification',
       'ViTMSNPreTrainedModel',
   ]

if TYPE_CHECKING:
   from .configuration_vit_msn import VIT_MSN_PRETRAINED_CONFIG_ARCHIVE_MAP, ViTMSNConfig

   try:
      if not is_torch_available():
         raise OptionalDependencyNotAvailable()
   except OptionalDependencyNotAvailable:
      pass
   else:
      from .modeling_vit_msn import (
          VIT_MSN_PRETRAINED_MODEL_ARCHIVE_LIST,
          ViTMSNForImageClassification,
          ViTMSNModel,
          ViTMSNPreTrainedModel,
      )

else:
   import sys

   UpperCAmelCase_				:			Dict   =	_LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)


