


"""simple docstring"""
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'''

							__lowercase:		int				   =     StableDiffusionImageVariationPipeline
							__lowercase:		Optional[int]				   =     IMAGE_VARIATION_PARAMS
							__lowercase:		str				   =     IMAGE_VARIATION_BATCH_PARAMS
							__lowercase:		Tuple				   =     frozenset([])
							# TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
							__lowercase:		List[str]				   =     frozenset([])



							def 		lowerCAmelCase							(					self				:   int							)       ->List[Any]:



														"""simple docstring"""



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

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



							def 		lowerCAmelCase							(					self				:   Dict		,							UpperCAmelCase_				:   int		,							UpperCAmelCase_				:   List[str]=0							)       ->int:



														"""simple docstring"""



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



							def 		lowerCAmelCase							(					self				:   Tuple							)       ->str:



														"""simple docstring"""



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

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

														assert image.shape == (1, 64, 64, 3)
														snake_case_     							=		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 		lowerCAmelCase							(					self				:   Dict							)       ->str:



														"""simple docstring"""



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

														snake_case_     							=		self.get_dummy_inputs(UpperCAmelCase_							)
														snake_case_     							=		2 * [inputs["""image"""]]
														snake_case_     							=		sd_pipe(**UpperCAmelCase_							)

														snake_case_     							=		output.images

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

														assert image.shape == (2, 64, 64, 3)
														snake_case_     							=		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 		lowerCAmelCase							(					self				:   Union[str, Any]							)       ->Any:



														"""simple docstring"""



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





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





							'''simple docstring'''

							def 		lowerCAmelCase							(					self				:   Any							)       ->int:



														"""simple docstring"""



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



							def 		lowerCAmelCase							(					self				:   Tuple		,							UpperCAmelCase_				:   Any		,							UpperCAmelCase_				:   Dict="cpu"		,							UpperCAmelCase_				:   Optional[Any]=torch.floataa		,							UpperCAmelCase_				:   List[str]=0							)       ->str:



														"""simple docstring"""



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



							def 		lowerCAmelCase							(					self				:   List[Any]							)       ->int:



														"""simple docstring"""



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

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

														assert image.shape == (1, 512, 512, 3)
														snake_case_     							=		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 		lowerCAmelCase							(					self				:   str							)       ->Tuple:



														"""simple docstring"""



														snake_case_     							=		0

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

														snake_case_     							=		False

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

														snake_case_     							=		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 		lowerCAmelCase							(					self				:   List[str]							)       ->int:



														"""simple docstring"""



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

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

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

														snake_case_     							=		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 		lowerCAmelCase							(					self				:   int							)       ->Dict:



														"""simple docstring"""



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



							def 		lowerCAmelCase							(					self				:   Tuple		,							UpperCAmelCase_				:   Tuple		,							UpperCAmelCase_				:   Optional[Any]="cpu"		,							UpperCAmelCase_				:   int=torch.floataa		,							UpperCAmelCase_				:   Tuple=0							)       ->Dict:



														"""simple docstring"""



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



							def 		lowerCAmelCase							(					self				:   str							)       ->Union[str, Any]:



														"""simple docstring"""



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

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

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



							def 		lowerCAmelCase							(					self				:   Any							)       ->Any:



														"""simple docstring"""



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

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

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







"""simple docstring"""
from typing import Any





class   __A  :





							'''simple docstring'''

							def __init__(					self				:   List[Any]							)       ->int:



														"""simple docstring"""



														snake_case_     							=		[]
														snake_case_     							=		0



							def __str__(					self				:   List[str]							)       ->Union[str, Any]:



														"""simple docstring"""



														snake_case_     							=		"""<""" + str(self.stack							)[1:-1] + """>"""
														return printed



							def 		lowerCAmelCase							(					self				:   List[Any]		,							UpperCAmelCase_				:   Any							)       ->None:



														"""simple docstring"""



														self.stack.append(UpperCAmelCase_							)
														snake_case_     							=		self.length + 1



							def 		lowerCAmelCase							(					self				:   List[str]							)       ->Any:



														"""simple docstring"""



														self.rotate(1							)
														snake_case_     							=		self.stack[self.length - 1]
														snake_case_     							=		self.stack[:-1]
														self.rotate(self.length - 1							)
														snake_case_     							=		self.length - 1
														return dequeued



							def 		lowerCAmelCase							(					self				:   List[str]		,							UpperCAmelCase_				:   int							)       ->None:



														"""simple docstring"""



														for _ in range(UpperCAmelCase_							):
																					snake_case_     							=		self.stack[0]
																					snake_case_     							=		self.stack[1:]
																					self.put(UpperCAmelCase_							)
																					snake_case_     							=		self.length - 1



							def 		lowerCAmelCase							(					self				:   Dict							)       ->Any:



														"""simple docstring"""



														snake_case_     							=		self.get()
														self.put(UpperCAmelCase_							)
														self.rotate(self.length - 1							)
														return front



							def 		lowerCAmelCase							(					self				:   Optional[Any]							)       ->int:



														"""simple docstring"""



														return self.length



