





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    __snake_case  (    _lowerCAmelCase       :       int , _lowerCAmelCase       :       List[str]		)       ->  Tuple:
   A_    : List[str]								=	checkpoint

   A_    : Any								=	{}

   A_    : Optional[int]								=	vae_state_dict["encoder.conv_in.weight"]
   A_    : Any								=	vae_state_dict["encoder.conv_in.bias"]
   A_    : Any								=	vae_state_dict["encoder.conv_out.weight"]
   A_    : Dict								=	vae_state_dict["encoder.conv_out.bias"]
   A_    : List[str]								=	vae_state_dict["encoder.norm_out.weight"]
   A_    : Union[str, Any]								=	vae_state_dict["encoder.norm_out.bias"]

   A_    : Dict								=	vae_state_dict["decoder.conv_in.weight"]
   A_    : Tuple								=	vae_state_dict["decoder.conv_in.bias"]
   A_    : Tuple								=	vae_state_dict["decoder.conv_out.weight"]
   A_    : str								=	vae_state_dict["decoder.conv_out.bias"]
   A_    : int								=	vae_state_dict["decoder.norm_out.weight"]
   A_    : Dict								=	vae_state_dict["decoder.norm_out.bias"]

   A_    : Any								=	vae_state_dict["quant_conv.weight"]
   A_    : Dict								=	vae_state_dict["quant_conv.bias"]
   A_    : Dict								=	vae_state_dict["post_quant_conv.weight"]
   A_    : List[str]								=	vae_state_dict["post_quant_conv.bias"]

   # Retrieves the keys for the encoder down blocks only
   A_    : Optional[Any]								=	len({".".join(layer.split("."		)[:3]		) for layer in vae_state_dict if "encoder.down" in layer}		)
   A_    : Tuple								=	{
       layer_id: [key for key in vae_state_dict if f"down.{layer_id}" in key] for layer_id in range(__lowercase		)
   }

   # Retrieves the keys for the decoder up blocks only
   A_    : Union[str, Any]								=	len({".".join(layer.split("."		)[:3]		) for layer in vae_state_dict if "decoder.up" in layer}		)
   A_    : int								=	{
       layer_id: [key for key in vae_state_dict if f"up.{layer_id}" in key] for layer_id in range(__lowercase		)
   }

   for i in range(__lowercase		):
      A_    : Dict								=	[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:
         A_    : Union[str, Any]								=	vae_state_dict.pop(
             f"encoder.down.{i}.downsample.conv.weight"		)
         A_    : Optional[int]								=	vae_state_dict.pop(
             f"encoder.down.{i}.downsample.conv.bias"		)

      A_    : List[Any]								=	renew_vae_resnet_paths(__lowercase		)
      A_    : Optional[Any]								=	{"old": f"down.{i}.block", "new": f"down_blocks.{i}.resnets"}
      assign_to_checkpoint(__lowercase , __lowercase , __lowercase , additional_replacements=[meta_path] , config=__lowercase		)

   A_    : List[str]								=	[key for key in vae_state_dict if "encoder.mid.block" in key]
   A_    : int								=	2
   for i in range(1 , num_mid_res_blocks + 1		):
      A_    : Tuple								=	[key for key in mid_resnets if f"encoder.mid.block_{i}" in key]

      A_    : Any								=	renew_vae_resnet_paths(__lowercase		)
      A_    : Tuple								=	{"old": f"mid.block_{i}", "new": f"mid_block.resnets.{i - 1}"}
      assign_to_checkpoint(__lowercase , __lowercase , __lowercase , additional_replacements=[meta_path] , config=__lowercase		)

   A_    : Optional[Any]								=	[key for key in vae_state_dict if "encoder.mid.attn" in key]
   A_    : Union[str, Any]								=	renew_vae_attention_paths(__lowercase		)
   A_    : Optional[int]								=	{"old": "mid.attn_1", "new": "mid_block.attentions.0"}
   assign_to_checkpoint(__lowercase , __lowercase , __lowercase , additional_replacements=[meta_path] , config=__lowercase		)
   conv_attn_to_linear(__lowercase		)

   for i in range(__lowercase		):
      A_    : Any								=	num_up_blocks - 1 - i
      A_    : Dict								=	[
          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:
         A_    : Union[str, Any]								=	vae_state_dict[
             f"decoder.up.{block_id}.upsample.conv.weight"
         ]
         A_    : List[Any]								=	vae_state_dict[
             f"decoder.up.{block_id}.upsample.conv.bias"
         ]

      A_    : Optional[Any]								=	renew_vae_resnet_paths(__lowercase		)
      A_    : Union[str, Any]								=	{"old": f"up.{block_id}.block", "new": f"up_blocks.{i}.resnets"}
      assign_to_checkpoint(__lowercase , __lowercase , __lowercase , additional_replacements=[meta_path] , config=__lowercase		)

   A_    : List[str]								=	[key for key in vae_state_dict if "decoder.mid.block" in key]
   A_    : Dict								=	2
   for i in range(1 , num_mid_res_blocks + 1		):
      A_    : List[str]								=	[key for key in mid_resnets if f"decoder.mid.block_{i}" in key]

      A_    : Optional[Any]								=	renew_vae_resnet_paths(__lowercase		)
      A_    : str								=	{"old": f"mid.block_{i}", "new": f"mid_block.resnets.{i - 1}"}
      assign_to_checkpoint(__lowercase , __lowercase , __lowercase , additional_replacements=[meta_path] , config=__lowercase		)

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

def    __snake_case  (    _lowerCAmelCase       :       str , _lowerCAmelCase       :       str , )       ->  int:
   # Only support V1
   A_    : Dict								=	requests.get(
       " https://raw.githubusercontent.com/CompVis/stable-diffusion/main/configs/stable-diffusion/v1-inference.yaml"		)
   A_    : List[str]								=	io.BytesIO(r.content		)

   A_    : List[Any]								=	OmegaConf.load(__lowercase		)
   A_    : Tuple								=	512
   A_    : Optional[Any]								=	"cuda" if torch.cuda.is_available() else "cpu"
   if checkpoint_path.endswith("safetensors"		):
      from safetensors import safe_open

      A_    : Dict								=	{}
      with safe_open(__lowercase , framework="pt" , device="cpu"		) as f:
         for key in f.keys():
            A_    : Tuple								=	f.get_tensor(__lowercase		)
   else:
      A_    : Optional[Any]								=	torch.load(__lowercase , map_location=__lowercase		)["state_dict"]

   # Convert the VAE model.
   A_    : Union[str, Any]								=	create_vae_diffusers_config(__lowercase , image_size=__lowercase		)
   A_    : str								=	custom_convert_ldm_vae_checkpoint(__lowercase , __lowercase		)

   A_    : str								=	AutoencoderKL(**__lowercase		)
   vae.load_state_dict(__lowercase		)
   vae.save_pretrained(__lowercase		)


if __name__ == "__main__":
 _lowerCAmelCase							:					Any		    =		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.''')

 _lowerCAmelCase							:					int		    =		parser.parse_args()

 vae_pt_to_vae_diffuser(args.vae_pt_path, args.dump_path)


import numpy as np






def        a_		(       __lowercase   :	np.array ) ->  np.array:
			return 1 / (1 + np.exp(-vector ))


if __name__ == "__main__":
			import doctest

			doctest.testmod()