



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 snake_case__  (			__SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE )		->							Optional[Any]:
		UpperCAmelCase_						=				checkpoint

		UpperCAmelCase_						=				{}

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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







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

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

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

		# Convert the VAE model.
		UpperCAmelCase_						=				create_vae_diffusers_config(__SCREAMING_SNAKE_CASE			,     image_size=__SCREAMING_SNAKE_CASE )
		UpperCAmelCase_						=				custom_convert_ldm_vae_checkpoint(__SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE )

		UpperCAmelCase_						=				AutoencoderKL(**__SCREAMING_SNAKE_CASE )
		vae.load_state_dict(__SCREAMING_SNAKE_CASE )
		vae.save_pretrained(__SCREAMING_SNAKE_CASE )


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

			SCREAMING_SNAKE_CASE     				=						parser.parse_args()

			vae_pt_to_vae_diffuser(args.vae_pt_path, args.dump_path)







def snake_case__  (			__SCREAMING_SNAKE_CASE )		->							str:
		UpperCAmelCase_						=				int(__SCREAMING_SNAKE_CASE )
		if decimal in (0, 1):  # Exit cases for the recursion
				return str(__SCREAMING_SNAKE_CASE )
		UpperCAmelCase_					,						UpperCAmelCase_						=				divmod(__SCREAMING_SNAKE_CASE			,     2 )
		return binary_recursive(__SCREAMING_SNAKE_CASE ) + str(__SCREAMING_SNAKE_CASE )







def snake_case__  (			__SCREAMING_SNAKE_CASE )		->							str:
		UpperCAmelCase_						=				str(__SCREAMING_SNAKE_CASE ).strip()
		if not number:
				raise ValueError("No input value was provided" )
		UpperCAmelCase_						=				"-" if number.startswith("-" ) else ""
		UpperCAmelCase_						=				number.lstrip("-" )
		if not number.isnumeric():
				raise ValueError("Input value is not an integer" )
		return f'''{negative}0b{binary_recursive(int(__SCREAMING_SNAKE_CASE ) )}'''


if __name__ == "__main__":
			from doctest import testmod

			testmod()

