"""simple docstring"""







import gc
import os
import tempfile
import unittest

import torch
import torch.nn as nn
import torch.nn.functional as F
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import AutoencoderKL, DDIMScheduler, StableDiffusionPipeline, UNetaDConditionModel
from diffusers.loaders import AttnProcsLayers, LoraLoaderMixin
from diffusers.models.attention_processor import (
    Attention,
    AttnProcessor,
    AttnProcessor2_0,
    LoRAAttnProcessor,
    LoRAAttnProcessor2_0,
    LoRAXFormersAttnProcessor,
    XFormersAttnProcessor,
)
from diffusers.utils import TEXT_ENCODER_ATTN_MODULE, floats_tensor, torch_device
def     __UpperCAmelCase							(       UpperCAmelCase_       :							Optional[int]    )				->     List[str]:

	'''simple docstring'''





	__snake_case     :				Union[str, Any] 						=					{}
	for name in unet.attn_processors.keys():
		__snake_case     :				List[str] 						=					None if name.endswith('attn1.processor'    ) else unet.config.cross_attention_dim
		if name.startswith('mid_block'    ):
			__snake_case     :				Union[str, Any] 						=					unet.config.block_out_channels[-1]
		elif name.startswith('up_blocks'    ):
			__snake_case     :				str 						=					int(name[len('up_blocks.'    )]    )
			__snake_case     :				Tuple 						=					list(reversed(unet.config.block_out_channels    )    )[block_id]
		elif name.startswith('down_blocks'    ):
			__snake_case     :				List[Any] 						=					int(name[len('down_blocks.'    )]    )
			__snake_case     :				Optional[int] 						=					unet.config.block_out_channels[block_id]
		__snake_case     :				str 						=					(
		    LoRAAttnProcessor2_0 if hasattr(_lowercase     ,				'scaled_dot_product_attention'    ) else LoRAAttnProcessor
		)
		__snake_case     :				str 						=					lora_attn_processor_class(
		    hidden_size=_lowercase     ,				cross_attention_dim=_lowercase    )
	__snake_case     :				Any 						=					AttnProcsLayers(_lowercase    )
	return lora_attn_procs, unet_lora_layers
def     __UpperCAmelCase							(       UpperCAmelCase_       :							Union[str, Any]    )				->     str:

	'''simple docstring'''





	__snake_case     :				List[Any] 						=					{}
	__snake_case     :				Optional[Any] 						=					(
	    LoRAAttnProcessor2_0 if hasattr(_lowercase     ,				'scaled_dot_product_attention'    ) else LoRAAttnProcessor
	)
	for name, module in text_encoder.named_modules():
		if name.endswith(_lowercase    ):
			__snake_case     :				List[Any] 						=					lora_attn_processor_class(
			    hidden_size=module.out_proj.out_features     ,				cross_attention_dim=_lowercase    )
	return text_lora_attn_procs
def     __UpperCAmelCase							(       UpperCAmelCase_       :							Optional[int]    )				->     int:

	'''simple docstring'''





	__snake_case     :				int 						=					create_text_encoder_lora_attn_procs(_lowercase    )
	__snake_case     :				List[str] 						=					AttnProcsLayers(_lowercase    )
	return text_encoder_lora_layers
def     __UpperCAmelCase							(       UpperCAmelCase_       :							Tuple     ,				UpperCAmelCase_       :							List[str]=False    )				->     Tuple:

	'''simple docstring'''





	for _, attn_proc in text_lora_attn_procs.items():
		# set up.weights
		for layer_name, layer_module in attn_proc.named_modules():
			if layer_name.endswith('_lora'    ):
				__snake_case     :				Dict 						=					(
				    torch.randn_like(layer_module.up.weight    )
				    if randn_weight
				    else torch.zeros_like(layer_module.up.weight    )
				)
				__snake_case     :				List[str] 						=					torch.nn.Parameter(_lowercase    )

class 						UpperCamelCase     (       unittest.TestCase		):



				def    _lowercase	(self							: Any)       ->      Optional[int]:
					torch.manual_seed(0)
					__snake_case     :				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      ,    )
					__snake_case     :				str 						=					DDIMScheduler(
					    beta_start=0.00_085      ,    beta_end=0.012      ,    beta_schedule='scaled_linear'      ,    clip_sample=_A      ,    set_alpha_to_one=_A      ,    steps_offset=1      ,    )
					torch.manual_seed(0)
					__snake_case     :				Optional[Any] 						=					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      ,    )
					__snake_case     :				Union[str, 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=10_00      ,    )
					__snake_case     :				Dict 						=					CLIPTextModel(_A)
					__snake_case     :				Union[str, Any] 						=					CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip')

					__snake_case     :				Optional[int] 						=					create_unet_lora_layers(_A)
					__snake_case     :				Optional[Any] 						=					create_text_encoder_lora_layers(_A)

					__snake_case     :				List[Any] 						=					{
					    """unet""": unet,
					    """scheduler""": scheduler,
					    """vae""": vae,
					    """text_encoder""": text_encoder,
					    """tokenizer""": tokenizer,
					    """safety_checker""": None,
					    """feature_extractor""": None,
					}
					__snake_case     :				Any 						=					{
					    """unet_lora_layers""": unet_lora_layers,
					    """text_encoder_lora_layers""": text_encoder_lora_layers,
					    """unet_lora_attn_procs""": unet_lora_attn_procs,
					}
					return pipeline_components, lora_components



				def    _lowercase	(self							: Optional[Any])       ->      str:
					__snake_case     :				Dict 						=					1
					__snake_case     :				Tuple 						=					10
					__snake_case     :				str 						=					4
					__snake_case     :				Any 						=					(32, 32)

					__snake_case     :				str 						=					torch.manual_seed(0)
					__snake_case     :				str 						=					floats_tensor((batch_size, num_channels) + sizes)
					__snake_case     :				Any 						=					torch.randint(1      ,    _A      ,    size=(batch_size, sequence_length)      ,    generator=_A)

					__snake_case     :				int 						=					{
					    """prompt""": """A painting of a squirrel eating a burger""",
					    """generator""": generator,
					    """num_inference_steps""": 2,
					    """guidance_scale""": 6.0,
					    """output_type""": """numpy""",
					}

					return noise, input_ids, pipeline_inputs

					# copied from: https://colab.research.google.com/gist/sayakpaul/df2ef6e1ae6d8c10a49d859883b10860/scratchpad.ipynb



				def    _lowercase	(self							: int)       ->      Union[str, Any]:
					__snake_case     :				str 						=					77

					__snake_case     :				Union[str, Any] 						=					torch.randint(2      ,    56      ,    size=(1, max_seq_length)      ,    generator=torch.manual_seed(0))

					__snake_case     :				Any 						=					{}
					__snake_case     :				Optional[int] 						=					inputs
					return prepared_inputs



				def    _lowercase	(self							: List[str]      ,    _A							: Optional[Any])       ->      Union[str, Any]:
					__snake_case     :				List[Any] 						=					self.get_dummy_components()
					LoraLoaderMixin.save_lora_weights(
					    save_directory=_A      ,    unet_lora_layers=lora_components['unet_lora_layers']      ,    text_encoder_lora_layers=lora_components['text_encoder_lora_layers']      ,    )
					self.assertTrue(os.path.isfile(os.path.join(_A      ,    'pytorch_lora_weights.bin')))



				def    _lowercase	(self							: str)       ->      List[str]:
					__snake_case     :				List[Any] 						=					self.get_dummy_components()
					__snake_case     :				Optional[Any] 						=					StableDiffusionPipeline(**_A)
					__snake_case     :				Dict 						=					sd_pipe.to(_A)
					sd_pipe.set_progress_bar_config(disable=_A)

					__snake_case     :				Any 						=					self.get_dummy_inputs()

					__snake_case     :				str 						=					sd_pipe(**_A).images
					__snake_case     :				str 						=					original_images[0, -3:, -3:, -1]

					with tempfile.TemporaryDirectory() as tmpdirname:
						LoraLoaderMixin.save_lora_weights(
						    save_directory=_A      ,    unet_lora_layers=lora_components['unet_lora_layers']      ,    text_encoder_lora_layers=lora_components['text_encoder_lora_layers']      ,    )
						self.assertTrue(os.path.isfile(os.path.join(_A      ,    'pytorch_lora_weights.bin')))
						sd_pipe.load_lora_weights(_A)

					__snake_case     :				Tuple 						=					sd_pipe(**_A).images
					__snake_case     :				Optional[Any] 						=					lora_images[0, -3:, -3:, -1]

					# Outputs shouldn't match.
					self.assertFalse(torch.allclose(torch.from_numpy(_A)      ,    torch.from_numpy(_A)))



				def    _lowercase	(self							: List[Any])       ->      List[str]:
					__snake_case     :				List[Any] 						=					self.get_dummy_components()
					__snake_case     :				str 						=					StableDiffusionPipeline(**_A)
					__snake_case     :				List[str] 						=					sd_pipe.to(_A)
					sd_pipe.set_progress_bar_config(disable=_A)

					__snake_case     :				Union[str, Any] 						=					self.get_dummy_inputs()

					__snake_case     :				List[str] 						=					sd_pipe(**_A).images
					__snake_case     :				Any 						=					original_images[0, -3:, -3:, -1]

					with tempfile.TemporaryDirectory() as tmpdirname:
						LoraLoaderMixin.save_lora_weights(
						    save_directory=_A      ,    unet_lora_layers=lora_components['unet_lora_layers']      ,    text_encoder_lora_layers=lora_components['text_encoder_lora_layers']      ,    safe_serialization=_A      ,    )
						self.assertTrue(os.path.isfile(os.path.join(_A      ,    'pytorch_lora_weights.safetensors')))
						sd_pipe.load_lora_weights(_A)

					__snake_case     :				List[Any] 						=					sd_pipe(**_A).images
					__snake_case     :				str 						=					lora_images[0, -3:, -3:, -1]

					# Outputs shouldn't match.
					self.assertFalse(torch.allclose(torch.from_numpy(_A)      ,    torch.from_numpy(_A)))



				def    _lowercase	(self							: Optional[Any])       ->      Optional[int]:
					__snake_case     :				List[Any] 						=					self.get_dummy_components()
					__snake_case     :				int 						=					lora_components["""unet_lora_attn_procs"""]
					__snake_case     :				Optional[int] 						=					StableDiffusionPipeline(**_A)
					__snake_case     :				List[Any] 						=					sd_pipe.to(_A)
					sd_pipe.set_progress_bar_config(disable=_A)

					__snake_case     :				Dict 						=					self.get_dummy_inputs()

					__snake_case     :				Optional[int] 						=					sd_pipe(**_A).images
					__snake_case     :				Optional[Any] 						=					original_images[0, -3:, -3:, -1]

					with tempfile.TemporaryDirectory() as tmpdirname:
						__snake_case     :				Tuple 						=					sd_pipe.unet
						unet.set_attn_processor(_A)
						unet.save_attn_procs(_A)
						self.assertTrue(os.path.isfile(os.path.join(_A      ,    'pytorch_lora_weights.bin')))
						sd_pipe.load_lora_weights(_A)

					__snake_case     :				int 						=					sd_pipe(**_A).images
					__snake_case     :				Optional[Any] 						=					lora_images[0, -3:, -3:, -1]

					# Outputs shouldn't match.
					self.assertFalse(torch.allclose(torch.from_numpy(_A)      ,    torch.from_numpy(_A)))



				def    _lowercase	(self							: str)       ->      Dict:
					__snake_case     :				Optional[int] 						=					self.get_dummy_components()
					__snake_case     :				List[Any] 						=					StableDiffusionPipeline(**_A)

					__snake_case     :				Optional[Any] 						=					self.get_dummy_tokens()

					# inference without lora
					__snake_case     :				Dict 						=					pipe.text_encoder(**_A)[0]
					assert outputs_without_lora.shape == (1, 77, 32)

					# create lora_attn_procs with zeroed out up.weights
					__snake_case     :				Optional[int] 						=					create_text_encoder_lora_attn_procs(pipe.text_encoder)
					set_lora_up_weights(_A      ,    randn_weight=_A)

					# monkey patch
					pipe._modify_text_encoder(_A)

					# verify that it's okay to release the text_attn_procs which holds the LoRAAttnProcessor.
					del text_attn_procs
					gc.collect()

					# inference with lora
					__snake_case     :				Optional[int] 						=					pipe.text_encoder(**_A)[0]
					assert outputs_with_lora.shape == (1, 77, 32)

					assert torch.allclose(
					    _A      ,    _A), "lora_up_weight are all zero, so the lora outputs should be the same to without lora outputs"

					# create lora_attn_procs with randn up.weights
					__snake_case     :				str 						=					create_text_encoder_lora_attn_procs(pipe.text_encoder)
					set_lora_up_weights(_A      ,    randn_weight=_A)

					# monkey patch
					pipe._modify_text_encoder(_A)

					# verify that it's okay to release the text_attn_procs which holds the LoRAAttnProcessor.
					del text_attn_procs
					gc.collect()

					# inference with lora
					__snake_case     :				Tuple 						=					pipe.text_encoder(**_A)[0]
					assert outputs_with_lora.shape == (1, 77, 32)

					assert not torch.allclose(
					    _A      ,    _A), "lora_up_weight are not zero, so the lora outputs should be different to without lora outputs"



				def    _lowercase	(self							: str)       ->      Dict:
					__snake_case     :				Dict 						=					self.get_dummy_components()
					__snake_case     :				str 						=					StableDiffusionPipeline(**_A)

					__snake_case     :				Optional[Any] 						=					self.get_dummy_tokens()

					# inference without lora
					__snake_case     :				Union[str, Any] 						=					pipe.text_encoder(**_A)[0]
					assert outputs_without_lora.shape == (1, 77, 32)

					# create lora_attn_procs with randn up.weights
					__snake_case     :				Dict 						=					create_text_encoder_lora_attn_procs(pipe.text_encoder)
					set_lora_up_weights(_A      ,    randn_weight=_A)

					# monkey patch
					pipe._modify_text_encoder(_A)

					# verify that it's okay to release the text_attn_procs which holds the LoRAAttnProcessor.
					del text_attn_procs
					gc.collect()

					# inference with lora
					__snake_case     :				str 						=					pipe.text_encoder(**_A)[0]
					assert outputs_with_lora.shape == (1, 77, 32)

					assert not torch.allclose(
					    _A      ,    _A), "lora outputs should be different to without lora outputs"

					# remove monkey patch
					pipe._remove_text_encoder_monkey_patch()

					# inference with removed lora
					__snake_case     :				Dict 						=					pipe.text_encoder(**_A)[0]
					assert outputs_without_lora_removed.shape == (1, 77, 32)

					assert torch.allclose(
					    _A      ,    _A), "remove lora monkey patch should restore the original outputs"



				def    _lowercase	(self							: Dict)       ->      Optional[int]:
					__snake_case     :				Any 						=					self.get_dummy_components()
					__snake_case     :				str 						=					StableDiffusionPipeline(**_A)
					__snake_case     :				int 						=					sd_pipe.to(_A)
					sd_pipe.set_progress_bar_config(disable=_A)

					__snake_case     :				Any 						=					self.get_dummy_inputs()

					with tempfile.TemporaryDirectory() as tmpdirname:
						LoraLoaderMixin.save_lora_weights(
						    save_directory=_A      ,    unet_lora_layers=lora_components['unet_lora_layers']      ,    text_encoder_lora_layers=lora_components['text_encoder_lora_layers']      ,    )
						self.assertTrue(os.path.isfile(os.path.join(_A      ,    'pytorch_lora_weights.bin')))
						sd_pipe.load_lora_weights(_A)

					__snake_case     :				Any 						=					sd_pipe(**_A).images
					__snake_case     :				List[Any] 						=					lora_images[0, -3:, -3:, -1]

					__snake_case     :				List[str] 						=					sd_pipe(**_A      ,    cross_attention_kwargs={'scale': 0.5}).images
					__snake_case     :				List[str] 						=					lora_images_with_scale[0, -3:, -3:, -1]

					# Outputs shouldn't match.
					self.assertFalse(
					    torch.allclose(torch.from_numpy(_A)      ,    torch.from_numpy(_A)))



				def    _lowercase	(self							: List[Any])       ->      Tuple:
					with tempfile.TemporaryDirectory() as tmpdirname:
						self.create_lora_weight_file(_A)

						__snake_case     :				str 						=					self.get_dummy_components()
						__snake_case     :				Any 						=					StableDiffusionPipeline(**_A)
						__snake_case     :				Optional[int] 						=					sd_pipe.to(_A)
						sd_pipe.set_progress_bar_config(disable=_A)

						# check if vanilla attention processors are used
						for _, module in sd_pipe.unet.named_modules():
							if isinstance(_A      ,    _A):
								self.assertIsInstance(module.processor      ,    (AttnProcessor, AttnProcessor2_0))

            # load LoRA weight file
						sd_pipe.load_lora_weights(_A)

						# check if lora attention processors are used
						for _, module in sd_pipe.unet.named_modules():
							if isinstance(_A      ,    _A):
								__snake_case     :				Optional[Any] 						=					(
								    LoRAAttnProcessor2_0 if hasattr(_A      ,    'scaled_dot_product_attention') else LoRAAttnProcessor
								)
								self.assertIsInstance(module.processor      ,    _A)



				@unittest.skipIf(torch_device != 'cuda'      ,    'This test is supposed to run on GPU')
				def    _lowercase	(self							: int)       ->      int:
					with tempfile.TemporaryDirectory() as tmpdirname:
						self.create_lora_weight_file(_A)

						__snake_case     :				List[Any] 						=					self.get_dummy_components()
						__snake_case     :				List[Any] 						=					StableDiffusionPipeline(**_A)
						__snake_case     :				Tuple 						=					sd_pipe.to(_A)
						sd_pipe.set_progress_bar_config(disable=_A)

						# enable XFormers
						sd_pipe.enable_xformers_memory_efficient_attention()

						# check if xFormers attention processors are used
						for _, module in sd_pipe.unet.named_modules():
							if isinstance(_A      ,    _A):
								self.assertIsInstance(module.processor      ,    _A)

            # load LoRA weight file
						sd_pipe.load_lora_weights(_A)

						# check if lora attention processors are used
						for _, module in sd_pipe.unet.named_modules():
							if isinstance(_A      ,    _A):
								self.assertIsInstance(module.processor      ,    _A)






				@unittest.skipIf(torch_device != 'cuda'      ,    'This test is supposed to run on GPU')
				def    _lowercase	(self							: str)       ->      List[Any]:
					__snake_case     :				str 						=					self.get_dummy_components()
					__snake_case     :				Union[str, Any] 						=					StableDiffusionPipeline(**_A)
					__snake_case     :				List[Any] 						=					sd_pipe.to(_A)
					sd_pipe.set_progress_bar_config(disable=_A)

					__snake_case     :				Union[str, Any] 						=					self.get_dummy_inputs()

					# enable XFormers
					sd_pipe.enable_xformers_memory_efficient_attention()

					__snake_case     :				Tuple 						=					sd_pipe(**_A).images
					__snake_case     :				Any 						=					original_images[0, -3:, -3:, -1]

					with tempfile.TemporaryDirectory() as tmpdirname:
						LoraLoaderMixin.save_lora_weights(
						    save_directory=_A      ,    unet_lora_layers=lora_components['unet_lora_layers']      ,    text_encoder_lora_layers=lora_components['text_encoder_lora_layers']      ,    )
						self.assertTrue(os.path.isfile(os.path.join(_A      ,    'pytorch_lora_weights.bin')))
						sd_pipe.load_lora_weights(_A)

					__snake_case     :				Dict 						=					sd_pipe(**_A).images
					__snake_case     :				int 						=					lora_images[0, -3:, -3:, -1]

					# Outputs shouldn't match.
					self.assertFalse(torch.allclose(torch.from_numpy(_A)      ,    torch.from_numpy(_A)))










'''simple docstring'''




import sys



def  __lowerCamelCase    (   _lowercase      ,							_lowercase						) ->     int:
							if not numbers:
														return 0
							if target == 0 and target in numbers:
														return 0
							if not isinstance(_lowercase      ,							(list, tuple)						) or not all(
							    isinstance(_lowercase      ,							_lowercase						) for number in numbers						):
														raise ValueError("""numbers must be an iterable of integers"""						)

							UpperCAmelCase	:						Optional[Any]					=						0
							UpperCAmelCase	:						List[str]					=						sys.maxsize

							while right < len(_lowercase						):
														curr_sum += numbers[right]
														while curr_sum >= target and left <= right:
																					UpperCAmelCase	:						Optional[Any]					=						min(_lowercase      ,							right - left + 1						)
																					curr_sum -= numbers[left]
																					left += 1
														right += 1

							return 0 if min_len == sys.maxsize else min_len

