

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_(    snake_case__:						nn.Module     ) ->				str:
 UpperCAmelCase__           =       {}
 for name in unet.attn_processors.keys():
  UpperCAmelCase__           =       None if name.endswith('attn1.processor'     ) else unet.config.cross_attention_dim
  if name.startswith('mid_block'     ):
   UpperCAmelCase__           =       unet.config.block_out_channels[-1]
  elif name.startswith('up_blocks'     ):
   UpperCAmelCase__           =       int(name[len('up_blocks.'     )]     )
   UpperCAmelCase__           =       list(reversed(unet.config.block_out_channels     )     )[block_id]
  elif name.startswith('down_blocks'     ):
   UpperCAmelCase__           =       int(name[len('down_blocks.'     )]     )
   UpperCAmelCase__           =       unet.config.block_out_channels[block_id]
  UpperCAmelCase__           =       (
      LoRAAttnProcessor2_0 if hasattr(snake_case__						,					'scaled_dot_product_attention'     ) else LoRAAttnProcessor
  )
  UpperCAmelCase__           =       lora_attn_processor_class(
      hidden_size=snake_case__						,					cross_attention_dim=snake_case__     )
 UpperCAmelCase__           =       AttnProcsLayers(snake_case__     )
 return lora_attn_procs, unet_lora_layers
def 		UpperCamelCase_(    snake_case__:						nn.Module     ) ->				Any:
 UpperCAmelCase__           =       {}
 UpperCAmelCase__           =       (
     LoRAAttnProcessor2_0 if hasattr(snake_case__						,					'scaled_dot_product_attention'     ) else LoRAAttnProcessor
 )
 for name, module in text_encoder.named_modules():
  if name.endswith(snake_case__     ):
   UpperCAmelCase__           =       lora_attn_processor_class(
       hidden_size=module.out_proj.out_features						,					cross_attention_dim=snake_case__     )
 return text_lora_attn_procs
def 		UpperCamelCase_(    snake_case__:						nn.Module     ) ->				Optional[int]:
 UpperCAmelCase__           =       create_text_encoder_lora_attn_procs(snake_case__     )
 UpperCAmelCase__           =       AttnProcsLayers(snake_case__     )
 return text_encoder_lora_layers
def 		UpperCamelCase_(    snake_case__:						str						,					snake_case__:						Tuple=False     ) ->				Any:
 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'     ):
    UpperCAmelCase__           =       (
        torch.randn_like(layer_module.up.weight     )
        if randn_weight
        else torch.zeros_like(layer_module.up.weight     )
    )
    UpperCAmelCase__           =       torch.nn.Parameter(snake_case__     )







class 					lowercase		( unittest.TestCase      ):



 '''simple docstring'''


 def 						UpperCamelCase__							(self				)				->      Any:






  """simple docstring"""





  torch.manual_seed(0				)
  UpperCAmelCase__           =       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 , )
  UpperCAmelCase__           =       DDIMScheduler(
      beta_start=0.0_00_85 , beta_end=0.0_12 , beta_schedule='scaled_linear' , clip_sample=__a , set_alpha_to_one=__a , steps_offset=1 , )
  torch.manual_seed(0				)
  UpperCAmelCase__           =       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 , )
  UpperCAmelCase__           =       CLIPTextConfig(
      bos_token_id=0 , eos_token_id=2 , hidden_size=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 , )
  UpperCAmelCase__           =       CLIPTextModel(__a				)
  UpperCAmelCase__           =       CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'				)

  UpperCAmelCase__       ,					UpperCAmelCase__           =       create_unet_lora_layers(__a				)
  UpperCAmelCase__           =       create_text_encoder_lora_layers(__a				)

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






  """simple docstring"""





  UpperCAmelCase__           =       1
  UpperCAmelCase__           =       10
  UpperCAmelCase__           =       4
  UpperCAmelCase__           =       (32, 32)

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

  UpperCAmelCase__           =       {
      '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 						UpperCamelCase__							(self				)				->      Optional[int]:






  """simple docstring"""





  UpperCAmelCase__           =       77

  UpperCAmelCase__           =       torch.randint(2 , 56 , size=(1, max_seq_length) , generator=torch.manual_seed(0				)				)

  UpperCAmelCase__           =       {}
  UpperCAmelCase__           =       inputs
  return prepared_inputs




 def 						UpperCamelCase__							(self , __a				)				->      Optional[Any]:






  """simple docstring"""





  UpperCAmelCase__       ,					UpperCAmelCase__           =       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 						UpperCamelCase__							(self				)				->      Any:






  """simple docstring"""





  UpperCAmelCase__       ,					UpperCAmelCase__           =       self.get_dummy_components()
  UpperCAmelCase__           =       StableDiffusionPipeline(**__a				)
  UpperCAmelCase__           =       sd_pipe.to(__a				)
  sd_pipe.set_progress_bar_config(disable=__a				)

  UpperCAmelCase__       ,					UpperCAmelCase__       ,					UpperCAmelCase__           =       self.get_dummy_inputs()

  UpperCAmelCase__           =       sd_pipe(**__a				).images
  UpperCAmelCase__           =       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				)

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

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




 def 						UpperCamelCase__							(self				)				->      List[Any]:






  """simple docstring"""





  UpperCAmelCase__       ,					UpperCAmelCase__           =       self.get_dummy_components()
  UpperCAmelCase__           =       StableDiffusionPipeline(**__a				)
  UpperCAmelCase__           =       sd_pipe.to(__a				)
  sd_pipe.set_progress_bar_config(disable=__a				)

  UpperCAmelCase__       ,					UpperCAmelCase__       ,					UpperCAmelCase__           =       self.get_dummy_inputs()

  UpperCAmelCase__           =       sd_pipe(**__a				).images
  UpperCAmelCase__           =       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				)

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

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




 def 						UpperCamelCase__							(self				)				->      str:






  """simple docstring"""





  UpperCAmelCase__       ,					UpperCAmelCase__           =       self.get_dummy_components()
  UpperCAmelCase__           =       lora_components['unet_lora_attn_procs']
  UpperCAmelCase__           =       StableDiffusionPipeline(**__a				)
  UpperCAmelCase__           =       sd_pipe.to(__a				)
  sd_pipe.set_progress_bar_config(disable=__a				)

  UpperCAmelCase__       ,					UpperCAmelCase__       ,					UpperCAmelCase__           =       self.get_dummy_inputs()

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

  with tempfile.TemporaryDirectory() as tmpdirname:
   UpperCAmelCase__           =       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				)

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

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




 def 						UpperCamelCase__							(self				)				->      List[str]:






  """simple docstring"""





  UpperCAmelCase__       ,					UpperCAmelCase__           =       self.get_dummy_components()
  UpperCAmelCase__           =       StableDiffusionPipeline(**__a				)

  UpperCAmelCase__           =       self.get_dummy_tokens()

  # inference without lora
  UpperCAmelCase__           =       pipe.text_encoder(**__a				)[0]
  assert outputs_without_lora.shape == (1, 77, 32)

  # create lora_attn_procs with zeroed out up.weights
  UpperCAmelCase__           =       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
  UpperCAmelCase__           =       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
  UpperCAmelCase__           =       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
  UpperCAmelCase__           =       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 						UpperCamelCase__							(self				)				->      Union[str, Any]:






  """simple docstring"""





  UpperCAmelCase__       ,					UpperCAmelCase__           =       self.get_dummy_components()
  UpperCAmelCase__           =       StableDiffusionPipeline(**__a				)

  UpperCAmelCase__           =       self.get_dummy_tokens()

  # inference without lora
  UpperCAmelCase__           =       pipe.text_encoder(**__a				)[0]
  assert outputs_without_lora.shape == (1, 77, 32)

  # create lora_attn_procs with randn up.weights
  UpperCAmelCase__           =       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
  UpperCAmelCase__           =       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
  UpperCAmelCase__           =       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 						UpperCamelCase__							(self				)				->      Dict:






  """simple docstring"""





  UpperCAmelCase__       ,					UpperCAmelCase__           =       self.get_dummy_components()
  UpperCAmelCase__           =       StableDiffusionPipeline(**__a				)
  UpperCAmelCase__           =       sd_pipe.to(__a				)
  sd_pipe.set_progress_bar_config(disable=__a				)

  UpperCAmelCase__       ,					UpperCAmelCase__       ,					UpperCAmelCase__           =       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				)

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

  UpperCAmelCase__           =       sd_pipe(**__a , cross_attention_kwargs={'scale': 0.5}				).images
  UpperCAmelCase__           =       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 						UpperCamelCase__							(self				)				->      int:






  """simple docstring"""





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

   UpperCAmelCase__       ,					UpperCAmelCase__           =       self.get_dummy_components()
   UpperCAmelCase__           =       StableDiffusionPipeline(**__a				)
   UpperCAmelCase__           =       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				):
     UpperCAmelCase__           =       (
         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 						UpperCamelCase__							(self				)				->      str:






  """simple docstring"""





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

   UpperCAmelCase__       ,					UpperCAmelCase__           =       self.get_dummy_components()
   UpperCAmelCase__           =       StableDiffusionPipeline(**__a				)
   UpperCAmelCase__           =       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 						UpperCamelCase__							(self				)				->      int:






  """simple docstring"""





  UpperCAmelCase__       ,					UpperCAmelCase__           =       self.get_dummy_components()
  UpperCAmelCase__           =       StableDiffusionPipeline(**__a				)
  UpperCAmelCase__           =       sd_pipe.to(__a				)
  sd_pipe.set_progress_bar_config(disable=__a				)

  UpperCAmelCase__       ,					UpperCAmelCase__       ,					UpperCAmelCase__           =       self.get_dummy_inputs()

  # enable XFormers
  sd_pipe.enable_xformers_memory_efficient_attention()

  UpperCAmelCase__           =       sd_pipe(**__a				).images
  UpperCAmelCase__           =       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				)

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

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






from collections import OrderedDict
from dataclasses import fields
from typing import Any, Tuple

import numpy as np

from .import_utils import is_torch_available
def 		UpperCamelCase_(    snake_case__:						Optional[int]     ) ->				Tuple:
 if is_torch_available():
  import torch

  if isinstance(snake_case__						,					torch.Tensor     ):
   return True

 return isinstance(snake_case__						,					np.ndarray     )







class 					lowercase		( _UpperCamelCase      ):



 '''simple docstring'''


 def 						UpperCamelCase__							(self				)				->      int:






  """simple docstring"""





  UpperCAmelCase__           =       fields(self				)

  # Safety and consistency checks
  if not len(__a				):
   raise ValueError(F"{self.__class__.__name__} has no fields."				)

  UpperCAmelCase__           =       getattr(self , class_fields[0].name				)
  UpperCAmelCase__           =       all(getattr(self , field.name				) is None for field in class_fields[1:]				)

  if other_fields_are_none and isinstance(__a , __a				):
   for key, value in first_field.items():
    UpperCAmelCase__           =       value
  else:
   for field in class_fields:
    UpperCAmelCase__           =       getattr(self , field.name				)
    if v is not None:
     UpperCAmelCase__           =       v




 def __delitem__(self , *__a , **__a				)				->      Optional[int]:






  """simple docstring"""





  raise Exception(F"You cannot use ``__delitem__`` on a {self.__class__.__name__} instance."				)




 def 						UpperCamelCase__							(self , *__a , **__a				)				->      str:






  """simple docstring"""





  raise Exception(F"You cannot use ``setdefault`` on a {self.__class__.__name__} instance."				)




 def 						UpperCamelCase__							(self , *__a , **__a				)				->      Optional[int]:






  """simple docstring"""





  raise Exception(F"You cannot use ``pop`` on a {self.__class__.__name__} instance."				)




 def 						UpperCamelCase__							(self , *__a , **__a				)				->      Tuple:






  """simple docstring"""





  raise Exception(F"You cannot use ``update`` on a {self.__class__.__name__} instance."				)




 def __getitem__(self , __a				)				->      Tuple:






  """simple docstring"""





  if isinstance(__a , __a				):
   UpperCAmelCase__           =       dict(self.items()				)
   return inner_dict[k]
  else:
   return self.to_tuple()[k]




 def __setattr__(self , __a , __a				)				->      Any:






  """simple docstring"""





  if name in self.keys() and value is not None:
   # Don't call self.__setitem__ to avoid recursion errors
   super().__setitem__(__a , __a				)
  super().__setattr__(__a , __a				)




 def __setitem__(self , __a , __a				)				->      Union[str, Any]:






  """simple docstring"""





  super().__setitem__(__a , __a				)
  # Don't call self.__setattr__ to avoid recursion errors
  super().__setattr__(__a , __a				)




 def 						UpperCamelCase__							(self				)				->      Tuple[Any]:






  """simple docstring"""





  return tuple(self[k] for k in self.keys()				)



