



"""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				: Any  ,       _lowerCamelCase				: str					):
							__a		:		Dict									=       checkpoint

							__a		:		int									=       {}

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

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

							__a		:		Union[str, Any]									=       vae_state_dict["""quant_conv.weight"""]
							__a		:		Dict									=       vae_state_dict["""quant_conv.bias"""]
							__a		:		List[Any]									=       vae_state_dict["""post_quant_conv.weight"""]
							__a		:		Optional[int]									=       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		:		Optional[Any]									=       {
							    layer_id: [key for key in vae_state_dict if F'''down.{layer_id}''' in key] for layer_id in range(__UpperCamelCase					)
							}

							# 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		:		Dict									=       {
							    layer_id: [key for key in vae_state_dict if F'''up.{layer_id}''' in key] for layer_id in range(__UpperCamelCase					)
							}

							for i in range(__UpperCamelCase					):
														__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		:		str									=       vae_state_dict.pop(
																					    F'''encoder.down.{i}.downsample.conv.bias'''					)

														__a		:		int									=       renew_vae_resnet_paths(__UpperCamelCase					)
														__a		:		Dict									=       {"""old""": F'''down.{i}.block''', """new""": F'''down_blocks.{i}.resnets'''}
														assign_to_checkpoint(__UpperCamelCase  ,       __UpperCamelCase  ,       __UpperCamelCase  ,       additional_replacements=[meta_path]  ,       config=__UpperCamelCase					)

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

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

							__a		:		Union[str, Any]									=       [key for key in vae_state_dict if """encoder.mid.attn""" in key]
							__a		:		Dict									=       renew_vae_attention_paths(__UpperCamelCase					)
							__a		:		Tuple									=       {"""old""": """mid.attn_1""", """new""": """mid_block.attentions.0"""}
							assign_to_checkpoint(__UpperCamelCase  ,       __UpperCamelCase  ,       __UpperCamelCase  ,       additional_replacements=[meta_path]  ,       config=__UpperCamelCase					)
							conv_attn_to_linear(__UpperCamelCase					)

							for i in range(__UpperCamelCase					):
														__a		:		Union[str, Any]									=       num_up_blocks - 1 - i
														__a		:		Optional[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:
																					__a		:		Dict									=       vae_state_dict[
																					    F'''decoder.up.{block_id}.upsample.conv.weight'''
																					]
																					__a		:		Dict									=       vae_state_dict[
																					    F'''decoder.up.{block_id}.upsample.conv.bias'''
																					]

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

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

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

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





def       __magic_name__    (   _lowerCamelCase				: Dict  ,       _lowerCamelCase				: Dict  ,       ):
							# Only support V1
							__a		:		List[Any]									=       requests.get(
							    """ https://raw.githubusercontent.com/CompVis/stable-diffusion/main/configs/stable-diffusion/v1-inference.yaml"""					)
							__a		:		Optional[Any]									=       io.BytesIO(r.content					)

							__a		:		List[str]									=       OmegaConf.load(__UpperCamelCase					)
							__a		:		Tuple									=       5_1_2
							__a		:		Optional[int]									=       """cuda""" if torch.cuda.is_available() else """cpu"""
							if checkpoint_path.endswith("""safetensors"""					):
														from safetensors import safe_open

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

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

							__a		:		Dict									=       AutoencoderKL(**__UpperCamelCase					)
							vae.load_state_dict(__UpperCamelCase					)
							vae.save_pretrained(__UpperCamelCase					)


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_sentencepiece_available,
    is_tokenizers_available,
    is_torch_available,
)


lowercase__			   =  {
    "configuration_llama": ["LLAMA_PRETRAINED_CONFIG_ARCHIVE_MAP", "LlamaConfig"],
}

try:
	if not is_sentencepiece_available():
		raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
	pass
else:
	lowercase__			   =  ["LlamaTokenizer"]

try:
	if not is_tokenizers_available():
		raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
	pass
else:
	lowercase__			   =  ["LlamaTokenizerFast"]

try:
	if not is_torch_available():
		raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
	pass
else:
	lowercase__			   =  [
	    "LlamaForCausalLM",
	    "LlamaModel",
	    "LlamaPreTrainedModel",
	    "LlamaForSequenceClassification",
	]


if TYPE_CHECKING:
	from .configuration_llama import LLAMA_PRETRAINED_CONFIG_ARCHIVE_MAP, LlamaConfig

	try:
		if not is_sentencepiece_available():
			raise OptionalDependencyNotAvailable()
	except OptionalDependencyNotAvailable:
		pass
	else:
		from .tokenization_llama import LlamaTokenizer

	try:
		if not is_tokenizers_available():
			raise OptionalDependencyNotAvailable()
	except OptionalDependencyNotAvailable:
		pass
	else:
		from .tokenization_llama_fast import LlamaTokenizerFast

	try:
		if not is_torch_available():
			raise OptionalDependencyNotAvailable()
	except OptionalDependencyNotAvailable:
		pass
	else:
		from .modeling_llama import LlamaForCausalLM, LlamaForSequenceClassification, LlamaModel, LlamaPreTrainedModel


else:
	import sys

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

