




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 				lowerCAmelCase_	(     snake_case_,snake_case_     ):
      _A							:     List[Any]			=						checkpoint

      _A							:     Optional[Any]			=						{}

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

      _A							:     Optional[int]			=						vae_state_dict["""decoder.conv_in.weight"""]
      _A							:     Tuple			=						vae_state_dict["""decoder.conv_in.bias"""]
      _A							:     Union[str, Any]			=						vae_state_dict["""decoder.conv_out.weight"""]
      _A							:     Tuple			=						vae_state_dict["""decoder.conv_out.bias"""]
      _A							:     Tuple			=						vae_state_dict["""decoder.norm_out.weight"""]
      _A							:     List[str]			=						vae_state_dict["""decoder.norm_out.bias"""]

      _A							:     Optional[int]			=						vae_state_dict["""quant_conv.weight"""]
      _A							:     List[Any]			=						vae_state_dict["""quant_conv.bias"""]
      _A							:     Optional[Any]			=						vae_state_dict["""post_quant_conv.weight"""]
      _A							:     int			=						vae_state_dict["""post_quant_conv.bias"""]

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

      # Retrieves the keys for the decoder up blocks only
      _A							:     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(snake_case_     )
      }

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

            _A							:     Optional[Any]			=						renew_vae_resnet_paths(snake_case_     )
            _A							:     str			=						{"""old""": f'''down.{i}.block''', """new""": f'''down_blocks.{i}.resnets'''}
            assign_to_checkpoint(snake_case_,snake_case_,snake_case_,additional_replacements=[meta_path],config=snake_case_     )

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

            _A							:     str			=						renew_vae_resnet_paths(snake_case_     )
            _A							:     str			=						{"""old""": f'''mid.block_{i}''', """new""": f'''mid_block.resnets.{i - 1}'''}
            assign_to_checkpoint(snake_case_,snake_case_,snake_case_,additional_replacements=[meta_path],config=snake_case_     )

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

      for i in range(snake_case_     ):
            _A							:     List[str]			=						num_up_blocks - 1 - i
            _A							:     List[str]			=						[
                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[str]			=						vae_state_dict[
                      f'''decoder.up.{block_id}.upsample.conv.bias'''
                  ]

            _A							:     str			=						renew_vae_resnet_paths(snake_case_     )
            _A							:     int			=						{"""old""": f'''up.{block_id}.block''', """new""": f'''up_blocks.{i}.resnets'''}
            assign_to_checkpoint(snake_case_,snake_case_,snake_case_,additional_replacements=[meta_path],config=snake_case_     )

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

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

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







def 				lowerCAmelCase_	(     snake_case_,snake_case_,):
      # Only support V1
      _A							:     Tuple			=						requests.get(
          """ https://raw.githubusercontent.com/CompVis/stable-diffusion/main/configs/stable-diffusion/v1-inference.yaml"""     )
      _A							:     Tuple			=						io.BytesIO(r.content     )

      _A							:     List[Any]			=						OmegaConf.load(snake_case_     )
      _A							:     Optional[int]			=						512
      _A							:     Dict			=						"""cuda""" if torch.cuda.is_available() else """cpu"""
      if checkpoint_path.endswith("""safetensors"""     ):
            from safetensors import safe_open

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

      # Convert the VAE model.
      _A							:     List[str]			=						create_vae_diffusers_config(snake_case_,image_size=snake_case_     )
      _A							:     Any			=						custom_convert_ldm_vae_checkpoint(snake_case_,snake_case_     )

      _A							:     List[Any]			=						AutoencoderKL(**snake_case_     )
      vae.load_state_dict(snake_case_     )
      vae.save_pretrained(snake_case_     )


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

  _snake_case										=       parser.parse_args()

  vae_pt_to_vae_diffuser(args.vae_pt_path, args.dump_path)









def 						_a    (      a :float	,							a :float )							->			float:
 return price * (1 + tax_rate)


if __name__ == "__main__":
     print(f"""{price_plus_tax(100, 0.25) = }""")
     print(f"""{price_plus_tax(125.50, 0.05) = }""")


