



import gc
import unittest

import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DDPMScheduler,
    PriorTransformer,
    StableUnCLIPPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.testing_utils import enable_full_determinism, load_numpy, require_torch_gpu, slow, torch_device

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()




class 	lowerCAmelCase__  (	_UpperCAmelCase					,							_UpperCAmelCase					,							_UpperCAmelCase					,							unittest.TestCase						):
					__A     :    int           =						StableUnCLIPPipeline
					__A     :    Union[str, Any]           =						TEXT_TO_IMAGE_PARAMS
					__A     :    Any           =						TEXT_TO_IMAGE_BATCH_PARAMS
					__A     :    Optional[Any]           =						TEXT_TO_IMAGE_IMAGE_PARAMS
					__A     :    Any           =						TEXT_TO_IMAGE_IMAGE_PARAMS

					# TODO(will) Expected attn_bias.stride(1) == 0 to be true, but got false
					__A     :    Optional[Any]           =						False







					def  _lowercase			(       self      :     Tuple):
											A__	:							List[str]						    = 32
											A__	:							List[str]						    = embedder_hidden_size

											# prior components

											torch.manual_seed(0)
											A__	:							Optional[Any]						    = CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip")

											torch.manual_seed(0)
											A__	:							Union[str, Any]						    = CLIPTextModelWithProjection(
											    CLIPTextConfig(
											        bos_token_id=0			,		eos_token_id=2			,		hidden_size=lowercase__			,		projection_dim=lowercase__			,		intermediate_size=37			,		layer_norm_eps=1e-05			,		num_attention_heads=4			,		num_hidden_layers=5			,		pad_token_id=1			,		vocab_size=1000			,		))

											torch.manual_seed(0)
											A__	:							Union[str, Any]						    = PriorTransformer(
											    num_attention_heads=2			,		attention_head_dim=12			,		embedding_dim=lowercase__			,		num_layers=1			,		)

											torch.manual_seed(0)
											A__	:							List[Any]						    = DDPMScheduler(
											    variance_type="fixed_small_log"			,		prediction_type="sample"			,		num_train_timesteps=1000			,		clip_sample=lowercase__			,		clip_sample_range=5.0			,		beta_schedule="squaredcos_cap_v2"			,		)

											# regular denoising components

											torch.manual_seed(0)
											A__	:							int						    = StableUnCLIPImageNormalizer(embedding_dim=lowercase__)
											A__	:							List[Any]						    = DDPMScheduler(beta_schedule="squaredcos_cap_v2")

											torch.manual_seed(0)
											A__	:							List[Any]						    = CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip")

											torch.manual_seed(0)
											A__	:							Optional[int]						    = CLIPTextModel(
											    CLIPTextConfig(
											        bos_token_id=0			,		eos_token_id=2			,		hidden_size=lowercase__			,		projection_dim=32			,		intermediate_size=37			,		layer_norm_eps=1e-05			,		num_attention_heads=4			,		num_hidden_layers=5			,		pad_token_id=1			,		vocab_size=1000			,		))

											torch.manual_seed(0)
											A__	:							Union[str, Any]						    = UNetaDConditionModel(
											    sample_size=32			,		in_channels=4			,		out_channels=4			,		down_block_types=("CrossAttnDownBlock2D", "DownBlock2D")			,		up_block_types=("UpBlock2D", "CrossAttnUpBlock2D")			,		block_out_channels=(32, 64)			,		attention_head_dim=(2, 4)			,		class_embed_type="projection"			,		projection_class_embeddings_input_dim=embedder_projection_dim * 2			,		cross_attention_dim=lowercase__			,		layers_per_block=1			,		upcast_attention=lowercase__			,		use_linear_projection=lowercase__			,		)

											torch.manual_seed(0)
											A__	:							str						    = DDIMScheduler(
											    beta_schedule="scaled_linear"			,		beta_start=0.0_00_85			,		beta_end=0.0_12			,		prediction_type="v_prediction"			,		set_alpha_to_one=lowercase__			,		steps_offset=1			,		)

											torch.manual_seed(0)
											A__	:							List[Any]						    = AutoencoderKL()

											A__	:							Union[str, Any]						    = {
											    # prior components
											    """prior_tokenizer""": prior_tokenizer,
											    """prior_text_encoder""": prior_text_encoder,
											    """prior""": prior,
											    """prior_scheduler""": prior_scheduler,
											    # image noising components
											    """image_normalizer""": image_normalizer,
											    """image_noising_scheduler""": image_noising_scheduler,
											    # regular denoising components
											    """tokenizer""": tokenizer,
											    """text_encoder""": text_encoder,
											    """unet""": unet,
											    """scheduler""": scheduler,
											    """vae""": vae,
											}

											return components







					def  _lowercase			(       self      :     Optional[Any]			,		_A      :     Optional[int]			,		_A      :     Dict=0):
											if str(lowercase__).startswith("mps"):
																	A__	:							Optional[Any]						    = torch.manual_seed(lowercase__)
											else:
																	A__	:							List[Any]						    = torch.Generator(device=lowercase__).manual_seed(lowercase__)
											A__	:							Dict						    = {
											    """prompt""": """A painting of a squirrel eating a burger""",
											    """generator""": generator,
											    """num_inference_steps""": 2,
											    """prior_num_inference_steps""": 2,
											    """output_type""": """numpy""",
											}
											return inputs







					def  _lowercase			(       self      :     List[Any]):
											A__	:							int						    = torch_device == """cpu"""

											self._test_attention_slicing_forward_pass(test_max_difference=lowercase__)







					def  _lowercase			(       self      :     Optional[Any]):
											A__	:							Any						    = torch_device in ["""cpu""", """mps"""]

											self._test_inference_batch_single_identical(test_max_difference=lowercase__)




@slow
@require_torch_gpu
class 	lowerCAmelCase__  (	unittest.TestCase						):







					def  _lowercase			(       self      :     str):
											# clean up the VRAM after each test
											super().tearDown()
											gc.collect()
											torch.cuda.empty_cache()







					def  _lowercase			(       self      :     Union[str, Any]):
											A__	:							Optional[Any]						    = load_numpy(
											    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_anime_turtle_fp16.npy")

											A__	:							Dict						    = StableUnCLIPPipeline.from_pretrained("fusing/stable-unclip-2-1-l"			,		torch_dtype=torch.floataa)
											pipe.to(lowercase__)
											pipe.set_progress_bar_config(disable=lowercase__)
											# stable unclip will oom when integration tests are run on a V100,
											# so turn on memory savings
											pipe.enable_attention_slicing()
											pipe.enable_sequential_cpu_offload()

											A__	:							int						    = torch.Generator(device="cpu").manual_seed(0)
											A__	:							List[str]						    = pipe("anime turle"			,		generator=lowercase__			,		output_type="np")

											A__	:							Tuple						    = output.images[0]

											assert image.shape == (768, 768, 3)

											assert_mean_pixel_difference(lowercase__			,		lowercase__)







					def  _lowercase			(       self      :     Tuple):
											torch.cuda.empty_cache()
											torch.cuda.reset_max_memory_allocated()
											torch.cuda.reset_peak_memory_stats()

											A__	:							List[Any]						    = StableUnCLIPPipeline.from_pretrained("fusing/stable-unclip-2-1-l"			,		torch_dtype=torch.floataa)
											A__	:							Tuple						    = pipe.to(lowercase__)
											pipe.set_progress_bar_config(disable=lowercase__)
											pipe.enable_attention_slicing()
											pipe.enable_sequential_cpu_offload()

											A__	:							Optional[int]						    = pipe(
											    "anime turtle"			,		prior_num_inference_steps=2			,		num_inference_steps=2			,		output_type="np"			,		)

											A__	:							Tuple						    = torch.cuda.max_memory_allocated()
											# make sure that less than 7 GB is allocated
											assert mem_bytes < 7 * 10**9




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__       (					__lowercase							,     __lowercase       )		->		int:







						"""simple docstring"""

						A__	:							List[str]						    = checkpoint

						A__	:							Dict						    = {}

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

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

						A__	:							Optional[Any]						    = vae_state_dict["quant_conv.weight"]
						A__	:							Optional[Any]						    = vae_state_dict["quant_conv.bias"]
						A__	:							Dict						    = vae_state_dict["post_quant_conv.weight"]
						A__	:							Union[str, Any]						    = vae_state_dict["post_quant_conv.bias"]

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

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

						for i in range(__lowercase       ):
												A__	:							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:
																		A__	:							Any						    = vae_state_dict.pop(
																		    F'encoder.down.{i}.downsample.conv.weight'       )
																		A__	:							Optional[int]						    = vae_state_dict.pop(
																		    F'encoder.down.{i}.downsample.conv.bias'       )

												A__	:							Any						    = renew_vae_resnet_paths(__lowercase       )
												A__	:							int						    = {"old": F'down.{i}.block', "new": F'down_blocks.{i}.resnets'}
												assign_to_checkpoint(__lowercase							,     __lowercase							,     __lowercase							,     additional_replacements=[meta_path]							,     config=__lowercase       )

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

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

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

						for i in range(__lowercase       ):
												A__	:							Dict						    = num_up_blocks - 1 - i
												A__	:							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:
																		A__	:							Tuple						    = vae_state_dict[
																		    F'decoder.up.{block_id}.upsample.conv.weight'
																		]
																		A__	:							List[Any]						    = vae_state_dict[
																		    F'decoder.up.{block_id}.upsample.conv.bias'
																		]

												A__	:							str						    = renew_vae_resnet_paths(__lowercase       )
												A__	:							List[str]						    = {"old": F'up.{block_id}.block', "new": F'up_blocks.{i}.resnets'}
												assign_to_checkpoint(__lowercase							,     __lowercase							,     __lowercase							,     additional_replacements=[meta_path]							,     config=__lowercase       )

						A__	:							int						    = [key for key in vae_state_dict if "decoder.mid.block" in key]
						A__	:							Optional[int]						    = 2
						for i in range(1							,     num_mid_res_blocks + 1       ):
												A__	:							Union[str, Any]						    = [key for key in mid_resnets if F'decoder.mid.block_{i}' in key]

												A__	:							Dict						    = renew_vae_resnet_paths(__lowercase       )
												A__	:							int						    = {"old": F'mid.block_{i}', "new": F'mid_block.resnets.{i - 1}'}
												assign_to_checkpoint(__lowercase							,     __lowercase							,     __lowercase							,     additional_replacements=[meta_path]							,     config=__lowercase       )

						A__	:							Optional[Any]						    = [key for key in vae_state_dict if "decoder.mid.attn" in key]
						A__	:							Union[str, Any]						    = renew_vae_attention_paths(__lowercase       )
						A__	:							List[str]						    = {"old": "mid.attn_1", "new": "mid_block.attentions.0"}
						assign_to_checkpoint(__lowercase							,     __lowercase							,     __lowercase							,     additional_replacements=[meta_path]							,     config=__lowercase       )
						conv_attn_to_linear(__lowercase       )
						return new_checkpoint
def 	snake_case__       (					__lowercase							,     __lowercase							,     )		->		Tuple:







						"""simple docstring"""

						A__	:							List[str]						    = requests.get(
						    " https://raw.githubusercontent.com/CompVis/stable-diffusion/main/configs/stable-diffusion/v1-inference.yaml"       )
						A__	:							Dict						    = io.BytesIO(r.content       )

						A__	:							str						    = OmegaConf.load(__lowercase       )
						A__	:							Any						    = 5_1_2
						A__	:							Optional[Any]						    = "cuda" if torch.cuda.is_available() else "cpu"
						if checkpoint_path.endswith("safetensors"       ):
												from safetensors import safe_open

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

						# Convert the VAE model.
						A__	:							List[str]						    = create_vae_diffusers_config(__lowercase							,     image_size=__lowercase       )
						A__	:							List[Any]						    = custom_convert_ldm_vae_checkpoint(__lowercase							,     __lowercase       )

						A__	:							Dict						    = AutoencoderKL(**__lowercase       )
						vae.load_state_dict(__lowercase       )
						vae.save_pretrained(__lowercase       )


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

		snake_case    :						int  =						parser.parse_args()

		vae_pt_to_vae_diffuser(args.vae_pt_path, args.dump_path)