



import gc
import unittest

import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DDPMScheduler,
    PriorTransformer,
    StableUnCLIPPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.testing_utils import enable_full_determinism, load_numpy, require_torch_gpu, slow, torch_device

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()

class 	__SCREAMING_SNAKE_CASE     (				lowerCAmelCase_		,       lowerCAmelCase_		,       lowerCAmelCase_		,       unittest.TestCase       ):


					'''simple docstring'''





					_a             =							StableUnCLIPPipeline
					_a             =							TEXT_TO_IMAGE_PARAMS
					_a             =							TEXT_TO_IMAGE_BATCH_PARAMS
					_a             =							TEXT_TO_IMAGE_IMAGE_PARAMS
					_a             =							TEXT_TO_IMAGE_IMAGE_PARAMS

					# TODO(will) Expected attn_bias.stride(1) == 0 to be true, but got false
					_a             =							False



					def 		snake_case	(	self							:  int					)-> Dict:
								lowerCamelCase__		:		int		     =32
								lowerCamelCase__		:		List[str]		     =embedder_hidden_size

								# prior components

								torch.manual_seed(0					)
								lowerCamelCase__		:		List[Any]		     =CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''					)

								torch.manual_seed(0					)
								lowerCamelCase__		:		Optional[int]		     =CLIPTextModelWithProjection(
								    CLIPTextConfig(
								        bos_token_id=0,  eos_token_id=2,  hidden_size=lowerCAmelCase__,  projection_dim=lowerCAmelCase__,  intermediate_size=37,  layer_norm_eps=1E-05,  num_attention_heads=4,  num_hidden_layers=5,  pad_token_id=1,  vocab_size=1000,  )					)

								torch.manual_seed(0					)
								lowerCamelCase__		:		List[Any]		     =PriorTransformer(
								    num_attention_heads=2,  attention_head_dim=12,  embedding_dim=lowerCAmelCase__,  num_layers=1,  )

								torch.manual_seed(0					)
								lowerCamelCase__		:		Tuple		     =DDPMScheduler(
								    variance_type='''fixed_small_log''',  prediction_type='''sample''',  num_train_timesteps=1000,  clip_sample=lowerCAmelCase__,  clip_sample_range=5.0,  beta_schedule='''squaredcos_cap_v2''',  )

								# regular denoising components

								torch.manual_seed(0					)
								lowerCamelCase__		:		int		     =StableUnCLIPImageNormalizer(embedding_dim=lowerCAmelCase__					)
								lowerCamelCase__		:		Tuple		     =DDPMScheduler(beta_schedule='''squaredcos_cap_v2'''					)

								torch.manual_seed(0					)
								lowerCamelCase__		:		List[Any]		     =CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''					)

								torch.manual_seed(0					)
								lowerCamelCase__		:		str		     =CLIPTextModel(
								    CLIPTextConfig(
								        bos_token_id=0,  eos_token_id=2,  hidden_size=lowerCAmelCase__,  projection_dim=32,  intermediate_size=37,  layer_norm_eps=1E-05,  num_attention_heads=4,  num_hidden_layers=5,  pad_token_id=1,  vocab_size=1000,  )					)

								torch.manual_seed(0					)
								lowerCamelCase__		:		List[str]		     =UNetaDConditionModel(
								    sample_size=32,  in_channels=4,  out_channels=4,  down_block_types=('''CrossAttnDownBlock2D''', '''DownBlock2D'''),  up_block_types=('''UpBlock2D''', '''CrossAttnUpBlock2D'''),  block_out_channels=(32, 64),  attention_head_dim=(2, 4),  class_embed_type='''projection''',  projection_class_embeddings_input_dim=embedder_projection_dim * 2,  cross_attention_dim=lowerCAmelCase__,  layers_per_block=1,  upcast_attention=lowerCAmelCase__,  use_linear_projection=lowerCAmelCase__,  )

								torch.manual_seed(0					)
								lowerCamelCase__		:		Optional[int]		     =DDIMScheduler(
								    beta_schedule='''scaled_linear''',  beta_start=0.00_085,  beta_end=0.012,  prediction_type='''v_prediction''',  set_alpha_to_one=lowerCAmelCase__,  steps_offset=1,  )

								torch.manual_seed(0					)
								lowerCamelCase__		:		Any		     =AutoencoderKL()

								lowerCamelCase__		:		int		     ={
								    # prior components
								    "prior_tokenizer": prior_tokenizer,
								    "prior_text_encoder": prior_text_encoder,
								    "prior": prior,
								    "prior_scheduler": prior_scheduler,
								    # image noising components
								    "image_normalizer": image_normalizer,
								    "image_noising_scheduler": image_noising_scheduler,
								    # regular denoising components
								    "tokenizer": tokenizer,
								    "text_encoder": text_encoder,
								    "unet": unet,
								    "scheduler": scheduler,
								    "vae": vae,
								}

								return components



					def 		snake_case	(	self							:  Dict,  lowerCamelCase							:  Optional[Any],  lowerCamelCase							:  Dict=0					)-> Tuple:




								if str(lowerCAmelCase__					).startswith('''mps'''					):
											lowerCamelCase__		:		Union[str, Any]		     =torch.manual_seed(lowerCAmelCase__					)
								else:
											lowerCamelCase__		:		Optional[int]		     =torch.Generator(device=lowerCAmelCase__					).manual_seed(lowerCAmelCase__					)
								lowerCamelCase__		:		List[str]		     ={
								    "prompt": "A painting of a squirrel eating a burger",
								    "generator": generator,
								    "num_inference_steps": 2,
								    "prior_num_inference_steps": 2,
								    "output_type": "numpy",
								}
								return inputs



					def 		snake_case	(	self							:  Any					)-> str:
								lowerCamelCase__		:		Optional[Any]		     =torch_device == "cpu"

								self._test_attention_slicing_forward_pass(test_max_difference=lowerCAmelCase__					)




					def 		snake_case	(	self							:  List[Any]					)-> Optional[int]:
								lowerCamelCase__		:		List[str]		     =torch_device in ["cpu", "mps"]

								self._test_inference_batch_single_identical(test_max_difference=lowerCAmelCase__					)





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


					'''simple docstring'''





					def 		snake_case	(	self							:  Dict					)-> Optional[int]:
								super().tearDown()
								gc.collect()
								torch.cuda.empty_cache()



					def 		snake_case	(	self							:  Dict					)-> List[Any]:
								lowerCamelCase__		:		Dict		     =load_numpy(
								    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_anime_turtle_fp16.npy'''					)

								lowerCamelCase__		:		Tuple		     =StableUnCLIPPipeline.from_pretrained('''fusing/stable-unclip-2-1-l''',  torch_dtype=torch.floataa					)
								pipe.to(lowerCAmelCase__					)
								pipe.set_progress_bar_config(disable=lowerCAmelCase__					)
								# stable unclip will oom when integration tests are run on a V100,
								# so turn on memory savings
								pipe.enable_attention_slicing()
								pipe.enable_sequential_cpu_offload()

								lowerCamelCase__		:		Optional[int]		     =torch.Generator(device='''cpu'''					).manual_seed(0					)
								lowerCamelCase__		:		Tuple		     =pipe('''anime turle''',  generator=lowerCAmelCase__,  output_type='''np'''					)

								lowerCamelCase__		:		str		     =output.images[0]

								assert image.shape == (768, 768, 3)

								assert_mean_pixel_difference(lowerCAmelCase__,  lowerCAmelCase__					)




					def 		snake_case	(	self							:  Tuple					)-> List[str]:
								torch.cuda.empty_cache()
								torch.cuda.reset_max_memory_allocated()
								torch.cuda.reset_peak_memory_stats()

								lowerCamelCase__		:		Tuple		     =StableUnCLIPPipeline.from_pretrained('''fusing/stable-unclip-2-1-l''',  torch_dtype=torch.floataa					)
								lowerCamelCase__		:		Dict		     =pipe.to(lowerCAmelCase__					)
								pipe.set_progress_bar_config(disable=lowerCAmelCase__					)
								pipe.enable_attention_slicing()
								pipe.enable_sequential_cpu_offload()

								lowerCamelCase__		:		List[str]		     =pipe(
								    '''anime turtle''',  prior_num_inference_steps=2,  num_inference_steps=2,  output_type='''np''',  )

								lowerCamelCase__		:		Optional[Any]		     =torch.cuda.max_memory_allocated()
								# make sure that less than 7 GB is allocated
								assert mem_bytes < 7 * 10**9









"""simple docstring"""






import numpy as np
from PIL import Image






def 					snake_case__       (		__lowerCamelCase    :   np.ndarray							,		__lowerCamelCase    :   int							,		__lowerCamelCase    :   int     ):
			"""simple docstring"""
			lowerCamelCase__		:		List[Any]		     =np.array(__lowerCamelCase     )
			if arr.shape[0] != arr.shape[1]:
						raise ValueError('''The input array is not a square matrix'''     )
			lowerCamelCase__		:		int		     =0
			lowerCamelCase__		:		int		     =0
			lowerCamelCase__		:		Optional[int]		     =0
			lowerCamelCase__		:		List[Any]		     =0

			# compute the shape of the output matrix
			lowerCamelCase__		:		Union[str, Any]		     =(arr.shape[0] - size) // stride + 1
			# initialize the output matrix with zeros of shape maxpool_shape
			lowerCamelCase__		:		Union[str, Any]		     =np.zeros((maxpool_shape, maxpool_shape)     )

			while i < arr.shape[0]:
						if i + size > arr.shape[0]:
									# if the end of the matrix is reached, break
									break
						while j < arr.shape[1]:
									# if the end of the matrix is reached, break
									if j + size > arr.shape[1]:
												break
									# compute the maximum of the pooling matrix
									lowerCamelCase__		:		str		     =np.max(arr[i : i + size, j : j + size]     )
									# shift the pooling matrix by stride of column pixels
									j += stride
									mat_j += 1

						# shift the pooling matrix by stride of row pixels
						i += stride
						mat_i += 1

						# reset the column index to 0
						lowerCamelCase__		:		Optional[int]		     =0
						lowerCamelCase__		:		Optional[int]		     =0

			return updated_arr






def 					snake_case__       (		__lowerCamelCase    :   np.ndarray							,		__lowerCamelCase    :   int							,		__lowerCamelCase    :   int     ):
			"""simple docstring"""
			lowerCamelCase__		:		List[Any]		     =np.array(__lowerCamelCase     )
			if arr.shape[0] != arr.shape[1]:
						raise ValueError('''The input array is not a square matrix'''     )
			lowerCamelCase__		:		str		     =0
			lowerCamelCase__		:		List[Any]		     =0
			lowerCamelCase__		:		Optional[int]		     =0
			lowerCamelCase__		:		List[Any]		     =0

			# compute the shape of the output matrix
			lowerCamelCase__		:		Dict		     =(arr.shape[0] - size) // stride + 1
			# initialize the output matrix with zeros of shape avgpool_shape
			lowerCamelCase__		:		Optional[Any]		     =np.zeros((avgpool_shape, avgpool_shape)     )

			while i < arr.shape[0]:
						# if the end of the matrix is reached, break
						if i + size > arr.shape[0]:
									break
						while j < arr.shape[1]:
									# if the end of the matrix is reached, break
									if j + size > arr.shape[1]:
												break
									# compute the average of the pooling matrix
									lowerCamelCase__		:		Optional[int]		     =int(np.average(arr[i : i + size, j : j + size]     )     )
									# shift the pooling matrix by stride of column pixels
									j += stride
									mat_j += 1

						# shift the pooling matrix by stride of row pixels
						i += stride
						mat_i += 1
						# reset the column index to 0
						lowerCamelCase__		:		Optional[Any]		     =0
						lowerCamelCase__		:		int		     =0

			return updated_arr


# Main Function
if __name__ == "__main__":
		from doctest import testmod

		testmod(name="avgpooling", verbose=True)

		# Loading the image
		_lowercase			:			int										=						Image.open("path_to_image")

		# Converting the image to numpy array and maxpooling, displaying the result
		# Ensure that the image is a square matrix

		Image.fromarray(maxpooling(np.array(image), size=3, stride=2)).show()

		# Converting the image to numpy array and averagepooling, displaying the result
		# Ensure that the image is a square matrix

		Image.fromarray(avgpooling(np.array(image), size=3, stride=2)).show()




