





# Copyright (c) 2021-, NVIDIA CORPORATION.  All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

####################################################################################################

#
# Note: If when running this conversion script you're getting an exception:
#     ModuleNotFoundError: No module named 'megatron.model.enums'
# you need to tell python where to find the clone of Megatron-LM, e.g.:
#
# cd /tmp
# git clone https://github.com/NVIDIA/Megatron-LM
# PYTHONPATH=/tmp/Megatron-LM python src/transformers/models/megatron_bert/convert_megatron_bert_checkpoint.py ...
#
# if you already have it cloned elsewhere, simply adjust the path to the existing path
#
# If the training was done using a Megatron-LM fork, e.g.,
# https://github.com/microsoft/Megatron-DeepSpeed/ then chances are that you need to have that one
# in your path, i.e., /path/to/Megatron-DeepSpeed/
#

import argparse
import os
import re
import zipfile

import torch

from transformers import MegatronBertConfig
def 			__UpperCamelCase				(			lowercase__	:       Tuple		,    lowercase__	:       List[Any]		,    lowercase__	:       Optional[int]=0		)	->      Union[str, Any]:
    '''simple docstring'''





    if name is None:
        lowerCAmelCase_					:	int						=   None
    else:
        lowerCAmelCase_					:	Optional[int]						=   """.""" * max(0		,    spaces - 2		) + """# {:""" + str(50 - spaces		) + """s}"""
        lowerCAmelCase_					:	List[Any]						=   fmt.format(lowercase__		)

    # Print and recurse (if needed).
    if isinstance(lowercase__		,    lowercase__		):
        if msg is not None:
            print(lowercase__		)
        for k in val.keys():
            recursive_print(lowercase__		,    val[k]		,    spaces + 2		)
    elif isinstance(lowercase__		,    torch.Tensor		):
        print(lowercase__		,    """:"""		,    val.size()		)
    else:
        print(lowercase__		,    """:"""		,    lowercase__		)
def 			__UpperCamelCase				(			lowercase__	:       List[Any]		,    lowercase__	:       str		,    lowercase__	:       int		,    lowercase__	:       int		,    lowercase__	:       Tuple		)	->      List[Any]:
    '''simple docstring'''





    lowerCAmelCase_					:	List[str]						=   param.size()
    if checkpoint_version == 1.0:
        # version 1.0 stores [num_heads * hidden_size * num_splits, :]
        lowerCAmelCase_					:	Any						=   (num_heads, hidden_size, num_splits) + input_shape[1:]
        lowerCAmelCase_					:	Union[str, Any]						=   param.view(*lowercase__		)
        lowerCAmelCase_					:	str						=   param.transpose(0		,    2		)
        lowerCAmelCase_					:	Union[str, Any]						=   param.transpose(1		,    2		).contiguous()
    elif checkpoint_version >= 2.0:
        # other versions store [num_heads * num_splits * hidden_size, :]
        lowerCAmelCase_					:	Optional[int]						=   (num_heads, num_splits, hidden_size) + input_shape[1:]
        lowerCAmelCase_					:	Optional[Any]						=   param.view(*lowercase__		)
        lowerCAmelCase_					:	List[str]						=   param.transpose(0		,    1		).contiguous()
    lowerCAmelCase_					:	Optional[int]						=   param.view(*lowercase__		)
    return param
def 			__UpperCamelCase				(			lowercase__	:       int		,    lowercase__	:       List[str]		,    lowercase__	:       Optional[Any]		)	->      Optional[Any]:
    '''simple docstring'''





    lowerCAmelCase_					:	List[Any]						=   {}

    # old versions did not store training args
    lowerCAmelCase_					:	List[Any]						=   input_state_dict.get("""args"""		,    lowercase__		)
    if ds_args is not None:
        # do not make the user write a config file when the exact dimensions/sizes are already in the checkpoint
        # from pprint import pprint
        # pprint(vars(ds_args))

        lowerCAmelCase_					:	Optional[Any]						=   ds_args.tokenizer_type
        lowerCAmelCase_					:	Any						=   ds_args.padded_vocab_size
        lowerCAmelCase_					:	Optional[Any]						=   ds_args.max_position_embeddings
        lowerCAmelCase_					:	Union[str, Any]						=   ds_args.hidden_size
        lowerCAmelCase_					:	List[Any]						=   ds_args.num_layers
        lowerCAmelCase_					:	Union[str, Any]						=   ds_args.num_attention_heads
        lowerCAmelCase_					:	Dict						=   ds_args.ffn_hidden_size if """ffn_hidden_size""" in ds_args else 4 * ds_args.hidden_size
        # pprint(config)

    # The number of heads.
    lowerCAmelCase_					:	List[str]						=   config.num_attention_heads
    # The hidden_size per head.
    lowerCAmelCase_					:	Union[str, Any]						=   config.hidden_size // heads
    # Megatron-LM checkpoint version
    if "checkpoint_version" in input_state_dict.keys():
        lowerCAmelCase_					:	int						=   input_state_dict["""checkpoint_version"""]
    else:
        lowerCAmelCase_					:	List[Any]						=   0.0

    # The model.
    lowerCAmelCase_					:	Any						=   input_state_dict["""model"""]
    # The language model.
    lowerCAmelCase_					:	Optional[Any]						=   model["""language_model"""]
    # The embeddings.
    lowerCAmelCase_					:	str						=   lm["""embedding"""]

    # The word embeddings.
    lowerCAmelCase_					:	List[str]						=   embeddings["""word_embeddings"""]["""weight"""]
    # Truncate the embedding table to vocab_size rows.
    lowerCAmelCase_					:	Optional[Any]						=   word_embeddings[: config.vocab_size, :]
    # Store the word embeddings.
    lowerCAmelCase_					:	Optional[Any]						=   word_embeddings

    # The position embeddings.
    lowerCAmelCase_					:	Dict						=   embeddings["""position_embeddings"""]["""weight"""]
    assert pos_embeddings.size(0		) == config.max_position_embeddings and pos_embeddings.size(1		) == config.hidden_size
    # Store the position embeddings.
    lowerCAmelCase_					:	Union[str, Any]						=   pos_embeddings

    # The token-type embeddings.
    lowerCAmelCase_					:	Union[str, Any]						=   embeddings["""tokentype_embeddings"""]["""weight"""]
    # Store the position embeddings.
    lowerCAmelCase_					:	List[str]						=   tokentype_embeddings

    # The transformer.
    lowerCAmelCase_					:	Tuple						=   lm["""transformer"""] if """transformer""" in lm.keys() else lm["""encoder"""]

    # The regex to extract layer names.
    lowerCAmelCase_					:	int						=   re.compile(R"""layers\.(\d+)\.([a-z0-9_.]+)\.([a-z]+)"""		)

    # The simple map of names for "automated" rules.
    lowerCAmelCase_					:	Dict						=   {
        """attention.dense""": """.attention.output.dense.""",
        """self_attention.dense""": """.attention.output.dense.""",
        """mlp.dense_h_to_4h""": """.intermediate.dense.""",
        """mlp.dense_4h_to_h""": """.output.dense.""",
    }

    # Keep track of the attention/query/value tensor.
    lowerCAmelCase_					:	Optional[int]						=   None

    # Extract the layers.
    for key, val in transformer.items():
        # Match the name.
        lowerCAmelCase_					:	List[Any]						=   layer_re.match(lowercase__		)

        # Stop if that's not a layer
        if m is None:
            break

        # The index of the layer.
        lowerCAmelCase_					:	List[str]						=   int(m.group(1		)		)
        # The name of the operation.
        lowerCAmelCase_					:	Union[str, Any]						=   m.group(2		)
        # Is it a weight or a bias?
        lowerCAmelCase_					:	Tuple						=   m.group(3		)

        # The name of the layer.
        lowerCAmelCase_					:	Tuple						=   f'bert.encoder.layer.{layer_idx}'

        # For layernorm(s), simply store the layer norm.
        if op_name.endswith("""layernorm"""		):
            lowerCAmelCase_					:	str						=   """attention.ln""" if op_name.startswith("""input"""		) else """ln"""
            lowerCAmelCase_					:	Tuple						=   val

        # Transpose the QKV matrix.
        elif (
            op_name == "attention.query_key_value" or op_name == "self_attention.query_key_value"
        ) and weight_or_bias == "weight":
            # Make sure the QKV pointer is nil.
            assert attention_qkv_weight is None, ""

            lowerCAmelCase_					:	Dict						=   fix_query_key_value_ordering(lowercase__		,    lowercase__		,    3		,    lowercase__		,    lowercase__		)
            # Store the tensor as we need the bias as well to interleave QKV and biases.
            lowerCAmelCase_					:	Dict						=   out_val

        # Transpose the bias.
        elif (
            op_name == "attention.query_key_value" or op_name == "self_attention.query_key_value"
        ) and weight_or_bias == "bias":
            # Make sure we read the weight tensor.
            assert attention_qkv_weight is not None, ""

            # Split the QKV matrix into Q, K and V. Megatron stores Q,K,V interleaved.
            lowerCAmelCase_					:	Tuple						=   attention_qkv_weight[0 * config.hidden_size : 1 * config.hidden_size, :]
            lowerCAmelCase_					:	str						=   attention_qkv_weight[1 * config.hidden_size : 2 * config.hidden_size, :]
            lowerCAmelCase_					:	int						=   attention_qkv_weight[2 * config.hidden_size : 3 * config.hidden_size, :]

            lowerCAmelCase_					:	Optional[int]						=   fix_query_key_value_ordering(lowercase__		,    lowercase__		,    3		,    lowercase__		,    lowercase__		)
            # Split the bias.
            lowerCAmelCase_					:	List[str]						=   out_val[0 * config.hidden_size : 1 * config.hidden_size]
            lowerCAmelCase_					:	Optional[Any]						=   out_val[1 * config.hidden_size : 2 * config.hidden_size]
            lowerCAmelCase_					:	List[Any]						=   out_val[2 * config.hidden_size : 3 * config.hidden_size]

            # Store.
            lowerCAmelCase_					:	Any						=   q
            lowerCAmelCase_					:	List[Any]						=   q_bias
            lowerCAmelCase_					:	Optional[Any]						=   k
            lowerCAmelCase_					:	str						=   k_bias
            lowerCAmelCase_					:	List[str]						=   v
            lowerCAmelCase_					:	List[str]						=   v_bias

            # Clear the stored tensor.
            lowerCAmelCase_					:	Optional[Any]						=   None

        # Copy weights and biases as is.
        elif weight_or_bias in ["weight", "bias"]:
            lowerCAmelCase_					:	Optional[int]						=   megatron_to_transformers[op_name]
            lowerCAmelCase_					:	Any						=   val

    # The final layernorm.
    lowerCAmelCase_					:	Any						=   transformer["""final_layernorm.weight"""]
    lowerCAmelCase_					:	str						=   transformer["""final_layernorm.bias"""]

    # The pooler.
    lowerCAmelCase_					:	Optional[int]						=   lm["""pooler"""]

    # Store the matrix and the bias.
    lowerCAmelCase_					:	str						=   pooler["""dense.weight"""]
    lowerCAmelCase_					:	str						=   pooler["""dense.bias"""]

    # The LM head from Megatron (for RACE).
    lowerCAmelCase_					:	List[Any]						=   model["""lm_head"""]

    # The transform matrix.
    lowerCAmelCase_					:	List[Any]						=   lm_head["""dense.weight"""]
    lowerCAmelCase_					:	List[str]						=   lm_head["""dense.bias"""]

    # The transform LN.
    lowerCAmelCase_					:	Dict						=   lm_head["""layernorm.weight"""]
    lowerCAmelCase_					:	Dict						=   lm_head["""layernorm.bias"""]

    # For the decoder, we replicate the weights.
    lowerCAmelCase_					:	Optional[int]						=   word_embeddings
    lowerCAmelCase_					:	Dict						=   lm_head["""bias"""]

    # The classifier from Megatron (for MLNI).
    lowerCAmelCase_					:	Tuple						=   model["""binary_head"""]

    # Store the classifier.
    lowerCAmelCase_					:	str						=   binary_head["""weight"""]
    lowerCAmelCase_					:	Optional[int]						=   binary_head["""bias"""]

    # It should be done!
    return output_state_dict
def 			__UpperCamelCase				(			)	->      List[Any]:
    '''simple docstring'''





    lowerCAmelCase_					:	List[str]						=   argparse.ArgumentParser()
    parser.add_argument("""--print-checkpoint-structure"""		,    action="""store_true"""		)
    parser.add_argument("""path_to_checkpoint"""		,    type=lowercase__		,    help="""Path to the ZIP file containing the checkpoint"""		)
    parser.add_argument(
        """--config_file"""		,    default=""""""		,    type=lowercase__		,    help="""An optional config json file describing the pre-trained model."""		,    )
    lowerCAmelCase_					:	Optional[int]						=   parser.parse_args()

    # Extract the basename.
    lowerCAmelCase_					:	Tuple						=   os.path.dirname(args.path_to_checkpoint		)

    # Load the model.
    # the .zip is very optional, let's keep it for backward compatibility
    print(f'Extracting PyTorch state dictionary from "{args.path_to_checkpoint}"'		)
    if args.path_to_checkpoint.endswith(""".zip"""		):
        with zipfile.ZipFile(args.path_to_checkpoint		,    """r"""		) as checkpoint:
            with checkpoint.open("""release/mp_rank_00/model_optim_rng.pt"""		) as pytorch_dict:
                lowerCAmelCase_					:	Union[str, Any]						=   torch.load(lowercase__		,    map_location="""cpu"""		)
    else:
        lowerCAmelCase_					:	Union[str, Any]						=   torch.load(args.path_to_checkpoint		,    map_location="""cpu"""		)

    if args.config_file == "":
        # Default config of megatron-bert 345m
        lowerCAmelCase_					:	Any						=   MegatronBertConfig()

        # different megatron-bert-*-345m models have different vocab sizes, so override the default
        # config (which is for megatron-bert-cased-345m) with the actual vocab dimension
        lowerCAmelCase_					:	Dict						=   input_state_dict["""model"""]["""lm_head"""]["""bias"""].numel()
    else:
        lowerCAmelCase_					:	Optional[Any]						=   MegatronBertConfig.from_json_file(args.config_file		)

    # Convert.
    print("""Converting"""		)
    lowerCAmelCase_					:	Union[str, Any]						=   convert_megatron_checkpoint(lowercase__		,    lowercase__		,    lowercase__		)

    # Print the structure of converted state dict.
    if args.print_checkpoint_structure:
        recursive_print(lowercase__		,    lowercase__		)

    # Store the config to file.
    print("""Saving config"""		)
    config.save_pretrained(lowercase__		)

    # Store the state_dict to file.
    lowerCAmelCase_					:	Any						=   os.path.join(lowercase__		,    """pytorch_model.bin"""		)
    print(f'Saving checkpoint to "{output_checkpoint_file}"'		)
    torch.save(lowercase__		,    lowercase__		)


####################################################################################################

if __name__ == "__main__":
  main()

####################################################################################################









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				(			lowercase__	:       nn.Module		)	->      str:
    '''simple docstring'''





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





    lowerCAmelCase_					:	Optional[int]						=   {}
    lowerCAmelCase_					:	str						=   (
        LoRAAttnProcessor2_0 if hasattr(lowercase__		,    """scaled_dot_product_attention"""		) else LoRAAttnProcessor
    )
    for name, module in text_encoder.named_modules():
        if name.endswith(lowercase__		):
            lowerCAmelCase_					:	Optional[int]						=   lora_attn_processor_class(
                hidden_size=module.out_proj.out_features		,    cross_attention_dim=lowercase__		)
    return text_lora_attn_procs
def 			__UpperCamelCase				(			lowercase__	:       nn.Module		)	->      List[str]:
    '''simple docstring'''





    lowerCAmelCase_					:	int						=   create_text_encoder_lora_attn_procs(lowercase__		)
    lowerCAmelCase_					:	str						=   AttnProcsLayers(lowercase__		)
    return text_encoder_lora_layers
def 			__UpperCamelCase				(			lowercase__	:       int		,    lowercase__	:       Union[str, Any]=False		)	->      Any:
    '''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"""		):
                lowerCAmelCase_					:	Union[str, Any]						=   (
                    torch.randn_like(layer_module.up.weight		)
                    if randn_weight
                    else torch.zeros_like(layer_module.up.weight		)
                )
                lowerCAmelCase_					:	Optional[Any]						=   torch.nn.Parameter(lowercase__		)

class 		__a							(	unittest.TestCase      ):



       def  A							(     self	:   List[Any]			):
           torch.manual_seed(0			)
           lowerCAmelCase_					:	Any						=   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					,			)
           lowerCAmelCase_					:	Dict						=   DDIMScheduler(
               beta_start=0.0_0085					,			beta_end=0.012					,			beta_schedule="""scaled_linear"""					,			clip_sample=UpperCAmelCase					,			set_alpha_to_one=UpperCAmelCase					,			steps_offset=1					,			)
           torch.manual_seed(0			)
           lowerCAmelCase_					:	List[str]						=   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					,			)
           lowerCAmelCase_					:	int						=   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=10_00					,			)
           lowerCAmelCase_					:	str						=   CLIPTextModel(UpperCAmelCase			)
           lowerCAmelCase_					:	Union[str, Any]						=   CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""			)

           lowerCAmelCase_		,		lowerCAmelCase_					:	Union[str, Any]						=   create_unet_lora_layers(UpperCAmelCase			)
           lowerCAmelCase_					:	int						=   create_text_encoder_lora_layers(UpperCAmelCase			)

           lowerCAmelCase_					:	str						=   {
               """unet""": unet,
               """scheduler""": scheduler,
               """vae""": vae,
               """text_encoder""": text_encoder,
               """tokenizer""": tokenizer,
               """safety_checker""": None,
               """feature_extractor""": None,
           }
           lowerCAmelCase_					:	List[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	:   Tuple			):
           lowerCAmelCase_					:	Dict						=   1
           lowerCAmelCase_					:	Optional[Any]						=   10
           lowerCAmelCase_					:	List[str]						=   4
           lowerCAmelCase_					:	Tuple						=   (32, 32)

           lowerCAmelCase_					:	int						=   torch.manual_seed(0			)
           lowerCAmelCase_					:	str						=   floats_tensor((batch_size, num_channels) + sizes			)
           lowerCAmelCase_					:	int						=   torch.randint(1					,			UpperCAmelCase					,			size=(batch_size, sequence_length)					,			generator=UpperCAmelCase			)

           lowerCAmelCase_					:	Union[str, Any]						=   {
               """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	:   Dict			):
           lowerCAmelCase_					:	Union[str, Any]						=   77

           lowerCAmelCase_					:	List[Any]						=   torch.randint(2					,			56					,			size=(1, max_seq_length)					,			generator=torch.manual_seed(0			)			)

           lowerCAmelCase_					:	List[Any]						=   {}
           lowerCAmelCase_					:	Optional[Any]						=   inputs
           return prepared_inputs



       def  A							(     self	:   List[str]					,			UpperCAmelCase	:   List[str]			):
           lowerCAmelCase_		,		lowerCAmelCase_					:	List[Any]						=   self.get_dummy_components()
           LoraLoaderMixin.save_lora_weights(
               save_directory=UpperCAmelCase					,			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(UpperCAmelCase					,			"""pytorch_lora_weights.bin"""			)			)			)



       def  A							(     self	:   Optional[int]			):
           lowerCAmelCase_		,		lowerCAmelCase_					:	Optional[Any]						=   self.get_dummy_components()
           lowerCAmelCase_					:	Optional[Any]						=   StableDiffusionPipeline(**UpperCAmelCase			)
           lowerCAmelCase_					:	str						=   sd_pipe.to(UpperCAmelCase			)
           sd_pipe.set_progress_bar_config(disable=UpperCAmelCase			)

           lowerCAmelCase_		,		lowerCAmelCase_		,		lowerCAmelCase_					:	List[Any]						=   self.get_dummy_inputs()

           lowerCAmelCase_					:	Tuple						=   sd_pipe(**UpperCAmelCase			).images
           lowerCAmelCase_					:	Dict						=   original_images[0, -3:, -3:, -1]

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

           lowerCAmelCase_					:	Any						=   sd_pipe(**UpperCAmelCase			).images
           lowerCAmelCase_					:	Any						=   lora_images[0, -3:, -3:, -1]

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



       def  A							(     self	:   Tuple			):
           lowerCAmelCase_		,		lowerCAmelCase_					:	Optional[Any]						=   self.get_dummy_components()
           lowerCAmelCase_					:	Union[str, Any]						=   StableDiffusionPipeline(**UpperCAmelCase			)
           lowerCAmelCase_					:	Optional[Any]						=   sd_pipe.to(UpperCAmelCase			)
           sd_pipe.set_progress_bar_config(disable=UpperCAmelCase			)

           lowerCAmelCase_		,		lowerCAmelCase_		,		lowerCAmelCase_					:	Optional[Any]						=   self.get_dummy_inputs()

           lowerCAmelCase_					:	Optional[int]						=   sd_pipe(**UpperCAmelCase			).images
           lowerCAmelCase_					:	int						=   original_images[0, -3:, -3:, -1]

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

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

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



       def  A							(     self	:   Dict			):
           lowerCAmelCase_		,		lowerCAmelCase_					:	Optional[Any]						=   self.get_dummy_components()
           lowerCAmelCase_					:	Any						=   lora_components["""unet_lora_attn_procs"""]
           lowerCAmelCase_					:	str						=   StableDiffusionPipeline(**UpperCAmelCase			)
           lowerCAmelCase_					:	Union[str, Any]						=   sd_pipe.to(UpperCAmelCase			)
           sd_pipe.set_progress_bar_config(disable=UpperCAmelCase			)

           lowerCAmelCase_		,		lowerCAmelCase_		,		lowerCAmelCase_					:	Dict						=   self.get_dummy_inputs()

           lowerCAmelCase_					:	int						=   sd_pipe(**UpperCAmelCase			).images
           lowerCAmelCase_					:	int						=   original_images[0, -3:, -3:, -1]

           with tempfile.TemporaryDirectory() as tmpdirname:
               lowerCAmelCase_					:	List[Any]						=   sd_pipe.unet
               unet.set_attn_processor(UpperCAmelCase			)
               unet.save_attn_procs(UpperCAmelCase			)
               self.assertTrue(os.path.isfile(os.path.join(UpperCAmelCase					,			"""pytorch_lora_weights.bin"""			)			)			)
               sd_pipe.load_lora_weights(UpperCAmelCase			)

           lowerCAmelCase_					:	List[str]						=   sd_pipe(**UpperCAmelCase			).images
           lowerCAmelCase_					:	List[str]						=   lora_images[0, -3:, -3:, -1]

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



       def  A							(     self	:   int			):
           lowerCAmelCase_		,		lowerCAmelCase_					:	Dict						=   self.get_dummy_components()
           lowerCAmelCase_					:	Dict						=   StableDiffusionPipeline(**UpperCAmelCase			)

           lowerCAmelCase_					:	Optional[int]						=   self.get_dummy_tokens()

           # inference without lora
           lowerCAmelCase_					:	List[Any]						=   pipe.text_encoder(**UpperCAmelCase			)[0]
           assert outputs_without_lora.shape == (1, 77, 32)

           # create lora_attn_procs with zeroed out up.weights
           lowerCAmelCase_					:	Tuple						=   create_text_encoder_lora_attn_procs(pipe.text_encoder			)
           set_lora_up_weights(UpperCAmelCase					,			randn_weight=UpperCAmelCase			)

           # monkey patch
           pipe._modify_text_encoder(UpperCAmelCase			)

           # 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_					:	str						=   pipe.text_encoder(**UpperCAmelCase			)[0]
           assert outputs_with_lora.shape == (1, 77, 32)

           assert torch.allclose(
               UpperCAmelCase					,			UpperCAmelCase			), "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_					:	Optional[Any]						=   create_text_encoder_lora_attn_procs(pipe.text_encoder			)
           set_lora_up_weights(UpperCAmelCase					,			randn_weight=UpperCAmelCase			)

           # monkey patch
           pipe._modify_text_encoder(UpperCAmelCase			)

           # 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_					:	Optional[int]						=   pipe.text_encoder(**UpperCAmelCase			)[0]
           assert outputs_with_lora.shape == (1, 77, 32)

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



       def  A							(     self	:   Union[str, Any]			):
           lowerCAmelCase_		,		lowerCAmelCase_					:	Optional[int]						=   self.get_dummy_components()
           lowerCAmelCase_					:	List[Any]						=   StableDiffusionPipeline(**UpperCAmelCase			)

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

           # inference without lora
           lowerCAmelCase_					:	Optional[Any]						=   pipe.text_encoder(**UpperCAmelCase			)[0]
           assert outputs_without_lora.shape == (1, 77, 32)

           # create lora_attn_procs with randn up.weights
           lowerCAmelCase_					:	int						=   create_text_encoder_lora_attn_procs(pipe.text_encoder			)
           set_lora_up_weights(UpperCAmelCase					,			randn_weight=UpperCAmelCase			)

           # monkey patch
           pipe._modify_text_encoder(UpperCAmelCase			)

           # 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_					:	List[str]						=   pipe.text_encoder(**UpperCAmelCase			)[0]
           assert outputs_with_lora.shape == (1, 77, 32)

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

           # remove monkey patch
           pipe._remove_text_encoder_monkey_patch()

           # inference with removed lora
           lowerCAmelCase_					:	Tuple						=   pipe.text_encoder(**UpperCAmelCase			)[0]
           assert outputs_without_lora_removed.shape == (1, 77, 32)

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



       def  A							(     self	:   Any			):
           lowerCAmelCase_		,		lowerCAmelCase_					:	Any						=   self.get_dummy_components()
           lowerCAmelCase_					:	int						=   StableDiffusionPipeline(**UpperCAmelCase			)
           lowerCAmelCase_					:	int						=   sd_pipe.to(UpperCAmelCase			)
           sd_pipe.set_progress_bar_config(disable=UpperCAmelCase			)

           lowerCAmelCase_		,		lowerCAmelCase_		,		lowerCAmelCase_					:	Optional[int]						=   self.get_dummy_inputs()

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

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

           lowerCAmelCase_					:	Dict						=   sd_pipe(**UpperCAmelCase					,			cross_attention_kwargs={"""scale""": 0.5}			).images
           lowerCAmelCase_					:	str						=   lora_images_with_scale[0, -3:, -3:, -1]

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



       def  A							(     self	:   List[str]			):
           with tempfile.TemporaryDirectory() as tmpdirname:
               self.create_lora_weight_file(UpperCAmelCase			)

               lowerCAmelCase_		,		lowerCAmelCase_					:	List[str]						=   self.get_dummy_components()
               lowerCAmelCase_					:	int						=   StableDiffusionPipeline(**UpperCAmelCase			)
               lowerCAmelCase_					:	Union[str, Any]						=   sd_pipe.to(UpperCAmelCase			)
               sd_pipe.set_progress_bar_config(disable=UpperCAmelCase			)

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

            # load LoRA weight file
               sd_pipe.load_lora_weights(UpperCAmelCase			)

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



       @unittest.skipIf(torch_device != """cuda"""					,			"""This test is supposed to run on GPU"""			)
       def  A							(     self	:   Any			):
           with tempfile.TemporaryDirectory() as tmpdirname:
               self.create_lora_weight_file(UpperCAmelCase			)

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

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

            # load LoRA weight file
               sd_pipe.load_lora_weights(UpperCAmelCase			)

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



       @unittest.skipIf(torch_device != """cuda"""					,			"""This test is supposed to run on GPU"""			)
       def  A							(     self	:   List[str]			):
           lowerCAmelCase_		,		lowerCAmelCase_					:	int						=   self.get_dummy_components()
           lowerCAmelCase_					:	Optional[int]						=   StableDiffusionPipeline(**UpperCAmelCase			)
           lowerCAmelCase_					:	Dict						=   sd_pipe.to(UpperCAmelCase			)
           sd_pipe.set_progress_bar_config(disable=UpperCAmelCase			)

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

           # enable XFormers
           sd_pipe.enable_xformers_memory_efficient_attention()

           lowerCAmelCase_					:	Any						=   sd_pipe(**UpperCAmelCase			).images
           lowerCAmelCase_					:	Union[str, Any]						=   original_images[0, -3:, -3:, -1]

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

           lowerCAmelCase_					:	Optional[Any]						=   sd_pipe(**UpperCAmelCase			).images
           lowerCAmelCase_					:	Union[str, Any]						=   lora_images[0, -3:, -3:, -1]

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


