






"""simple docstring"""





import argparse
import os
import re

import packaging.version


UpperCamelCase_	:				str										= """examples/"""
UpperCamelCase_	:				int										= {
    """examples""": (re.compile(r"""^check_min_version\(\"[^\"]+\"\)\s*$""", re.MULTILINE), """check_min_version(\"VERSION\")\n"""),
    """init""": (re.compile(r"""^__version__\s+=\s+\"([^\"]+)\"\s*$""", re.MULTILINE), """__version__ = \"VERSION\"\n"""),
    """setup""": (re.compile(r"""^(\s*)version\s*=\s*\"[^\"]+\",""", re.MULTILINE), r"""\1version=\"VERSION\","""),
    """doc""": (re.compile(r"""^(\s*)release\s*=\s*\"[^\"]+\"$""", re.MULTILINE), """release = \"VERSION\"\n"""),
}
UpperCamelCase_	:				Tuple										= {
    """init""": """src/transformers/__init__.py""",
    """setup""": """setup.py""",
}
UpperCamelCase_	:				Any										= """README.md"""



def     __lowercase					(    a :				Optional[Any]       ,		a :				List[str]       ,		a :				Any )					->					Dict:






      with open(__SCREAMING_SNAKE_CASE       ,		'''r'''       ,		encoding='''utf-8'''       ,		newline='''\n''' ) as f:
            __snake_case      :    Dict  			=f.read()
      __snake_case      :    int  			=REPLACE_PATTERNS[pattern]
      __snake_case      :    int  			=replace.replace('''VERSION'''       ,		__SCREAMING_SNAKE_CASE )
      __snake_case      :    Any  			=re_pattern.sub(__SCREAMING_SNAKE_CASE       ,		__SCREAMING_SNAKE_CASE )
      with open(__SCREAMING_SNAKE_CASE       ,		'''w'''       ,		encoding='''utf-8'''       ,		newline='''\n''' ) as f:
            f.write(__SCREAMING_SNAKE_CASE )



def     __lowercase					(    a :				Optional[Any] )					->					Any:






      for folder, directories, fnames in os.walk(__SCREAMING_SNAKE_CASE ):
            # Removing some of the folders with non-actively maintained examples from the walk
            if "research_projects" in directories:
                  directories.remove('''research_projects''' )
            if "legacy" in directories:
                  directories.remove('''legacy''' )
            for fname in fnames:
                  if fname.endswith('''.py''' ):
                        update_version_in_file(os.path.join(__SCREAMING_SNAKE_CASE       ,		__SCREAMING_SNAKE_CASE )       ,		__SCREAMING_SNAKE_CASE       ,		pattern='''examples''' )



def     __lowercase					(    a :				str       ,		a :				int=False )					->					Dict:






      for pattern, fname in REPLACE_FILES.items():
            update_version_in_file(__SCREAMING_SNAKE_CASE       ,		__SCREAMING_SNAKE_CASE       ,		__SCREAMING_SNAKE_CASE )
      if not patch:
            update_version_in_examples(__SCREAMING_SNAKE_CASE )



def     __lowercase					(    )					->					Tuple:
      __snake_case      :    Union[str, Any]  			="🤗 Transformers currently provides the following architectures"
      __snake_case      :    str  			="1. Want to contribute a new model?"
      with open(__SCREAMING_SNAKE_CASE       ,		'''r'''       ,		encoding='''utf-8'''       ,		newline='''\n''' ) as f:
            __snake_case      :    int  			=f.readlines()

      # Find the start of the list.
      __snake_case      :    str  			=0
      while not lines[start_index].startswith(_start_prompt ):
            start_index += 1
      start_index += 1

      __snake_case      :    List[str]  			=start_index
      # Update the lines in the model list.
      while not lines[index].startswith(_end_prompt ):
            if lines[index].startswith('''1.''' ):
                  __snake_case      :    int  			=lines[index].replace(
                      '''https://huggingface.co/docs/transformers/main/model_doc'''       ,		'''https://huggingface.co/docs/transformers/model_doc'''       ,		)
            index += 1

      with open(__SCREAMING_SNAKE_CASE       ,		'''w'''       ,		encoding='''utf-8'''       ,		newline='''\n''' ) as f:
            f.writelines(__SCREAMING_SNAKE_CASE )



def     __lowercase					(    )					->					Tuple:






      with open(REPLACE_FILES['''init''']       ,		'''r''' ) as f:
            __snake_case      :    Tuple  			=f.read()
      __snake_case      :    List[Any]  			=REPLACE_PATTERNS["init"][0].search(__SCREAMING_SNAKE_CASE ).groups()[0]
      return packaging.version.parse(__SCREAMING_SNAKE_CASE )



def     __lowercase					(    a :				Optional[int]=False )					->					List[Any]:
      __snake_case      :    List[Any]  			=get_version()
      if patch and default_version.is_devrelease:
            raise ValueError('''Can\'t create a patch version from the dev branch, checkout a released version!''' )
      if default_version.is_devrelease:
            __snake_case      :    Any  			=default_version.base_version
      elif patch:
            __snake_case      :    List[Any]  			=f'''{default_version.major}.{default_version.minor}.{default_version.micro + 1}'''
      else:
            __snake_case      :    str  			=f'''{default_version.major}.{default_version.minor + 1}.0'''

      # Now let's ask nicely if that's the right one.
      __snake_case      :    Optional[Any]  			=input(f'''Which version are you releasing? [{default_version}]''' )
      if len(__SCREAMING_SNAKE_CASE ) == 0:
            __snake_case      :    Any  			=default_version

      print(f'''Updating version to {version}.''' )
      global_version_update(__SCREAMING_SNAKE_CASE       ,		patch=__SCREAMING_SNAKE_CASE )
      if not patch:
            print('''Cleaning main README, don\'t forget to run `make fix-copies`.''' )
            clean_main_ref_in_model_list()



def     __lowercase					(    )					->					List[Any]:
      __snake_case      :    int  			=get_version()
      __snake_case      :    Tuple  			=f'''{current_version.major}.{current_version.minor + 1}.0.dev0'''
      __snake_case      :    Optional[Any]  			=current_version.base_version

      # Check with the user we got that right.
      __snake_case      :    Tuple  			=input(f'''Which version are we developing now? [{dev_version}]''' )
      if len(__SCREAMING_SNAKE_CASE ) == 0:
            __snake_case      :    Union[str, Any]  			=dev_version

      print(f'''Updating version to {version}.''' )
      global_version_update(__SCREAMING_SNAKE_CASE )
      print('''Cleaning main README, don\'t forget to run `make fix-copies`.''' )
      clean_main_ref_in_model_list()


if __name__ == "__main__":
    UpperCamelCase_	:				str										= argparse.ArgumentParser()
    parser.add_argument("""--post_release""", action="""store_true""", help="""Whether this is pre or post release.""")
    parser.add_argument("""--patch""", action="""store_true""", help="""Whether or not this is a patch release.""")
    UpperCamelCase_	:				Any										= parser.parse_args()
    if not args.post_release:
        pre_release_work(patch=args.patch)
    elif args.patch:
        print("""Nothing to do after a patch :-)""")
    else:
        post_release_work()








"""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     __lowercase					(    a :				Any       ,		a :				List[Any] )					->					Dict:
      __snake_case      :    Any  			=checkpoint

      __snake_case      :    Dict  			={}

      __snake_case      :    List[Any]  			=vae_state_dict['''encoder.conv_in.weight''']
      __snake_case      :    List[str]  			=vae_state_dict['''encoder.conv_in.bias''']
      __snake_case      :    Union[str, Any]  			=vae_state_dict['''encoder.conv_out.weight''']
      __snake_case      :    Union[str, Any]  			=vae_state_dict['''encoder.conv_out.bias''']
      __snake_case      :    str  			=vae_state_dict['''encoder.norm_out.weight''']
      __snake_case      :    str  			=vae_state_dict['''encoder.norm_out.bias''']

      __snake_case      :    Tuple  			=vae_state_dict['''decoder.conv_in.weight''']
      __snake_case      :    str  			=vae_state_dict['''decoder.conv_in.bias''']
      __snake_case      :    List[str]  			=vae_state_dict['''decoder.conv_out.weight''']
      __snake_case      :    Tuple  			=vae_state_dict['''decoder.conv_out.bias''']
      __snake_case      :    Union[str, Any]  			=vae_state_dict['''decoder.norm_out.weight''']
      __snake_case      :    List[str]  			=vae_state_dict['''decoder.norm_out.bias''']

      __snake_case      :    Tuple  			=vae_state_dict['''quant_conv.weight''']
      __snake_case      :    List[str]  			=vae_state_dict['''quant_conv.bias''']
      __snake_case      :    Optional[int]  			=vae_state_dict['''post_quant_conv.weight''']
      __snake_case      :    Tuple  			=vae_state_dict['''post_quant_conv.bias''']

      # Retrieves the keys for the encoder down blocks only
      __snake_case      :    Union[str, Any]  			=len({'''.'''.join(layer.split('''.''' )[:3] ) for layer in vae_state_dict if '''encoder.down''' in layer} )
      __snake_case      :    Union[str, Any]  			={
          layer_id: [key for key in vae_state_dict if f'''down.{layer_id}''' in key] for layer_id in range(a )
      }

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

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

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

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

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

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

      for i in range(a ):
            __snake_case      :    List[Any]  			=num_up_blocks - 1 - i
            __snake_case      :    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:
                  __snake_case      :    int  			=vae_state_dict[
                      f'''decoder.up.{block_id}.upsample.conv.weight'''
                  ]
                  __snake_case      :    Optional[Any]  			=vae_state_dict[
                      f'''decoder.up.{block_id}.upsample.conv.bias'''
                  ]

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

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

            __snake_case      :    Tuple  			=renew_vae_resnet_paths(a )
            __snake_case      :    Tuple  			={'''old''': f'''mid.block_{i}''', '''new''': f'''mid_block.resnets.{i - 1}'''}
            assign_to_checkpoint(a       ,		a       ,		a       ,		additional_replacements=[meta_path]       ,		config=a )

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



def     __lowercase					(    a :				str       ,		a :				str       ,		)					->					Optional[int]:
      # Only support V1
      __snake_case      :    Union[str, Any]  			=requests.get(
          ''' https://raw.githubusercontent.com/CompVis/stable-diffusion/main/configs/stable-diffusion/v1-inference.yaml''' )
      __snake_case      :    Union[str, Any]  			=io.BytesIO(r.content )

      __snake_case      :    Any  			=OmegaConf.load(a )
      __snake_case      :    Union[str, Any]  			=512
      __snake_case      :    List[str]  			='''cuda''' if torch.cuda.is_available() else '''cpu'''
      if checkpoint_path.endswith('''safetensors''' ):
            from safetensors import safe_open

            __snake_case      :    Any  			={}
            with safe_open(a       ,		framework='''pt'''       ,		device='''cpu''' ) as f:
                  for key in f.keys():
                        __snake_case      :    Optional[int]  			=f.get_tensor(a )
      else:
            __snake_case      :    Optional[int]  			=torch.load(a       ,		map_location=a )['''state_dict''']

      # Convert the VAE model.
      __snake_case      :    Dict  			=create_vae_diffusers_config(a       ,		image_size=a )
      __snake_case      :    Union[str, Any]  			=custom_convert_ldm_vae_checkpoint(a       ,		a )

      __snake_case      :    str  			=AutoencoderKL(**a )
      vae.load_state_dict(a )
      vae.save_pretrained(a )


if __name__ == "__main__":
    UpperCamelCase_	:				Optional[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)


