



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  ):
			_UpperCamelCase:   Tuple	     =		StableDiffusionImageVariationPipeline
			_UpperCamelCase:   Tuple	     =		IMAGE_VARIATION_PARAMS
			_UpperCamelCase:   Tuple	     =		IMAGE_VARIATION_BATCH_PARAMS
			_UpperCamelCase:   Union[str, Any]	     =		frozenset([]  )
			# TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
			_UpperCamelCase:   int	     =		frozenset([]  )






			def 							_snake_case   (    self )	->					str:
									torch.manual_seed(0 )
									lowerCAmelCase	:   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    ,  )
									lowerCAmelCase	:   Union[str, Any]					     =					PNDMScheduler(skip_prk_steps=lowercase_ )
									torch.manual_seed(0 )
									lowerCAmelCase	:   Dict					     =					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 )
									lowerCAmelCase	:   Optional[int]					     =					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    ,  )
									lowerCAmelCase	:   int					     =					CLIPVisionModelWithProjection(lowercase_ )
									lowerCAmelCase	:   Dict					     =					CLIPImageProcessor(crop_size=32    ,  size=32 )

									lowerCAmelCase	:   List[str]					     =					{
									    """unet""": unet,
									    """scheduler""": scheduler,
									    """vae""": vae,
									    """image_encoder""": image_encoder,
									    """feature_extractor""": feature_extractor,
									    """safety_checker""": None,
									}
									return components






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






			def 							_snake_case   (    self )	->					Optional[Any]:
									lowerCAmelCase	:   List[str]					     =					"""cpu"""  # ensure determinism for the device-dependent torch.Generator
									lowerCAmelCase	:   Dict					     =					self.get_dummy_components()
									lowerCAmelCase	:   Dict					     =					StableDiffusionImageVariationPipeline(**lowercase_ )
									lowerCAmelCase	:   Optional[int]					     =					sd_pipe.to(lowercase_ )
									sd_pipe.set_progress_bar_config(disable=lowercase_ )

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

									assert image.shape == (1, 64, 64, 3)
									lowerCAmelCase	:   Optional[int]					     =					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 							_snake_case   (    self )	->					Optional[int]:
									lowerCAmelCase	:   Union[str, Any]					     =					"""cpu"""  # ensure determinism for the device-dependent torch.Generator
									lowerCAmelCase	:   Union[str, Any]					     =					self.get_dummy_components()
									lowerCAmelCase	:   Union[str, Any]					     =					StableDiffusionImageVariationPipeline(**lowercase_ )
									lowerCAmelCase	:   int					     =					sd_pipe.to(lowercase_ )
									sd_pipe.set_progress_bar_config(disable=lowercase_ )

									lowerCAmelCase	:   List[str]					     =					self.get_dummy_inputs(lowercase_ )
									lowerCAmelCase	:   str					     =					2 * [inputs["""image"""]]
									lowerCAmelCase	:   List[str]					     =					sd_pipe(**lowercase_ )

									lowerCAmelCase	:   Optional[int]					     =					output.images

									lowerCAmelCase	:   List[Any]					     =					image[-1, -3:, -3:, -1]

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






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






			def 							_snake_case   (    self )	->					Optional[Any]:
									super().tearDown()
									gc.collect()
									torch.cuda.empty_cache()






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






			def 							_snake_case   (    self )	->					Any:
									lowerCAmelCase	:   Union[str, Any]					     =					StableDiffusionImageVariationPipeline.from_pretrained(
									    """lambdalabs/sd-image-variations-diffusers"""    ,  safety_checker=lowercase_ )
									lowerCAmelCase	:   Tuple					     =					sd_pipe.to(lowercase_ )
									sd_pipe.set_progress_bar_config(disable=lowercase_ )

									lowerCAmelCase	:   Optional[Any]					     =					self.get_inputs(lowercase_ )
									lowerCAmelCase	:   Dict					     =					sd_pipe(**lowercase_ ).images
									lowerCAmelCase	:   int					     =					image[0, -3:, -3:, -1].flatten()

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

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

									lowerCAmelCase	:   int					     =					False

									lowerCAmelCase	:   List[Any]					     =					StableDiffusionImageVariationPipeline.from_pretrained(
									    """fusing/sd-image-variations-diffusers"""    ,  safety_checker=lowercase_    ,  torch_dtype=torch.floataa    ,  )
									pipe.to(lowercase_ )
									pipe.set_progress_bar_config(disable=lowercase_ )
									pipe.enable_attention_slicing()

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






			def 							_snake_case   (    self )	->					List[Any]:
									torch.cuda.empty_cache()
									torch.cuda.reset_max_memory_allocated()
									torch.cuda.reset_peak_memory_stats()

									lowerCAmelCase	:   List[str]					     =					"""fusing/sd-image-variations-diffusers"""
									lowerCAmelCase	:   str					     =					StableDiffusionImageVariationPipeline.from_pretrained(
									    lowercase_    ,  safety_checker=lowercase_    ,  torch_dtype=torch.floataa )
									lowerCAmelCase	:   List[str]					     =					pipe.to(lowercase_ )
									pipe.set_progress_bar_config(disable=lowercase_ )
									pipe.enable_attention_slicing(1 )
									pipe.enable_sequential_cpu_offload()

									lowerCAmelCase	:   int					     =					self.get_inputs(lowercase_    ,  dtype=torch.floataa )
									lowerCAmelCase	:   Optional[int]					     =					pipe(**lowercase_ )

									lowerCAmelCase	:   Tuple					     =					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  ):






			def 							_snake_case   (    self )	->					Union[str, Any]:
									super().tearDown()
									gc.collect()
									torch.cuda.empty_cache()






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






			def 							_snake_case   (    self )	->					Optional[int]:
									lowerCAmelCase	:   int					     =					StableDiffusionImageVariationPipeline.from_pretrained("""fusing/sd-image-variations-diffusers""" )
									sd_pipe.to(lowercase_ )
									sd_pipe.set_progress_bar_config(disable=lowercase_ )

									lowerCAmelCase	:   str					     =					self.get_inputs(lowercase_ )
									lowerCAmelCase	:   Tuple					     =					sd_pipe(**lowercase_ ).images[0]

									lowerCAmelCase	:   Optional[Any]					     =					load_numpy(
									    """https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"""
									    """/stable_diffusion_imgvar/lambdalabs_variations_pndm.npy""" )
									lowerCAmelCase	:   Union[str, Any]					     =					np.abs(expected_image - image ).max()
									assert max_diff < 1e-3






			def 							_snake_case   (    self )	->					Union[str, Any]:
									lowerCAmelCase	:   str					     =					StableDiffusionImageVariationPipeline.from_pretrained("""fusing/sd-image-variations-diffusers""" )
									lowerCAmelCase	:   Any					     =					DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config )
									sd_pipe.to(lowercase_ )
									sd_pipe.set_progress_bar_config(disable=lowercase_ )

									lowerCAmelCase	:   Union[str, Any]					     =					self.get_inputs(lowercase_ )
									lowerCAmelCase	:   Any					     =					25
									lowerCAmelCase	:   Optional[Any]					     =					sd_pipe(**lowercase_ ).images[0]

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






import cva
import numpy as np







def  _UpperCAmelCase  (				SCREAMING_SNAKE_CASE__			,SCREAMING_SNAKE_CASE__			,SCREAMING_SNAKE_CASE__			,SCREAMING_SNAKE_CASE__  ):


						'''simple docstring'''




						try:
												return int(image[x_coordinate][y_coordinate] >= center  )
						except (IndexError, TypeError):
												return 0







def  _UpperCAmelCase  (				SCREAMING_SNAKE_CASE__			,SCREAMING_SNAKE_CASE__			,SCREAMING_SNAKE_CASE__  ):


						'''simple docstring'''




						lowerCAmelCase	:   Any					     =					image[x_coordinate][y_coordinate]
						lowerCAmelCase	:   int					     =					[1, 2, 4, 8, 1_6, 3_2, 6_4, 1_2_8]

						# skip get_neighbors_pixel if center is null
						if center is None:
												return 0

						# Starting from the top right, assigning value to pixels clockwise
						lowerCAmelCase	:   Union[str, Any]					     =					[
						    get_neighbors_pixel(SCREAMING_SNAKE_CASE__			,x_coordinate - 1			,y_coordinate + 1			,SCREAMING_SNAKE_CASE__  ),
						    get_neighbors_pixel(SCREAMING_SNAKE_CASE__			,SCREAMING_SNAKE_CASE__			,y_coordinate + 1			,SCREAMING_SNAKE_CASE__  ),
						    get_neighbors_pixel(SCREAMING_SNAKE_CASE__			,x_coordinate - 1			,SCREAMING_SNAKE_CASE__			,SCREAMING_SNAKE_CASE__  ),
						    get_neighbors_pixel(SCREAMING_SNAKE_CASE__			,x_coordinate + 1			,y_coordinate + 1			,SCREAMING_SNAKE_CASE__  ),
						    get_neighbors_pixel(SCREAMING_SNAKE_CASE__			,x_coordinate + 1			,SCREAMING_SNAKE_CASE__			,SCREAMING_SNAKE_CASE__  ),
						    get_neighbors_pixel(SCREAMING_SNAKE_CASE__			,x_coordinate + 1			,y_coordinate - 1			,SCREAMING_SNAKE_CASE__  ),
						    get_neighbors_pixel(SCREAMING_SNAKE_CASE__			,SCREAMING_SNAKE_CASE__			,y_coordinate - 1			,SCREAMING_SNAKE_CASE__  ),
						    get_neighbors_pixel(SCREAMING_SNAKE_CASE__			,x_coordinate - 1			,y_coordinate - 1			,SCREAMING_SNAKE_CASE__  ),
						]

						# Converting the binary value to decimal.
						return sum(
						    binary_value * power for binary_value, power in zip(SCREAMING_SNAKE_CASE__			,SCREAMING_SNAKE_CASE__  )  )


if __name__ == "__main__":
		# Reading the image and converting it to grayscale.
		lowerCAmelCase					:				Dict   	=cva.imread(
		    'digital_image_processing/image_data/lena.jpg', cva.IMREAD_GRAYSCALE
		)

		# Create a numpy array as the same height and width of read image
		lowerCAmelCase					:				Dict   	=np.zeros((image.shape[0], image.shape[1]))

		# Iterating through the image and calculating the
		# local binary pattern value for each pixel.
		for i in range(0, image.shape[0]):
				for j in range(0, image.shape[1]):
						lowerCAmelCase					:				Optional[int]   	=local_binary_value(image, i, j)

		cva.imshow('local binary pattern', lbp_image)
		cva.waitKey(0)
		cva.destroyAllWindows()

