



'''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 				lowerCAmelCase_   (	_lowerCamelCase:   Optional[Any]       ,      _lowerCamelCase:   Dict					):
		__SCREAMING_SNAKE_CASE       :		Dict           =		checkpoint

		__SCREAMING_SNAKE_CASE       :		List[str]           =		{}

		__SCREAMING_SNAKE_CASE       :		Any           =		vae_state_dict["""encoder.conv_in.weight"""]
		__SCREAMING_SNAKE_CASE       :		Optional[int]           =		vae_state_dict["""encoder.conv_in.bias"""]
		__SCREAMING_SNAKE_CASE       :		str           =		vae_state_dict["""encoder.conv_out.weight"""]
		__SCREAMING_SNAKE_CASE       :		Optional[int]           =		vae_state_dict["""encoder.conv_out.bias"""]
		__SCREAMING_SNAKE_CASE       :		Union[str, Any]           =		vae_state_dict["""encoder.norm_out.weight"""]
		__SCREAMING_SNAKE_CASE       :		List[Any]           =		vae_state_dict["""encoder.norm_out.bias"""]

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

		__SCREAMING_SNAKE_CASE       :		int           =		vae_state_dict["""quant_conv.weight"""]
		__SCREAMING_SNAKE_CASE       :		Optional[int]           =		vae_state_dict["""quant_conv.bias"""]
		__SCREAMING_SNAKE_CASE       :		Dict           =		vae_state_dict["""post_quant_conv.weight"""]
		__SCREAMING_SNAKE_CASE       :		List[str]           =		vae_state_dict["""post_quant_conv.bias"""]

		# Retrieves the keys for the encoder down blocks only
		__SCREAMING_SNAKE_CASE       :		Optional[int]           =		len({""".""".join(layer.split("""."""					)[:3]					) for layer in vae_state_dict if """encoder.down""" in layer}					)
		__SCREAMING_SNAKE_CASE       :		Tuple           =		{
		    layer_id: [key for key in vae_state_dict if F"down.{layer_id}" in key] for layer_id in range(_lowerCamelCase					)
		}

		# Retrieves the keys for the decoder up blocks only
		__SCREAMING_SNAKE_CASE       :		int           =		len({""".""".join(layer.split("""."""					)[:3]					) for layer in vae_state_dict if """decoder.up""" in layer}					)
		__SCREAMING_SNAKE_CASE       :		List[str]           =		{
		    layer_id: [key for key in vae_state_dict if F"up.{layer_id}" in key] for layer_id in range(_lowerCamelCase					)
		}

		for i in range(_lowerCamelCase					):
				__SCREAMING_SNAKE_CASE       :		str           =		[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       :		Dict           =		vae_state_dict.pop(
						    F"encoder.down.{i}.downsample.conv.weight"					)
						__SCREAMING_SNAKE_CASE       :		List[Any]           =		vae_state_dict.pop(
						    F"encoder.down.{i}.downsample.conv.bias"					)

				__SCREAMING_SNAKE_CASE       :		Optional[int]           =		renew_vae_resnet_paths(_lowerCamelCase					)
				__SCREAMING_SNAKE_CASE       :		Optional[int]           =		{"""old""": F"down.{i}.block", """new""": F"down_blocks.{i}.resnets"}
				assign_to_checkpoint(_lowerCamelCase       ,      _lowerCamelCase       ,      _lowerCamelCase       ,      additional_replacements=[meta_path]       ,      config=_lowerCamelCase					)

		__SCREAMING_SNAKE_CASE       :		List[Any]           =		[key for key in vae_state_dict if """encoder.mid.block""" in key]
		__SCREAMING_SNAKE_CASE       :		List[Any]           =		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(_lowerCamelCase					)
				__SCREAMING_SNAKE_CASE       :		Optional[int]           =		{"""old""": F"mid.block_{i}", """new""": F"mid_block.resnets.{i - 1}"}
				assign_to_checkpoint(_lowerCamelCase       ,      _lowerCamelCase       ,      _lowerCamelCase       ,      additional_replacements=[meta_path]       ,      config=_lowerCamelCase					)

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

		for i in range(_lowerCamelCase					):
				__SCREAMING_SNAKE_CASE       :		str           =		num_up_blocks - 1 - i
				__SCREAMING_SNAKE_CASE       :		Dict           =		[
				    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       :		Union[str, Any]           =		vae_state_dict[
						    F"decoder.up.{block_id}.upsample.conv.bias"
						]

				__SCREAMING_SNAKE_CASE       :		Optional[Any]           =		renew_vae_resnet_paths(_lowerCamelCase					)
				__SCREAMING_SNAKE_CASE       :		Tuple           =		{"""old""": F"up.{block_id}.block", """new""": F"up_blocks.{i}.resnets"}
				assign_to_checkpoint(_lowerCamelCase       ,      _lowerCamelCase       ,      _lowerCamelCase       ,      additional_replacements=[meta_path]       ,      config=_lowerCamelCase					)

		__SCREAMING_SNAKE_CASE       :		Optional[int]           =		[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       :		str           =		[key for key in mid_resnets if F"decoder.mid.block_{i}" in key]

				__SCREAMING_SNAKE_CASE       :		Optional[Any]           =		renew_vae_resnet_paths(_lowerCamelCase					)
				__SCREAMING_SNAKE_CASE       :		Dict           =		{"""old""": F"mid.block_{i}", """new""": F"mid_block.resnets.{i - 1}"}
				assign_to_checkpoint(_lowerCamelCase       ,      _lowerCamelCase       ,      _lowerCamelCase       ,      additional_replacements=[meta_path]       ,      config=_lowerCamelCase					)

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


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

		__SCREAMING_SNAKE_CASE       :		str           =		OmegaConf.load(_lowerCamelCase					)
		__SCREAMING_SNAKE_CASE       :		List[str]           =		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       :		Tuple           =		{}
				with safe_open(_lowerCamelCase       ,      framework="""pt"""       ,      device="""cpu"""					) as f:
						for key in f.keys():
								__SCREAMING_SNAKE_CASE       :		Union[str, Any]           =		f.get_tensor(_lowerCamelCase					)
		else:
				__SCREAMING_SNAKE_CASE       :		str           =		torch.load(_lowerCamelCase       ,      map_location=_lowerCamelCase					)["""state_dict"""]

		# Convert the VAE model.
		__SCREAMING_SNAKE_CASE       :		Tuple           =		create_vae_diffusers_config(_lowerCamelCase       ,      image_size=_lowerCamelCase					)
		__SCREAMING_SNAKE_CASE       :		Dict           =		custom_convert_ldm_vae_checkpoint(_lowerCamelCase       ,      _lowerCamelCase					)

		__SCREAMING_SNAKE_CASE       :		Optional[Any]           =		AutoencoderKL(**_lowerCamelCase					)
		vae.load_state_dict(_lowerCamelCase					)
		vae.save_pretrained(_lowerCamelCase					)


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

						UpperCamelCase__       :	Optional[Any]    = parser.parse_args()

						vae_pt_to_vae_diffuser(args.vae_pt_path, args.dump_path)




'''simple docstring'''




def 				lowerCAmelCase_   (	_lowerCamelCase:   list					):
		if any(not isinstance(_lowerCamelCase       ,      _lowerCamelCase					) or x < 0 for x in sequence					):
				raise TypeError("""Sequence must be list of non-negative integers"""					)
		for _ in range(len(_lowerCamelCase					)					):
				for i, (rod_upper, rod_lower) in enumerate(zip(_lowerCamelCase       ,      sequence[1:]					)					):
						if rod_upper > rod_lower:
								sequence[i] -= rod_upper - rod_lower
								sequence[i + 1] += rod_upper - rod_lower
		return sequence


if __name__ == "__main__":
						assert bead_sort([5, 4, 3, 2, 1]) == [1, 2, 3, 4, 5]
						assert bead_sort([7, 9, 4, 3, 5]) == [3, 4, 5, 7, 9]