


"""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						(						lowercase  :  str						,							lowercase  :  Union[str, Any]			)     ->						int:


    """simple docstring"""





    snake_case   :				Dict				 =   checkpoint

    snake_case   :				Optional[Any]				 =   {}

    snake_case   :				Union[str, Any]				 =   vae_state_dict["encoder.conv_in.weight"]
    snake_case   :				Tuple				 =   vae_state_dict["encoder.conv_in.bias"]
    snake_case   :				List[str]				 =   vae_state_dict["encoder.conv_out.weight"]
    snake_case   :				List[str]				 =   vae_state_dict["encoder.conv_out.bias"]
    snake_case   :				List[str]				 =   vae_state_dict["encoder.norm_out.weight"]
    snake_case   :				Union[str, Any]				 =   vae_state_dict["encoder.norm_out.bias"]

    snake_case   :				List[str]				 =   vae_state_dict["decoder.conv_in.weight"]
    snake_case   :				List[str]				 =   vae_state_dict["decoder.conv_in.bias"]
    snake_case   :				int				 =   vae_state_dict["decoder.conv_out.weight"]
    snake_case   :				List[str]				 =   vae_state_dict["decoder.conv_out.bias"]
    snake_case   :				Dict				 =   vae_state_dict["decoder.norm_out.weight"]
    snake_case   :				Optional[int]				 =   vae_state_dict["decoder.norm_out.bias"]

    snake_case   :				Tuple				 =   vae_state_dict["quant_conv.weight"]
    snake_case   :				List[Any]				 =   vae_state_dict["quant_conv.bias"]
    snake_case   :				Dict				 =   vae_state_dict["post_quant_conv.weight"]
    snake_case   :				Optional[int]				 =   vae_state_dict["post_quant_conv.bias"]

    # Retrieves the keys for the encoder down blocks only
    snake_case   :				Optional[Any]				 =   len({".".join(layer.split("."			)[:3]			) for layer in vae_state_dict if "encoder.down" in layer}			)
    snake_case   :				List[Any]				 =   {
        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
    snake_case   :				Optional[Any]				 =   len({".".join(layer.split("."			)[:3]			) for layer in vae_state_dict if "decoder.up" in layer}			)
    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(lowercase			)
    }

    for i in range(lowercase			):
        snake_case   :				int				 =   [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:
            snake_case   :				Tuple				 =   vae_state_dict.pop(
                F'encoder.down.{i}.downsample.conv.weight'			)
            snake_case   :				Tuple				 =   vae_state_dict.pop(
                F'encoder.down.{i}.downsample.conv.bias'			)

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

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

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

    snake_case   :				int				 =   [key for key in vae_state_dict if "encoder.mid.attn" in key]
    snake_case   :				Dict				 =   renew_vae_attention_paths(lowercase			)
    snake_case   :				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			):
        snake_case   :				int				 =   num_up_blocks - 1 - i
        snake_case   :				Optional[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:
            snake_case   :				Any				 =   vae_state_dict[
                F'decoder.up.{block_id}.upsample.conv.weight'
            ]
            snake_case   :				Optional[Any]				 =   vae_state_dict[
                F'decoder.up.{block_id}.upsample.conv.bias'
            ]

        snake_case   :				Any				 =   renew_vae_resnet_paths(lowercase			)
        snake_case   :				Any				 =   {"old": F'up.{block_id}.block', "new": F'up_blocks.{i}.resnets'}
        assign_to_checkpoint(lowercase						,							lowercase						,							lowercase						,							additional_replacements=[meta_path]						,							config=lowercase			)

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

        snake_case   :				Union[str, Any]				 =   renew_vae_resnet_paths(lowercase			)
        snake_case   :				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			)

    snake_case   :				Optional[Any]				 =   [key for key in vae_state_dict if "decoder.mid.attn" in key]
    snake_case   :				List[str]				 =   renew_vae_attention_paths(lowercase			)
    snake_case   :				Dict				 =   {"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   __lowerCAmelCase						(						lowercase  :  str						,							lowercase  :  str						,							)     ->						Tuple:


    """simple docstring"""





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

    snake_case   :				List[str]				 =   OmegaConf.load(lowercase			)
    snake_case   :				List[Any]				 =   512
    snake_case   :				Optional[Any]				 =   "cuda" if torch.cuda.is_available() else "cpu"
    if checkpoint_path.endswith("safetensors"			):
        from safetensors import safe_open

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

    # Convert the VAE model.
    snake_case   :				Tuple				 =   create_vae_diffusers_config(lowercase						,							image_size=lowercase			)
    snake_case   :				str				 =   custom_convert_ldm_vae_checkpoint(lowercase						,							lowercase			)

    snake_case   :				List[Any]				 =   AutoencoderKL(**lowercase			)
    vae.load_state_dict(lowercase			)
    vae.save_pretrained(lowercase			)


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"""



import unittest

import numpy as np
import torch

from diffusers import KarrasVePipeline, KarrasVeScheduler, UNetaDModel
from diffusers.utils.testing_utils import enable_full_determinism, require_torch, slow, torch_device


enable_full_determinism()







class _lowerCAmelCase    (				unittest.TestCase     ):







   @property
   def 			lowerCamelCase		(     self						)     ->		str:

       '''simple docstring'''




       torch.manual_seed(0						)
       snake_case   :				int				 =   UNetaDModel(
           block_out_channels=(32, 64)						,     layers_per_block=2						,     sample_size=32						,     in_channels=3						,     out_channels=3						,     down_block_types=("DownBlock2D", "AttnDownBlock2D")						,     up_block_types=("AttnUpBlock2D", "UpBlock2D")						,     )
       return model







   def 			lowerCamelCase		(     self						)     ->		Any:

       '''simple docstring'''




       snake_case   :				Any				 =   self.dummy_uncond_unet
       snake_case   :				Tuple				 =   KarrasVeScheduler()

       snake_case   :				int				 =   KarrasVePipeline(unet=UpperCamelCase__						,     scheduler=UpperCamelCase__						)
       pipe.to(UpperCamelCase__						)
       pipe.set_progress_bar_config(disable=UpperCamelCase__						)

       snake_case   :				Optional[Any]				 =   torch.manual_seed(0						)
       snake_case   :				List[Any]				 =   pipe(num_inference_steps=2						,     generator=UpperCamelCase__						,     output_type="numpy"						).images

       snake_case   :				Dict				 =   torch.manual_seed(0						)
       snake_case   :				Dict				 =   pipe(num_inference_steps=2						,     generator=UpperCamelCase__						,     output_type="numpy"						,     return_dict=UpperCamelCase__						)[0]

       snake_case   :				Tuple				 =   image[0, -3:, -3:, -1]
       snake_case   :				Tuple				 =   image_from_tuple[0, -3:, -3:, -1]

       assert image.shape == (1, 32, 32, 3)
       snake_case   :				int				 =   np.array([0.0, 1.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0]						)

       assert np.abs(image_slice.flatten() - expected_slice						).max() < 1e-2
       assert np.abs(image_from_tuple_slice.flatten() - expected_slice						).max() < 1e-2







@slow
@require_torch
class _lowerCAmelCase    (				unittest.TestCase     ):







   def 			lowerCamelCase		(     self						)     ->		List[Any]:

       '''simple docstring'''




       snake_case   :				Optional[Any]				 =   "google/ncsnpp-celebahq-256"
       snake_case   :				List[str]				 =   UNetaDModel.from_pretrained(UpperCamelCase__						)
       snake_case   :				Optional[Any]				 =   KarrasVeScheduler()

       snake_case   :				Optional[int]				 =   KarrasVePipeline(unet=UpperCamelCase__						,     scheduler=UpperCamelCase__						)
       pipe.to(UpperCamelCase__						)
       pipe.set_progress_bar_config(disable=UpperCamelCase__						)

       snake_case   :				Dict				 =   torch.manual_seed(0						)
       snake_case   :				Union[str, Any]				 =   pipe(num_inference_steps=20						,     generator=UpperCamelCase__						,     output_type="numpy"						).images

       snake_case   :				int				 =   image[0, -3:, -3:, -1]
       assert image.shape == (1, 256, 256, 3)
       snake_case   :				Any				 =   np.array([0.578, 0.5811, 0.5924, 0.5809, 0.587, 0.5886, 0.5861, 0.5802, 0.586]						)

       assert np.abs(image_slice.flatten() - expected_slice						).max() < 1e-2





