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 (  lowercase_       , lowercase_       , lowercase_       , unittest.TestCase ):
 __lowerCamelCase     : Any 	=   StableDiffusionImageVariationPipeline
 __lowerCamelCase     : List[str] 	=   IMAGE_VARIATION_PARAMS
 __lowerCamelCase     : str 	=   IMAGE_VARIATION_BATCH_PARAMS
 __lowerCamelCase     : Dict 	=   frozenset([] )
 # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
 __lowerCamelCase     : List[str] 	=   frozenset([] )
 def 						UpperCAmelCase__							(	self    )						->				Any:
       '''simple docstring'''





       torch.manual_seed(0    )
       UpperCAmelCase     :			Optional[Any]     	=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     :			Optional[int]     	=PNDMScheduler(skip_prk_steps=_lowerCAmelCase    )
       torch.manual_seed(0    )
       UpperCAmelCase     :			List[str]     	=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     :			str     	=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     :			Tuple     	=CLIPVisionModelWithProjection(_lowerCAmelCase    )
       UpperCAmelCase     :			int     	=CLIPImageProcessor(crop_size=32		, size=32    )

       UpperCAmelCase     :			Union[str, Any]     	={
           '''unet''': unet,
           '''scheduler''': scheduler,
           '''vae''': vae,
           '''image_encoder''': image_encoder,
           '''feature_extractor''': feature_extractor,
           '''safety_checker''': None,
       }
       return components
 def 						UpperCAmelCase__							(	self		, snake_case__		, snake_case__=0    )						->				str:
       '''simple docstring'''





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





       UpperCAmelCase     :			List[str]     	='''cpu'''  # ensure determinism for the device-dependent torch.Generator
       UpperCAmelCase     :			str     	=self.get_dummy_components()
       UpperCAmelCase     :			Any     	=StableDiffusionImageVariationPipeline(**_lowerCAmelCase    )
       UpperCAmelCase     :			Optional[int]     	=sd_pipe.to(_lowerCAmelCase    )
       sd_pipe.set_progress_bar_config(disable=_lowerCAmelCase    )

       UpperCAmelCase     :			Any     	=self.get_dummy_inputs(_lowerCAmelCase    )
       UpperCAmelCase     :			Optional[int]     	=sd_pipe(**_lowerCAmelCase    ).images
       UpperCAmelCase     :			Dict     	=image[0, -3:, -3:, -1]

       assert image.shape == (1, 64, 64, 3)
       UpperCAmelCase     :			List[Any]     	=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    )						->				Tuple:
       '''simple docstring'''





       UpperCAmelCase     :			Any     	='''cpu'''  # ensure determinism for the device-dependent torch.Generator
       UpperCAmelCase     :			List[Any]     	=self.get_dummy_components()
       UpperCAmelCase     :			Optional[int]     	=StableDiffusionImageVariationPipeline(**_lowerCAmelCase    )
       UpperCAmelCase     :			Optional[int]     	=sd_pipe.to(_lowerCAmelCase    )
       sd_pipe.set_progress_bar_config(disable=_lowerCAmelCase    )

       UpperCAmelCase     :			List[Any]     	=self.get_dummy_inputs(_lowerCAmelCase    )
       UpperCAmelCase     :			Any     	=2 * [inputs['''image''']]
       UpperCAmelCase     :			int     	=sd_pipe(**_lowerCAmelCase    )

       UpperCAmelCase     :			List[str]     	=output.images

       UpperCAmelCase     :			Optional[Any]     	=image[-1, -3:, -3:, -1]

       assert image.shape == (2, 64, 64, 3)
       UpperCAmelCase     :			Optional[int]     	=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    )						->				Optional[int]:
       '''simple docstring'''





       super().test_inference_batch_single_identical(expected_max_diff=3e-3    )







@slow
@require_torch_gpu
class       __snake_case (  unittest.TestCase ):
 def 						UpperCAmelCase__							(	self    )						->				Optional[Any]:
       '''simple docstring'''





       super().tearDown()
       gc.collect()
       torch.cuda.empty_cache()
 def 						UpperCAmelCase__							(	self		, snake_case__		, snake_case__="cpu"		, snake_case__=torch.floataa		, snake_case__=0    )						->				Any:
       '''simple docstring'''





       UpperCAmelCase     :			int     	=torch.Generator(device=_lowerCAmelCase    ).manual_seed(_lowerCAmelCase    )
       UpperCAmelCase     :			int     	=load_image(
           '''https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'''
           '''/stable_diffusion_imgvar/input_image_vermeer.png'''    )
       UpperCAmelCase     :			str     	=np.random.RandomState(_lowerCAmelCase    ).standard_normal((1, 4, 64, 64)    )
       UpperCAmelCase     :			Dict     	=torch.from_numpy(_lowerCAmelCase    ).to(device=_lowerCAmelCase		, dtype=_lowerCAmelCase    )
       UpperCAmelCase     :			List[Any]     	={
           '''image''': init_image,
           '''latents''': latents,
           '''generator''': generator,
           '''num_inference_steps''': 3,
           '''guidance_scale''': 7.5,
           '''output_type''': '''numpy''',
       }
       return inputs
 def 						UpperCAmelCase__							(	self    )						->				Union[str, Any]:
       '''simple docstring'''





       UpperCAmelCase     :			Optional[Any]     	=StableDiffusionImageVariationPipeline.from_pretrained(
           '''lambdalabs/sd-image-variations-diffusers'''		, safety_checker=_lowerCAmelCase    )
       UpperCAmelCase     :			Tuple     	=sd_pipe.to(_lowerCAmelCase    )
       sd_pipe.set_progress_bar_config(disable=_lowerCAmelCase    )

       UpperCAmelCase     :			Union[str, Any]     	=self.get_inputs(_lowerCAmelCase    )
       UpperCAmelCase     :			Any     	=sd_pipe(**_lowerCAmelCase    ).images
       UpperCAmelCase     :			Optional[int]     	=image[0, -3:, -3:, -1].flatten()

       assert image.shape == (1, 512, 512, 3)
       UpperCAmelCase     :			Optional[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    )						->				List[Any]:
       '''simple docstring'''





       UpperCAmelCase     :			Tuple     	=0

       def callback_fn(snake_case__		, snake_case__		, snake_case__    ) -> None:
             UpperCAmelCase     :			Dict     	=True
             nonlocal number_of_steps
             number_of_steps += 1
             if step == 1:
                   UpperCAmelCase     :			Tuple     	=latents.detach().cpu().numpy()
                   assert latents.shape == (1, 4, 64, 64)
                   UpperCAmelCase     :			Tuple     	=latents[0, -3:, -3:, -1]
                   UpperCAmelCase     :			Dict     	=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     :			Union[str, Any]     	=latents.detach().cpu().numpy()
                   assert latents.shape == (1, 4, 64, 64)
                   UpperCAmelCase     :			str     	=latents[0, -3:, -3:, -1]
                   UpperCAmelCase     :			Tuple     	=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     :			List[str]     	=False

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

       UpperCAmelCase     :			Optional[Any]     	=self.get_inputs(_lowerCAmelCase		, dtype=torch.floataa    )
       pipe(**_lowerCAmelCase		, callback=_lowerCAmelCase		, callback_steps=1    )
       assert callback_fn.has_been_called
       assert number_of_steps == inputs["num_inference_steps"]






 def 						UpperCAmelCase__							(	self    )						->				Union[str, Any]:
       '''simple docstring'''





       torch.cuda.empty_cache()
       torch.cuda.reset_max_memory_allocated()
       torch.cuda.reset_peak_memory_stats()

       UpperCAmelCase     :			int     	='''fusing/sd-image-variations-diffusers'''
       UpperCAmelCase     :			Optional[Any]     	=StableDiffusionImageVariationPipeline.from_pretrained(
           _lowerCAmelCase		, safety_checker=_lowerCAmelCase		, torch_dtype=torch.floataa    )
       UpperCAmelCase     :			Any     	=pipe.to(_lowerCAmelCase    )
       pipe.set_progress_bar_config(disable=_lowerCAmelCase    )
       pipe.enable_attention_slicing(1    )
       pipe.enable_sequential_cpu_offload()

       UpperCAmelCase     :			Union[str, Any]     	=self.get_inputs(_lowerCAmelCase		, dtype=torch.floataa    )
       UpperCAmelCase     :			List[Any]     	=pipe(**_lowerCAmelCase    )

       UpperCAmelCase     :			Any     	=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 ):
 def 						UpperCAmelCase__							(	self    )						->				List[Any]:
       '''simple docstring'''





       super().tearDown()
       gc.collect()
       torch.cuda.empty_cache()
 def 						UpperCAmelCase__							(	self		, snake_case__		, snake_case__="cpu"		, snake_case__=torch.floataa		, snake_case__=0    )						->				List[str]:
       '''simple docstring'''





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





       UpperCAmelCase     :			Any     	=StableDiffusionImageVariationPipeline.from_pretrained('''fusing/sd-image-variations-diffusers'''    )
       sd_pipe.to(_lowerCAmelCase    )
       sd_pipe.set_progress_bar_config(disable=_lowerCAmelCase    )

       UpperCAmelCase     :			Tuple     	=self.get_inputs(_lowerCAmelCase    )
       UpperCAmelCase     :			int     	=sd_pipe(**_lowerCAmelCase    ).images[0]

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






 def 						UpperCAmelCase__							(	self    )						->				str:
       '''simple docstring'''





       UpperCAmelCase     :			str     	=StableDiffusionImageVariationPipeline.from_pretrained('''fusing/sd-image-variations-diffusers'''    )
       UpperCAmelCase     :			int     	=DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config    )
       sd_pipe.to(_lowerCAmelCase    )
       sd_pipe.set_progress_bar_config(disable=_lowerCAmelCase    )

       UpperCAmelCase     :			str     	=self.get_inputs(_lowerCAmelCase    )
       UpperCAmelCase     :			Optional[int]     	=25
       UpperCAmelCase     :			Tuple     	=sd_pipe(**_lowerCAmelCase    ).images[0]

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







def  __UpperCamelCase   (     a = 998001)			->int:

					# fetches the next number
					for number in range(n - 1,  9999,  -1):
										lowerCamelCase__      =							str(a)

										# checks whether 'str_number' is a palindrome.
										if str_number == str_number[::-1]:
															lowerCamelCase__      =							999

															# if 'number' is a product of two 3-digit numbers
															# then number is the answer otherwise fetch next number.
															while divisor != 99:
																				if (number % divisor == 0) and (len(str(number // divisor)) == 3.0):
																									return number
																				divisor -= 1
					raise ValueError("That number is larger than our acceptable range.")


if __name__ == "__main__":
						print(f"""{solution() = }""")

