





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__    (							__UpperCAmelCase ,  __UpperCAmelCase ,  __UpperCAmelCase ,  unittest.TestCase	):




   """simple docstring"""



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

   def 							__lowercase		(					self)			->						Optional[int]:







          '''simple docstring'''

          a__						:	Any =				32
          a__						:	Any =				embedder_hidden_size

          # image encoding components

          a__						:	Optional[Any] =				CLIPImageProcessor(crop_size=32					,  size=32)

          torch.manual_seed(0)
          a__						:	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)
          a__						:	Union[str, Any] =				StableUnCLIPImageNormalizer(embedding_dim=lowercase)
          a__						:	str =				DDPMScheduler(beta_schedule='squaredcos_cap_v2')

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

          torch.manual_seed(0)
          a__						:	List[str] =				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=1000					,  ))

          torch.manual_seed(0)
          a__						:	Optional[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)
          a__						:	List[str] =				DDIMScheduler(
              beta_schedule='scaled_linear'					,  beta_start=0.0_00_85					,  beta_end=0.0_12					,  prediction_type='v_prediction'					,  set_alpha_to_one=lowercase					,  steps_offset=1					,  )

          torch.manual_seed(0)
          a__						:	str =				AutoencoderKL()

          a__						:	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					,  lowercase					,  lowercase=0					,  lowercase=True)			->						Union[str, Any]:







          '''simple docstring'''

          if str(lowercase).startswith('mps'):
                 a__						:	List[Any] =				torch.manual_seed(lowercase)
          else:
                 a__						:	Optional[Any] =				torch.Generator(device=lowercase).manual_seed(lowercase)

          a__						:	Union[str, Any] =				floats_tensor((1, 3, 32, 32)					,  rng=random.Random(lowercase)).to(lowercase)

          if pil_image:
                 a__						:	Any =				input_image * 0.5 + 0.5
                 a__						:	int =				input_image.clamp(0					,  1)
                 a__						:	Any =				input_image.cpu().permute(0					,  2					,  3					,  1).float().numpy()
                 a__						:	Tuple =				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 							__lowercase		(					self)			->						Dict:







          '''simple docstring'''

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

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

          assert image.shape == (1, 32, 32, 3)
          a__						:	List[Any] =				np.array([0.38_72, 0.72_24, 0.56_01, 0.47_41, 0.68_72, 0.58_14, 0.46_36, 0.38_67, 0.50_78])

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

   def 							__lowercase		(					self)			->						List[str]:







          '''simple docstring'''

          a__						:	str =				torch_device in ['cpu', 'mps']

          self._test_attention_slicing_forward_pass(test_max_difference=lowercase)

   def 							__lowercase		(					self)			->						Optional[int]:







          '''simple docstring'''

          a__						:	int =				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 							__lowercase		(					self)			->						int:







          '''simple docstring'''

          self._test_xformers_attention_forwardGenerator_pass(test_max_difference=lowercase)





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




   """simple docstring"""



   def 							__lowercase		(					self)			->						int:







          '''simple docstring'''

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

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







          '''simple docstring'''

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

          a__						:	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__						:	Any =				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()

          a__						:	Any =				torch.Generator(device='cpu').manual_seed(0)
          a__						:	Dict =				pipe(lowercase					,  'anime turle'					,  generator=lowercase					,  output_type='np')

          a__						:	Any =				output.images[0]

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

          assert_mean_pixel_difference(lowercase					,  lowercase)

   def 							__lowercase		(					self)			->						List[Any]:







          '''simple docstring'''

          a__						:	Optional[int] =				load_image(
              'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png')

          a__						:	Tuple =				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__						:	int =				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()

          a__						:	Dict =				torch.Generator(device='cpu').manual_seed(0)
          a__						:	Dict =				pipe(lowercase					,  'anime turle'					,  generator=lowercase					,  output_type='np')

          a__						:	Any =				output.images[0]

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

          assert_mean_pixel_difference(lowercase					,  lowercase)

   def 							__lowercase		(					self)			->						List[str]:







          '''simple docstring'''

          a__						:	Optional[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()

          a__						:	Optional[int] =				StableUnCLIPImgaImgPipeline.from_pretrained(
              'fusing/stable-unclip-2-1-h-img2img'					,  torch_dtype=torch.floataa)
          a__						:	Optional[int] =				pipe.to(lowercase)
          pipe.set_progress_bar_config(disable=lowercase)
          pipe.enable_attention_slicing()
          pipe.enable_sequential_cpu_offload()

          a__						:	List[Any] =				pipe(
              lowercase					,  'anime turtle'					,  num_inference_steps=2					,  output_type='np'					,  )

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




'''simple docstring'''







from ...utils import is_torch_available, is_transformers_available


if is_transformers_available() and is_torch_available():
					from .pipeline_vq_diffusion import LearnedClassifierFreeSamplingEmbeddings, VQDiffusionPipeline

