





'''simple docstring'''


import argparse
import io

import requests
import torch
from omegaconf import OmegaConf

from diffusers import AutoencoderKL
from diffusers.pipelines.stable_diffusion.convert_from_ckpt import (
    assign_to_checkpoint,
    conv_attn_to_linear,
    create_vae_diffusers_config,
    renew_vae_attention_paths,
    renew_vae_resnet_paths,
)


def 						__a(SCREAMING_SNAKE_CASE_	:       Any		,							SCREAMING_SNAKE_CASE_	:       List[Any]     ):



    '''simple docstring'''





    _lowerCAmelCase						      =   checkpoint

    _lowerCAmelCase						      =   {}

    _lowerCAmelCase						      =   vae_state_dict["encoder.conv_in.weight"]
    _lowerCAmelCase						      =   vae_state_dict["encoder.conv_in.bias"]
    _lowerCAmelCase						      =   vae_state_dict["encoder.conv_out.weight"]
    _lowerCAmelCase						      =   vae_state_dict["encoder.conv_out.bias"]
    _lowerCAmelCase						      =   vae_state_dict["encoder.norm_out.weight"]
    _lowerCAmelCase						      =   vae_state_dict["encoder.norm_out.bias"]

    _lowerCAmelCase						      =   vae_state_dict["decoder.conv_in.weight"]
    _lowerCAmelCase						      =   vae_state_dict["decoder.conv_in.bias"]
    _lowerCAmelCase						      =   vae_state_dict["decoder.conv_out.weight"]
    _lowerCAmelCase						      =   vae_state_dict["decoder.conv_out.bias"]
    _lowerCAmelCase						      =   vae_state_dict["decoder.norm_out.weight"]
    _lowerCAmelCase						      =   vae_state_dict["decoder.norm_out.bias"]

    _lowerCAmelCase						      =   vae_state_dict["quant_conv.weight"]
    _lowerCAmelCase						      =   vae_state_dict["quant_conv.bias"]
    _lowerCAmelCase						      =   vae_state_dict["post_quant_conv.weight"]
    _lowerCAmelCase						      =   vae_state_dict["post_quant_conv.bias"]

    # Retrieves the keys for the encoder down blocks only
    _lowerCAmelCase						      =   len({".".join(layer.split("."     )[:3]     ) for layer in vae_state_dict if "encoder.down" in layer}     )
    _lowerCAmelCase						      =   {
        layer_id: [key for key in vae_state_dict if F'''down.{layer_id}''' in key] for layer_id in range(SCREAMING_SNAKE_CASE_     )
    }

    # Retrieves the keys for the decoder up blocks only
    _lowerCAmelCase						      =   len({".".join(layer.split("."     )[:3]     ) for layer in vae_state_dict if "decoder.up" in layer}     )
    _lowerCAmelCase						      =   {
        layer_id: [key for key in vae_state_dict if F'''up.{layer_id}''' in key] for layer_id in range(SCREAMING_SNAKE_CASE_     )
    }

    for i in range(SCREAMING_SNAKE_CASE_     ):
        _lowerCAmelCase						      =   [key for key in down_blocks[i] if F'''down.{i}''' in key and F'''down.{i}.downsample''' not in key]

        if F'''encoder.down.{i}.downsample.conv.weight''' in vae_state_dict:
            _lowerCAmelCase						      =   vae_state_dict.pop(
                F'''encoder.down.{i}.downsample.conv.weight'''     )
            _lowerCAmelCase						      =   vae_state_dict.pop(
                F'''encoder.down.{i}.downsample.conv.bias'''     )

        _lowerCAmelCase						      =   renew_vae_resnet_paths(SCREAMING_SNAKE_CASE_     )
        _lowerCAmelCase						      =   {"old": F'''down.{i}.block''', "new": F'''down_blocks.{i}.resnets'''}
        assign_to_checkpoint(SCREAMING_SNAKE_CASE_		,							SCREAMING_SNAKE_CASE_		,							SCREAMING_SNAKE_CASE_		,							additional_replacements=[meta_path]		,							config=SCREAMING_SNAKE_CASE_     )

    _lowerCAmelCase						      =   [key for key in vae_state_dict if "encoder.mid.block" in key]
    _lowerCAmelCase						      =   2
    for i in range(1		,							num_mid_res_blocks + 1     ):
        _lowerCAmelCase						      =   [key for key in mid_resnets if F'''encoder.mid.block_{i}''' in key]

        _lowerCAmelCase						      =   renew_vae_resnet_paths(SCREAMING_SNAKE_CASE_     )
        _lowerCAmelCase						      =   {"old": F'''mid.block_{i}''', "new": F'''mid_block.resnets.{i - 1}'''}
        assign_to_checkpoint(SCREAMING_SNAKE_CASE_		,							SCREAMING_SNAKE_CASE_		,							SCREAMING_SNAKE_CASE_		,							additional_replacements=[meta_path]		,							config=SCREAMING_SNAKE_CASE_     )

    _lowerCAmelCase						      =   [key for key in vae_state_dict if "encoder.mid.attn" in key]
    _lowerCAmelCase						      =   renew_vae_attention_paths(SCREAMING_SNAKE_CASE_     )
    _lowerCAmelCase						      =   {"old": "mid.attn_1", "new": "mid_block.attentions.0"}
    assign_to_checkpoint(SCREAMING_SNAKE_CASE_		,							SCREAMING_SNAKE_CASE_		,							SCREAMING_SNAKE_CASE_		,							additional_replacements=[meta_path]		,							config=SCREAMING_SNAKE_CASE_     )
    conv_attn_to_linear(SCREAMING_SNAKE_CASE_     )

    for i in range(SCREAMING_SNAKE_CASE_     ):
        _lowerCAmelCase						      =   num_up_blocks - 1 - i
        _lowerCAmelCase						      =   [
            key for key in up_blocks[block_id] if F'''up.{block_id}''' in key and F'''up.{block_id}.upsample''' not in key
        ]

        if F'''decoder.up.{block_id}.upsample.conv.weight''' in vae_state_dict:
            _lowerCAmelCase						      =   vae_state_dict[
                F'''decoder.up.{block_id}.upsample.conv.weight'''
            ]
            _lowerCAmelCase						      =   vae_state_dict[
                F'''decoder.up.{block_id}.upsample.conv.bias'''
            ]

        _lowerCAmelCase						      =   renew_vae_resnet_paths(SCREAMING_SNAKE_CASE_     )
        _lowerCAmelCase						      =   {"old": F'''up.{block_id}.block''', "new": F'''up_blocks.{i}.resnets'''}
        assign_to_checkpoint(SCREAMING_SNAKE_CASE_		,							SCREAMING_SNAKE_CASE_		,							SCREAMING_SNAKE_CASE_		,							additional_replacements=[meta_path]		,							config=SCREAMING_SNAKE_CASE_     )

    _lowerCAmelCase						      =   [key for key in vae_state_dict if "decoder.mid.block" in key]
    _lowerCAmelCase						      =   2
    for i in range(1		,							num_mid_res_blocks + 1     ):
        _lowerCAmelCase						      =   [key for key in mid_resnets if F'''decoder.mid.block_{i}''' in key]

        _lowerCAmelCase						      =   renew_vae_resnet_paths(SCREAMING_SNAKE_CASE_     )
        _lowerCAmelCase						      =   {"old": F'''mid.block_{i}''', "new": F'''mid_block.resnets.{i - 1}'''}
        assign_to_checkpoint(SCREAMING_SNAKE_CASE_		,							SCREAMING_SNAKE_CASE_		,							SCREAMING_SNAKE_CASE_		,							additional_replacements=[meta_path]		,							config=SCREAMING_SNAKE_CASE_     )

    _lowerCAmelCase						      =   [key for key in vae_state_dict if "decoder.mid.attn" in key]
    _lowerCAmelCase						      =   renew_vae_attention_paths(SCREAMING_SNAKE_CASE_     )
    _lowerCAmelCase						      =   {"old": "mid.attn_1", "new": "mid_block.attentions.0"}
    assign_to_checkpoint(SCREAMING_SNAKE_CASE_		,							SCREAMING_SNAKE_CASE_		,							SCREAMING_SNAKE_CASE_		,							additional_replacements=[meta_path]		,							config=SCREAMING_SNAKE_CASE_     )
    conv_attn_to_linear(SCREAMING_SNAKE_CASE_     )
    return new_checkpoint


def 						__a(SCREAMING_SNAKE_CASE_	:       str		,							SCREAMING_SNAKE_CASE_	:       str		,							):



    '''simple docstring'''





    _lowerCAmelCase						      =   requests.get(
        " https://raw.githubusercontent.com/CompVis/stable-diffusion/main/configs/stable-diffusion/v1-inference.yaml"     )
    _lowerCAmelCase						      =   io.BytesIO(r.content     )

    _lowerCAmelCase						      =   OmegaConf.load(SCREAMING_SNAKE_CASE_     )
    _lowerCAmelCase						      =   512
    _lowerCAmelCase						      =   "cuda" if torch.cuda.is_available() else "cpu"
    if checkpoint_path.endswith("safetensors"     ):
        from safetensors import safe_open

        _lowerCAmelCase						      =   {}
        with safe_open(SCREAMING_SNAKE_CASE_		,							framework="pt"		,							device="cpu"     ) as f:
            for key in f.keys():
                _lowerCAmelCase						      =   f.get_tensor(SCREAMING_SNAKE_CASE_     )
    else:
        _lowerCAmelCase						      =   torch.load(SCREAMING_SNAKE_CASE_		,							map_location=SCREAMING_SNAKE_CASE_     )["state_dict"]

    # Convert the VAE model.
    _lowerCAmelCase						      =   create_vae_diffusers_config(SCREAMING_SNAKE_CASE_		,							image_size=SCREAMING_SNAKE_CASE_     )
    _lowerCAmelCase						      =   custom_convert_ldm_vae_checkpoint(SCREAMING_SNAKE_CASE_		,							SCREAMING_SNAKE_CASE_     )

    _lowerCAmelCase						      =   AutoencoderKL(**SCREAMING_SNAKE_CASE_     )
    vae.load_state_dict(SCREAMING_SNAKE_CASE_     )
    vae.save_pretrained(SCREAMING_SNAKE_CASE_     )


if __name__ == "__main__":
       _SCREAMING_SNAKE_CASE									=				argparse.ArgumentParser()

       parser.add_argument("--vae_pt_path", default=None, type=str, required=True, help="Path to the VAE.pt to convert.")
       parser.add_argument("--dump_path", default=None, type=str, required=True, help="Path to the VAE.pt to convert.")

       _SCREAMING_SNAKE_CASE									=				parser.parse_args()

       vae_pt_to_vae_diffuser(args.vae_pt_path, args.dump_path)



'''simple docstring'''


from ...utils import (
    OptionalDependencyNotAvailable,
    is_torch_available,
    is_transformers_available,
    is_transformers_version,
)


try:
					if not (is_transformers_available() and is_torch_available()):
										raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
					from ...utils.dummy_torch_and_transformers_objects import (
					    ImageTextPipelineOutput,
					    UniDiffuserPipeline,
					)
else:
					from .modeling_text_decoder import UniDiffuserTextDecoder
					from .modeling_uvit import UniDiffuserModel, UTransformeraDModel
					from .pipeline_unidiffuser import ImageTextPipelineOutput, UniDiffuserPipeline
