
import gc
import random
import unittest

import numpy as np
import torch
from transformers import (
    CLIPImageProcessor,
    CLIPTextConfig,
    CLIPTextModel,
    CLIPTokenizer,
    CLIPVisionConfig,
    CLIPVisionModelWithProjection,
)

from diffusers import AutoencoderKL, DDIMScheduler, DDPMScheduler, StableUnCLIPImgaImgPipeline, UNetaDConditionModel
from diffusers.pipelines.pipeline_utils import DiffusionPipeline
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    floats_tensor,
    load_image,
    load_numpy,
    require_torch_gpu,
    skip_mps,
    slow,
    torch_device,
)

from ..pipeline_params import TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS, TEXT_GUIDED_IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()




class 						lowercase__  (						_UpperCAmelCase     ,       _UpperCAmelCase     ,       _UpperCAmelCase     ,       unittest.TestCase  ):
	A__    :   List[str]				       =StableUnCLIPImgaImgPipeline
	A__    :   Dict				       =TEXT_GUIDED_IMAGE_VARIATION_PARAMS
	A__    :   Optional[Any]				       =TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
	A__    :   Union[str, Any]				       =frozenset(
	    []  )  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
	A__    :   int				       =frozenset([]  )
	def      A_				(			self							:   Union[str, Any]							):
						SCREAMING_SNAKE_CASE__													= 32
						SCREAMING_SNAKE_CASE__													= embedder_hidden_size

						# image encoding components

						SCREAMING_SNAKE_CASE__													= CLIPImageProcessor(crop_size=32       ,   size=32							)

						torch.manual_seed(0							)
						SCREAMING_SNAKE_CASE__													= CLIPVisionModelWithProjection(
						    CLIPVisionConfig(
						        hidden_size=UpperCAmelCase_       ,   projection_dim=UpperCAmelCase_       ,   num_hidden_layers=5       ,   num_attention_heads=4       ,   image_size=32       ,   intermediate_size=37       ,   patch_size=1       ,   )							)

						# regular denoising components

						torch.manual_seed(0							)
						SCREAMING_SNAKE_CASE__													= StableUnCLIPImageNormalizer(embedding_dim=UpperCAmelCase_							)
						SCREAMING_SNAKE_CASE__													= DDPMScheduler(beta_schedule='squaredcos_cap_v2'							)

						torch.manual_seed(0							)
						SCREAMING_SNAKE_CASE__													= CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'							)

						torch.manual_seed(0							)
						SCREAMING_SNAKE_CASE__													= CLIPTextModel(
						    CLIPTextConfig(
						        bos_token_id=0       ,   eos_token_id=2       ,   hidden_size=UpperCAmelCase_       ,   projection_dim=32       ,   intermediate_size=37       ,   layer_norm_eps=1e-0_5       ,   num_attention_heads=4       ,   num_hidden_layers=5       ,   pad_token_id=1       ,   vocab_size=1000       ,   )							)

						torch.manual_seed(0							)
						SCREAMING_SNAKE_CASE__													= 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=UpperCAmelCase_       ,   layers_per_block=1       ,   upcast_attention=UpperCAmelCase_       ,   use_linear_projection=UpperCAmelCase_       ,   )

						torch.manual_seed(0							)
						SCREAMING_SNAKE_CASE__													= DDIMScheduler(
						    beta_schedule='scaled_linear'       ,   beta_start=0.00_085       ,   beta_end=0.012       ,   prediction_type='v_prediction'       ,   set_alpha_to_one=UpperCAmelCase_       ,   steps_offset=1       ,   )

						torch.manual_seed(0							)
						SCREAMING_SNAKE_CASE__													= AutoencoderKL()

						SCREAMING_SNAKE_CASE__													= {
						    # image encoding components
						    'feature_extractor': feature_extractor,
						    'image_encoder': image_encoder.eval(),
						    # image noising components
						    'image_normalizer': image_normalizer.eval(),
						    'image_noising_scheduler': image_noising_scheduler,
						    # regular denoising components
						    'tokenizer': tokenizer,
						    'text_encoder': text_encoder.eval(),
						    'unet': unet.eval(),
						    'scheduler': scheduler,
						    'vae': vae.eval(),
						}

						return components
	def      A_				(			self							:   Tuple       ,   UpperCAmelCase_							:   Tuple       ,   UpperCAmelCase_							:   Optional[int]=0       ,   UpperCAmelCase_							:   int=True							):
						if str(UpperCAmelCase_							).startswith('mps'							):
											SCREAMING_SNAKE_CASE__													= torch.manual_seed(UpperCAmelCase_							)
						else:
											SCREAMING_SNAKE_CASE__													= torch.Generator(device=UpperCAmelCase_							).manual_seed(UpperCAmelCase_							)

						SCREAMING_SNAKE_CASE__													= floats_tensor((1, 3, 32, 32)       ,   rng=random.Random(UpperCAmelCase_							)							).to(UpperCAmelCase_							)

						if pil_image:
											SCREAMING_SNAKE_CASE__													= input_image * 0.5 + 0.5
											SCREAMING_SNAKE_CASE__													= input_image.clamp(0       ,   1							)
											SCREAMING_SNAKE_CASE__													= input_image.cpu().permute(0       ,   2       ,   3       ,   1							).float().numpy()
											SCREAMING_SNAKE_CASE__													= DiffusionPipeline.numpy_to_pil(UpperCAmelCase_							)[0]

						return {
						    "prompt": "An anime racoon running a marathon",
						    "image": input_image,
						    "generator": generator,
						    "num_inference_steps": 2,
						    "output_type": "np",
						}
	@skip_mps
	def      A_				(			self							:   List[Any]							):
						SCREAMING_SNAKE_CASE__													= 'cpu'  # ensure determinism for the device-dependent torch.Generator
						SCREAMING_SNAKE_CASE__													= self.get_dummy_components()
						SCREAMING_SNAKE_CASE__													= StableUnCLIPImgaImgPipeline(**UpperCAmelCase_							)
						SCREAMING_SNAKE_CASE__													= sd_pipe.to(UpperCAmelCase_							)
						sd_pipe.set_progress_bar_config(disable=UpperCAmelCase_							)

						SCREAMING_SNAKE_CASE__													= self.get_dummy_inputs(UpperCAmelCase_							)
						inputs.update({'image_embeds': None}							)
						SCREAMING_SNAKE_CASE__													= sd_pipe(**UpperCAmelCase_							).images
						SCREAMING_SNAKE_CASE__													= image[0, -3:, -3:, -1]

						assert image.shape == (1, 32, 32, 3)
						SCREAMING_SNAKE_CASE__													= np.array([0.3_872, 0.7_224, 0.5_601, 0.4_741, 0.6_872, 0.5_814, 0.4_636, 0.3_867, 0.5_078]							)

						assert np.abs(image_slice.flatten() - expected_slice							).max() < 1e-3
	def      A_				(			self							:   Dict							):
						SCREAMING_SNAKE_CASE__													= torch_device in ['cpu', 'mps']

						self._test_attention_slicing_forward_pass(test_max_difference=UpperCAmelCase_							)
	def      A_				(			self							:   Dict							):
						SCREAMING_SNAKE_CASE__													= torch_device in ['cpu', 'mps']

						self._test_inference_batch_single_identical(test_max_difference=UpperCAmelCase_							)





	@unittest.skipIf(
	    torch_device != 'cuda' or not is_xformers_available()       ,   reason='XFormers attention is only available with CUDA and `xformers` installed'       ,   )
	def      A_				(			self							:   int							):
						self._test_xformers_attention_forwardGenerator_pass(test_max_difference=UpperCAmelCase_							)





@slow
@require_torch_gpu
class 						lowercase__  (						unittest.TestCase  ):
	def      A_				(			self							:   Tuple							):
						# clean up the VRAM after each test
						super().tearDown()
						gc.collect()
						torch.cuda.empty_cache()
	def      A_				(			self							:   str							):
						SCREAMING_SNAKE_CASE__													= load_image(
						    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png'							)

						SCREAMING_SNAKE_CASE__													= load_numpy(
						    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_img2img_anime_turtle_fp16.npy'							)

						SCREAMING_SNAKE_CASE__													= StableUnCLIPImgaImgPipeline.from_pretrained(
						    'fusing/stable-unclip-2-1-l-img2img'       ,   torch_dtype=torch.floataa							)
						pipe.to(UpperCAmelCase_							)
						pipe.set_progress_bar_config(disable=UpperCAmelCase_							)
						# 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()

						SCREAMING_SNAKE_CASE__													= torch.Generator(device='cpu'							).manual_seed(0							)
						SCREAMING_SNAKE_CASE__													= pipe(UpperCAmelCase_       ,   'anime turle'       ,   generator=UpperCAmelCase_       ,   output_type='np'							)

						SCREAMING_SNAKE_CASE__													= output.images[0]

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

						assert_mean_pixel_difference(UpperCAmelCase_       ,   UpperCAmelCase_							)
	def      A_				(			self							:   List[str]							):
						SCREAMING_SNAKE_CASE__													= load_image(
						    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png'							)

						SCREAMING_SNAKE_CASE__													= load_numpy(
						    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_h_img2img_anime_turtle_fp16.npy'							)

						SCREAMING_SNAKE_CASE__													= StableUnCLIPImgaImgPipeline.from_pretrained(
						    'fusing/stable-unclip-2-1-h-img2img'       ,   torch_dtype=torch.floataa							)
						pipe.to(UpperCAmelCase_							)
						pipe.set_progress_bar_config(disable=UpperCAmelCase_							)
						# 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()

						SCREAMING_SNAKE_CASE__													= torch.Generator(device='cpu'							).manual_seed(0							)
						SCREAMING_SNAKE_CASE__													= pipe(UpperCAmelCase_       ,   'anime turle'       ,   generator=UpperCAmelCase_       ,   output_type='np'							)

						SCREAMING_SNAKE_CASE__													= output.images[0]

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

						assert_mean_pixel_difference(UpperCAmelCase_       ,   UpperCAmelCase_							)





	def      A_				(			self							:   List[Any]							):
						SCREAMING_SNAKE_CASE__													= load_image(
						    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png'							)

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

						SCREAMING_SNAKE_CASE__													= StableUnCLIPImgaImgPipeline.from_pretrained(
						    'fusing/stable-unclip-2-1-h-img2img'       ,   torch_dtype=torch.floataa							)
						SCREAMING_SNAKE_CASE__													= pipe.to(UpperCAmelCase_							)
						pipe.set_progress_bar_config(disable=UpperCAmelCase_							)
						pipe.enable_attention_slicing()
						pipe.enable_sequential_cpu_offload()

						SCREAMING_SNAKE_CASE__													= pipe(
						    UpperCAmelCase_       ,   'anime turtle'       ,   num_inference_steps=2       ,   output_type='np'       ,   )

						SCREAMING_SNAKE_CASE__													= torch.cuda.max_memory_allocated()
						# make sure that less than 7 GB is allocated
						assert mem_bytes < 7 * 10**9







def 							_lowercase							(     UpperCamelCase_	,	UpperCamelCase_	,	UpperCamelCase_		)		->		int:
					'''simple docstring'''


					def update_area_of_max_square(UpperCamelCase_	,	UpperCamelCase_		) -> int:
										# BASE CASE
										if row >= rows or col >= cols:
															return 0

										SCREAMING_SNAKE_CASE__													= update_area_of_max_square(UpperCamelCase_	,	col + 1		)
										SCREAMING_SNAKE_CASE__													= update_area_of_max_square(row + 1	,	col + 1		)
										SCREAMING_SNAKE_CASE__													= update_area_of_max_square(row + 1	,	UpperCamelCase_		)

										if mat[row][col]:
															SCREAMING_SNAKE_CASE__													= 1 + min([right, diagonal, down]		)
															SCREAMING_SNAKE_CASE__													= max(largest_square_area[0]	,	UpperCamelCase_		)
															return sub_problem_sol
										else:
															return 0

					SCREAMING_SNAKE_CASE__													= [0]
					update_area_of_max_square(0	,	0		)
					return largest_square_area[0]






def 							_lowercase							(     UpperCamelCase_	,	UpperCamelCase_	,	UpperCamelCase_		)		->		int:
					'''simple docstring'''


					def update_area_of_max_square_using_dp_array(
					    UpperCamelCase_	,	UpperCamelCase_	,	UpperCamelCase_		) -> int:
										if row >= rows or col >= cols:
															return 0
										if dp_array[row][col] != -1:
															return dp_array[row][col]

										SCREAMING_SNAKE_CASE__													= update_area_of_max_square_using_dp_array(UpperCamelCase_	,	col + 1	,	UpperCamelCase_		)
										SCREAMING_SNAKE_CASE__													= update_area_of_max_square_using_dp_array(row + 1	,	col + 1	,	UpperCamelCase_		)
										SCREAMING_SNAKE_CASE__													= update_area_of_max_square_using_dp_array(row + 1	,	UpperCamelCase_	,	UpperCamelCase_		)

										if mat[row][col]:
															SCREAMING_SNAKE_CASE__													= 1 + min([right, diagonal, down]		)
															SCREAMING_SNAKE_CASE__													= max(largest_square_area[0]	,	UpperCamelCase_		)
															SCREAMING_SNAKE_CASE__													= sub_problem_sol
															return sub_problem_sol
										else:
															return 0

					SCREAMING_SNAKE_CASE__													= [0]
					SCREAMING_SNAKE_CASE__													= [[-1] * cols for _ in range(UpperCamelCase_		)]
					update_area_of_max_square_using_dp_array(0	,	0	,	UpperCamelCase_		)

					return largest_square_area[0]






def 							_lowercase							(     UpperCamelCase_	,	UpperCamelCase_	,	UpperCamelCase_		)		->		int:
					'''simple docstring'''


					SCREAMING_SNAKE_CASE__													= [[0] * (cols + 1) for _ in range(rows + 1		)]
					SCREAMING_SNAKE_CASE__													= 0
					for row in range(rows - 1	,	-1	,	-1		):
										for col in range(cols - 1	,	-1	,	-1		):
															SCREAMING_SNAKE_CASE__													= dp_array[row][col + 1]
															SCREAMING_SNAKE_CASE__													= dp_array[row + 1][col + 1]
															SCREAMING_SNAKE_CASE__													= dp_array[row + 1][col]

															if mat[row][col] == 1:
																				SCREAMING_SNAKE_CASE__													= 1 + min(UpperCamelCase_	,	UpperCamelCase_	,	UpperCamelCase_		)
																				SCREAMING_SNAKE_CASE__													= max(dp_array[row][col]	,	UpperCamelCase_		)
															else:
																				SCREAMING_SNAKE_CASE__													= 0

					return largest_square_area






def 							_lowercase							(     UpperCamelCase_	,	UpperCamelCase_	,	UpperCamelCase_		)		->		int:
					'''simple docstring'''


					SCREAMING_SNAKE_CASE__													= [0] * (cols + 1)
					SCREAMING_SNAKE_CASE__													= [0] * (cols + 1)
					SCREAMING_SNAKE_CASE__													= 0
					for row in range(rows - 1	,	-1	,	-1		):
										for col in range(cols - 1	,	-1	,	-1		):
															SCREAMING_SNAKE_CASE__													= current_row[col + 1]
															SCREAMING_SNAKE_CASE__													= next_row[col + 1]
															SCREAMING_SNAKE_CASE__													= next_row[col]

															if mat[row][col] == 1:
																				SCREAMING_SNAKE_CASE__													= 1 + min(UpperCamelCase_	,	UpperCamelCase_	,	UpperCamelCase_		)
																				SCREAMING_SNAKE_CASE__													= max(current_row[col]	,	UpperCamelCase_		)
															else:
																				SCREAMING_SNAKE_CASE__													= 0
										SCREAMING_SNAKE_CASE__													= current_row

					return largest_square_area


if __name__ == "__main__":
				import doctest

				doctest.testmod()
				print(largest_square_area_in_matrix_bottom_up(2, 2, [[1, 1], [1, 1]]))





