



"""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       __magic_name__    (   _lowerCamelCase				: Union[str, Any]  ,       _lowerCamelCase				: List[Any]					):
							__a		:		str									=       checkpoint

							__a		:		Any									=       {}

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

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

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

							# Retrieves the keys for the encoder down blocks only
							__a		:		Optional[Any]									=       len({""".""".join(layer.split("""."""					)[:3]					) for layer in vae_state_dict if """encoder.down""" in layer}					)
							__a		:		Any									=       {
							    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
							__a		:		Tuple									=       len({""".""".join(layer.split("""."""					)[:3]					) for layer in vae_state_dict if """decoder.up""" in layer}					)
							__a		:		Optional[Any]									=       {
							    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					):
														__a		:		Union[str, 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:
																					__a		:		Union[str, 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		:		List[str]									=       renew_vae_resnet_paths(_lowerCamelCase					)
														__a		:		int									=       {"""old""": F'''down.{i}.block''', """new""": F'''down_blocks.{i}.resnets'''}
														assign_to_checkpoint(_lowerCamelCase  ,       _lowerCamelCase  ,       _lowerCamelCase  ,       additional_replacements=[meta_path]  ,       config=_lowerCamelCase					)

							__a		:		str									=       [key for key in vae_state_dict if """encoder.mid.block""" in key]
							__a		:		List[str]									=       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		:		Tuple									=       renew_vae_resnet_paths(_lowerCamelCase					)
														__a		:		List[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					)

							__a		:		List[str]									=       [key for key in vae_state_dict if """encoder.mid.attn""" in key]
							__a		:		Dict									=       renew_vae_attention_paths(_lowerCamelCase					)
							__a		:		List[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					):
														__a		:		str									=       num_up_blocks - 1 - i
														__a		:		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		:		Any									=       vae_state_dict[
																					    F'''decoder.up.{block_id}.upsample.conv.weight'''
																					]
																					__a		:		Any									=       vae_state_dict[
																					    F'''decoder.up.{block_id}.upsample.conv.bias'''
																					]

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

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

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

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

							__a		:		Dict									=       OmegaConf.load(_lowerCamelCase					)
							__a		:		Optional[Any]									=       5_1_2
							__a		:		List[Any]									=       """cuda""" if torch.cuda.is_available() else """cpu"""
							if checkpoint_path.endswith("""safetensors"""					):
														from safetensors import safe_open

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

							# Convert the VAE model.
							__a		:		Optional[int]									=       create_vae_diffusers_config(_lowerCamelCase  ,       image_size=_lowerCamelCase					)
							__a		:		Dict									=       custom_convert_ldm_vae_checkpoint(_lowerCamelCase  ,       _lowerCamelCase					)

							__a		:		Optional[Any]									=       AutoencoderKL(**_lowerCamelCase					)
							vae.load_state_dict(_lowerCamelCase					)
							vae.save_pretrained(_lowerCamelCase					)


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"""

from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_tensorflow_text_available, is_torch_available


lowercase__			   =  {
    "configuration_ernie": ["ERNIE_PRETRAINED_CONFIG_ARCHIVE_MAP", "ErnieConfig", "ErnieOnnxConfig"],
}

try:
	if not is_torch_available():
		raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
	pass
else:
	lowercase__			   =  [
	    "ERNIE_PRETRAINED_MODEL_ARCHIVE_LIST",
	    "ErnieForCausalLM",
	    "ErnieForMaskedLM",
	    "ErnieForMultipleChoice",
	    "ErnieForNextSentencePrediction",
	    "ErnieForPreTraining",
	    "ErnieForQuestionAnswering",
	    "ErnieForSequenceClassification",
	    "ErnieForTokenClassification",
	    "ErnieModel",
	    "ErniePreTrainedModel",
	]

if TYPE_CHECKING:
	from .configuration_ernie import ERNIE_PRETRAINED_CONFIG_ARCHIVE_MAP, ErnieConfig, ErnieOnnxConfig

	try:
		if not is_torch_available():
			raise OptionalDependencyNotAvailable()
	except OptionalDependencyNotAvailable:
		pass
	else:
		from .modeling_ernie import (
		    ERNIE_PRETRAINED_MODEL_ARCHIVE_LIST,
		    ErnieForCausalLM,
		    ErnieForMaskedLM,
		    ErnieForMultipleChoice,
		    ErnieForNextSentencePrediction,
		    ErnieForPreTraining,
		    ErnieForQuestionAnswering,
		    ErnieForSequenceClassification,
		    ErnieForTokenClassification,
		    ErnieModel,
		    ErniePreTrainedModel,
		)

else:
	import sys

	lowercase__			   =  _LazyModule(__name__, globals()["__file__"], _import_structure, module_spec=__spec__)

