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

						_lowerCAmelCase  						=			{}

						_lowerCAmelCase  						=			vae_state_dict["""encoder.conv_in.weight"""]
						_lowerCAmelCase  						=			vae_state_dict["""encoder.conv_in.bias"""]
						_lowerCAmelCase  						=			vae_state_dict["""encoder.conv_out.weight"""]
						_lowerCAmelCase  						=			vae_state_dict["""encoder.conv_out.bias"""]
						_lowerCAmelCase  						=			vae_state_dict["""encoder.norm_out.weight"""]
						_lowerCAmelCase  						=			vae_state_dict["""encoder.norm_out.bias"""]

						_lowerCAmelCase  						=			vae_state_dict["""decoder.conv_in.weight"""]
						_lowerCAmelCase  						=			vae_state_dict["""decoder.conv_in.bias"""]
						_lowerCAmelCase  						=			vae_state_dict["""decoder.conv_out.weight"""]
						_lowerCAmelCase  						=			vae_state_dict["""decoder.conv_out.bias"""]
						_lowerCAmelCase  						=			vae_state_dict["""decoder.norm_out.weight"""]
						_lowerCAmelCase  						=			vae_state_dict["""decoder.norm_out.bias"""]

						_lowerCAmelCase  						=			vae_state_dict["""quant_conv.weight"""]
						_lowerCAmelCase  						=			vae_state_dict["""quant_conv.bias"""]
						_lowerCAmelCase  						=			vae_state_dict["""post_quant_conv.weight"""]
						_lowerCAmelCase  						=			vae_state_dict["""post_quant_conv.bias"""]

						# Retrieves the keys for the encoder down blocks only
						_lowerCAmelCase  						=			len({""".""".join(layer.split("""."""		)[:3]		) for layer in vae_state_dict if """encoder.down""" in layer}		)
						_lowerCAmelCase  						=			{
						    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
						_lowerCAmelCase  						=			len({""".""".join(layer.split("""."""		)[:3]		) for layer in vae_state_dict if """decoder.up""" in layer}		)
						_lowerCAmelCase  						=			{
						    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_		):
												_lowerCAmelCase  						=			[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  						=			vae_state_dict.pop(
																		    f'encoder.down.{i}.downsample.conv.weight'		)
																		_lowerCAmelCase  						=			vae_state_dict.pop(
																		    f'encoder.down.{i}.downsample.conv.bias'		)

												_lowerCAmelCase  						=			renew_vae_resnet_paths(lowercase_		)
												_lowerCAmelCase  						=			{"""old""": f'down.{i}.block', """new""": f'down_blocks.{i}.resnets'}
												assign_to_checkpoint(lowercase_			,     lowercase_			,     lowercase_			,     additional_replacements=[meta_path]			,     config=lowercase_		)

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

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

						_lowerCAmelCase  						=			[key for key in vae_state_dict if """encoder.mid.attn""" in key]
						_lowerCAmelCase  						=			renew_vae_attention_paths(lowercase_		)
						_lowerCAmelCase  						=			{"""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_		):
												_lowerCAmelCase  						=			num_up_blocks - 1 - i
												_lowerCAmelCase  						=			[
												    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  						=			vae_state_dict[
																		    f'decoder.up.{block_id}.upsample.conv.weight'
																		]
																		_lowerCAmelCase  						=			vae_state_dict[
																		    f'decoder.up.{block_id}.upsample.conv.bias'
																		]

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

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

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

						_lowerCAmelCase  						=			[key for key in vae_state_dict if """decoder.mid.attn""" in key]
						_lowerCAmelCase  						=			renew_vae_attention_paths(lowercase_		)
						_lowerCAmelCase  						=			{"""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 A				(__lowerCamelCase				:str			,     __lowerCamelCase				:str			,     ):
						# Only support V1
						_lowerCAmelCase  						=			requests.get(
						    """ https://raw.githubusercontent.com/CompVis/stable-diffusion/main/configs/stable-diffusion/v1-inference.yaml"""		)
						_lowerCAmelCase  						=			io.BytesIO(r.content		)

						_lowerCAmelCase  						=			OmegaConf.load(lowercase_		)
						_lowerCAmelCase  						=			512
						_lowerCAmelCase  						=			"""cuda""" if torch.cuda.is_available() else """cpu"""
						if checkpoint_path.endswith("""safetensors"""		):
												from safetensors import safe_open

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

						# Convert the VAE model.
						_lowerCAmelCase  						=			create_vae_diffusers_config(lowercase_			,     image_size=lowercase_		)
						_lowerCAmelCase  						=			custom_convert_ldm_vae_checkpoint(lowercase_			,     lowercase_		)

						_lowerCAmelCase  						=			AutoencoderKL(**lowercase_		)
						vae.load_state_dict(lowercase_		)
						vae.save_pretrained(lowercase_		)


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

						_lowercase       =		parser.parse_args()

						vae_pt_to_vae_diffuser(args.vae_pt_path, args.dump_path)


'''simple docstring'''



import inspect
import re

from transformers.utils import direct_transformers_import


# All paths are set with the intent you should run this script from the root of the repo with the command
# python utils/check_config_docstrings.py
_lowercase    =		"""src/transformers"""


# This is to make sure the transformers module imported is the one in the repo.
_lowercase    =		direct_transformers_import(PATH_TO_TRANSFORMERS)

_lowercase    =		transformers.models.auto.configuration_auto.CONFIG_MAPPING

# Regex pattern used to find the checkpoint mentioned in the docstring of `config_class`.
# For example, `[bert-base-uncased](https://huggingface.co/bert-base-uncased)`
_lowercase    =		re.compile(R"""\[(.+?)\]\((https://huggingface\.co/.+?)\)""")


_lowercase    =		{
    """DecisionTransformerConfig""",
    """EncoderDecoderConfig""",
    """MusicgenConfig""",
    """RagConfig""",
    """SpeechEncoderDecoderConfig""",
    """TimmBackboneConfig""",
    """VisionEncoderDecoderConfig""",
    """VisionTextDualEncoderConfig""",
    """LlamaConfig""",
}


def A				(__lowerCamelCase				:str		):
						_lowerCAmelCase  						=			None

						# source code of `config_class`
						_lowerCAmelCase  						=			inspect.getsource(__lowerCamelCase		)
						_lowerCAmelCase  						=			_re_checkpoint.findall(__lowerCamelCase		)

						# Each `checkpoint` is a tuple of a checkpoint name and a checkpoint link.
						# For example, `('bert-base-uncased', 'https://huggingface.co/bert-base-uncased')`
						for ckpt_name, ckpt_link in checkpoints:
												# allow the link to end with `/`
												if ckpt_link.endswith("""/"""		):
																		_lowerCAmelCase  						=			ckpt_link[:-1]

												# verify the checkpoint name corresponds to the checkpoint link
												_lowerCAmelCase  						=			f'https://huggingface.co/{ckpt_name}'
												if ckpt_link == ckpt_link_from_name:
																		_lowerCAmelCase  						=			ckpt_name
																		break

						return checkpoint


def A				():
						_lowerCAmelCase  						=			[]

						for config_class in list(CONFIG_MAPPING.values()		):
												# Skip deprecated models
												if "models.deprecated" in config_class.__module__:
																		continue
												_lowerCAmelCase  						=			get_checkpoint_from_config_class(__lowerCamelCase		)

												_lowerCAmelCase  						=			config_class.__name__
												if checkpoint is None and name not in CONFIG_CLASSES_TO_IGNORE_FOR_DOCSTRING_CHECKPOINT_CHECK:
																		configs_without_checkpoint.append(__lowerCamelCase		)

						if len(__lowerCamelCase		) > 0:
												_lowerCAmelCase  						=			"""\n""".join(sorted(__lowerCamelCase		)		)
												raise ValueError(f'The following configurations don\'t contain any valid checkpoint:\n{message}'		)


if __name__ == "__main__":
						check_config_docstrings_have_checkpoints()
