

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_      (    _UpperCAmelCase				,	_UpperCAmelCase    ):
						SCREAMING_SNAKE_CASE_: str  		=       checkpoint

						SCREAMING_SNAKE_CASE_: Union[str, Any]  		=       {}

						SCREAMING_SNAKE_CASE_: Any  		=       vae_state_dict["encoder.conv_in.weight"]
						SCREAMING_SNAKE_CASE_: Tuple  		=       vae_state_dict["encoder.conv_in.bias"]
						SCREAMING_SNAKE_CASE_: Dict  		=       vae_state_dict["encoder.conv_out.weight"]
						SCREAMING_SNAKE_CASE_: Optional[Any]  		=       vae_state_dict["encoder.conv_out.bias"]
						SCREAMING_SNAKE_CASE_: Any  		=       vae_state_dict["encoder.norm_out.weight"]
						SCREAMING_SNAKE_CASE_: Dict  		=       vae_state_dict["encoder.norm_out.bias"]

						SCREAMING_SNAKE_CASE_: Tuple  		=       vae_state_dict["decoder.conv_in.weight"]
						SCREAMING_SNAKE_CASE_: Union[str, Any]  		=       vae_state_dict["decoder.conv_in.bias"]
						SCREAMING_SNAKE_CASE_: int  		=       vae_state_dict["decoder.conv_out.weight"]
						SCREAMING_SNAKE_CASE_: Optional[int]  		=       vae_state_dict["decoder.conv_out.bias"]
						SCREAMING_SNAKE_CASE_: Any  		=       vae_state_dict["decoder.norm_out.weight"]
						SCREAMING_SNAKE_CASE_: Optional[int]  		=       vae_state_dict["decoder.norm_out.bias"]

						SCREAMING_SNAKE_CASE_: List[Any]  		=       vae_state_dict["quant_conv.weight"]
						SCREAMING_SNAKE_CASE_: int  		=       vae_state_dict["quant_conv.bias"]
						SCREAMING_SNAKE_CASE_: List[Any]  		=       vae_state_dict["post_quant_conv.weight"]
						SCREAMING_SNAKE_CASE_: Optional[int]  		=       vae_state_dict["post_quant_conv.bias"]

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

												SCREAMING_SNAKE_CASE_: Dict  		=       renew_vae_resnet_paths(_UpperCAmelCase    )
												SCREAMING_SNAKE_CASE_: str  		=       {"old": f"down.{i}.block", "new": f"down_blocks.{i}.resnets"}
												assign_to_checkpoint(_UpperCAmelCase				,	_UpperCAmelCase				,	_UpperCAmelCase				,	additional_replacements=[meta_path]				,	config=_UpperCAmelCase    )

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

												SCREAMING_SNAKE_CASE_: List[Any]  		=       renew_vae_resnet_paths(_UpperCAmelCase    )
												SCREAMING_SNAKE_CASE_: int  		=       {"old": f"mid.block_{i}", "new": f"mid_block.resnets.{i - 1}"}
												assign_to_checkpoint(_UpperCAmelCase				,	_UpperCAmelCase				,	_UpperCAmelCase				,	additional_replacements=[meta_path]				,	config=_UpperCAmelCase    )

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

												SCREAMING_SNAKE_CASE_: Optional[Any]  		=       renew_vae_resnet_paths(_UpperCAmelCase    )
												SCREAMING_SNAKE_CASE_: List[str]  		=       {"old": f"up.{block_id}.block", "new": f"up_blocks.{i}.resnets"}
												assign_to_checkpoint(_UpperCAmelCase				,	_UpperCAmelCase				,	_UpperCAmelCase				,	additional_replacements=[meta_path]				,	config=_UpperCAmelCase    )

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

												SCREAMING_SNAKE_CASE_: Any  		=       renew_vae_resnet_paths(_UpperCAmelCase    )
												SCREAMING_SNAKE_CASE_: Optional[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    )

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

						SCREAMING_SNAKE_CASE_: Any  		=       OmegaConf.load(_UpperCAmelCase    )
						SCREAMING_SNAKE_CASE_: Optional[Any]  		=       5_12
						SCREAMING_SNAKE_CASE_: Tuple  		=       "cuda" if torch.cuda.is_available() else "cpu"
						if checkpoint_path.endswith("safetensors"    ):
												from safetensors import safe_open

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

						# Convert the VAE model.
						SCREAMING_SNAKE_CASE_: Optional[int]  		=       create_vae_diffusers_config(_UpperCAmelCase				,	image_size=_UpperCAmelCase    )
						SCREAMING_SNAKE_CASE_: Optional[int]  		=       custom_convert_ldm_vae_checkpoint(_UpperCAmelCase				,	_UpperCAmelCase    )

						SCREAMING_SNAKE_CASE_: int  		=       AutoencoderKL(**_UpperCAmelCase    )
						vae.load_state_dict(_UpperCAmelCase    )
						vae.save_pretrained(_UpperCAmelCase    )


if __name__ == "__main__":
						lowerCAmelCase							:  Optional[Any]   	=						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.""")

						lowerCAmelCase							:  Optional[int]   	=						parser.parse_args()

						vae_pt_to_vae_diffuser(args.vae_pt_path, args.dump_path)








'''simple docstring'''







def 					snake_case_		(				__SCREAMING_SNAKE_CASE					: int							):



						"""simple docstring"""





						lowercase_	:     Optional[int]													=       int(__SCREAMING_SNAKE_CASE							)
						if decimal in (0, 1):  # Exit cases for the recursion
												return str(__SCREAMING_SNAKE_CASE							)
						lowercase_						, lowercase_	:     List[str]													=       divmod(__SCREAMING_SNAKE_CASE      ,      2							)
						return binary_recursive(__SCREAMING_SNAKE_CASE							) + str(__SCREAMING_SNAKE_CASE							)
def 					snake_case_		(				__SCREAMING_SNAKE_CASE					: str							):



						"""simple docstring"""





						lowercase_	:     str													=       str(__SCREAMING_SNAKE_CASE							).strip()
						if not number:
												raise ValueError('''No input value was provided'''							)
						lowercase_	:     Optional[int]													=       '''-''' if number.startswith('''-'''							) else ''''''
						lowercase_	:     Union[str, Any]													=       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()



