

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 _lowercase					(       _UpperCAmelCase   ,				_UpperCAmelCase   ,				_UpperCAmelCase   ,				unittest.TestCase						):





							"""simple docstring"""




							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     _UpperCAmelCase       (    self  ):







														'''simple docstring'''






														torch.manual_seed(0  )
														_lowercase				=	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     ,   )
														_lowercase				=	PNDMScheduler(skip_prk_steps=UpperCAmelCase_  )
														torch.manual_seed(0  )
														_lowercase				=	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  )
														_lowercase				=	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     ,   )
														_lowercase				=	CLIPVisionModelWithProjection(UpperCAmelCase_  )
														_lowercase				=	CLIPImageProcessor(crop_size=32     ,   size=32  )

														_lowercase				=	{
														    '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  ):







														'''simple docstring'''






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



							def     _UpperCAmelCase       (    self  ):







														'''simple docstring'''






														_lowercase				=	'cpu'  # ensure determinism for the device-dependent torch.Generator
														_lowercase				=	self.get_dummy_components()
														_lowercase				=	StableDiffusionImageVariationPipeline(**UpperCAmelCase_  )
														_lowercase				=	sd_pipe.to(UpperCAmelCase_  )
														sd_pipe.set_progress_bar_config(disable=UpperCAmelCase_  )

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

														assert image.shape == (1, 64, 64, 3)
														_lowercase				=	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     _UpperCAmelCase       (    self  ):







														'''simple docstring'''






														_lowercase				=	'cpu'  # ensure determinism for the device-dependent torch.Generator
														_lowercase				=	self.get_dummy_components()
														_lowercase				=	StableDiffusionImageVariationPipeline(**UpperCAmelCase_  )
														_lowercase				=	sd_pipe.to(UpperCAmelCase_  )
														sd_pipe.set_progress_bar_config(disable=UpperCAmelCase_  )

														_lowercase				=	self.get_dummy_inputs(UpperCAmelCase_  )
														_lowercase				=	2 * [inputs['image']]
														_lowercase				=	sd_pipe(**UpperCAmelCase_  )

														_lowercase				=	output.images

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

														assert image.shape == (2, 64, 64, 3)
														_lowercase				=	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     _UpperCAmelCase       (    self  ):







														'''simple docstring'''






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






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





							"""simple docstring"""




							def     _UpperCAmelCase       (    self  ):







														'''simple docstring'''






														super().tearDown()
														gc.collect()
														torch.cuda.empty_cache()



							def     _UpperCAmelCase       (    self     ,   UpperCAmelCase     ,   UpperCAmelCase="cpu"     ,   UpperCAmelCase=torch.floataa     ,   UpperCAmelCase=0  ):







														'''simple docstring'''






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



							def     _UpperCAmelCase       (    self  ):







														'''simple docstring'''






														_lowercase				=	StableDiffusionImageVariationPipeline.from_pretrained(
														    """lambdalabs/sd-image-variations-diffusers"""     ,   safety_checker=UpperCAmelCase_  )
														_lowercase				=	sd_pipe.to(UpperCAmelCase_  )
														sd_pipe.set_progress_bar_config(disable=UpperCAmelCase_  )

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

														assert image.shape == (1, 512, 512, 3)
														_lowercase				=	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     _UpperCAmelCase       (    self  ):







														'''simple docstring'''






														_lowercase				=	0

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

														_lowercase				=	False

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

														_lowercase				=	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  ):







														'''simple docstring'''






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

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

														_lowercase				=	self.get_inputs(UpperCAmelCase_     ,   dtype=torch.floataa  )
														_lowercase				=	pipe(**UpperCAmelCase_  )

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





							"""simple docstring"""




							def     _UpperCAmelCase       (    self  ):







														'''simple docstring'''






														super().tearDown()
														gc.collect()
														torch.cuda.empty_cache()



							def     _UpperCAmelCase       (    self     ,   UpperCAmelCase     ,   UpperCAmelCase="cpu"     ,   UpperCAmelCase=torch.floataa     ,   UpperCAmelCase=0  ):







														'''simple docstring'''






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



							def     _UpperCAmelCase       (    self  ):







														'''simple docstring'''






														_lowercase				=	StableDiffusionImageVariationPipeline.from_pretrained("""fusing/sd-image-variations-diffusers"""  )
														sd_pipe.to(UpperCAmelCase_  )
														sd_pipe.set_progress_bar_config(disable=UpperCAmelCase_  )

														_lowercase				=	self.get_inputs(UpperCAmelCase_  )
														_lowercase				=	sd_pipe(**UpperCAmelCase_  ).images[0]

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



							def     _UpperCAmelCase       (    self  ):







														'''simple docstring'''






														_lowercase				=	StableDiffusionImageVariationPipeline.from_pretrained("""fusing/sd-image-variations-diffusers"""  )
														_lowercase				=	DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config  )
														sd_pipe.to(UpperCAmelCase_  )
														sd_pipe.set_progress_bar_config(disable=UpperCAmelCase_  )

														_lowercase				=	self.get_inputs(UpperCAmelCase_  )
														_lowercase				=	25
														_lowercase				=	sd_pipe(**UpperCAmelCase_  ).images[0]

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



def 					__lowerCAmelCase			(  _A = 998_001					):



							"""simple docstring"""


							_lowercase				=	0
							for i in range(999 ,99 ,-1					):  # 3 digit numbers range from 999 down to 100
														for j in range(999 ,99 ,-1					):
																					_lowercase				=	str(i * j					)
																					if product_string == product_string[::-1] and i * j < n:
																												_lowercase				=	max(_A ,i * j					)
							return answer


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


