
'''simple docstring'''







import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import CLIPImageProcessor, CLIPVisionConfig, CLIPVisionModelWithProjection

from diffusers import (
    AutoencoderKL,
    DPMSolverMultistepScheduler,
    PNDMScheduler,
    StableDiffusionImageVariationPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import floats_tensor, load_image, load_numpy, nightly, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import IMAGE_VARIATION_BATCH_PARAMS, IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()







class 		snake_case_       (						A__		,	A__		,	A__		,	unittest.TestCase						):



   """simple docstring"""





   __lowerCAmelCase	:							Tuple								=StableDiffusionImageVariationPipeline
   __lowerCAmelCase	:							Dict								=IMAGE_VARIATION_PARAMS
   __lowerCAmelCase	:							Optional[int]								=IMAGE_VARIATION_BATCH_PARAMS
   __lowerCAmelCase	:							str								=frozenset([]						)
   # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
   __lowerCAmelCase	:							Any								=frozenset([]						)







   def     __UpperCAmelCase      ( self):
          torch.manual_seed(0)
          lowerCamelCase__				   =      UNetaDConditionModel(
              block_out_channels=(32, 64)			,   layers_per_block=2			,   sample_size=32			,   in_channels=4			,   out_channels=4			,   down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")			,   up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")			,   cross_attention_dim=32			,   )
          lowerCamelCase__				   =      PNDMScheduler(skip_prk_steps=UpperCAmelCase__)
          torch.manual_seed(0)
          lowerCamelCase__				   =      AutoencoderKL(
              block_out_channels=[32, 64]			,   in_channels=3			,   out_channels=3			,   down_block_types=["DownEncoderBlock2D", "DownEncoderBlock2D"]			,   up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D"]			,   latent_channels=4			,   )
          torch.manual_seed(0)
          lowerCamelCase__				   =      CLIPVisionConfig(
              hidden_size=32			,   projection_dim=32			,   intermediate_size=37			,   layer_norm_eps=1E-0_5			,   num_attention_heads=4			,   num_hidden_layers=5			,   image_size=32			,   patch_size=4			,   )
          lowerCamelCase__				   =      CLIPVisionModelWithProjection(UpperCAmelCase__)
          lowerCamelCase__				   =      CLIPImageProcessor(crop_size=32			,   size=32)

          lowerCamelCase__				   =      {
              "unet": unet,
              "scheduler": scheduler,
              "vae": vae,
              "image_encoder": image_encoder,
              "feature_extractor": feature_extractor,
              "safety_checker": None,
          }
          return components







   def     __UpperCAmelCase      ( self			,   UpperCamelCase			,   UpperCamelCase=0):
          lowerCamelCase__				   =      floats_tensor((1, 3, 32, 32)			,   rng=random.Random(UpperCAmelCase__))
          lowerCamelCase__				   =      image.cpu().permute(0			,   2			,   3			,   1)[0]
          lowerCamelCase__				   =      Image.fromarray(np.uinta(UpperCAmelCase__)).convert("RGB").resize((32, 32))
          if str(UpperCAmelCase__).startswith("mps"):
                 lowerCamelCase__				   =      torch.manual_seed(UpperCAmelCase__)
          else:
                 lowerCamelCase__				   =      torch.Generator(device=UpperCAmelCase__).manual_seed(UpperCAmelCase__)
          lowerCamelCase__				   =      {
              "image": image,
              "generator": generator,
              "num_inference_steps": 2,
              "guidance_scale": 6.0,
              "output_type": "numpy",
          }
          return inputs







   def     __UpperCAmelCase      ( self):
          lowerCamelCase__				   =      "cpu"  # ensure determinism for the device-dependent torch.Generator
          lowerCamelCase__				   =      self.get_dummy_components()
          lowerCamelCase__				   =      StableDiffusionImageVariationPipeline(**UpperCAmelCase__)
          lowerCamelCase__				   =      sd_pipe.to(UpperCAmelCase__)
          sd_pipe.set_progress_bar_config(disable=UpperCAmelCase__)

          lowerCamelCase__				   =      self.get_dummy_inputs(UpperCAmelCase__)
          lowerCamelCase__				   =      sd_pipe(**UpperCAmelCase__).images
          lowerCamelCase__				   =      image[0, -3:, -3:, -1]

          assert image.shape == (1, 64, 64, 3)
          lowerCamelCase__				   =      np.array([0.5_2_3_9, 0.5_7_2_3, 0.4_7_9_6, 0.5_0_4_9, 0.5_5_5_0, 0.4_6_8_5, 0.5_3_2_9, 0.4_8_9_1, 0.4_9_2_1])

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







   def     __UpperCAmelCase      ( self):
          lowerCamelCase__				   =      "cpu"  # ensure determinism for the device-dependent torch.Generator
          lowerCamelCase__				   =      self.get_dummy_components()
          lowerCamelCase__				   =      StableDiffusionImageVariationPipeline(**UpperCAmelCase__)
          lowerCamelCase__				   =      sd_pipe.to(UpperCAmelCase__)
          sd_pipe.set_progress_bar_config(disable=UpperCAmelCase__)

          lowerCamelCase__				   =      self.get_dummy_inputs(UpperCAmelCase__)
          lowerCamelCase__				   =      2 * [inputs["image"]]
          lowerCamelCase__				   =      sd_pipe(**UpperCAmelCase__)

          lowerCamelCase__				   =      output.images

          lowerCamelCase__				   =      image[-1, -3:, -3:, -1]

          assert image.shape == (2, 64, 64, 3)
          lowerCamelCase__				   =      np.array([0.6_8_9_2, 0.5_6_3_7, 0.5_8_3_6, 0.5_7_7_1, 0.6_2_5_4, 0.6_4_0_9, 0.5_5_8_0, 0.5_5_6_9, 0.5_2_8_9])

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







   def     __UpperCAmelCase      ( self):
          super().test_inference_batch_single_identical(expected_max_diff=3E-3)







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



   """simple docstring"""





   def     __UpperCAmelCase      ( self):
          super().tearDown()
          gc.collect()
          torch.cuda.empty_cache()







   def     __UpperCAmelCase      ( self			,   UpperCamelCase			,   UpperCamelCase="cpu"			,   UpperCamelCase=torch.floataa			,   UpperCamelCase=0):
          lowerCamelCase__				   =      torch.Generator(device=UpperCAmelCase__).manual_seed(UpperCAmelCase__)
          lowerCamelCase__				   =      load_image(
              "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
              "/stable_diffusion_imgvar/input_image_vermeer.png")
          lowerCamelCase__				   =      np.random.RandomState(UpperCAmelCase__).standard_normal((1, 4, 64, 64))
          lowerCamelCase__				   =      torch.from_numpy(UpperCAmelCase__).to(device=UpperCAmelCase__			,   dtype=UpperCAmelCase__)
          lowerCamelCase__				   =      {
              "image": init_image,
              "latents": latents,
              "generator": generator,
              "num_inference_steps": 3,
              "guidance_scale": 7.5,
              "output_type": "numpy",
          }
          return inputs







   def     __UpperCAmelCase      ( self):
          lowerCamelCase__				   =      StableDiffusionImageVariationPipeline.from_pretrained(
              "lambdalabs/sd-image-variations-diffusers"			,   safety_checker=UpperCAmelCase__)
          lowerCamelCase__				   =      sd_pipe.to(UpperCAmelCase__)
          sd_pipe.set_progress_bar_config(disable=UpperCAmelCase__)

          lowerCamelCase__				   =      self.get_inputs(UpperCAmelCase__)
          lowerCamelCase__				   =      sd_pipe(**UpperCAmelCase__).images
          lowerCamelCase__				   =      image[0, -3:, -3:, -1].flatten()

          assert image.shape == (1, 5_12, 5_12, 3)
          lowerCamelCase__				   =      np.array([0.8_4_4_9_1, 0.9_0_7_8_9, 0.7_5_7_0_8, 0.7_8_7_3_4, 0.8_3_4_8_5, 0.7_0_0_9_9, 0.6_6_9_3_8, 0.6_8_7_2_7, 0.6_1_3_7_9])
          assert np.abs(image_slice - expected_slice).max() < 6E-3







   def     __UpperCAmelCase      ( self):
          lowerCamelCase__				   =      0

          def callback_fn(UpperCamelCase			,   UpperCamelCase			,   UpperCamelCase) -> None:
                 lowerCamelCase__				   =      True
                 nonlocal number_of_steps
                 number_of_steps += 1
                 if step == 1:
                        lowerCamelCase__				   =      latents.detach().cpu().numpy()
                        assert latents.shape == (1, 4, 64, 64)
                        lowerCamelCase__				   =      latents[0, -3:, -3:, -1]
                        lowerCamelCase__				   =      np.array(
                            [-0.1_6_2_1, 0.2_8_3_7, -0.7_9_7_9, -0.1_2_2_1, -1.3_0_5_7, 0.7_6_8_1, -2.1_1_9_1, 0.0_4_6_4, 1.6_3_0_9])

                        assert np.abs(latents_slice.flatten() - expected_slice).max() < 5E-2
                 elif step == 2:
                        lowerCamelCase__				   =      latents.detach().cpu().numpy()
                        assert latents.shape == (1, 4, 64, 64)
                        lowerCamelCase__				   =      latents[0, -3:, -3:, -1]
                        lowerCamelCase__				   =      np.array([0.6_2_9_9, 1.7_5_0_0, 1.1_9_9_2, -2.1_5_8_2, -1.8_9_9_4, 0.7_3_3_4, -0.7_0_9_0, 1.0_1_3_7, 1.5_2_7_3])

                        assert np.abs(latents_slice.flatten() - expected_slice).max() < 5E-2

          lowerCamelCase__				   =      False

          lowerCamelCase__				   =      StableDiffusionImageVariationPipeline.from_pretrained(
              "fusing/sd-image-variations-diffusers"			,   safety_checker=UpperCAmelCase__			,   torch_dtype=torch.floataa			,   )
          pipe.to(UpperCAmelCase__)
          pipe.set_progress_bar_config(disable=UpperCAmelCase__)
          pipe.enable_attention_slicing()

          lowerCamelCase__				   =      self.get_inputs(UpperCAmelCase__			,   dtype=torch.floataa)
          pipe(**UpperCAmelCase__			,   callback=UpperCAmelCase__			,   callback_steps=1)
          assert callback_fn.has_been_called
          assert number_of_steps == inputs["num_inference_steps"]







   def     __UpperCAmelCase      ( self):
          torch.cuda.empty_cache()
          torch.cuda.reset_max_memory_allocated()
          torch.cuda.reset_peak_memory_stats()

          lowerCamelCase__				   =      "fusing/sd-image-variations-diffusers"
          lowerCamelCase__				   =      StableDiffusionImageVariationPipeline.from_pretrained(
              UpperCAmelCase__			,   safety_checker=UpperCAmelCase__			,   torch_dtype=torch.floataa)
          lowerCamelCase__				   =      pipe.to(UpperCAmelCase__)
          pipe.set_progress_bar_config(disable=UpperCAmelCase__)
          pipe.enable_attention_slicing(1)
          pipe.enable_sequential_cpu_offload()

          lowerCamelCase__				   =      self.get_inputs(UpperCAmelCase__			,   dtype=torch.floataa)
          lowerCamelCase__				   =      pipe(**UpperCAmelCase__)

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







@nightly
@require_torch_gpu
class 		snake_case_       (						unittest.TestCase						):



   """simple docstring"""





   def     __UpperCAmelCase      ( self):
          super().tearDown()
          gc.collect()
          torch.cuda.empty_cache()







   def     __UpperCAmelCase      ( self			,   UpperCamelCase			,   UpperCamelCase="cpu"			,   UpperCamelCase=torch.floataa			,   UpperCamelCase=0):
          lowerCamelCase__				   =      torch.Generator(device=UpperCAmelCase__).manual_seed(UpperCAmelCase__)
          lowerCamelCase__				   =      load_image(
              "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
              "/stable_diffusion_imgvar/input_image_vermeer.png")
          lowerCamelCase__				   =      np.random.RandomState(UpperCAmelCase__).standard_normal((1, 4, 64, 64))
          lowerCamelCase__				   =      torch.from_numpy(UpperCAmelCase__).to(device=UpperCAmelCase__			,   dtype=UpperCAmelCase__)
          lowerCamelCase__				   =      {
              "image": init_image,
              "latents": latents,
              "generator": generator,
              "num_inference_steps": 50,
              "guidance_scale": 7.5,
              "output_type": "numpy",
          }
          return inputs







   def     __UpperCAmelCase      ( self):
          lowerCamelCase__				   =      StableDiffusionImageVariationPipeline.from_pretrained("fusing/sd-image-variations-diffusers")
          sd_pipe.to(UpperCAmelCase__)
          sd_pipe.set_progress_bar_config(disable=UpperCAmelCase__)

          lowerCamelCase__				   =      self.get_inputs(UpperCAmelCase__)
          lowerCamelCase__				   =      sd_pipe(**UpperCAmelCase__).images[0]

          lowerCamelCase__				   =      load_numpy(
              "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
              "/stable_diffusion_imgvar/lambdalabs_variations_pndm.npy")
          lowerCamelCase__				   =      np.abs(expected_image - image).max()
          assert max_diff < 1E-3







   def     __UpperCAmelCase      ( self):
          lowerCamelCase__				   =      StableDiffusionImageVariationPipeline.from_pretrained("fusing/sd-image-variations-diffusers")
          lowerCamelCase__				   =      DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config)
          sd_pipe.to(UpperCAmelCase__)
          sd_pipe.set_progress_bar_config(disable=UpperCAmelCase__)

          lowerCamelCase__				   =      self.get_inputs(UpperCAmelCase__)
          lowerCamelCase__				   =      25
          lowerCamelCase__				   =      sd_pipe(**UpperCAmelCase__).images[0]

          lowerCamelCase__				   =      load_numpy(
              "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
              "/stable_diffusion_imgvar/lambdalabs_variations_dpm_multi.npy")
          lowerCamelCase__				   =      np.abs(expected_image - image).max()
          assert max_diff < 1E-3





"""simple docstring"""






class       lowerCamelCase_		:





				"""simple docstring"""







				class       lowerCamelCase_		:





								"""simple docstring"""







								_lowerCAmelCase   :   Dict   				=       "_prev", "_data", "_next"



								def __init__(    self							,		UpperCAmelCase__							,		UpperCAmelCase__							,		UpperCAmelCase__  ):
												SCREAMING_SNAKE_CASE__  				=					link_p
												SCREAMING_SNAKE_CASE__  				=					element
												SCREAMING_SNAKE_CASE__  				=					link_n






								def 	lowerCAmelCase__		(    self  ):
												return (
												    f''' Prev -> {self._prev is not None}, Next -> {self._next is not None}'''
												)



				def __init__(    self  ):
								SCREAMING_SNAKE_CASE__  				=					self._Node(UpperCAmelCase__							,		UpperCAmelCase__							,		UpperCAmelCase__  )
								SCREAMING_SNAKE_CASE__  				=					self._Node(UpperCAmelCase__							,		UpperCAmelCase__							,		UpperCAmelCase__  )
								SCREAMING_SNAKE_CASE__  				=					self._trailer
								SCREAMING_SNAKE_CASE__  				=					self._header
								SCREAMING_SNAKE_CASE__  				=					0



				def __len__(    self  ):
								return self._size



				def 	lowerCAmelCase__		(    self  ):
								return self.__len__() == 0



				def 	lowerCAmelCase__		(    self							,		UpperCAmelCase__							,		UpperCAmelCase__							,		UpperCAmelCase__  ):
								# Create new_node by setting it's prev.link -> header
								# setting it's next.link -> trailer
								SCREAMING_SNAKE_CASE__  				=					self._Node(UpperCAmelCase__							,		UpperCAmelCase__							,		UpperCAmelCase__  )
								SCREAMING_SNAKE_CASE__  				=					new_node
								SCREAMING_SNAKE_CASE__  				=					new_node
								self._size += 1
								return self






				def 	lowerCAmelCase__		(    self							,		UpperCAmelCase__  ):
								SCREAMING_SNAKE_CASE__  				=					node._prev
								SCREAMING_SNAKE_CASE__  				=					node._next

								SCREAMING_SNAKE_CASE__  				=					successor
								SCREAMING_SNAKE_CASE__  				=					predecessor
								self._size -= 1
								SCREAMING_SNAKE_CASE__  				=					node._data
								SCREAMING_SNAKE_CASE__  				=					SCREAMING_SNAKE_CASE__  				=					SCREAMING_SNAKE_CASE__  				=					None
								del node
								return temp


class       lowerCamelCase_		(      _DoublyLinkedBase ):





				"""simple docstring"""







				def 	lowerCAmelCase__		(    self  ):
								if self.is_empty():
												raise Exception("List is empty"  )
								return self._header._next._data



				def 	lowerCAmelCase__		(    self  ):
								if self.is_empty():
												raise Exception("List is empty"  )
								return self._trailer._prev._data



				def 	lowerCAmelCase__		(    self							,		UpperCAmelCase__  ):
								return self._insert(self._header							,		UpperCAmelCase__							,		self._header._next  )



				def 	lowerCAmelCase__		(    self							,		UpperCAmelCase__  ):
								return self._insert(self._trailer._prev							,		UpperCAmelCase__							,		self._trailer  )



				def 	lowerCAmelCase__		(    self  ):
								if self.is_empty():
												raise IndexError("remove_first from empty list"  )
								return self._delete(self._header._next  )






				def 	lowerCAmelCase__		(    self  ):
								if self.is_empty():
												raise IndexError("remove_first from empty list"  )
								return self._delete(self._trailer._prev  )
