


'''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       a				(      __a			, __a       )  ->     int:



					'''simple docstring'''






					UpperCamelCase__       :Optional[Any]							 =			checkpoint

					UpperCamelCase__       :Optional[int]							 =			{}

					UpperCamelCase__       :Any							 =			vae_state_dict['''encoder.conv_in.weight''']
					UpperCamelCase__       :Union[str, Any]							 =			vae_state_dict['''encoder.conv_in.bias''']
					UpperCamelCase__       :List[Any]							 =			vae_state_dict['''encoder.conv_out.weight''']
					UpperCamelCase__       :List[str]							 =			vae_state_dict['''encoder.conv_out.bias''']
					UpperCamelCase__       :Optional[Any]							 =			vae_state_dict['''encoder.norm_out.weight''']
					UpperCamelCase__       :Any							 =			vae_state_dict['''encoder.norm_out.bias''']

					UpperCamelCase__       :Any							 =			vae_state_dict['''decoder.conv_in.weight''']
					UpperCamelCase__       :List[Any]							 =			vae_state_dict['''decoder.conv_in.bias''']
					UpperCamelCase__       :Union[str, Any]							 =			vae_state_dict['''decoder.conv_out.weight''']
					UpperCamelCase__       :Union[str, Any]							 =			vae_state_dict['''decoder.conv_out.bias''']
					UpperCamelCase__       :Any							 =			vae_state_dict['''decoder.norm_out.weight''']
					UpperCamelCase__       :List[str]							 =			vae_state_dict['''decoder.norm_out.bias''']

					UpperCamelCase__       :Tuple							 =			vae_state_dict['''quant_conv.weight''']
					UpperCamelCase__       :Optional[int]							 =			vae_state_dict['''quant_conv.bias''']
					UpperCamelCase__       :List[str]							 =			vae_state_dict['''post_quant_conv.weight''']
					UpperCamelCase__       :int							 =			vae_state_dict['''post_quant_conv.bias''']

					# Retrieves the keys for the encoder down blocks only
					UpperCamelCase__       :List[Any]							 =			len({'''.'''.join(layer.split('''.'''       )[:3]       ) for layer in vae_state_dict if '''encoder.down''' in layer}       )
					UpperCamelCase__       :int							 =			{
					    layer_id: [key for key in vae_state_dict if f'''down.{layer_id}''' in key] for layer_id in range(__a       )
					}

					# Retrieves the keys for the decoder up blocks only
					UpperCamelCase__       :Tuple							 =			len({'''.'''.join(layer.split('''.'''       )[:3]       ) for layer in vae_state_dict if '''decoder.up''' in layer}       )
					UpperCamelCase__       :Optional[Any]							 =			{
					    layer_id: [key for key in vae_state_dict if f'''up.{layer_id}''' in key] for layer_id in range(__a       )
					}

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

										UpperCamelCase__       :Optional[Any]							 =			renew_vae_resnet_paths(__a       )
										UpperCamelCase__       :Union[str, Any]							 =			{'''old''': f'''down.{i}.block''', '''new''': f'''down_blocks.{i}.resnets'''}
										assign_to_checkpoint(__a			, __a			, __a			, additional_replacements=[meta_path]			, config=__a       )

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

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

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

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

										UpperCamelCase__       :Union[str, Any]							 =			renew_vae_resnet_paths(__a       )
										UpperCamelCase__       :Union[str, Any]							 =			{'''old''': f'''up.{block_id}.block''', '''new''': f'''up_blocks.{i}.resnets'''}
										assign_to_checkpoint(__a			, __a			, __a			, additional_replacements=[meta_path]			, config=__a       )

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

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

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


def       a				(      __a			, __a			, )  ->     List[str]:



					'''simple docstring'''






					UpperCamelCase__       :Optional[int]							 =			requests.get(
					    ''' https://raw.githubusercontent.com/CompVis/stable-diffusion/main/configs/stable-diffusion/v1-inference.yaml'''       )
					UpperCamelCase__       :Union[str, Any]							 =			io.BytesIO(r.content       )

					UpperCamelCase__       :Any							 =			OmegaConf.load(__a       )
					UpperCamelCase__       :List[Any]							 =			512
					UpperCamelCase__       :int							 =			'''cuda''' if torch.cuda.is_available() else '''cpu'''
					if checkpoint_path.endswith('''safetensors'''       ):
										from safetensors import safe_open

										UpperCamelCase__       :Union[str, Any]							 =			{}
										with safe_open(__a			, framework='''pt'''			, device='''cpu'''       ) as f:
															for key in f.keys():
																				UpperCamelCase__       :List[Any]							 =			f.get_tensor(__a       )
					else:
										UpperCamelCase__       :List[str]							 =			torch.load(__a			, map_location=__a       )['''state_dict''']

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

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


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

		__snake_case											=					parser.parse_args()

		vae_pt_to_vae_diffuser(args.vae_pt_path, args.dump_path)






'''simple docstring'''
from math import ceil


def       a				(      __a			, __a       )  ->     Any:



					'''simple docstring'''






					UpperCamelCase__       :str							 =			list(range(0			, __a       )       )

					UpperCamelCase__       :Optional[int]							 =			[item for sublist in list(device_map.values()       ) for item in sublist]

					# Duplicate check
					UpperCamelCase__       :Optional[int]							 =			[]
					for i in device_map_blocks:
										if device_map_blocks.count(__a       ) > 1 and i not in duplicate_blocks:
															duplicate_blocks.append(__a       )
    # Missing blocks
					UpperCamelCase__       :List[str]							 =			[i for i in blocks if i not in device_map_blocks]
					UpperCamelCase__       :Optional[Any]							 =			[i for i in device_map_blocks if i not in blocks]

					if len(__a       ) != 0:
										raise ValueError(
										    '''Duplicate attention blocks specified in device_map. Attention blocks must be specified to one device.'''
										    ''' These attention blocks were specified more than once: ''' + str(__a       )       )
					if len(__a       ) != 0:
										raise ValueError(
										    '''There are attention blocks for this model that are not specified in the device_map. Add these attention '''
										    '''blocks to a device on the device_map: ''' + str(__a       )       )
					if len(__a       ) != 0:
										raise ValueError(
										    '''The device_map contains more attention blocks than this model has. Remove these from the device_map:'''
										    + str(__a       )       )







def       a				(      __a			, __a       )  ->     Tuple:



					'''simple docstring'''






					UpperCamelCase__       :Optional[Any]							 =			list(range(__a       )       )
					UpperCamelCase__       :Any							 =			int(ceil(n_layers / len(__a       )       )       )
					UpperCamelCase__       :List[Any]							 =			[layers[i : i + n_blocks] for i in range(0			, __a			, __a       )]

					return dict(zip(__a			, __a       )       )