





"""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       (							a_					:		str			,       a_					:		Optional[Any]							)      ->				List[str]:
      __SCREAMING_SNAKE_CASE       :Dict		    =  checkpoint

      __SCREAMING_SNAKE_CASE       :str		    =  {}

      __SCREAMING_SNAKE_CASE       :Any		    =  vae_state_dict['''encoder.conv_in.weight''']
      __SCREAMING_SNAKE_CASE       :List[str]		    =  vae_state_dict['''encoder.conv_in.bias''']
      __SCREAMING_SNAKE_CASE       :Dict		    =  vae_state_dict['''encoder.conv_out.weight''']
      __SCREAMING_SNAKE_CASE       :Dict		    =  vae_state_dict['''encoder.conv_out.bias''']
      __SCREAMING_SNAKE_CASE       :Tuple		    =  vae_state_dict['''encoder.norm_out.weight''']
      __SCREAMING_SNAKE_CASE       :Any		    =  vae_state_dict['''encoder.norm_out.bias''']

      __SCREAMING_SNAKE_CASE       :Optional[Any]		    =  vae_state_dict['''decoder.conv_in.weight''']
      __SCREAMING_SNAKE_CASE       :Any		    =  vae_state_dict['''decoder.conv_in.bias''']
      __SCREAMING_SNAKE_CASE       :List[Any]		    =  vae_state_dict['''decoder.conv_out.weight''']
      __SCREAMING_SNAKE_CASE       :str		    =  vae_state_dict['''decoder.conv_out.bias''']
      __SCREAMING_SNAKE_CASE       :Tuple		    =  vae_state_dict['''decoder.norm_out.weight''']
      __SCREAMING_SNAKE_CASE       :Optional[Any]		    =  vae_state_dict['''decoder.norm_out.bias''']

      __SCREAMING_SNAKE_CASE       :Tuple		    =  vae_state_dict['''quant_conv.weight''']
      __SCREAMING_SNAKE_CASE       :Optional[int]		    =  vae_state_dict['''quant_conv.bias''']
      __SCREAMING_SNAKE_CASE       :Optional[Any]		    =  vae_state_dict['''post_quant_conv.weight''']
      __SCREAMING_SNAKE_CASE       :Optional[int]		    =  vae_state_dict['''post_quant_conv.bias''']

      # Retrieves the keys for the encoder down blocks only
      __SCREAMING_SNAKE_CASE       :List[Any]		    =  len({'''.'''.join(layer.split('''.'''							)[:3]							) for layer in vae_state_dict if '''encoder.down''' in layer}							)
      __SCREAMING_SNAKE_CASE       :str		    =  {
          layer_id: [key for key in vae_state_dict if f'''down.{layer_id}''' in key] for layer_id in range(a_							)
      }

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

      for i in range(a_							):
            __SCREAMING_SNAKE_CASE       :List[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:
                  __SCREAMING_SNAKE_CASE       :Optional[int]		    =  vae_state_dict.pop(
                      f'''encoder.down.{i}.downsample.conv.weight'''							)
                  __SCREAMING_SNAKE_CASE       :Union[str, Any]		    =  vae_state_dict.pop(
                      f'''encoder.down.{i}.downsample.conv.bias'''							)

            __SCREAMING_SNAKE_CASE       :int		    =  renew_vae_resnet_paths(a_							)
            __SCREAMING_SNAKE_CASE       :str		    =  {'''old''': f'''down.{i}.block''', '''new''': f'''down_blocks.{i}.resnets'''}
            assign_to_checkpoint(a_			,       a_			,       a_			,       additional_replacements=[meta_path]			,       config=a_							)

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

            __SCREAMING_SNAKE_CASE       :List[str]		    =  renew_vae_resnet_paths(a_							)
            __SCREAMING_SNAKE_CASE       :Dict		    =  {'''old''': f'''mid.block_{i}''', '''new''': f'''mid_block.resnets.{i - 1}'''}
            assign_to_checkpoint(a_			,       a_			,       a_			,       additional_replacements=[meta_path]			,       config=a_							)

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

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

            __SCREAMING_SNAKE_CASE       :List[str]		    =  renew_vae_resnet_paths(a_							)
            __SCREAMING_SNAKE_CASE       :Any		    =  {'''old''': f'''up.{block_id}.block''', '''new''': f'''up_blocks.{i}.resnets'''}
            assign_to_checkpoint(a_			,       a_			,       a_			,       additional_replacements=[meta_path]			,       config=a_							)

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

            __SCREAMING_SNAKE_CASE       :Any		    =  renew_vae_resnet_paths(a_							)
            __SCREAMING_SNAKE_CASE       :Union[str, Any]		    =  {'''old''': f'''mid.block_{i}''', '''new''': f'''mid_block.resnets.{i - 1}'''}
            assign_to_checkpoint(a_			,       a_			,       a_			,       additional_replacements=[meta_path]			,       config=a_							)

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



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

      __SCREAMING_SNAKE_CASE       :Dict		    =  OmegaConf.load(a_							)
      __SCREAMING_SNAKE_CASE       :List[Any]		    =  5_12
      __SCREAMING_SNAKE_CASE       :Dict		    =  '''cuda''' if torch.cuda.is_available() else '''cpu'''
      if checkpoint_path.endswith('''safetensors'''							):
            from safetensors import safe_open

            __SCREAMING_SNAKE_CASE       :Optional[Any]		    =  {}
            with safe_open(a_			,       framework='''pt'''			,       device='''cpu'''							) as f:
                  for key in f.keys():
                        __SCREAMING_SNAKE_CASE       :Optional[Any]		    =  f.get_tensor(a_							)
      else:
            __SCREAMING_SNAKE_CASE       :Union[str, Any]		    =  torch.load(a_			,       map_location=a_							)['''state_dict''']

      # Convert the VAE model.
      __SCREAMING_SNAKE_CASE       :Optional[Any]		    =  create_vae_diffusers_config(a_			,       image_size=a_							)
      __SCREAMING_SNAKE_CASE       :List[Any]		    =  custom_convert_ldm_vae_checkpoint(a_			,       a_							)

      __SCREAMING_SNAKE_CASE       :int		    =  AutoencoderKL(**a_							)
      vae.load_state_dict(a_							)
      vae.save_pretrained(a_							)


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

  lowerCamelCase_       =		parser.parse_args()

  vae_pt_to_vae_diffuser(args.vae_pt_path, args.dump_path)






"""simple docstring"""
from ...configuration_utils import PretrainedConfig
from ...utils import logging


lowerCamelCase_       =		logging.get_logger(__name__)

lowerCamelCase_       =		{
    "google/vivit-b-16x2-kinetics400": (
        "https://huggingface.co/google/vivit-b-16x2-kinetics400/resolve/main/config.json"
    ),
    # See all Vivit models at https://huggingface.co/models?filter=vivit
}






class  _SCREAMING_SNAKE_CASE(       A  ):
       SCREAMING_SNAKE_CASE_       :				Optional[Any]     						=  '''vivit'''




       def __init__(						self		,SCREAMING_SNAKE_CASE__=2_24		,SCREAMING_SNAKE_CASE__=32		,SCREAMING_SNAKE_CASE__=[2, 16, 16]		,SCREAMING_SNAKE_CASE__=3		,SCREAMING_SNAKE_CASE__=7_68		,SCREAMING_SNAKE_CASE__=12		,SCREAMING_SNAKE_CASE__=12		,SCREAMING_SNAKE_CASE__=30_72		,SCREAMING_SNAKE_CASE__="gelu_fast"		,SCREAMING_SNAKE_CASE__=0.0		,SCREAMING_SNAKE_CASE__=0.0		,SCREAMING_SNAKE_CASE__=0.0_2		,SCREAMING_SNAKE_CASE__=1E-06		,SCREAMING_SNAKE_CASE__=True		,**SCREAMING_SNAKE_CASE__		,)      ->				str:





             """simple docstring"""






             __SCREAMING_SNAKE_CASE       :Optional[Any]		    =  hidden_size
             __SCREAMING_SNAKE_CASE       :List[str]		    =  num_hidden_layers
             __SCREAMING_SNAKE_CASE       :List[str]		    =  num_attention_heads
             __SCREAMING_SNAKE_CASE       :Any		    =  intermediate_size
             __SCREAMING_SNAKE_CASE       :Optional[Any]		    =  hidden_act
             __SCREAMING_SNAKE_CASE       :int		    =  hidden_dropout_prob
             __SCREAMING_SNAKE_CASE       :Union[str, Any]		    =  attention_probs_dropout_prob
             __SCREAMING_SNAKE_CASE       :Dict		    =  initializer_range
             __SCREAMING_SNAKE_CASE       :Dict		    =  layer_norm_eps

             __SCREAMING_SNAKE_CASE       :Any		    =  image_size
             __SCREAMING_SNAKE_CASE       :Any		    =  num_frames
             __SCREAMING_SNAKE_CASE       :Any		    =  tubelet_size
             __SCREAMING_SNAKE_CASE       :Tuple		    =  num_channels
             __SCREAMING_SNAKE_CASE       :List[str]		    =  qkv_bias

             super().__init__(**SCREAMING_SNAKE_CASE__							)