


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   _UpperCAmelCase       (     UpperCAmelCase      :			str						,	UpperCAmelCase      :			List[str]      ):
      """simple docstring"""




      __lowerCamelCase     :      Tuple       						=						checkpoint

      __lowerCamelCase     :      str       						=						{}

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

      __lowerCamelCase     :      Optional[int]       						=						vae_state_dict["""decoder.conv_in.weight"""]
      __lowerCamelCase     :      Optional[Any]       						=						vae_state_dict["""decoder.conv_in.bias"""]
      __lowerCamelCase     :      List[str]       						=						vae_state_dict["""decoder.conv_out.weight"""]
      __lowerCamelCase     :      Tuple       						=						vae_state_dict["""decoder.conv_out.bias"""]
      __lowerCamelCase     :      Optional[Any]       						=						vae_state_dict["""decoder.norm_out.weight"""]
      __lowerCamelCase     :      Dict       						=						vae_state_dict["""decoder.norm_out.bias"""]

      __lowerCamelCase     :      Any       						=						vae_state_dict["""quant_conv.weight"""]
      __lowerCamelCase     :      int       						=						vae_state_dict["""quant_conv.bias"""]
      __lowerCamelCase     :      List[Any]       						=						vae_state_dict["""post_quant_conv.weight"""]
      __lowerCamelCase     :      Union[str, Any]       						=						vae_state_dict["""post_quant_conv.bias"""]

      # Retrieves the keys for the encoder down blocks only
      __lowerCamelCase     :      Dict       						=						len({""".""".join(layer.split("""."""      )[:3]      ) for layer in vae_state_dict if """encoder.down""" in layer}      )
      __lowerCamelCase     :      List[str]       						=						{
          layer_id: [key for key in vae_state_dict if f"""down.{layer_id}""" in key] for layer_id in range(UpperCAmelCase      )
      }

      # Retrieves the keys for the decoder up blocks only
      __lowerCamelCase     :      List[Any]       						=						len({""".""".join(layer.split("""."""      )[:3]      ) for layer in vae_state_dict if """decoder.up""" in layer}      )
      __lowerCamelCase     :      Dict       						=						{
          layer_id: [key for key in vae_state_dict if f"""up.{layer_id}""" in key] for layer_id in range(UpperCAmelCase      )
      }

      for i in range(UpperCAmelCase      ):
            __lowerCamelCase     :      Optional[Any]       						=						[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     :      Any       						=						vae_state_dict.pop(
                      f"""encoder.down.{i}.downsample.conv.weight"""      )
                  __lowerCamelCase     :      List[Any]       						=						vae_state_dict.pop(
                      f"""encoder.down.{i}.downsample.conv.bias"""      )

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

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

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

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

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

            __lowerCamelCase     :      List[Any]       						=						renew_vae_resnet_paths(UpperCAmelCase      )
            __lowerCamelCase     :      str       						=						{"""old""": f"""up.{block_id}.block""", """new""": f"""up_blocks.{i}.resnets"""}
            assign_to_checkpoint(UpperCAmelCase						,	UpperCAmelCase						,	UpperCAmelCase						,	additional_replacements=[meta_path]						,	config=UpperCAmelCase      )

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

            __lowerCamelCase     :      Optional[int]       						=						renew_vae_resnet_paths(UpperCAmelCase      )
            __lowerCamelCase     :      Dict       						=						{"""old""": f"""mid.block_{i}""", """new""": f"""mid_block.resnets.{i - 1}"""}
            assign_to_checkpoint(UpperCAmelCase						,	UpperCAmelCase						,	UpperCAmelCase						,	additional_replacements=[meta_path]						,	config=UpperCAmelCase      )

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





def   _UpperCAmelCase       (     UpperCAmelCase      :			str						,	UpperCAmelCase      :			str						,	):
      """simple docstring"""




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

      __lowerCamelCase     :      Optional[int]       						=						OmegaConf.load(UpperCAmelCase      )
      __lowerCamelCase     :      str       						=						512
      __lowerCamelCase     :      Union[str, Any]       						=						"""cuda""" if torch.cuda.is_available() else """cpu"""
      if checkpoint_path.endswith("""safetensors"""      ):
            from safetensors import safe_open

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

      # Convert the VAE model.
      __lowerCamelCase     :      Tuple       						=						create_vae_diffusers_config(UpperCAmelCase						,	image_size=UpperCAmelCase      )
      __lowerCamelCase     :      Any       						=						custom_convert_ldm_vae_checkpoint(UpperCAmelCase						,	UpperCAmelCase      )

      __lowerCamelCase     :      int       						=						AutoencoderKL(**UpperCAmelCase      )
      vae.load_state_dict(UpperCAmelCase      )
      vae.save_pretrained(UpperCAmelCase      )


if __name__ == "__main__":
    __UpperCamelCase    :						List[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.')

    __UpperCamelCase    :						Optional[Any]     =							parser.parse_args()

    vae_pt_to_vae_diffuser(args.vae_pt_path, args.dump_path)







from typing import Any





def   _UpperCAmelCase       (     UpperCAmelCase      :			list      ):
      """simple docstring"""




      if not input_list:
            return []
      __lowerCamelCase     :      int       						=						[input_list.count(UpperCAmelCase      ) for value in input_list]
      __lowerCamelCase     :      Tuple       						=						max(UpperCAmelCase      )  # Gets the maximum count in the input list.
      # Gets values of modes
      return sorted({input_list[i] for i, value in enumerate(UpperCAmelCase      ) if value == y}      )


if __name__ == "__main__":
    import doctest

    doctest.testmod()



