

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      lowerCamelCase_ (			_lowerCAmelCase	,				_lowerCAmelCase	,				_lowerCAmelCase	,				unittest.TestCase):
 _lowerCAmelCase									=	StableDiffusionImageVariationPipeline
 _lowerCAmelCase									=	IMAGE_VARIATION_PARAMS
 _lowerCAmelCase									=	IMAGE_VARIATION_BATCH_PARAMS
 _lowerCAmelCase									=	frozenset([])
 # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
 _lowerCAmelCase									=	frozenset([])




 def     a       (			self							):
   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=_UpperCAmelCase							)
   torch.manual_seed(0							)
   __UpperCAmelCase	:  Union[str, Any]		=     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	:  Tuple		=     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	:  int		=     CLIPVisionModelWithProjection(_UpperCAmelCase							)
   __UpperCAmelCase	:  Tuple		=     CLIPImageProcessor(crop_size=32  ,  size=32							)

   __UpperCAmelCase	:  Tuple		=     {
       'unet': unet,
       'scheduler': scheduler,
       'vae': vae,
       'image_encoder': image_encoder,
       'feature_extractor': feature_extractor,
       'safety_checker': None,
   }
   return components




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




 def     a       (			self							):
   __UpperCAmelCase	:  Optional[Any]		=     'cpu'  # ensure determinism for the device-dependent torch.Generator
   __UpperCAmelCase	:  Tuple		=     self.get_dummy_components()
   __UpperCAmelCase	:  Optional[int]		=     StableDiffusionImageVariationPipeline(**_UpperCAmelCase							)
   __UpperCAmelCase	:  Tuple		=     sd_pipe.to(_UpperCAmelCase							)
   sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase							)

   __UpperCAmelCase	:  List[Any]		=     self.get_dummy_inputs(_UpperCAmelCase							)
   __UpperCAmelCase	:  Tuple		=     sd_pipe(**_UpperCAmelCase							).images
   __UpperCAmelCase	:  Optional[int]		=     image[0, -3:, -3:, -1]

   assert image.shape == (1, 64, 64, 3)
   __UpperCAmelCase	:  List[str]		=     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     a       (			self							):
   __UpperCAmelCase	:  List[Any]		=     'cpu'  # ensure determinism for the device-dependent torch.Generator
   __UpperCAmelCase	:  Dict		=     self.get_dummy_components()
   __UpperCAmelCase	:  List[str]		=     StableDiffusionImageVariationPipeline(**_UpperCAmelCase							)
   __UpperCAmelCase	:  int		=     sd_pipe.to(_UpperCAmelCase							)
   sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase							)

   __UpperCAmelCase	:  Optional[int]		=     self.get_dummy_inputs(_UpperCAmelCase							)
   __UpperCAmelCase	:  Dict		=     2 * [inputs['image']]
   __UpperCAmelCase	:  Tuple		=     sd_pipe(**_UpperCAmelCase							)

   __UpperCAmelCase	:  List[Any]		=     output.images

   __UpperCAmelCase	:  Tuple		=     image[-1, -3:, -3:, -1]

   assert image.shape == (2, 64, 64, 3)
   __UpperCAmelCase	:  int		=     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     a       (			self							):
   super().test_inference_batch_single_identical(expected_max_diff=3E-3							)



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




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




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




 def     a       (			self							):
   __UpperCAmelCase	:  List[str]		=     StableDiffusionImageVariationPipeline.from_pretrained(
       'lambdalabs/sd-image-variations-diffusers'  ,  safety_checker=_UpperCAmelCase							)
   __UpperCAmelCase	:  List[str]		=     sd_pipe.to(_UpperCAmelCase							)
   sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase							)

   __UpperCAmelCase	:  str		=     self.get_inputs(_UpperCAmelCase							)
   __UpperCAmelCase	:  Tuple		=     sd_pipe(**_UpperCAmelCase							).images
   __UpperCAmelCase	:  Union[str, Any]		=     image[0, -3:, -3:, -1].flatten()

   assert image.shape == (1, 512, 512, 3)
   __UpperCAmelCase	:  List[str]		=     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     a       (			self							):
   __UpperCAmelCase	:  List[Any]		=     0

   def callback_fn(_UpperCAmelCase  ,  _UpperCAmelCase  ,  _UpperCAmelCase							) -> None:
     __UpperCAmelCase	:  Dict		=     True
     nonlocal number_of_steps
     number_of_steps += 1
     if step == 1:
       __UpperCAmelCase	:  Optional[Any]		=     latents.detach().cpu().numpy()
       assert latents.shape == (1, 4, 64, 64)
       __UpperCAmelCase	:  int		=     latents[0, -3:, -3:, -1]
       __UpperCAmelCase	:  Tuple		=     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:
       __UpperCAmelCase	:  Dict		=     latents.detach().cpu().numpy()
       assert latents.shape == (1, 4, 64, 64)
       __UpperCAmelCase	:  Any		=     latents[0, -3:, -3:, -1]
       __UpperCAmelCase	:  Optional[int]		=     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

   __UpperCAmelCase	:  Union[str, Any]		=     False

   __UpperCAmelCase	:  Optional[int]		=     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()

   __UpperCAmelCase	:  Optional[int]		=     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							):
   torch.cuda.empty_cache()
   torch.cuda.reset_max_memory_allocated()
   torch.cuda.reset_peak_memory_stats()

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

   __UpperCAmelCase	:  Optional[int]		=     self.get_inputs(_UpperCAmelCase  ,  dtype=torch.floataa							)
   __UpperCAmelCase	:  Dict		=     pipe(**_UpperCAmelCase							)

   __UpperCAmelCase	:  str		=     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      lowerCamelCase_ (			unittest.TestCase):




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




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




 def     a       (			self							):
   __UpperCAmelCase	:  List[str]		=     StableDiffusionImageVariationPipeline.from_pretrained('fusing/sd-image-variations-diffusers'							)
   sd_pipe.to(_UpperCAmelCase							)
   sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase							)

   __UpperCAmelCase	:  List[str]		=     self.get_inputs(_UpperCAmelCase							)
   __UpperCAmelCase	:  int		=     sd_pipe(**_UpperCAmelCase							).images[0]

   __UpperCAmelCase	:  str		=     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     a       (			self							):
   __UpperCAmelCase	:  Any		=     StableDiffusionImageVariationPipeline.from_pretrained('fusing/sd-image-variations-diffusers'							)
   __UpperCAmelCase	:  Dict		=     DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config							)
   sd_pipe.to(_UpperCAmelCase							)
   sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase							)

   __UpperCAmelCase	:  Tuple		=     self.get_inputs(_UpperCAmelCase							)
   __UpperCAmelCase	:  Tuple		=     25
   __UpperCAmelCase	:  List[Any]		=     sd_pipe(**_UpperCAmelCase							).images[0]

   __UpperCAmelCase	:  Union[str, Any]		=     load_numpy(
       'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
       '/stable_diffusion_imgvar/lambdalabs_variations_dpm_multi.npy'							)
   __UpperCAmelCase	:  Union[str, Any]		=     np.abs(expected_image - image							).max()
   assert max_diff < 1E-3







from __future__ import annotations
def lowerCamelCase_       (			A			:      list[int]						):




      """simple docstring"""
      if not list_of_ints:
            return []

      if min(A						) < 0:
            raise ValueError('''All numbers must be positive'''						)

      lowerCAmelCase_						=      max(len(bin(A						)[2:]						) for x in list_of_ints						)
      return _msd_radix_sort(A		,					A						)
def lowerCamelCase_       (			A			:      list[int]		,					A			:      int						):




      """simple docstring"""
      if bit_position == 0 or len(A						) in [0, 1]:
            return list_of_ints

      lowerCAmelCase_						=      []
      lowerCAmelCase_						=      []
      # Split numbers based on bit at bit_position from the right
      for number in list_of_ints:
            if (number >> (bit_position - 1)) & 1:
                  # number has a one at bit bit_position
                  ones.append(A						)
            else:
                  # number has a zero at bit bit_position
                  zeros.append(A						)

    # recursively split both lists further
      lowerCAmelCase_						=      _msd_radix_sort(A		,					bit_position - 1						)
      lowerCAmelCase_						=      _msd_radix_sort(A		,					bit_position - 1						)

      # recombine lists
      lowerCAmelCase_						=      zeros
      res.extend(A						)

      return res
def lowerCamelCase_       (			A			:      list[int]						):




      """simple docstring"""
      lowerCAmelCase_						=      len(A						)
      if not list_of_ints or length == 1:
            return

      if min(A						) < 0:
            raise ValueError('''All numbers must be positive'''						)

      lowerCAmelCase_						=      max(len(bin(A						)[2:]						) for x in list_of_ints						)
      _msd_radix_sort_inplace(A		,					A		,					0		,					A						)
def lowerCamelCase_       (			A			:      list[int]		,					A			:      int		,					A			:      int		,					A			:      int						):




      """simple docstring"""
      if bit_position == 0 or end_index - begin_index <= 1:
            return

      bit_position -= 1

      lowerCAmelCase_						=      begin_index
      lowerCAmelCase_						=      end_index - 1
      while i <= j:
            lowerCAmelCase_						=      False
            if not (list_of_ints[i] >> bit_position) & 1:
                  # found zero at the beginning
                  i += 1
                  lowerCAmelCase_						=      True
            if (list_of_ints[j] >> bit_position) & 1:
                  # found one at the end
                  j -= 1
                  lowerCAmelCase_						=      True

            if changed:
                  continue

            lowerCAmelCase_			,		lowerCAmelCase_						=      list_of_ints[j], list_of_ints[i]
            j -= 1
            if j != i:
                  i += 1

      _msd_radix_sort_inplace(A		,					A		,					A		,					A						)
      _msd_radix_sort_inplace(A		,					A		,					A		,					A						)


if __name__ == "__main__":
 import doctest

 doctest.testmod()



