




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       _a (		SCREAMING_SNAKE_CASE_				:							Dict	):
					__lowerCAmelCase							  =			{}
					for name in unet.attn_processors.keys():
										__lowerCAmelCase							  =			None if name.endswith("attn1.processor"	) else unet.config.cross_attention_dim
										if name.startswith("mid_block"	):
															__lowerCAmelCase							  =			unet.config.block_out_channels[-1]
										elif name.startswith("up_blocks"	):
															__lowerCAmelCase							  =			int(name[len("up_blocks."	)]	)
															__lowerCAmelCase							  =			list(reversed(unet.config.block_out_channels	)	)[block_id]
										elif name.startswith("down_blocks"	):
															__lowerCAmelCase							  =			int(name[len("down_blocks."	)]	)
															__lowerCAmelCase							  =			unet.config.block_out_channels[block_id]
										__lowerCAmelCase							  =			(
										    LoRAAttnProcessor2_0 if hasattr(lowerCamelCase__ , "scaled_dot_product_attention"	) else LoRAAttnProcessor
										)
										__lowerCAmelCase							  =			lora_attn_processor_class(
										    hidden_size=lowerCamelCase__ , cross_attention_dim=lowerCamelCase__	)
					__lowerCAmelCase							  =			AttnProcsLayers(lowerCamelCase__	)
					return lora_attn_procs, unet_lora_layers







def       _a (		SCREAMING_SNAKE_CASE_				:							Tuple	):
					__lowerCAmelCase							  =			{}
					__lowerCAmelCase							  =			(
					    LoRAAttnProcessor2_0 if hasattr(lowerCamelCase__ , "scaled_dot_product_attention"	) else LoRAAttnProcessor
					)
					for name, module in text_encoder.named_modules():
										if name.endswith(lowerCamelCase__	):
															__lowerCAmelCase							  =			lora_attn_processor_class(
															    hidden_size=module.out_proj.out_features , cross_attention_dim=lowerCamelCase__	)
					return text_lora_attn_procs







def       _a (		SCREAMING_SNAKE_CASE_				:							str	):
					__lowerCAmelCase							  =			create_text_encoder_lora_attn_procs(lowerCamelCase__	)
					__lowerCAmelCase							  =			AttnProcsLayers(lowerCamelCase__	)
					return text_encoder_lora_layers







def       _a (		SCREAMING_SNAKE_CASE_				:							Any , SCREAMING_SNAKE_CASE_				:							str=False	):
					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"	):
																				__lowerCAmelCase							  =			(
																				    torch.randn_like(layer_module.up.weight	)
																				    if randn_weight
																				    else torch.zeros_like(layer_module.up.weight	)
																				)
																				__lowerCAmelCase							  =			torch.nn.Parameter(lowerCamelCase__	)


class      a__       (   unittest.TestCase ):






						def 							__SCREAMING_SNAKE_CASE(  self     ):




											"""simple docstring"""



											torch.manual_seed(0     )
											__lowerCAmelCase							  =			UNetaDConditionModel(
											    block_out_channels=(3_2, 6_4)						,     layers_per_block=2						,     sample_size=3_2						,     in_channels=4						,     out_channels=4						,     down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")						,     up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")						,     cross_attention_dim=3_2						,     )
											__lowerCAmelCase							  =			DDIMScheduler(
											    beta_start=0.0_00_85						,     beta_end=0.0_12						,     beta_schedule="scaled_linear"						,     clip_sample=_snake_case						,     set_alpha_to_one=_snake_case						,     steps_offset=1						,     )
											torch.manual_seed(0     )
											__lowerCAmelCase							  =			AutoencoderKL(
											    block_out_channels=[3_2, 6_4]						,     in_channels=3						,     out_channels=3						,     down_block_types=["DownEncoderBlock2D", "DownEncoderBlock2D"]						,     up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D"]						,     latent_channels=4						,     )
											__lowerCAmelCase							  =			CLIPTextConfig(
											    bos_token_id=0						,     eos_token_id=2						,     hidden_size=3_2						,     intermediate_size=3_7						,     layer_norm_eps=1E-0_5						,     num_attention_heads=4						,     num_hidden_layers=5						,     pad_token_id=1						,     vocab_size=1_0_0_0						,     )
											__lowerCAmelCase							  =			CLIPTextModel(_snake_case     )
											__lowerCAmelCase							  =			CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"     )

											__lowerCAmelCase							  =			create_unet_lora_layers(_snake_case     )
											__lowerCAmelCase							  =			create_text_encoder_lora_layers(_snake_case     )

											__lowerCAmelCase							  =			{
											    'unet': unet,
											    'scheduler': scheduler,
											    'vae': vae,
											    'text_encoder': text_encoder,
											    'tokenizer': tokenizer,
											    'safety_checker': None,
											    'feature_extractor': None,
											}
											__lowerCAmelCase							  =			{
											    '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 							__SCREAMING_SNAKE_CASE(  self     ):




											"""simple docstring"""



											__lowerCAmelCase							  =			1
											__lowerCAmelCase							  =			1_0
											__lowerCAmelCase							  =			4
											__lowerCAmelCase							  =			(3_2, 3_2)

											__lowerCAmelCase							  =			torch.manual_seed(0     )
											__lowerCAmelCase							  =			floats_tensor((batch_size, num_channels) + sizes     )
											__lowerCAmelCase							  =			torch.randint(1						,     _snake_case						,     size=(batch_size, sequence_length)						,     generator=_snake_case     )

											__lowerCAmelCase							  =			{
											    '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 							__SCREAMING_SNAKE_CASE(  self     ):




											"""simple docstring"""



											__lowerCAmelCase							  =			7_7

											__lowerCAmelCase							  =			torch.randint(2						,     5_6						,     size=(1, max_seq_length)						,     generator=torch.manual_seed(0     )     )

											__lowerCAmelCase							  =			{}
											__lowerCAmelCase							  =			inputs
											return prepared_inputs






						def 							__SCREAMING_SNAKE_CASE(  self						,     _A     ):




											"""simple docstring"""



											__lowerCAmelCase							  =			self.get_dummy_components()
											LoraLoaderMixin.save_lora_weights(
											    save_directory=_snake_case						,     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(_snake_case						,     "pytorch_lora_weights.bin"     )     )     )






						def 							__SCREAMING_SNAKE_CASE(  self     ):




											"""simple docstring"""



											__lowerCAmelCase							  =			self.get_dummy_components()
											__lowerCAmelCase							  =			StableDiffusionPipeline(**_snake_case     )
											__lowerCAmelCase							  =			sd_pipe.to(_snake_case     )
											sd_pipe.set_progress_bar_config(disable=_snake_case     )

											__lowerCAmelCase							  =			self.get_dummy_inputs()

											__lowerCAmelCase							  =			sd_pipe(**_snake_case     ).images
											__lowerCAmelCase							  =			original_images[0, -3:, -3:, -1]

											with tempfile.TemporaryDirectory() as tmpdirname:
																LoraLoaderMixin.save_lora_weights(
																    save_directory=_snake_case						,     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(_snake_case						,     "pytorch_lora_weights.bin"     )     )     )
																sd_pipe.load_lora_weights(_snake_case     )

											__lowerCAmelCase							  =			sd_pipe(**_snake_case     ).images
											__lowerCAmelCase							  =			lora_images[0, -3:, -3:, -1]

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






						def 							__SCREAMING_SNAKE_CASE(  self     ):




											"""simple docstring"""



											__lowerCAmelCase							  =			self.get_dummy_components()
											__lowerCAmelCase							  =			StableDiffusionPipeline(**_snake_case     )
											__lowerCAmelCase							  =			sd_pipe.to(_snake_case     )
											sd_pipe.set_progress_bar_config(disable=_snake_case     )

											__lowerCAmelCase							  =			self.get_dummy_inputs()

											__lowerCAmelCase							  =			sd_pipe(**_snake_case     ).images
											__lowerCAmelCase							  =			original_images[0, -3:, -3:, -1]

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

											__lowerCAmelCase							  =			sd_pipe(**_snake_case     ).images
											__lowerCAmelCase							  =			lora_images[0, -3:, -3:, -1]

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






						def 							__SCREAMING_SNAKE_CASE(  self     ):




											"""simple docstring"""



											__lowerCAmelCase							  =			self.get_dummy_components()
											__lowerCAmelCase							  =			lora_components['unet_lora_attn_procs']
											__lowerCAmelCase							  =			StableDiffusionPipeline(**_snake_case     )
											__lowerCAmelCase							  =			sd_pipe.to(_snake_case     )
											sd_pipe.set_progress_bar_config(disable=_snake_case     )

											__lowerCAmelCase							  =			self.get_dummy_inputs()

											__lowerCAmelCase							  =			sd_pipe(**_snake_case     ).images
											__lowerCAmelCase							  =			original_images[0, -3:, -3:, -1]

											with tempfile.TemporaryDirectory() as tmpdirname:
																__lowerCAmelCase							  =			sd_pipe.unet
																unet.set_attn_processor(_snake_case     )
																unet.save_attn_procs(_snake_case     )
																self.assertTrue(os.path.isfile(os.path.join(_snake_case						,     "pytorch_lora_weights.bin"     )     )     )
																sd_pipe.load_lora_weights(_snake_case     )

											__lowerCAmelCase							  =			sd_pipe(**_snake_case     ).images
											__lowerCAmelCase							  =			lora_images[0, -3:, -3:, -1]

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






						def 							__SCREAMING_SNAKE_CASE(  self     ):




											"""simple docstring"""



											__lowerCAmelCase							  =			self.get_dummy_components()
											__lowerCAmelCase							  =			StableDiffusionPipeline(**_snake_case     )

											__lowerCAmelCase							  =			self.get_dummy_tokens()

											# inference without lora
											__lowerCAmelCase							  =			pipe.text_encoder(**_snake_case     )[0]
											assert outputs_without_lora.shape == (1, 7_7, 3_2)

											# create lora_attn_procs with zeroed out up.weights
											__lowerCAmelCase							  =			create_text_encoder_lora_attn_procs(pipe.text_encoder     )
											set_lora_up_weights(_snake_case						,     randn_weight=_snake_case     )

											# monkey patch
											pipe._modify_text_encoder(_snake_case     )

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

											# inference with lora
											__lowerCAmelCase							  =			pipe.text_encoder(**_snake_case     )[0]
											assert outputs_with_lora.shape == (1, 7_7, 3_2)

											assert torch.allclose(
											    _snake_case						,     _snake_case     ), "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
											__lowerCAmelCase							  =			create_text_encoder_lora_attn_procs(pipe.text_encoder     )
											set_lora_up_weights(_snake_case						,     randn_weight=_snake_case     )

											# monkey patch
											pipe._modify_text_encoder(_snake_case     )

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

											# inference with lora
											__lowerCAmelCase							  =			pipe.text_encoder(**_snake_case     )[0]
											assert outputs_with_lora.shape == (1, 7_7, 3_2)

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






						def 							__SCREAMING_SNAKE_CASE(  self     ):




											"""simple docstring"""



											__lowerCAmelCase							  =			self.get_dummy_components()
											__lowerCAmelCase							  =			StableDiffusionPipeline(**_snake_case     )

											__lowerCAmelCase							  =			self.get_dummy_tokens()

											# inference without lora
											__lowerCAmelCase							  =			pipe.text_encoder(**_snake_case     )[0]
											assert outputs_without_lora.shape == (1, 7_7, 3_2)

											# create lora_attn_procs with randn up.weights
											__lowerCAmelCase							  =			create_text_encoder_lora_attn_procs(pipe.text_encoder     )
											set_lora_up_weights(_snake_case						,     randn_weight=_snake_case     )

											# monkey patch
											pipe._modify_text_encoder(_snake_case     )

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

											# inference with lora
											__lowerCAmelCase							  =			pipe.text_encoder(**_snake_case     )[0]
											assert outputs_with_lora.shape == (1, 7_7, 3_2)

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

											# remove monkey patch
											pipe._remove_text_encoder_monkey_patch()

											# inference with removed lora
											__lowerCAmelCase							  =			pipe.text_encoder(**_snake_case     )[0]
											assert outputs_without_lora_removed.shape == (1, 7_7, 3_2)

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






						def 							__SCREAMING_SNAKE_CASE(  self     ):




											"""simple docstring"""



											__lowerCAmelCase							  =			self.get_dummy_components()
											__lowerCAmelCase							  =			StableDiffusionPipeline(**_snake_case     )
											__lowerCAmelCase							  =			sd_pipe.to(_snake_case     )
											sd_pipe.set_progress_bar_config(disable=_snake_case     )

											__lowerCAmelCase							  =			self.get_dummy_inputs()

											with tempfile.TemporaryDirectory() as tmpdirname:
																LoraLoaderMixin.save_lora_weights(
																    save_directory=_snake_case						,     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(_snake_case						,     "pytorch_lora_weights.bin"     )     )     )
																sd_pipe.load_lora_weights(_snake_case     )

											__lowerCAmelCase							  =			sd_pipe(**_snake_case     ).images
											__lowerCAmelCase							  =			lora_images[0, -3:, -3:, -1]

											__lowerCAmelCase							  =			sd_pipe(**_snake_case						,     cross_attention_kwargs={"scale": 0.5}     ).images
											__lowerCAmelCase							  =			lora_images_with_scale[0, -3:, -3:, -1]

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






						def 							__SCREAMING_SNAKE_CASE(  self     ):




											"""simple docstring"""



											with tempfile.TemporaryDirectory() as tmpdirname:
																self.create_lora_weight_file(_snake_case     )

																__lowerCAmelCase							  =			self.get_dummy_components()
																__lowerCAmelCase							  =			StableDiffusionPipeline(**_snake_case     )
																__lowerCAmelCase							  =			sd_pipe.to(_snake_case     )
																sd_pipe.set_progress_bar_config(disable=_snake_case     )

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

																										  # load LoRA weight file
																sd_pipe.load_lora_weights(_snake_case     )

																# check if lora attention processors are used
																for _, module in sd_pipe.unet.named_modules():
																					if isinstance(_snake_case						,     _snake_case     ):
																										__lowerCAmelCase							  =			(
																										    LoRAAttnProcessor2_0 if hasattr(_snake_case						,     "scaled_dot_product_attention"     ) else LoRAAttnProcessor
																										)
																										self.assertIsInstance(module.processor						,     _snake_case     )






						@unittest.skipIf(torch_device != "cuda"						,     "This test is supposed to run on GPU"     )
						def 							__SCREAMING_SNAKE_CASE(  self     ):




											"""simple docstring"""



											with tempfile.TemporaryDirectory() as tmpdirname:
																self.create_lora_weight_file(_snake_case     )

																__lowerCAmelCase							  =			self.get_dummy_components()
																__lowerCAmelCase							  =			StableDiffusionPipeline(**_snake_case     )
																__lowerCAmelCase							  =			sd_pipe.to(_snake_case     )
																sd_pipe.set_progress_bar_config(disable=_snake_case     )

																# 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(_snake_case						,     _snake_case     ):
																										self.assertIsInstance(module.processor						,     _snake_case     )

																										  # load LoRA weight file
																sd_pipe.load_lora_weights(_snake_case     )

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






						@unittest.skipIf(torch_device != "cuda"						,     "This test is supposed to run on GPU"     )
						def 							__SCREAMING_SNAKE_CASE(  self     ):




											"""simple docstring"""



											__lowerCAmelCase							  =			self.get_dummy_components()
											__lowerCAmelCase							  =			StableDiffusionPipeline(**_snake_case     )
											__lowerCAmelCase							  =			sd_pipe.to(_snake_case     )
											sd_pipe.set_progress_bar_config(disable=_snake_case     )

											__lowerCAmelCase							  =			self.get_dummy_inputs()

											# enable XFormers
											sd_pipe.enable_xformers_memory_efficient_attention()

											__lowerCAmelCase							  =			sd_pipe(**_snake_case     ).images
											__lowerCAmelCase							  =			original_images[0, -3:, -3:, -1]

											with tempfile.TemporaryDirectory() as tmpdirname:
																LoraLoaderMixin.save_lora_weights(
																    save_directory=_snake_case						,     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(_snake_case						,     "pytorch_lora_weights.bin"     )     )     )
																sd_pipe.load_lora_weights(_snake_case     )

											__lowerCAmelCase							  =			sd_pipe(**_snake_case     ).images
											__lowerCAmelCase							  =			lora_images[0, -3:, -3:, -1]

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












from collections import namedtuple

UpperCamelCase											=      namedtuple('''from_to''', '''from_ to''')

UpperCamelCase											=      {
    '''millimeter''': '''mm''',
    '''centimeter''': '''cm''',
    '''meter''': '''m''',
    '''kilometer''': '''km''',
    '''inch''': '''in''',
    '''inche''': '''in''',  # Trailing 's' has been stripped off
    '''feet''': '''ft''',
    '''foot''': '''ft''',
    '''yard''': '''yd''',
    '''mile''': '''mi''',
}

UpperCamelCase											=      {
    '''mm''': from_to(0.001, 1_0_0_0),
    '''cm''': from_to(0.01, 1_0_0),
    '''m''': from_to(1, 1),
    '''km''': from_to(1_0_0_0, 0.001),
    '''in''': from_to(0.0254, 39.3701),
    '''ft''': from_to(0.3048, 3.2_8084),
    '''yd''': from_to(0.9144, 1.0_9361),
    '''mi''': from_to(1609.34, 0.0_0062_1371),
}




def       _a   (							lowerCamelCase__ ,			lowerCamelCase__ ,			lowerCamelCase__				)  ->      float:
 lowerCamelCase_							:    List[Any]							=     from_type.lower().rstrip('s'				)
 lowerCamelCase_							:    Optional[Any]							=     TYPE_CONVERSION.get(lowerCamelCase__ ,			lowerCamelCase__				)
 lowerCamelCase_							:    List[Any]							=     to_type.lower().rstrip('s'				)
 lowerCamelCase_							:    Optional[int]							=     TYPE_CONVERSION.get(lowerCamelCase__ ,			lowerCamelCase__				)
 if new_from not in METRIC_CONVERSION:
  lowerCamelCase_							:    List[Any]							=     (
      F'Invalid \'from_type\' value: {from_type!r}.\n'
      F'Conversion abbreviations are: {", ".join(lowerCamelCase__				)}'
  )
  raise ValueError(lowerCamelCase__				)
 if new_to not in METRIC_CONVERSION:
  lowerCamelCase_							:    List[str]							=     (
      F'Invalid \'to_type\' value: {to_type!r}.\n'
      F'Conversion abbreviations are: {", ".join(lowerCamelCase__				)}'
  )
  raise ValueError(lowerCamelCase__				)
 return value * METRIC_CONVERSION[new_from].from_ * METRIC_CONVERSION[new_to].to


if __name__ == "__main__":
       import doctest

       doctest.testmod()
