



import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionSAGPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()

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


					'''simple docstring'''





					_a             =							StableDiffusionSAGPipeline
					_a             =							TEXT_TO_IMAGE_PARAMS
					_a             =							TEXT_TO_IMAGE_BATCH_PARAMS
					_a             =							TEXT_TO_IMAGE_IMAGE_PARAMS
					_a             =							TEXT_TO_IMAGE_IMAGE_PARAMS
					_a             =							False



					def 		snake_case	(	self							:  List[Any]					)-> Dict:
								torch.manual_seed(0					)
								lowerCamelCase__		:		Tuple		     =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__		:		List[str]		     =DDIMScheduler(
								    beta_start=0.00_085,  beta_end=0.012,  beta_schedule='''scaled_linear''',  clip_sample=__lowercase,  set_alpha_to_one=__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__		:		Any		     =CLIPTextConfig(
								    bos_token_id=0,  eos_token_id=2,  hidden_size=32,  intermediate_size=37,  layer_norm_eps=1E-05,  num_attention_heads=4,  num_hidden_layers=5,  pad_token_id=1,  vocab_size=1000,  )
								lowerCamelCase__		:		Tuple		     =CLIPTextModel(__lowercase					)
								lowerCamelCase__		:		Optional[Any]		     =CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''					)

								lowerCamelCase__		:		int		     ={
								    '''unet''': unet,
								    '''scheduler''': scheduler,
								    '''vae''': vae,
								    '''text_encoder''': text_encoder,
								    '''tokenizer''': tokenizer,
								    '''safety_checker''': None,
								    '''feature_extractor''': None,
								}
								return components



					def 		snake_case	(	self							:  Union[str, Any],  lowerCamelCase							:  Any,  lowerCamelCase							:  Tuple=0					)-> Union[str, Any]:
								if str(__lowercase					).startswith('''mps'''					):
											lowerCamelCase__		:		Union[str, Any]		     =torch.manual_seed(__lowercase					)
								else:
											lowerCamelCase__		:		Optional[Any]		     =torch.Generator(device=__lowercase					).manual_seed(__lowercase					)
								lowerCamelCase__		:		Any		     ={
								    '''prompt''': '''.''',
								    '''generator''': generator,
								    '''num_inference_steps''': 2,
								    '''guidance_scale''': 1.0,
								    '''sag_scale''': 1.0,
								    '''output_type''': '''numpy''',
								}
								return inputs




					def 		snake_case	(	self							:  Tuple					)-> Optional[int]:
								super().test_inference_batch_single_identical(expected_max_diff=3E-3					)





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


					'''simple docstring'''





					def 		snake_case	(	self							:  Dict					)-> Union[str, Any]:
								# clean up the VRAM after each test
								super().tearDown()
								gc.collect()
								torch.cuda.empty_cache()



					def 		snake_case	(	self							:  Optional[Any]					)-> Tuple:
								lowerCamelCase__		:		str		     =StableDiffusionSAGPipeline.from_pretrained('''CompVis/stable-diffusion-v1-4'''					)
								lowerCamelCase__		:		Tuple		     =sag_pipe.to(__lowercase					)
								sag_pipe.set_progress_bar_config(disable=__lowercase					)

								lowerCamelCase__		:		Optional[Any]		     ='''.'''
								lowerCamelCase__		:		Union[str, Any]		     =torch.manual_seed(0					)
								lowerCamelCase__		:		List[str]		     =sag_pipe(
								    [prompt],  generator=__lowercase,  guidance_scale=7.5,  sag_scale=1.0,  num_inference_steps=20,  output_type='''np'''					)

								lowerCamelCase__		:		List[Any]		     =output.images

								lowerCamelCase__		:		Optional[int]		     =image[0, -3:, -3:, -1]

								assert image.shape == (1, 512, 512, 3)
								lowerCamelCase__		:		Any		     =np.array([0.1_568, 0.1_738, 0.1_695, 0.1_693, 0.1_507, 0.1_705, 0.1_547, 0.1_751, 0.1_949]					)

								assert np.abs(image_slice.flatten() - expected_slice					).max() < 5E-2



					def 		snake_case	(	self							:  int					)-> List[str]:
								lowerCamelCase__		:		Optional[int]		     =StableDiffusionSAGPipeline.from_pretrained('''stabilityai/stable-diffusion-2-1-base'''					)
								lowerCamelCase__		:		List[str]		     =sag_pipe.to(__lowercase					)
								sag_pipe.set_progress_bar_config(disable=__lowercase					)

								lowerCamelCase__		:		Dict		     ='''.'''
								lowerCamelCase__		:		List[Any]		     =torch.manual_seed(0					)
								lowerCamelCase__		:		Union[str, Any]		     =sag_pipe(
								    [prompt],  generator=__lowercase,  guidance_scale=7.5,  sag_scale=1.0,  num_inference_steps=20,  output_type='''np'''					)

								lowerCamelCase__		:		List[str]		     =output.images

								lowerCamelCase__		:		Union[str, Any]		     =image[0, -3:, -3:, -1]

								assert image.shape == (1, 512, 512, 3)
								lowerCamelCase__		:		Optional[int]		     =np.array([0.3_459, 0.2_876, 0.2_537, 0.3_002, 0.2_671, 0.2_160, 0.3_026, 0.2_262, 0.2_371]					)

								assert np.abs(image_slice.flatten() - expected_slice					).max() < 5E-2




					def 		snake_case	(	self							:  Optional[int]					)-> Dict:
								lowerCamelCase__		:		Dict		     =StableDiffusionSAGPipeline.from_pretrained('''stabilityai/stable-diffusion-2-1-base'''					)
								lowerCamelCase__		:		Optional[Any]		     =sag_pipe.to(__lowercase					)
								sag_pipe.set_progress_bar_config(disable=__lowercase					)

								lowerCamelCase__		:		Union[str, Any]		     ='''.'''
								lowerCamelCase__		:		int		     =torch.manual_seed(0					)
								lowerCamelCase__		:		Any		     =sag_pipe(
								    [prompt],  width=768,  height=512,  generator=__lowercase,  guidance_scale=7.5,  sag_scale=1.0,  num_inference_steps=20,  output_type='''np''',  )

								lowerCamelCase__		:		Optional[int]		     =output.images

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









"""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()




