






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       UpperCAmelCase_						( lowercase,     lowercase,     lowercase,     unittest.TestCase			):



      """simple docstring"""
      UpperCamelCase_    : Tuple									=StableDiffusionImageVariationPipeline
      UpperCamelCase_    : List[str]									=IMAGE_VARIATION_PARAMS
      UpperCamelCase_    : Optional[Any]									=IMAGE_VARIATION_BATCH_PARAMS
      UpperCamelCase_    : Optional[Any]									=frozenset([]			)
      # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
      UpperCamelCase_    : Optional[int]									=frozenset([]			)
      def    UpperCAmelCase						(    self			)    ->				Any:
          torch.manual_seed(0			)
          UpperCamelCase	:Optional[int]     							=       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   ,    )
          UpperCamelCase	:List[Any]     							=       PNDMScheduler(skip_prk_steps=SCREAMING_SNAKE_CASE_			)
          torch.manual_seed(0			)
          UpperCamelCase	:Tuple     							=       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			)
          UpperCamelCase	:Optional[Any]     							=       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   ,    )
          UpperCamelCase	:List[Any]     							=       CLIPVisionModelWithProjection(SCREAMING_SNAKE_CASE_			)
          UpperCamelCase	:Tuple     							=       CLIPImageProcessor(crop_size=32   ,    size=32			)

          UpperCamelCase	:Optional[Any]     							=       {
              '''unet''': unet,
              '''scheduler''': scheduler,
              '''vae''': vae,
              '''image_encoder''': image_encoder,
              '''feature_extractor''': feature_extractor,
              '''safety_checker''': None,
          }
          return components
      def    UpperCAmelCase						(    self   ,    SCREAMING_SNAKE_CASE_   ,    SCREAMING_SNAKE_CASE_=0			)    ->				Any:
          UpperCamelCase	:Any     							=       floats_tensor((1, 3, 32, 32)   ,    rng=random.Random(SCREAMING_SNAKE_CASE_			)			)
          UpperCamelCase	:List[Any]     							=       image.cpu().permute(0   ,    2   ,    3   ,    1			)[0]
          UpperCamelCase	:Optional[int]     							=       Image.fromarray(np.uinta(SCREAMING_SNAKE_CASE_			)			).convert('''RGB'''			).resize((32, 32)			)
          if str(SCREAMING_SNAKE_CASE_			).startswith('''mps'''			):
              UpperCamelCase	:List[str]     							=       torch.manual_seed(SCREAMING_SNAKE_CASE_			)
          else:
              UpperCamelCase	:List[str]     							=       torch.Generator(device=SCREAMING_SNAKE_CASE_			).manual_seed(SCREAMING_SNAKE_CASE_			)
          UpperCamelCase	:Union[str, Any]     							=       {
              '''image''': image,
              '''generator''': generator,
              '''num_inference_steps''': 2,
              '''guidance_scale''': 6.0,
              '''output_type''': '''numpy''',
          }
          return inputs
      def    UpperCAmelCase						(    self			)    ->				Optional[Any]:
          UpperCamelCase	:str     							=       '''cpu'''  # ensure determinism for the device-dependent torch.Generator
          UpperCamelCase	:List[str]     							=       self.get_dummy_components()
          UpperCamelCase	:Tuple     							=       StableDiffusionImageVariationPipeline(**SCREAMING_SNAKE_CASE_			)
          UpperCamelCase	:Any     							=       sd_pipe.to(SCREAMING_SNAKE_CASE_			)
          sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_			)

          UpperCamelCase	:List[str]     							=       self.get_dummy_inputs(SCREAMING_SNAKE_CASE_			)
          UpperCamelCase	:List[Any]     							=       sd_pipe(**SCREAMING_SNAKE_CASE_			).images
          UpperCamelCase	:Dict     							=       image[0, -3:, -3:, -1]

          assert image.shape == (1, 64, 64, 3)
          UpperCamelCase	:Tuple     							=       np.array([0.5239, 0.5723, 0.4796, 0.5049, 0.5550, 0.4685, 0.5329, 0.4891, 0.4921]			)

          assert np.abs(image_slice.flatten() - expected_slice			).max() < 1e-3
      def    UpperCAmelCase						(    self			)    ->				Optional[Any]:
          UpperCamelCase	:List[str]     							=       '''cpu'''  # ensure determinism for the device-dependent torch.Generator
          UpperCamelCase	:List[Any]     							=       self.get_dummy_components()
          UpperCamelCase	:Optional[int]     							=       StableDiffusionImageVariationPipeline(**SCREAMING_SNAKE_CASE_			)
          UpperCamelCase	:Tuple     							=       sd_pipe.to(SCREAMING_SNAKE_CASE_			)
          sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_			)

          UpperCamelCase	:Tuple     							=       self.get_dummy_inputs(SCREAMING_SNAKE_CASE_			)
          UpperCamelCase	:str     							=       2 * [inputs['''image''']]
          UpperCamelCase	:List[str]     							=       sd_pipe(**SCREAMING_SNAKE_CASE_			)

          UpperCamelCase	:List[str]     							=       output.images

          UpperCamelCase	:List[str]     							=       image[-1, -3:, -3:, -1]

          assert image.shape == (2, 64, 64, 3)
          UpperCamelCase	:Dict     							=       np.array([0.6892, 0.5637, 0.5836, 0.5771, 0.6254, 0.6409, 0.5580, 0.5569, 0.5289]			)

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




      def    UpperCAmelCase						(    self			)    ->				Any:
          super().test_inference_batch_single_identical(expected_max_diff=3e-3			)







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



      """simple docstring"""
      def    UpperCAmelCase						(    self			)    ->				Any:
          super().tearDown()
          gc.collect()
          torch.cuda.empty_cache()
      def    UpperCAmelCase						(    self   ,    SCREAMING_SNAKE_CASE_   ,    SCREAMING_SNAKE_CASE_="cpu"   ,    SCREAMING_SNAKE_CASE_=torch.floataa   ,    SCREAMING_SNAKE_CASE_=0			)    ->				str:
          UpperCamelCase	:int     							=       torch.Generator(device=SCREAMING_SNAKE_CASE_			).manual_seed(SCREAMING_SNAKE_CASE_			)
          UpperCamelCase	:int     							=       load_image(
              '''https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'''
              '''/stable_diffusion_imgvar/input_image_vermeer.png'''			)
          UpperCamelCase	:List[Any]     							=       np.random.RandomState(SCREAMING_SNAKE_CASE_			).standard_normal((1, 4, 64, 64)			)
          UpperCamelCase	:int     							=       torch.from_numpy(SCREAMING_SNAKE_CASE_			).to(device=SCREAMING_SNAKE_CASE_   ,    dtype=SCREAMING_SNAKE_CASE_			)
          UpperCamelCase	:List[str]     							=       {
              '''image''': init_image,
              '''latents''': latents,
              '''generator''': generator,
              '''num_inference_steps''': 3,
              '''guidance_scale''': 7.5,
              '''output_type''': '''numpy''',
          }
          return inputs
      def    UpperCAmelCase						(    self			)    ->				Optional[Any]:
          UpperCamelCase	:Optional[int]     							=       StableDiffusionImageVariationPipeline.from_pretrained(
              '''lambdalabs/sd-image-variations-diffusers'''   ,    safety_checker=SCREAMING_SNAKE_CASE_			)
          UpperCamelCase	:List[Any]     							=       sd_pipe.to(SCREAMING_SNAKE_CASE_			)
          sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_			)

          UpperCamelCase	:Dict     							=       self.get_inputs(SCREAMING_SNAKE_CASE_			)
          UpperCamelCase	:List[str]     							=       sd_pipe(**SCREAMING_SNAKE_CASE_			).images
          UpperCamelCase	:List[str]     							=       image[0, -3:, -3:, -1].flatten()

          assert image.shape == (1, 512, 512, 3)
          UpperCamelCase	:int     							=       np.array([0.8_4491, 0.9_0789, 0.7_5708, 0.7_8734, 0.8_3485, 0.7_0099, 0.6_6938, 0.6_8727, 0.6_1379]			)
          assert np.abs(image_slice - expected_slice			).max() < 6e-3
      def    UpperCAmelCase						(    self			)    ->				Dict:
          UpperCamelCase	:Any     							=       0

          def callback_fn(SCREAMING_SNAKE_CASE_   ,    SCREAMING_SNAKE_CASE_   ,    SCREAMING_SNAKE_CASE_			) -> None:
              UpperCamelCase	:Optional[Any]     							=       True
              nonlocal number_of_steps
              number_of_steps += 1
              if step == 1:
                  UpperCamelCase	:List[Any]     							=       latents.detach().cpu().numpy()
                  assert latents.shape == (1, 4, 64, 64)
                  UpperCamelCase	:Dict     							=       latents[0, -3:, -3:, -1]
                  UpperCamelCase	:Tuple     							=       np.array(
                      [-0.1621, 0.2837, -0.7979, -0.1221, -1.3057, 0.7681, -2.1191, 0.0464, 1.6309]			)

                  assert np.abs(latents_slice.flatten() - expected_slice			).max() < 5e-2
              elif step == 2:
                  UpperCamelCase	:List[str]     							=       latents.detach().cpu().numpy()
                  assert latents.shape == (1, 4, 64, 64)
                  UpperCamelCase	:Union[str, Any]     							=       latents[0, -3:, -3:, -1]
                  UpperCamelCase	:str     							=       np.array([0.6299, 1.7500, 1.1992, -2.1582, -1.8994, 0.7334, -0.7090, 1.0137, 1.5273]			)

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

          UpperCamelCase	:Tuple     							=       False

          UpperCamelCase	:List[str]     							=       StableDiffusionImageVariationPipeline.from_pretrained(
              '''fusing/sd-image-variations-diffusers'''   ,    safety_checker=SCREAMING_SNAKE_CASE_   ,    torch_dtype=torch.floataa   ,    )
          pipe.to(SCREAMING_SNAKE_CASE_			)
          pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_			)
          pipe.enable_attention_slicing()

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




      def    UpperCAmelCase						(    self			)    ->				Any:
          torch.cuda.empty_cache()
          torch.cuda.reset_max_memory_allocated()
          torch.cuda.reset_peak_memory_stats()

          UpperCamelCase	:Tuple     							=       '''fusing/sd-image-variations-diffusers'''
          UpperCamelCase	:Any     							=       StableDiffusionImageVariationPipeline.from_pretrained(
              SCREAMING_SNAKE_CASE_   ,    safety_checker=SCREAMING_SNAKE_CASE_   ,    torch_dtype=torch.floataa			)
          UpperCamelCase	:Tuple     							=       pipe.to(SCREAMING_SNAKE_CASE_			)
          pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_			)
          pipe.enable_attention_slicing(1			)
          pipe.enable_sequential_cpu_offload()

          UpperCamelCase	:int     							=       self.get_inputs(SCREAMING_SNAKE_CASE_   ,    dtype=torch.floataa			)
          UpperCamelCase	:str     							=       pipe(**SCREAMING_SNAKE_CASE_			)

          UpperCamelCase	:Dict     							=       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       UpperCAmelCase_						( unittest.TestCase			):



      """simple docstring"""
      def    UpperCAmelCase						(    self			)    ->				Dict:
          super().tearDown()
          gc.collect()
          torch.cuda.empty_cache()
      def    UpperCAmelCase						(    self   ,    SCREAMING_SNAKE_CASE_   ,    SCREAMING_SNAKE_CASE_="cpu"   ,    SCREAMING_SNAKE_CASE_=torch.floataa   ,    SCREAMING_SNAKE_CASE_=0			)    ->				Union[str, Any]:
          UpperCamelCase	:Dict     							=       torch.Generator(device=SCREAMING_SNAKE_CASE_			).manual_seed(SCREAMING_SNAKE_CASE_			)
          UpperCamelCase	:str     							=       load_image(
              '''https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'''
              '''/stable_diffusion_imgvar/input_image_vermeer.png'''			)
          UpperCamelCase	:List[str]     							=       np.random.RandomState(SCREAMING_SNAKE_CASE_			).standard_normal((1, 4, 64, 64)			)
          UpperCamelCase	:Optional[Any]     							=       torch.from_numpy(SCREAMING_SNAKE_CASE_			).to(device=SCREAMING_SNAKE_CASE_   ,    dtype=SCREAMING_SNAKE_CASE_			)
          UpperCamelCase	:str     							=       {
              '''image''': init_image,
              '''latents''': latents,
              '''generator''': generator,
              '''num_inference_steps''': 50,
              '''guidance_scale''': 7.5,
              '''output_type''': '''numpy''',
          }
          return inputs
      def    UpperCAmelCase						(    self			)    ->				Any:
          UpperCamelCase	:int     							=       StableDiffusionImageVariationPipeline.from_pretrained('''fusing/sd-image-variations-diffusers'''			)
          sd_pipe.to(SCREAMING_SNAKE_CASE_			)
          sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_			)

          UpperCamelCase	:List[str]     							=       self.get_inputs(SCREAMING_SNAKE_CASE_			)
          UpperCamelCase	:Tuple     							=       sd_pipe(**SCREAMING_SNAKE_CASE_			).images[0]

          UpperCamelCase	:Tuple     							=       load_numpy(
              '''https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'''
              '''/stable_diffusion_imgvar/lambdalabs_variations_pndm.npy'''			)
          UpperCamelCase	:List[str]     							=       np.abs(expected_image - image			).max()
          assert max_diff < 1e-3




      def    UpperCAmelCase						(    self			)    ->				Any:
          UpperCamelCase	:Tuple     							=       StableDiffusionImageVariationPipeline.from_pretrained('''fusing/sd-image-variations-diffusers'''			)
          UpperCamelCase	:int     							=       DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config			)
          sd_pipe.to(SCREAMING_SNAKE_CASE_			)
          sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_			)

          UpperCamelCase	:Dict     							=       self.get_inputs(SCREAMING_SNAKE_CASE_			)
          UpperCamelCase	:Tuple     							=       25
          UpperCamelCase	:Optional[Any]     							=       sd_pipe(**SCREAMING_SNAKE_CASE_			).images[0]

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













from __future__ import annotations

from enum import Enum







class       UpperCAmelCase_						( lowercase			):



      """simple docstring"""
      UpperCamelCase_    : Optional[Any]									=24
      UpperCamelCase_    : Any									=21
      UpperCamelCase_    : Union[str, Any]									=18
      UpperCamelCase_    : Tuple									=15
      UpperCamelCase_    : List[Any]									=12
      UpperCamelCase_    : Optional[Any]									=9
      UpperCamelCase_    : List[str]									=6
      UpperCamelCase_    : List[str]									=3
      UpperCamelCase_    : Any									=2
      UpperCamelCase_    : List[str]									=1
      UpperCamelCase_    : Optional[Any]									=-1
      UpperCamelCase_    : str									=-2
      UpperCamelCase_    : Any									=-3
      UpperCamelCase_    : Optional[int]									=-6
      UpperCamelCase_    : int									=-9
      UpperCamelCase_    : Optional[int]									=-12
      UpperCamelCase_    : Optional[Any]									=-15
      UpperCamelCase_    : str									=-18
      UpperCamelCase_    : int									=-21
      UpperCamelCase_    : Union[str, Any]									=-24







class       UpperCAmelCase_						( lowercase			):



      """simple docstring"""
      UpperCamelCase_    : Tuple									=8
      UpperCamelCase_    : str									=7
      UpperCamelCase_    : List[str]									=6
      UpperCamelCase_    : List[str]									=5
      UpperCamelCase_    : int									=4
      UpperCamelCase_    : Optional[Any]									=3
      UpperCamelCase_    : Optional[Any]									=2
      UpperCamelCase_    : List[str]									=1







def   _A       (    SCREAMING_SNAKE_CASE__  :     float					,  SCREAMING_SNAKE_CASE__  :     str | SIUnit					,  SCREAMING_SNAKE_CASE__  :     str | SIUnit					,  ):
    if isinstance(SCREAMING_SNAKE_CASE__					,  SCREAMING_SNAKE_CASE__	):
        UpperCamelCase	:str     							=       SIUnit[known_prefix.lower()]
    if isinstance(SCREAMING_SNAKE_CASE__					,  SCREAMING_SNAKE_CASE__	):
        UpperCamelCase	:Any     							=       SIUnit[unknown_prefix.lower()]
    UpperCamelCase	:float     							=       known_amount * (
        10 ** (known_prefix.value - unknown_prefix.value)
    )
    return unknown_amount


def   _A       (    SCREAMING_SNAKE_CASE__  :     float					,  SCREAMING_SNAKE_CASE__  :     str | BinaryUnit					,  SCREAMING_SNAKE_CASE__  :     str | BinaryUnit					,  ):
    if isinstance(SCREAMING_SNAKE_CASE__					,  SCREAMING_SNAKE_CASE__	):
        UpperCamelCase	:Dict     							=       BinaryUnit[known_prefix.lower()]
    if isinstance(SCREAMING_SNAKE_CASE__					,  SCREAMING_SNAKE_CASE__	):
        UpperCamelCase	:List[str]     							=       BinaryUnit[unknown_prefix.lower()]
    UpperCamelCase	:float     							=       known_amount * (
        2 ** ((known_prefix.value - unknown_prefix.value) * 10)
    )
    return unknown_amount


if __name__ == "__main__":
 import doctest

 doctest.testmod()





