'''simple docstring'''

import json
import os
from functools import lru_cache
from typing import TYPE_CHECKING, List, Optional, Tuple, Union

import numpy as np
import regex as re

from ...utils import is_tf_available, is_torch_available, logging


if TYPE_CHECKING:
						if is_torch_available():
												import torch
						if is_tf_available():
												import tensorflow as tf

from ...tokenization_utils import AddedToken, PreTrainedTokenizer


A_	   = logging.get_logger(__name__)

A_	   = {
    'vocab_file': 'vocab.json',
    'merges_file': 'merges.txt',
}

A_	   = {
    'vocab_file': {
        'Salesforce/codegen-350M-mono': 'https://huggingface.co/Salesforce/codegen-350M-mono/resolve/main/vocab.json',
    },
    'merges_file': {
        'Salesforce/codegen-350M-mono': 'https://huggingface.co/Salesforce/codegen-350M-mono/resolve/main/merges.txt',
    },
}

A_	   = {
    'Salesforce/codegen-350M-mono': 2048,
}


@lru_cache()
def        UpperCAmelCase__   (				)    ->		Union[str, Any]:



						"""simple docstring"""

						snake_case :       Dict				=    (
						    list(range(ord("!"),      ord("~") + 1)) + list(range(ord("¡"),      ord("¬") + 1)) + list(range(ord("®"),      ord("ÿ") + 1))
						)
						snake_case :       Any				=    bs[:]
						snake_case :       Union[str, Any]				=    0
						for b in range(2**8):
												if b not in bs:
																		bs.append(_UpperCAmelCase)
																		cs.append(2**8 + n)
																		n += 1
						snake_case :       str				=    [chr(_UpperCAmelCase) for n in cs]
						return dict(zip(_UpperCAmelCase,      _UpperCAmelCase))


def        UpperCAmelCase__   (				_lowerCAmelCase    :			str)    ->		Any:



						"""simple docstring"""

						snake_case :       Union[str, Any]				=    set()
						snake_case :       List[str]				=    word[0]
						for char in word[1:]:
												pairs.add((prev_char, char))
												snake_case :       Dict				=    char
						return pairs




class 		SCREAMING_SNAKE_CASE_					(__UpperCAmelCase      ):



			"""simple docstring"""






			__magic_name__ :				List[Any]   							=       VOCAB_FILES_NAMES
			__magic_name__ :				List[str]   							=       PRETRAINED_VOCAB_FILES_MAP
			__magic_name__ :				Tuple   							=       PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
			__magic_name__ :				str   							=       ["""input_ids""", """attention_mask"""]



			def __init__(							self				,			snake_case_				,			snake_case_				,			snake_case_="replace"				,			snake_case_="<|endoftext|>"				,			snake_case_="<|endoftext|>"				,			snake_case_="<|endoftext|>"				,			snake_case_=None				,			snake_case_=False				,			snake_case_=False				,			**snake_case_				,			):
									snake_case :       str				=    AddedToken(_lowerCamelCase				,			lstrip=_lowerCamelCase				,			rstrip=_lowerCamelCase					) if isinstance(_lowerCamelCase				,			_lowerCamelCase					) else bos_token
									snake_case :       Union[str, Any]				=    AddedToken(_lowerCamelCase				,			lstrip=_lowerCamelCase				,			rstrip=_lowerCamelCase					) if isinstance(_lowerCamelCase				,			_lowerCamelCase					) else eos_token
									snake_case :       int				=    AddedToken(_lowerCamelCase				,			lstrip=_lowerCamelCase				,			rstrip=_lowerCamelCase					) if isinstance(_lowerCamelCase				,			_lowerCamelCase					) else unk_token
									snake_case :       str				=    AddedToken(_lowerCamelCase				,			lstrip=_lowerCamelCase				,			rstrip=_lowerCamelCase					) if isinstance(_lowerCamelCase				,			_lowerCamelCase					) else pad_token
									super().__init__(
									    errors=_lowerCamelCase				,			unk_token=_lowerCamelCase				,			bos_token=_lowerCamelCase				,			eos_token=_lowerCamelCase				,			pad_token=_lowerCamelCase				,			add_prefix_space=_lowerCamelCase				,			add_bos_token=_lowerCamelCase				,			**_lowerCamelCase				,			)
									snake_case :       List[str]				=    add_bos_token

									with open(_lowerCamelCase				,			encoding="utf-8"					) as vocab_handle:
															snake_case :       List[Any]				=    json.load(_lowerCamelCase					)
									snake_case :       Dict				=    {v: k for k, v in self.encoder.items()}
									snake_case :       str				=    errors  # how to handle errors in decoding
									snake_case :       str				=    bytes_to_unicode()
									snake_case :       List[str]				=    {v: k for k, v in self.byte_encoder.items()}
									with open(_lowerCamelCase				,			encoding="utf-8"					) as merges_handle:
															snake_case :       Optional[int]				=    merges_handle.read().split("\n"					)[1:-1]
									snake_case :       List[Any]				=    [tuple(merge.split()					) for merge in bpe_merges]
									snake_case :       List[Any]				=    dict(zip(_lowerCamelCase				,			range(len(_lowerCamelCase					)					)					)					)
									snake_case :       List[str]				=    {}
									snake_case :       Dict				=    add_prefix_space

									# Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
									snake_case :       Optional[Any]				=    re.compile(r"\'s|\'t|\'re|\'ve|\'m|\'ll|\'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+"					)



			@property
			def   lowerCamelCase (							self					):
									return len(self.encoder					)



			def   lowerCamelCase (							self					):
									return dict(self.encoder				,			**self.added_tokens_encoder					)



			def   lowerCamelCase (							self				,			snake_case_					):

									if token in self.cache:
															return self.cache[token]
									snake_case :       int				=    tuple(_lowerCamelCase					)
									snake_case :       List[Any]				=    get_pairs(_lowerCamelCase					)

									if not pairs:
															return token

									while True:
															snake_case :       Any				=    min(_lowerCamelCase				,			key=lambda snake_case_					: self.bpe_ranks.get(_lowerCamelCase				,			float("inf"					)					)					)
															if bigram not in self.bpe_ranks:
																					break
															snake_case :       str				=    bigram
															snake_case :       Union[str, Any]				=    []
															snake_case :       Any				=    0
															while i < len(_lowerCamelCase					):
																					try:
																											snake_case :       str				=    word.index(_lowerCamelCase				,			_lowerCamelCase					)
																					except ValueError:
																											new_word.extend(word[i:]					)
																											break
																					else:
																											new_word.extend(word[i:j]					)
																											snake_case :       Optional[int]				=    j

																					if word[i] == first and i < len(_lowerCamelCase					) - 1 and word[i + 1] == second:
																											new_word.append(first + second					)
																											i += 2
																					else:
																											new_word.append(word[i]					)
																											i += 1
															snake_case :       List[str]				=    tuple(_lowerCamelCase					)
															snake_case :       Any				=    new_word
															if len(_lowerCamelCase					) == 1:
																					break
															else:
																					snake_case :       int				=    get_pairs(_lowerCamelCase					)
									snake_case :       int				=    ''' '''.join(_lowerCamelCase					)
									snake_case :       List[Any]				=    word
									return word



			def   lowerCamelCase (							self				,			snake_case_				,			snake_case_=None					):

									if self.add_bos_token:
															snake_case :       str				=    [self.bos_token_id]
									else:
															snake_case :       Optional[Any]				=    []

									snake_case :       List[Any]				=    bos_token_ids + token_ids_a

									if token_ids_a is None:
															return output

									return output + bos_token_ids + token_ids_a



			def   lowerCamelCase (							self				,			snake_case_					):
									snake_case :       Optional[int]				=    []
									for token in re.findall(self.pat				,			_lowerCamelCase					):
															snake_case :       Union[str, Any]				=    ''''''.join(
															    self.byte_encoder[b] for b in token.encode("utf-8"					)					)  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
															bpe_tokens.extend(bpe_token for bpe_token in self.bpe(_lowerCamelCase					).split(" "					)					)
									return bpe_tokens



			def   lowerCamelCase (							self				,			snake_case_					):
									return self.encoder.get(_lowerCamelCase				,			self.encoder.get(self.unk_token					)					)



			def   lowerCamelCase (							self				,			snake_case_					):
									return self.decoder.get(_lowerCamelCase					)



			def   lowerCamelCase (							self				,			snake_case_					):
									snake_case :       List[str]				=    ''''''.join(_lowerCamelCase					)
									snake_case :       List[str]				=    bytearray([self.byte_decoder[c] for c in text]					).decode("utf-8"				,			errors=self.errors					)
									return text



			def   lowerCamelCase (							self				,			snake_case_				,			snake_case_ = None					):

									if not os.path.isdir(_lowerCamelCase					):
															logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""					)
															return
									snake_case :       Any				=    os.path.join(
									    _lowerCamelCase				,			(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]					)
									snake_case :       Tuple				=    os.path.join(
									    _lowerCamelCase				,			(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["merges_file"]					)

									with open(_lowerCamelCase				,			"w"				,			encoding="utf-8"					) as f:
															f.write(json.dumps(self.encoder				,			indent=2				,			sort_keys=_lowerCamelCase				,			ensure_ascii=_lowerCamelCase					) + "\n"					)

									snake_case :       Union[str, Any]				=    0
									with open(_lowerCamelCase				,			"w"				,			encoding="utf-8"					) as writer:
															writer.write("#version: 0.2\n"					)
															for bpe_tokens, token_index in sorted(self.bpe_ranks.items()				,			key=lambda snake_case_					: kv[1]					):
																					if index != token_index:
																											logger.warning(
																											    F"""Saving vocabulary to {merge_file}: BPE merge indices are not consecutive."""
																											    " Please check that the tokenizer is not corrupted!"					)
																											snake_case :       Dict				=    token_index
																					writer.write(" ".join(_lowerCamelCase					) + "\n"					)
																					index += 1

									return vocab_file, merge_file



			def   lowerCamelCase (							self				,			snake_case_				,			snake_case_=False				,			**snake_case_					):
									snake_case :       str				=    kwargs.pop("add_prefix_space"				,			self.add_prefix_space					)
									if is_split_into_words or add_prefix_space:
															snake_case :       Optional[int]				=    ''' ''' + text
									return (text, kwargs)



			def   lowerCamelCase (							self				,			snake_case_				,			snake_case_ = False				,			snake_case_ = None				,			snake_case_ = None				,			**snake_case_				,			):
									snake_case :       List[str]				=    super()._decode(
									    token_ids=_lowerCamelCase				,			skip_special_tokens=_lowerCamelCase				,			clean_up_tokenization_spaces=_lowerCamelCase				,			**_lowerCamelCase				,			)

									if truncate_before_pattern is not None and len(_lowerCamelCase					) > 0:
															snake_case :       Optional[Any]				=    self.truncate(_lowerCamelCase				,			_lowerCamelCase					)

									return decoded_text






			def   lowerCamelCase (							self				,			snake_case_				,			snake_case_					):

									def find_re(snake_case_				,			snake_case_				,			snake_case_					):
															snake_case :       Tuple				=    pattern.search(_lowerCamelCase				,			_lowerCamelCase					)
															return m.start() if m else -1

									snake_case :       Union[str, Any]				=    [re.compile(_lowerCamelCase				,			re.MULTILINE					) for pattern in truncate_before_pattern]

									snake_case :       List[str]				=    list(re.finditer("^print"				,			_lowerCamelCase				,			re.MULTILINE					)					)

									if len(_lowerCamelCase					) > 1:
															snake_case :       Dict				=    completion[: prints[1].start()]

									snake_case :       Any				=    list(re.finditer("^def"				,			_lowerCamelCase				,			re.MULTILINE					)					)

									if len(_lowerCamelCase					) > 1:
															snake_case :       Optional[int]				=    completion[: defs[1].start()]

									snake_case :       int				=    0

									snake_case :       Dict				=    [
									    pos for pos in [find_re(_lowerCamelCase				,			_lowerCamelCase				,			_lowerCamelCase					) for terminal in terminals] if pos != -1
									]

									if len(_lowerCamelCase					) > 0:
															return completion[: min(_lowerCamelCase					)]
									else:
															return completion











"""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 		lowercase_   (		_UpperCAmelCase						):




	"""simple docstring"""






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

def 		lowercase_   (		_UpperCAmelCase						):




	"""simple docstring"""






	A_	:					Union[str, Any]						 =		{}
	A_	:					List[str]						 =		(
	    LoRAAttnProcessor2_0 if hasattr(_UpperCAmelCase	,   '''scaled_dot_product_attention'''						) else LoRAAttnProcessor
	)
	for name, module in text_encoder.named_modules():
		if name.endswith(_UpperCAmelCase						):
			A_	:					Optional[int]						 =		lora_attn_processor_class(
			    hidden_size=module.out_proj.out_features	,   cross_attention_dim=_UpperCAmelCase						)
	return text_lora_attn_procs

def 		lowercase_   (		_UpperCAmelCase						):




	"""simple docstring"""






	A_	:					Optional[Any]						 =		create_text_encoder_lora_attn_procs(_UpperCAmelCase						)
	A_	:					List[str]						 =		AttnProcsLayers(_UpperCAmelCase						)
	return text_encoder_lora_layers

def 		lowercase_   (		_UpperCAmelCase	,   _UpperCAmelCase=False						):




	"""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'''						):
				A_	:					List[Any]						 =		(
				    torch.randn_like(layer_module.up.weight						)
				    if randn_weight
				    else torch.zeros_like(layer_module.up.weight						)
				)
				A_	:					Tuple						 =		torch.nn.Parameter(_UpperCAmelCase						)

class 							lowercase   (     unittest.TestCase):







				def        a_	(				self :       List[Any]							):



					"""simple docstring"""

					torch.manual_seed(0							)
					A_	:					List[str]						 =		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   ,  )
					A_	:					Optional[int]						 =		DDIMScheduler(
					    beta_start=0.00085   ,  beta_end=0.012   ,  beta_schedule='''scaled_linear'''   ,  clip_sample=_lowerCamelCase   ,  set_alpha_to_one=_lowerCamelCase   ,  steps_offset=1   ,  )
					torch.manual_seed(0							)
					A_	:					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   ,  )
					A_	:					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   ,  )
					A_	:					List[Any]						 =		CLIPTextModel(_lowerCamelCase							)
					A_	:					str						 =		CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''							)

					A_ ,			A_	:					Any						 =		create_unet_lora_layers(_lowerCamelCase							)
					A_	:					Tuple						 =		create_text_encoder_lora_layers(_lowerCamelCase							)

					A_	:					Any						 =		{
					    '''unet''': unet,
					    '''scheduler''': scheduler,
					    '''vae''': vae,
					    '''text_encoder''': text_encoder,
					    '''tokenizer''': tokenizer,
					    '''safety_checker''': None,
					    '''feature_extractor''': None,
					}
					A_	:					Union[str, 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        a_	(				self :       str							):



					"""simple docstring"""

					A_	:					Optional[int]						 =		1
					A_	:					Tuple						 =		10
					A_	:					int						 =		4
					A_	:					List[Any]						 =		(32, 32)

					A_	:					str						 =		torch.manual_seed(0							)
					A_	:					Dict						 =		floats_tensor((batch_size, num_channels) + sizes							)
					A_	:					List[str]						 =		torch.randint(1   ,  _lowerCamelCase   ,  size=(batch_size, sequence_length)   ,  generator=_lowerCamelCase							)

					A_	:					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        a_	(				self :       str							):



					"""simple docstring"""

					A_	:					Union[str, Any]						 =		77

					A_	:					str						 =		torch.randint(2   ,  56   ,  size=(1, max_seq_length)   ,  generator=torch.manual_seed(0							)							)

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







				def        a_	(				self :       Optional[int]   ,  _lowerCamelCase :       List[Any]							):



					"""simple docstring"""

					A_ ,			A_	:					List[str]						 =		self.get_dummy_components()
					LoraLoaderMixin.save_lora_weights(
					    save_directory=_lowerCamelCase   ,  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(_lowerCamelCase   ,  '''pytorch_lora_weights.bin'''							)							)							)







				def        a_	(				self :       Any							):



					"""simple docstring"""

					A_ ,			A_	:					Optional[Any]						 =		self.get_dummy_components()
					A_	:					List[str]						 =		StableDiffusionPipeline(**_lowerCamelCase							)
					A_	:					List[str]						 =		sd_pipe.to(_lowerCamelCase							)
					sd_pipe.set_progress_bar_config(disable=_lowerCamelCase							)

					A_ ,			A_ ,			A_	:					str						 =		self.get_dummy_inputs()

					A_	:					Optional[int]						 =		sd_pipe(**_lowerCamelCase							).images
					A_	:					str						 =		original_images[0, -3:, -3:, -1]

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

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

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







				def        a_	(				self :       Optional[int]							):



					"""simple docstring"""

					A_ ,			A_	:					List[Any]						 =		self.get_dummy_components()
					A_	:					str						 =		StableDiffusionPipeline(**_lowerCamelCase							)
					A_	:					Any						 =		sd_pipe.to(_lowerCamelCase							)
					sd_pipe.set_progress_bar_config(disable=_lowerCamelCase							)

					A_ ,			A_ ,			A_	:					List[str]						 =		self.get_dummy_inputs()

					A_	:					Union[str, Any]						 =		sd_pipe(**_lowerCamelCase							).images
					A_	:					int						 =		original_images[0, -3:, -3:, -1]

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

					A_	:					Union[str, Any]						 =		sd_pipe(**_lowerCamelCase							).images
					A_	:					Dict						 =		lora_images[0, -3:, -3:, -1]

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







				def        a_	(				self :       Optional[int]							):



					"""simple docstring"""

					A_ ,			A_	:					List[Any]						 =		self.get_dummy_components()
					A_	:					Any						 =		lora_components['''unet_lora_attn_procs''']
					A_	:					Any						 =		StableDiffusionPipeline(**_lowerCamelCase							)
					A_	:					Tuple						 =		sd_pipe.to(_lowerCamelCase							)
					sd_pipe.set_progress_bar_config(disable=_lowerCamelCase							)

					A_ ,			A_ ,			A_	:					Tuple						 =		self.get_dummy_inputs()

					A_	:					Tuple						 =		sd_pipe(**_lowerCamelCase							).images
					A_	:					Optional[Any]						 =		original_images[0, -3:, -3:, -1]

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

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

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







				def        a_	(				self :       Tuple							):



					"""simple docstring"""

					A_ ,			A_	:					List[str]						 =		self.get_dummy_components()
					A_	:					Dict						 =		StableDiffusionPipeline(**_lowerCamelCase							)

					A_	:					Any						 =		self.get_dummy_tokens()

					# inference without lora
					A_	:					str						 =		pipe.text_encoder(**_lowerCamelCase							)[0]
					assert outputs_without_lora.shape == (1, 77, 32)

					# create lora_attn_procs with zeroed out up.weights
					A_	:					str						 =		create_text_encoder_lora_attn_procs(pipe.text_encoder							)
					set_lora_up_weights(_lowerCamelCase   ,  randn_weight=_lowerCamelCase							)

					# monkey patch
					pipe._modify_text_encoder(_lowerCamelCase							)

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

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

					assert torch.allclose(
					    _lowerCamelCase   ,  _lowerCamelCase							), "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
					A_	:					Tuple						 =		create_text_encoder_lora_attn_procs(pipe.text_encoder							)
					set_lora_up_weights(_lowerCamelCase   ,  randn_weight=_lowerCamelCase							)

					# monkey patch
					pipe._modify_text_encoder(_lowerCamelCase							)

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

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

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







				def        a_	(				self :       Optional[int]							):



					"""simple docstring"""

					A_ ,			A_	:					List[Any]						 =		self.get_dummy_components()
					A_	:					Optional[int]						 =		StableDiffusionPipeline(**_lowerCamelCase							)

					A_	:					List[Any]						 =		self.get_dummy_tokens()

					# inference without lora
					A_	:					Tuple						 =		pipe.text_encoder(**_lowerCamelCase							)[0]
					assert outputs_without_lora.shape == (1, 77, 32)

					# create lora_attn_procs with randn up.weights
					A_	:					List[Any]						 =		create_text_encoder_lora_attn_procs(pipe.text_encoder							)
					set_lora_up_weights(_lowerCamelCase   ,  randn_weight=_lowerCamelCase							)

					# monkey patch
					pipe._modify_text_encoder(_lowerCamelCase							)

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

					# inference with lora
					A_	:					List[Any]						 =		pipe.text_encoder(**_lowerCamelCase							)[0]
					assert outputs_with_lora.shape == (1, 77, 32)

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

					# remove monkey patch
					pipe._remove_text_encoder_monkey_patch()

					# inference with removed lora
					A_	:					Optional[int]						 =		pipe.text_encoder(**_lowerCamelCase							)[0]
					assert outputs_without_lora_removed.shape == (1, 77, 32)

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







				def        a_	(				self :       Tuple							):



					"""simple docstring"""

					A_ ,			A_	:					Dict						 =		self.get_dummy_components()
					A_	:					str						 =		StableDiffusionPipeline(**_lowerCamelCase							)
					A_	:					Dict						 =		sd_pipe.to(_lowerCamelCase							)
					sd_pipe.set_progress_bar_config(disable=_lowerCamelCase							)

					A_ ,			A_ ,			A_	:					Tuple						 =		self.get_dummy_inputs()

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

					A_	:					Dict						 =		sd_pipe(**_lowerCamelCase							).images
					A_	:					Union[str, Any]						 =		lora_images[0, -3:, -3:, -1]

					A_	:					int						 =		sd_pipe(**_lowerCamelCase   ,  cross_attention_kwargs={'''scale''': 0.5}							).images
					A_	:					int						 =		lora_images_with_scale[0, -3:, -3:, -1]

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







				def        a_	(				self :       Tuple							):



					"""simple docstring"""

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

						A_ ,			A_	:					str						 =		self.get_dummy_components()
						A_	:					Optional[Any]						 =		StableDiffusionPipeline(**_lowerCamelCase							)
						A_	:					List[Any]						 =		sd_pipe.to(_lowerCamelCase							)
						sd_pipe.set_progress_bar_config(disable=_lowerCamelCase							)

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

            # load LoRA weight file
						sd_pipe.load_lora_weights(_lowerCamelCase							)

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







				@unittest.skipIf(torch_device != '''cuda'''   ,  '''This test is supposed to run on GPU'''							)
				def        a_	(				self :       Optional[Any]							):



					"""simple docstring"""

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

						A_ ,			A_	:					Tuple						 =		self.get_dummy_components()
						A_	:					Optional[int]						 =		StableDiffusionPipeline(**_lowerCamelCase							)
						A_	:					Any						 =		sd_pipe.to(_lowerCamelCase							)
						sd_pipe.set_progress_bar_config(disable=_lowerCamelCase							)

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

            # load LoRA weight file
						sd_pipe.load_lora_weights(_lowerCamelCase							)

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







				@unittest.skipIf(torch_device != '''cuda'''   ,  '''This test is supposed to run on GPU'''							)
				def        a_	(				self :       Optional[Any]							):



					"""simple docstring"""

					A_ ,			A_	:					Any						 =		self.get_dummy_components()
					A_	:					Union[str, Any]						 =		StableDiffusionPipeline(**_lowerCamelCase							)
					A_	:					Optional[int]						 =		sd_pipe.to(_lowerCamelCase							)
					sd_pipe.set_progress_bar_config(disable=_lowerCamelCase							)

					A_ ,			A_ ,			A_	:					int						 =		self.get_dummy_inputs()

					# enable XFormers
					sd_pipe.enable_xformers_memory_efficient_attention()

					A_	:					str						 =		sd_pipe(**_lowerCamelCase							).images
					A_	:					Tuple						 =		original_images[0, -3:, -3:, -1]

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

					A_	:					List[Any]						 =		sd_pipe(**_lowerCamelCase							).images
					A_	:					int						 =		lora_images[0, -3:, -3:, -1]

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





