





from collections.abc import Iterator, MutableMapping
from dataclasses import dataclass
from typing import Generic, TypeVar

snake_case_				=			TypeVar('KEY')
snake_case_				=			TypeVar('VAL')

@dataclass(frozen=lowercase__						,      slots=lowercase__  )
class 		SCREAMING_SNAKE_CASE__				(       Generic[KEY, VAL]  ):
      A_						:				KEY
      A_						:				VAL






class 		SCREAMING_SNAKE_CASE__				(       _Item  ):


      def __init__(self       :      Tuple      ):





         """simple docstring"""



         super().__init__(lowercase_				,			lowercase_      )


      def __bool__(self       :      Tuple      ):





         """simple docstring"""



         return False






snake_case_				=			_DeletedItem()

class 		SCREAMING_SNAKE_CASE__				(       MutableMapping[KEY, VAL]  ):


      def __init__(self       :      str				,			a__       :      Dict = 8				,			a__       :      Optional[int] = 0.7_5      ):





         """simple docstring"""



         __snake_case							       =   initial_block_size
         __snake_case							       =   [None] * initial_block_size
         assert 0.0 < capacity_factor < 1.0
         __snake_case							       =   capacity_factor
         __snake_case							       =   0


      def    a (self       :      Tuple				,			a__       :      Any      ):





         """simple docstring"""



         return hash(lowercase_      ) % len(self._buckets      )


      def    a (self       :      Union[str, Any]				,			a__       :      str      ):





         """simple docstring"""



         return (ind + 1) % len(self._buckets      )


      def    a (self       :      Tuple				,			a__       :      Optional[int]				,			a__       :      List[str]				,			a__       :      Any      ):





         """simple docstring"""



         __snake_case							       =   self._buckets[ind]
         if not stored:
            __snake_case							       =   _Item(lowercase_				,			lowercase_      )
            self._len += 1
            return True
         elif stored.key == key:
            __snake_case							       =   _Item(lowercase_				,			lowercase_      )
            return True
         else:
            return False


      def    a (self       :      Any      ):





         """simple docstring"""



         __snake_case							       =   len(self._buckets      ) * self._capacity_factor
         return len(self      ) >= int(lowercase_      )


      def    a (self       :      Any      ):





         """simple docstring"""



         if len(self._buckets      ) <= self._initial_block_size:
            return False
         __snake_case							       =   len(self._buckets      ) * self._capacity_factor / 2
         return len(self      ) < limit


      def    a (self       :      List[str]				,			a__       :      Any      ):





         """simple docstring"""



         __snake_case							       =   self._buckets
         __snake_case							       =   [None] * new_size
         __snake_case							       =   0
         for item in old_buckets:
            if item:
               self._add_item(item.key				,			item.val      )


      def    a (self       :      Any      ):





         """simple docstring"""



         self._resize(len(self._buckets      ) * 2      )


      def    a (self       :      int      ):





         """simple docstring"""



         self._resize(len(self._buckets      ) // 2      )


      def    a (self       :      List[Any]				,			a__       :      Dict      ):





         """simple docstring"""



         __snake_case							       =   self._get_bucket_index(lowercase_      )
         for _ in range(len(self._buckets      )      ):
            yield ind
            __snake_case							       =   self._get_next_ind(lowercase_      )


      def    a (self       :      Union[str, Any]				,			a__       :      int				,			a__       :      List[Any]      ):





         """simple docstring"""



         for ind in self._iterate_buckets(lowercase_      ):
            if self._try_set(lowercase_				,			lowercase_				,			lowercase_      ):
               break


      def __setitem__(self       :      List[Any]				,			a__       :      Tuple				,			a__       :      str      ):





         """simple docstring"""



         if self._is_full():
            self._size_up()

         self._add_item(lowercase_				,			lowercase_      )


      def __delitem__(self       :      List[Any]				,			a__       :      Optional[Any]      ):





         """simple docstring"""



         for ind in self._iterate_buckets(lowercase_      ):
            __snake_case							       =   self._buckets[ind]
            if item is None:
               raise KeyError(lowercase_      )
            if item is _deleted:
               continue
            if item.key == key:
               __snake_case							       =   _deleted
               self._len -= 1
               break
         if self._is_sparse():
            self._size_down()


      def __getitem__(self       :      Dict				,			a__       :      Tuple      ):





         """simple docstring"""



         for ind in self._iterate_buckets(lowercase_      ):
            __snake_case							       =   self._buckets[ind]
            if item is None:
               break
            if item is _deleted:
               continue
            if item.key == key:
               return item.val
         raise KeyError(lowercase_      )


      def __len__(self       :      List[Any]      ):





         """simple docstring"""



         return self._len


      def __iter__(self       :      int      ):





         """simple docstring"""



         yield from (item.key for item in self._buckets if item)


      def __repr__(self       :      List[Any]      ):





         """simple docstring"""



         __snake_case							       =   " ,".join(
             f"""{item.key}: {item.val}""" for item in self._buckets if item      )
         return f"""HashMap({val_string})"""



"""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     (		__lowerCamelCase,		__lowerCamelCase  ):
   UpperCAmelCase_					:					str			     =    checkpoint

   UpperCAmelCase_					:					int			     =    {}

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

   UpperCAmelCase_					:					Any			     =    vae_state_dict["decoder.conv_in.weight"]
   UpperCAmelCase_					:					int			     =    vae_state_dict["decoder.conv_in.bias"]
   UpperCAmelCase_					:					Any			     =    vae_state_dict["decoder.conv_out.weight"]
   UpperCAmelCase_					:					Tuple			     =    vae_state_dict["decoder.conv_out.bias"]
   UpperCAmelCase_					:					List[Any]			     =    vae_state_dict["decoder.norm_out.weight"]
   UpperCAmelCase_					:					Tuple			     =    vae_state_dict["decoder.norm_out.bias"]

   UpperCAmelCase_					:					str			     =    vae_state_dict["quant_conv.weight"]
   UpperCAmelCase_					:					Optional[Any]			     =    vae_state_dict["quant_conv.bias"]
   UpperCAmelCase_					:					List[str]			     =    vae_state_dict["post_quant_conv.weight"]
   UpperCAmelCase_					:					List[Any]			     =    vae_state_dict["post_quant_conv.bias"]

   # Retrieves the keys for the encoder down blocks only
   UpperCAmelCase_					:					Optional[Any]			     =    len({".".join(layer.split("."  )[:3]  ) for layer in vae_state_dict if "encoder.down" in layer}  )
   UpperCAmelCase_					:					Optional[Any]			     =    {
       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
   UpperCAmelCase_					:					Dict			     =    len({".".join(layer.split("."  )[:3]  ) for layer in vae_state_dict if "decoder.up" in layer}  )
   UpperCAmelCase_					:					Optional[int]			     =    {
       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  ):
      UpperCAmelCase_					:					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_					:					Dict			     =    vae_state_dict.pop(
             f"""encoder.down.{i}.downsample.conv.bias"""  )

      UpperCAmelCase_					:					List[str]			     =    renew_vae_resnet_paths(__lowerCamelCase  )
      UpperCAmelCase_					:					Optional[Any]			     =    {"old": f"""down.{i}.block""", "new": f"""down_blocks.{i}.resnets"""}
      assign_to_checkpoint(__lowerCamelCase,		__lowerCamelCase,		__lowerCamelCase,		additional_replacements=[meta_path],		config=__lowerCamelCase  )

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

      UpperCAmelCase_					:					List[Any]			     =    renew_vae_resnet_paths(__lowerCamelCase  )
      UpperCAmelCase_					:					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  )

   UpperCAmelCase_					:					Optional[int]			     =    [key for key in vae_state_dict if "encoder.mid.attn" in key]
   UpperCAmelCase_					:					Union[str, Any]			     =    renew_vae_attention_paths(__lowerCamelCase  )
   UpperCAmelCase_					:					int			     =    {"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  ):
      UpperCAmelCase_					:					Optional[Any]			     =    num_up_blocks - 1 - i
      UpperCAmelCase_					:					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:
         UpperCAmelCase_					:					str			     =    vae_state_dict[
             f"""decoder.up.{block_id}.upsample.conv.weight"""
         ]
         UpperCAmelCase_					:					Optional[Any]			     =    vae_state_dict[
             f"""decoder.up.{block_id}.upsample.conv.bias"""
         ]

      UpperCAmelCase_					:					Dict			     =    renew_vae_resnet_paths(__lowerCamelCase  )
      UpperCAmelCase_					:					List[str]			     =    {"old": f"""up.{block_id}.block""", "new": f"""up_blocks.{i}.resnets"""}
      assign_to_checkpoint(__lowerCamelCase,		__lowerCamelCase,		__lowerCamelCase,		additional_replacements=[meta_path],		config=__lowerCamelCase  )

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

      UpperCAmelCase_					:					Tuple			     =    renew_vae_resnet_paths(__lowerCamelCase  )
      UpperCAmelCase_					:					Tuple			     =    {"old": f"""mid.block_{i}""", "new": f"""mid_block.resnets.{i - 1}"""}
      assign_to_checkpoint(__lowerCamelCase,		__lowerCamelCase,		__lowerCamelCase,		additional_replacements=[meta_path],		config=__lowerCamelCase  )

   UpperCAmelCase_					:					Optional[int]			     =    [key for key in vae_state_dict if "decoder.mid.attn" in key]
   UpperCAmelCase_					:					Any			     =    renew_vae_attention_paths(__lowerCamelCase  )
   UpperCAmelCase_					:					Union[str, 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  )
   return new_checkpoint







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

   UpperCAmelCase_					:					Any			     =    OmegaConf.load(__lowerCamelCase  )
   UpperCAmelCase_					:					Union[str, Any]			     =    512
   UpperCAmelCase_					:					Optional[Any]			     =    "cuda" if torch.cuda.is_available() else "cpu"
   if checkpoint_path.endswith("safetensors"  ):
      from safetensors import safe_open

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

   # Convert the VAE model.
   UpperCAmelCase_					:					Dict			     =    create_vae_diffusers_config(__lowerCamelCase,		image_size=__lowerCamelCase  )
   UpperCAmelCase_					:					Union[str, Any]			     =    custom_convert_ldm_vae_checkpoint(__lowerCamelCase,		__lowerCamelCase  )

   UpperCAmelCase_					:					int			     =    AutoencoderKL(**__lowerCamelCase  )
   vae.load_state_dict(__lowerCamelCase  )
   vae.save_pretrained(__lowerCamelCase  )


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

    _a		       =							parser.parse_args()

    vae_pt_to_vae_diffuser(args.vae_pt_path, args.dump_path)


