

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      _a				( snake_case_    ,							snake_case_    ,							snake_case_    ,							unittest.TestCase							):




      """simple docstring"""

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





      def 		__A  (      self       :							Union[str, Any]  ):
            torch.manual_seed(0  )
            A_	     =							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	, )
            A_	     =							PNDMScheduler(skip_prk_steps=UpperCAmelCase  )
            torch.manual_seed(0  )
            A_	     =							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  )
            A_	     =							CLIPVisionConfig(
                hidden_size=32	, projection_dim=32	, intermediate_size=37	, layer_norm_eps=1E-05	, num_attention_heads=4	, num_hidden_layers=5	, image_size=32	, patch_size=4	, )
            A_	     =							CLIPVisionModelWithProjection(UpperCAmelCase  )
            A_	     =							CLIPImageProcessor(crop_size=32	, size=32  )

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





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





      def 		__A  (      self       :							Optional[Any]  ):
            A_	     =							"cpu"  # ensure determinism for the device-dependent torch.Generator
            A_	     =							self.get_dummy_components()
            A_	     =							StableDiffusionImageVariationPipeline(**UpperCAmelCase  )
            A_	     =							sd_pipe.to(UpperCAmelCase  )
            sd_pipe.set_progress_bar_config(disable=UpperCAmelCase  )

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

            assert image.shape == (1, 64, 64, 3)
            A_	     =							np.array([0.5_239, 0.5_723, 0.4_796, 0.5_049, 0.5_550, 0.4_685, 0.5_329, 0.4_891, 0.4_921]  )

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





      def 		__A  (      self       :							str  ):
            A_	     =							"cpu"  # ensure determinism for the device-dependent torch.Generator
            A_	     =							self.get_dummy_components()
            A_	     =							StableDiffusionImageVariationPipeline(**UpperCAmelCase  )
            A_	     =							sd_pipe.to(UpperCAmelCase  )
            sd_pipe.set_progress_bar_config(disable=UpperCAmelCase  )

            A_	     =							self.get_dummy_inputs(UpperCAmelCase  )
            A_	     =							2 * [inputs["image"]]
            A_	     =							sd_pipe(**UpperCAmelCase  )

            A_	     =							output.images

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

            assert image.shape == (2, 64, 64, 3)
            A_	     =							np.array([0.6_892, 0.5_637, 0.5_836, 0.5_771, 0.6_254, 0.6_409, 0.5_580, 0.5_569, 0.5_289]  )

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







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




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




      """simple docstring"""

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





      def 		__A  (      self       :							Tuple	, UpperCAmelCase       :							Optional[Any]	, UpperCAmelCase       :							Optional[int]="cpu"	, UpperCAmelCase       :							str=torch.floataa	, UpperCAmelCase       :							Tuple=0  ):
            A_	     =							torch.Generator(device=UpperCAmelCase  ).manual_seed(UpperCAmelCase  )
            A_	     =							load_image(
                "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
                "/stable_diffusion_imgvar/input_image_vermeer.png"  )
            A_	     =							np.random.RandomState(UpperCAmelCase  ).standard_normal((1, 4, 64, 64)  )
            A_	     =							torch.from_numpy(UpperCAmelCase  ).to(device=UpperCAmelCase	, dtype=UpperCAmelCase  )
            A_	     =							{
                "image": init_image,
                "latents": latents,
                "generator": generator,
                "num_inference_steps": 3,
                "guidance_scale": 7.5,
                "output_type": "numpy",
            }
            return inputs





      def 		__A  (      self       :							Optional[int]  ):
            A_	     =							StableDiffusionImageVariationPipeline.from_pretrained(
                "lambdalabs/sd-image-variations-diffusers"	, safety_checker=UpperCAmelCase  )
            A_	     =							sd_pipe.to(UpperCAmelCase  )
            sd_pipe.set_progress_bar_config(disable=UpperCAmelCase  )

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

            assert image.shape == (1, 512, 512, 3)
            A_	     =							np.array([0.84_491, 0.90_789, 0.75_708, 0.78_734, 0.83_485, 0.70_099, 0.66_938, 0.68_727, 0.61_379]  )
            assert np.abs(image_slice - expected_slice  ).max() < 6E-3





      def 		__A  (      self       :							Tuple  ):
            A_	     =							0

            def callback_fn(UpperCAmelCase       :							int	, UpperCAmelCase       :							int	, UpperCAmelCase       :							torch.FloatTensor  ) -> None:
                  A_	     =							True
                  nonlocal number_of_steps
                  number_of_steps += 1
                  if step == 1:
                        A_	     =							latents.detach().cpu().numpy()
                        assert latents.shape == (1, 4, 64, 64)
                        A_	     =							latents[0, -3:, -3:, -1]
                        A_	     =							np.array(
                            [-0.1_621, 0.2_837, -0.7_979, -0.1_221, -1.3_057, 0.7_681, -2.1_191, 0.0_464, 1.6_309]  )

                        assert np.abs(latents_slice.flatten() - expected_slice  ).max() < 5E-2
                  elif step == 2:
                        A_	     =							latents.detach().cpu().numpy()
                        assert latents.shape == (1, 4, 64, 64)
                        A_	     =							latents[0, -3:, -3:, -1]
                        A_	     =							np.array([0.6_299, 1.7_500, 1.1_992, -2.1_582, -1.8_994, 0.7_334, -0.7_090, 1.0_137, 1.5_273]  )

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

            A_	     =							False

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

            A_	     =							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 		__A  (      self       :							Tuple  ):
            torch.cuda.empty_cache()
            torch.cuda.reset_max_memory_allocated()
            torch.cuda.reset_peak_memory_stats()

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

            A_	     =							self.get_inputs(UpperCAmelCase	, dtype=torch.floataa  )
            A_	     =							pipe(**UpperCAmelCase  )

            A_	     =							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      _a				( unittest.TestCase							):




      """simple docstring"""

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





      def 		__A  (      self       :							Any	, UpperCAmelCase       :							List[Any]	, UpperCAmelCase       :							Union[str, Any]="cpu"	, UpperCAmelCase       :							Any=torch.floataa	, UpperCAmelCase       :							Optional[int]=0  ):
            A_	     =							torch.Generator(device=UpperCAmelCase  ).manual_seed(UpperCAmelCase  )
            A_	     =							load_image(
                "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
                "/stable_diffusion_imgvar/input_image_vermeer.png"  )
            A_	     =							np.random.RandomState(UpperCAmelCase  ).standard_normal((1, 4, 64, 64)  )
            A_	     =							torch.from_numpy(UpperCAmelCase  ).to(device=UpperCAmelCase	, dtype=UpperCAmelCase  )
            A_	     =							{
                "image": init_image,
                "latents": latents,
                "generator": generator,
                "num_inference_steps": 50,
                "guidance_scale": 7.5,
                "output_type": "numpy",
            }
            return inputs





      def 		__A  (      self       :							List[str]  ):
            A_	     =							StableDiffusionImageVariationPipeline.from_pretrained("fusing/sd-image-variations-diffusers"  )
            sd_pipe.to(UpperCAmelCase  )
            sd_pipe.set_progress_bar_config(disable=UpperCAmelCase  )

            A_	     =							self.get_inputs(UpperCAmelCase  )
            A_	     =							sd_pipe(**UpperCAmelCase  ).images[0]

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







      def 		__A  (      self       :							Dict  ):
            A_	     =							StableDiffusionImageVariationPipeline.from_pretrained("fusing/sd-image-variations-diffusers"  )
            A_	     =							DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config  )
            sd_pipe.to(UpperCAmelCase  )
            sd_pipe.set_progress_bar_config(disable=UpperCAmelCase  )

            A_	     =							self.get_inputs(UpperCAmelCase  )
            A_	     =							25
            A_	     =							sd_pipe(**UpperCAmelCase  ).images[0]

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


__a						:Dict            = {hex(i)[2:]: i for i in range(16)}  # Use [:2] to strip off the leading '0x'

def 			__snake_case	(   __UpperCamelCase  :   str						):



      """simple docstring"""

      A_	     =							hex_string.strip().lower()
      if not hex_string:
            raise ValueError("Empty string was passed to the function"						)
      A_	     =							hex_string[0] == "-"
      if is_negative:
            A_	     =							hex_string[1:]
      if not all(char in hex_table for char in hex_string						):
            raise ValueError("Non-hexadecimal value was passed to the function"						)
      A_	     =							0
      for char in hex_string:
            A_	     =							16 * decimal_number + hex_table[char]
      return -decimal_number if is_negative else decimal_number


if __name__ == "__main__":
      from doctest import testmod

      testmod()