



'''simple docstring'''




import json
import os
import shutil
import warnings
from argparse import ArgumentParser, Namespace
from pathlib import Path
from typing import List

from ..utils import logging
from . import BaseTransformersCLICommand


try:
						from cookiecutter.main import cookiecutter

						lowerCamelCase							:Tuple				    =					True
except ImportError:
						lowerCamelCase							:List[Any]				    =					False

lowerCamelCase							:Optional[Any]				    =					logging.get_logger(__name__)  # pylint: disable=invalid-name



def 	a			( lowerCamelCase__    ):
			'''simple docstring'''





			return AddNewModelCommand(args.testing						,      args.testing_file						,      path=args.path    )







class 	_lowerCAmelCase							(			SCREAMING_SNAKE_CASE_ ):


	@staticmethod
	def     _a	(lowercase							):
				A_	:						List[str]	=     parser.add_parser("""add-new-model"""							)
				add_new_model_parser.add_argument("""--testing"""       ,      action="""store_true"""       ,      help="""If in testing mode."""							)
				add_new_model_parser.add_argument("""--testing_file"""       ,      type=_a       ,      help="""Configuration file on which to run."""							)
				add_new_model_parser.add_argument(
				    """--path"""       ,      type=_a       ,      help="""Path to cookiecutter. Should only be used for testing purposes."""							)
				add_new_model_parser.set_defaults(func=_a							)


	def __init__(self       ,      lowercase       ,      lowercase       ,      lowercase=None       ,      *lowercase							):
				A_	:						str	=     testing
				A_	:						List[Any]	=     testing_file
				A_	:						Tuple	=     path


	def     _a	(self							):
				warnings.warn(
				    """The command `transformers-cli add-new-model` is deprecated and will be removed in v5 of Transformers. """
				    """It is not actively maintained anymore, so might give a result that won\'t pass all tests and quality """
				    """checks, you should use `transformers-cli add-new-model-like` instead."""							)
				if not _has_cookiecutter:
							raise ImportError(
							    """Model creation dependencies are required to use the `add_new_model` command. Install them by running """
							    """the following at the root of your `transformers` clone:\n\n\t$ pip install -e .[modelcreation]\n"""							)
				# Ensure that there is no other `cookiecutter-template-xxx` directory in the current working directory
				A_	:						Any	=     [directory for directory in os.listdir() if 'cookiecutter-template-' == directory[:22]]
				if len(_a							) > 0:
							raise ValueError(
							    """Several directories starting with `cookiecutter-template-` in current working directory. """
							    """Please clean your directory by removing all folders starting with `cookiecutter-template-` or """
							    """change your working directory."""							)

				A_	:						List[Any]	=     (
				    Path(_a							).parent.parent.parent.parent if self._path is None else Path(self._path							).parent.parent
				)
				A_	:						List[str]	=     path_to_transformer_root / 'templates' / 'adding_a_new_model'

				# Execute cookiecutter
				if not self._testing:
							cookiecutter(str(_a							)							)
				else:
							with open(self._testing_file       ,      """r"""							) as configuration_file:
										A_	:						List[Any]	=     json.load(_a							)

							cookiecutter(
							    str(path_to_cookiecutter if self._path is None else self._path							)       ,      no_input=_a       ,      extra_context=_a       ,      )

				A_	:						List[str]	=     [directory for directory in os.listdir() if 'cookiecutter-template-' in directory[:22]][0]

				# Retrieve configuration
				with open(directory + """/configuration.json"""       ,      """r"""							) as configuration_file:
							A_	:						Optional[int]	=     json.load(_a							)

				A_	:						Tuple	=     configuration['lowercase_modelname']
				A_	:						List[Any]	=     configuration['generate_tensorflow_pytorch_and_flax']
				os.remove(F'{directory}/configuration.json'							)

				A_	:						Optional[Any]	=     'PyTorch' in generate_tensorflow_pytorch_and_flax
				A_	:						List[Any]	=     'TensorFlow' in generate_tensorflow_pytorch_and_flax
				A_	:						Optional[Any]	=     'Flax' in generate_tensorflow_pytorch_and_flax

				A_	:						Tuple	=     F'{path_to_transformer_root}/src/transformers/models/{lowercase_model_name}'
				os.makedirs(_a       ,      exist_ok=_a							)
				os.makedirs(F'{path_to_transformer_root}/tests/models/{lowercase_model_name}'       ,      exist_ok=_a							)

				# Tests require submodules as they have parent imports
				with open(F'{path_to_transformer_root}/tests/models/{lowercase_model_name}/__init__.py'       ,      """w"""							):
							pass

				shutil.move(
				    F'{directory}/__init__.py'       ,      F'{model_dir}/__init__.py'       ,      )
				shutil.move(
				    F'{directory}/configuration_{lowercase_model_name}.py'       ,      F'{model_dir}/configuration_{lowercase_model_name}.py'       ,      )

				def remove_copy_lines(lowercase							):
							with open(_a       ,      """r"""							) as f:
										A_	:						List[str]	=     f.readlines()
							with open(_a       ,      """w"""							) as f:
										for line in lines:
													if "# Copied from transformers." not in line:
																f.write(_a							)

				if output_pytorch:
							if not self._testing:
										remove_copy_lines(F'{directory}/modeling_{lowercase_model_name}.py'							)

							shutil.move(
							    F'{directory}/modeling_{lowercase_model_name}.py'       ,      F'{model_dir}/modeling_{lowercase_model_name}.py'       ,      )

							shutil.move(
							    F'{directory}/test_modeling_{lowercase_model_name}.py'       ,      F'{path_to_transformer_root}/tests/models/{lowercase_model_name}/test_modeling_{lowercase_model_name}.py'       ,      )
				else:
							os.remove(F'{directory}/modeling_{lowercase_model_name}.py'							)
							os.remove(F'{directory}/test_modeling_{lowercase_model_name}.py'							)

				if output_tensorflow:
							if not self._testing:
										remove_copy_lines(F'{directory}/modeling_tf_{lowercase_model_name}.py'							)

							shutil.move(
							    F'{directory}/modeling_tf_{lowercase_model_name}.py'       ,      F'{model_dir}/modeling_tf_{lowercase_model_name}.py'       ,      )

							shutil.move(
							    F'{directory}/test_modeling_tf_{lowercase_model_name}.py'       ,      F'{path_to_transformer_root}/tests/models/{lowercase_model_name}/test_modeling_tf_{lowercase_model_name}.py'       ,      )
				else:
							os.remove(F'{directory}/modeling_tf_{lowercase_model_name}.py'							)
							os.remove(F'{directory}/test_modeling_tf_{lowercase_model_name}.py'							)

				if output_flax:
							if not self._testing:
										remove_copy_lines(F'{directory}/modeling_flax_{lowercase_model_name}.py'							)

							shutil.move(
							    F'{directory}/modeling_flax_{lowercase_model_name}.py'       ,      F'{model_dir}/modeling_flax_{lowercase_model_name}.py'       ,      )

							shutil.move(
							    F'{directory}/test_modeling_flax_{lowercase_model_name}.py'       ,      F'{path_to_transformer_root}/tests/models/{lowercase_model_name}/test_modeling_flax_{lowercase_model_name}.py'       ,      )
				else:
							os.remove(F'{directory}/modeling_flax_{lowercase_model_name}.py'							)
							os.remove(F'{directory}/test_modeling_flax_{lowercase_model_name}.py'							)

				shutil.move(
				    F'{directory}/{lowercase_model_name}.md'       ,      F'{path_to_transformer_root}/docs/source/en/model_doc/{lowercase_model_name}.md'       ,      )

				shutil.move(
				    F'{directory}/tokenization_{lowercase_model_name}.py'       ,      F'{model_dir}/tokenization_{lowercase_model_name}.py'       ,      )

				shutil.move(
				    F'{directory}/tokenization_fast_{lowercase_model_name}.py'       ,      F'{model_dir}/tokenization_{lowercase_model_name}_fast.py'       ,      )

				from os import fdopen, remove
				from shutil import copymode, move
				from tempfile import mkstemp

				def replace(lowercase       ,      lowercase       ,      lowercase							):
							# Create temp file
							A_	:						Any	=     mkstemp()
							A_	:						int	=     False
							with fdopen(_a       ,      """w"""							) as new_file:
										with open(_a							) as old_file:
													for line in old_file:
																new_file.write(_a							)
																if line_to_copy_below in line:
																			A_	:						int	=     True
																			for line_to_copy in lines_to_copy:
																						new_file.write(_a							)

							if not line_found:
										raise ValueError(F'Line {line_to_copy_below} was not found in file.'							)

							# Copy the file permissions from the old file to the new file
							copymode(_a       ,      _a							)
							# Remove original file
							remove(_a							)
							# Move new file
							move(_a       ,      _a							)

				def skip_units(lowercase							):
							return (
							    ("generating PyTorch" in line and not output_pytorch)
							    or ("generating TensorFlow" in line and not output_tensorflow)
							    or ("generating Flax" in line and not output_flax)
							)

				def replace_in_files(lowercase							):
							with open(_a							) as datafile:
										A_	:						int	=     []
										A_	:						Optional[Any]	=     False
										A_	:						Optional[int]	=     False
										for line in datafile:
													if "# To replace in: " in line and "##" not in line:
																A_	:						Any	=     line.split("""\""""							)[1]
																A_	:						List[str]	=     skip_units(_a							)
													elif "# Below: " in line and "##" not in line:
																A_	:						str	=     line.split("""\""""							)[1]
																A_	:						Any	=     skip_units(_a							)
													elif "# End." in line and "##" not in line:
																if not skip_file and not skip_snippet:
																			replace(_a       ,      _a       ,      _a							)

																A_	:						str	=     []
													elif "# Replace with" in line and "##" not in line:
																A_	:						int	=     []
													elif "##" not in line:
																lines_to_copy.append(_a							)

							remove(_a							)

				replace_in_files(F'{directory}/to_replace_{lowercase_model_name}.py'							)
				os.rmdir(_a							)




'''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_     (							_lowerCAmelCase       ,_lowerCAmelCase       )      ->      Dict:
  __lowerCamelCase       :       Any           = checkpoint

  __lowerCamelCase       :       List[str]           = {}

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

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

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

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

  # Retrieves the keys for the decoder up blocks only
  __lowerCamelCase       :       Dict           = 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(_lowerCAmelCase       )
  }

  for i in range(_lowerCAmelCase       ):
    __lowerCamelCase       :       List[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:
      __lowerCamelCase       :       Any           = vae_state_dict.pop(
          F'encoder.down.{i}.downsample.conv.weight'       )
      __lowerCamelCase       :       Any           = vae_state_dict.pop(
          F'encoder.down.{i}.downsample.conv.bias'       )

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

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

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

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

  for i in range(_lowerCAmelCase       ):
    __lowerCamelCase       :       Optional[Any]           = 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[int]           = vae_state_dict[
          F'decoder.up.{block_id}.upsample.conv.weight'
      ]
      __lowerCamelCase       :       Optional[Any]           = vae_state_dict[
          F'decoder.up.{block_id}.upsample.conv.bias'
      ]

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

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

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

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

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

  __lowerCamelCase       :       Any           = OmegaConf.load(_lowerCAmelCase       )
  __lowerCamelCase       :       List[Any]           = 512
  __lowerCamelCase       :       Dict           = 'cuda' if torch.cuda.is_available() else 'cpu'
  if checkpoint_path.endswith('safetensors'       ):
    from safetensors import safe_open

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

  # Convert the VAE model.
  __lowerCamelCase       :       int           = create_vae_diffusers_config(_lowerCAmelCase       ,image_size=_lowerCAmelCase       )
  __lowerCamelCase       :       int           = custom_convert_ldm_vae_checkpoint(_lowerCAmelCase       ,_lowerCAmelCase       )

  __lowerCamelCase       :       Any           = AutoencoderKL(**_lowerCAmelCase       )
  vae.load_state_dict(_lowerCAmelCase       )
  vae.save_pretrained(_lowerCAmelCase       )


if __name__ == "__main__":
   _UpperCamelCase											=						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											=						parser.parse_args()

   vae_pt_to_vae_diffuser(args.vae_pt_path, args.dump_path)






