
"""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       _snake_case				(		snake_case__   :     Optional[int]	,      snake_case__   :     List[Any]     ):
	A						   =   checkpoint

	A						   =   {}

	A						   =   vae_state_dict['encoder.conv_in.weight']
	A						   =   vae_state_dict['encoder.conv_in.bias']
	A						   =   vae_state_dict['encoder.conv_out.weight']
	A						   =   vae_state_dict['encoder.conv_out.bias']
	A						   =   vae_state_dict['encoder.norm_out.weight']
	A						   =   vae_state_dict['encoder.norm_out.bias']

	A						   =   vae_state_dict['decoder.conv_in.weight']
	A						   =   vae_state_dict['decoder.conv_in.bias']
	A						   =   vae_state_dict['decoder.conv_out.weight']
	A						   =   vae_state_dict['decoder.conv_out.bias']
	A						   =   vae_state_dict['decoder.norm_out.weight']
	A						   =   vae_state_dict['decoder.norm_out.bias']

	A						   =   vae_state_dict['quant_conv.weight']
	A						   =   vae_state_dict['quant_conv.bias']
	A						   =   vae_state_dict['post_quant_conv.weight']
	A						   =   vae_state_dict['post_quant_conv.bias']

	# Retrieves the keys for the encoder down blocks only
	A						   =   len({'.'.join(layer.split('.'     )[:3]     ) for layer in vae_state_dict if 'encoder.down' in layer}     )
	A						   =   {
	    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
	A						   =   len({'.'.join(layer.split('.'     )[:3]     ) for layer in vae_state_dict if 'decoder.up' in layer}     )
	A						   =   {
	    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__     ):
		A						   =   [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						   =   vae_state_dict.pop(
			    F'encoder.down.{i}.downsample.conv.weight'     )
			A						   =   vae_state_dict.pop(
			    F'encoder.down.{i}.downsample.conv.bias'     )

		A						   =   renew_vae_resnet_paths(snake_case__     )
		A						   =   {'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__     )

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

		A						   =   renew_vae_resnet_paths(snake_case__     )
		A						   =   {'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__     )

	A						   =   [key for key in vae_state_dict if 'encoder.mid.attn' in key]
	A						   =   renew_vae_attention_paths(snake_case__     )
	A						   =   {'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__     ):
		A						   =   num_up_blocks - 1 - i
		A						   =   [
		    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						   =   vae_state_dict[
			    F'decoder.up.{block_id}.upsample.conv.weight'
			]
			A						   =   vae_state_dict[
			    F'decoder.up.{block_id}.upsample.conv.bias'
			]

		A						   =   renew_vae_resnet_paths(snake_case__     )
		A						   =   {'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__     )

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

		A						   =   renew_vae_resnet_paths(snake_case__     )
		A						   =   {'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__     )

	A						   =   [key for key in vae_state_dict if 'decoder.mid.attn' in key]
	A						   =   renew_vae_attention_paths(snake_case__     )
	A						   =   {'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       _snake_case				(		snake_case__   :     str	,      snake_case__   :     str	,      ):
	# Only support V1
	A						   =   requests.get(
	    ' https://raw.githubusercontent.com/CompVis/stable-diffusion/main/configs/stable-diffusion/v1-inference.yaml'     )
	A						   =   io.BytesIO(r.content     )

	A						   =   OmegaConf.load(snake_case__     )
	A						   =   512
	A						   =   'cuda' if torch.cuda.is_available() else 'cpu'
	if checkpoint_path.endswith('safetensors'     ):
		from safetensors import safe_open

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

	# Convert the VAE model.
	A						   =   create_vae_diffusers_config(snake_case__	,      image_size=snake_case__     )
	A						   =   custom_convert_ldm_vae_checkpoint(snake_case__	,      snake_case__     )

	A						   =   AutoencoderKL(**snake_case__     )
	vae.load_state_dict(snake_case__     )
	vae.save_pretrained(snake_case__     )


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 argparse import ArgumentParser

from . import BaseTransformersCLICommand






def       _snake_case				(		snake_case__   :     Optional[int]     ):
	return DownloadCommand(args.model	,      args.cache_dir	,      args.force	,      args.trust_remote_code     )







class 				lowerCAmelCase_				(      _lowercase    ):



						'''simple docstring'''


						@staticmethod
						def 						_SCREAMING_SNAKE_CASE      ( A_     :      ArgumentParser      )     ->   Any:
							A						   =   parser.add_parser('download'      )
							download_parser.add_argument(
							    '--cache-dir'  ,type=A_  ,default=A_  ,help='Path to location to store the models'      )
							download_parser.add_argument(
							    '--force'  ,action='store_true'  ,help='Force the model to be download even if already in cache-dir'      )
							download_parser.add_argument(
							    '--trust-remote-code'  ,action='store_true'  ,help='Whether or not to allow for custom models defined on the Hub in their own modeling files. Use only if you\'ve reviewed the code as it will execute on your local machine'  ,)
							download_parser.add_argument('model'  ,type=A_  ,help='Name of the model to download'      )
							download_parser.set_defaults(func=A_      )


						def __init__( self     :      Dict  ,A_     :      str  ,A_     :      str  ,A_     :      bool  ,A_     :      bool      )     ->   Union[str, Any]:
							A						   =   model
							A						   =   cache
							A						   =   force
							A						   =   trust_remote_code




						def 						_SCREAMING_SNAKE_CASE      ( self     :      Dict      )     ->   Optional[int]:
							from ..models.auto import AutoModel, AutoTokenizer

							AutoModel.from_pretrained(
							    self._model  ,cache_dir=self._cache  ,force_download=self._force  ,trust_remote_code=self._trust_remote_code      )
							AutoTokenizer.from_pretrained(
							    self._model  ,cache_dir=self._cache  ,force_download=self._force  ,trust_remote_code=self._trust_remote_code      )