




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 			UpperCAmelCase_				(      UpperCAmelCase__						,							UpperCAmelCase__  ):
					lowercase_  							=			checkpoint

					lowercase_  							=			{}

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

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

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

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

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

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

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

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

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

					lowercase_  							=			[key for key in vae_state_dict if """encoder.mid.attn""" in key]
					lowercase_  							=			renew_vae_attention_paths(__snake_case  )
					lowercase_  							=			{"""old""": """mid.attn_1""", """new""": """mid_block.attentions.0"""}
					assign_to_checkpoint(__snake_case						,							__snake_case						,							__snake_case						,							additional_replacements=[meta_path]						,							config=__snake_case  )
					conv_attn_to_linear(__snake_case  )

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

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

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

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

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


def 			UpperCAmelCase_				(      UpperCAmelCase__						,							UpperCAmelCase__						,							):
					# Only support V1
					lowercase_  							=			requests.get(
					    """ https://raw.githubusercontent.com/CompVis/stable-diffusion/main/configs/stable-diffusion/v1-inference.yaml"""  )
					lowercase_  							=			io.BytesIO(r.content  )

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

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

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

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


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

							a					  =     parser.parse_args()

							vae_pt_to_vae_diffuser(args.vae_pt_path, args.dump_path)






def 			UpperCAmelCase_				(      UpperCAmelCase__  ):
					return " ".join(input_str.split()[::-1]  )


if __name__ == "__main__":
							import doctest

							doctest.testmod()


