






"""simple docstring"""





import copy
from typing import TYPE_CHECKING, Any, Mapping, Optional, OrderedDict

from packaging import version

from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfig
from ...utils import logging
from ..auto.configuration_auto import AutoConfig


if TYPE_CHECKING:
    from ... import PreTrainedTokenizerBase, TensorType

UpperCamelCase_	:				List[Any]										= logging.get_logger(__name__)


class      _lowercase (     lowerCAmelCase						):
      _a   :     Optional[int]								=   '''vision-encoder-decoder'''
      _a   :     str								=   True


      def __init__(    self				:	int							,   **a				:	Union[str, Any]						):





            """simple docstring"""







            super().__init__(**a						)
            if "encoder" not in kwargs or "decoder" not in kwargs:
                  raise ValueError(
                      f'''A configuraton of type {self.model_type} cannot be instantiated because '''
                      f'''not both `encoder` and `decoder` sub-configurations are passed, but only {kwargs}'''						)

            __snake_case      :    Union[str, Any]  			=kwargs.pop('''encoder'''						)
            __snake_case      :    List[Any]  			=encoder_config.pop('''model_type'''						)
            __snake_case      :    str  			=kwargs.pop('''decoder'''						)
            __snake_case      :    Optional[Any]  			=decoder_config.pop('''model_type'''						)

            __snake_case      :    Optional[int]  			=AutoConfig.for_model(a							,   **a						)
            __snake_case      :    int  			=AutoConfig.for_model(a							,   **a						)
            __snake_case      :    Optional[int]  			=True


      @classmethod
      def _UpperCamelCase						(    cls				:	List[str]							,   a				:	PretrainedConfig							,   a				:	PretrainedConfig							,   **a				:	Dict						):





            """simple docstring"""







            logger.info('''Setting `config.is_decoder=True` and `config.add_cross_attention=True` for decoder_config'''						)
            __snake_case      :    str  			=True
            __snake_case      :    List[str]  			=True

            return cls(encoder=encoder_config.to_dict()							,   decoder=decoder_config.to_dict()							,   **a						)







      def _UpperCamelCase						(    self				:	Optional[int]						):





            """simple docstring"""







            __snake_case      :    Optional[Any]  			=copy.deepcopy(self.__dict__						)
            __snake_case      :    Tuple  			=self.encoder.to_dict()
            __snake_case      :    Union[str, Any]  			=self.decoder.to_dict()
            __snake_case      :    str  			=self.__class__.model_type
            return output



class      _lowercase (     lowerCAmelCase						):
      _a   :     List[Any]								=   version.parse('''1.11'''						)


      @property
      def _UpperCamelCase						(    self				:	Tuple						):





            """simple docstring"""







            return OrderedDict(
                [
                    ('''pixel_values''', {0: '''batch''', 1: '''num_channels''', 2: '''height''', 3: '''width'''}),
                ]						)


      @property
      def _UpperCamelCase						(    self				:	Tuple						):





            """simple docstring"""







            return 1e-4







      @property
      def _UpperCamelCase						(    self				:	Dict						):





            """simple docstring"""







            return OrderedDict({'''last_hidden_state''': {0: '''batch''', 1: '''encoder_sequence'''}}						)



class      _lowercase (     lowerCAmelCase						):


      @property
      def _UpperCamelCase						(    self				:	Any						):





            """simple docstring"""







            __snake_case      :    Dict  			=OrderedDict()
            __snake_case      :    Union[str, Any]  			={0: '''batch''', 1: '''past_decoder_sequence + sequence'''}
            __snake_case      :    Optional[int]  			={0: '''batch''', 1: '''past_decoder_sequence + sequence'''}
            __snake_case      :    List[Any]  			={0: '''batch''', 1: '''encoder_sequence'''}

            return common_inputs







      def _UpperCamelCase						(    self				:	List[Any]							,   a				:	"PreTrainedTokenizerBase"							,   a				:	int = -1							,   a				:	int = -1							,   a				:	bool = False							,   a				:	Optional["TensorType"] = None							,   ):





            """simple docstring"""







            import torch

            __snake_case      :    Union[str, Any]  			=OrderedDict()

            __snake_case      :    Any  			=super().generate_dummy_inputs(
                a							,   batch_size=a							,   seq_length=a							,   is_pair=a							,   framework=a						)

            __snake_case      :    List[Any]  			=dummy_input['''input_ids'''].shape
            __snake_case      :    int  			=(batch, encoder_sequence, self._config.encoder_hidden_size)
            __snake_case      :    Dict  			=dummy_input.pop('''input_ids'''						)
            __snake_case      :    Dict  			=dummy_input.pop('''attention_mask'''						)
            __snake_case      :    Union[str, Any]  			=torch.zeros(a						)

            return common_inputs



class      _lowercase (     lowerCAmelCase						):


      @property
      def _UpperCamelCase						(    self				:	List[str]						):





            """simple docstring"""







            pass


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





            """simple docstring"""







            return VisionEncoderDecoderEncoderOnnxConfig(a						)







      def _UpperCamelCase						(    self				:	List[Any]							,   a				:	PretrainedConfig							,   a				:	PretrainedConfig							,   a				:	str = "default"						):





            """simple docstring"""







            __snake_case      :    Optional[Any]  			=encoder_config.hidden_size
            return VisionEncoderDecoderDecoderOnnxConfig(a							,   a						)








"""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     __lowercase					(    a :				Any       ,		a :				List[Any] )					->					Dict:
      __snake_case      :    Any  			=checkpoint

      __snake_case      :    Dict  			={}

      __snake_case      :    List[Any]  			=vae_state_dict['''encoder.conv_in.weight''']
      __snake_case      :    List[str]  			=vae_state_dict['''encoder.conv_in.bias''']
      __snake_case      :    Union[str, Any]  			=vae_state_dict['''encoder.conv_out.weight''']
      __snake_case      :    Union[str, Any]  			=vae_state_dict['''encoder.conv_out.bias''']
      __snake_case      :    str  			=vae_state_dict['''encoder.norm_out.weight''']
      __snake_case      :    str  			=vae_state_dict['''encoder.norm_out.bias''']

      __snake_case      :    Tuple  			=vae_state_dict['''decoder.conv_in.weight''']
      __snake_case      :    str  			=vae_state_dict['''decoder.conv_in.bias''']
      __snake_case      :    List[str]  			=vae_state_dict['''decoder.conv_out.weight''']
      __snake_case      :    Tuple  			=vae_state_dict['''decoder.conv_out.bias''']
      __snake_case      :    Union[str, Any]  			=vae_state_dict['''decoder.norm_out.weight''']
      __snake_case      :    List[str]  			=vae_state_dict['''decoder.norm_out.bias''']

      __snake_case      :    Tuple  			=vae_state_dict['''quant_conv.weight''']
      __snake_case      :    List[str]  			=vae_state_dict['''quant_conv.bias''']
      __snake_case      :    Optional[int]  			=vae_state_dict['''post_quant_conv.weight''']
      __snake_case      :    Tuple  			=vae_state_dict['''post_quant_conv.bias''']

      # Retrieves the keys for the encoder down blocks only
      __snake_case      :    Union[str, Any]  			=len({'''.'''.join(layer.split('''.''' )[:3] ) for layer in vae_state_dict if '''encoder.down''' in layer} )
      __snake_case      :    Union[str, Any]  			={
          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
      __snake_case      :    List[Any]  			=len({'''.'''.join(layer.split('''.''' )[:3] ) for layer in vae_state_dict if '''decoder.up''' in layer} )
      __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 ):
            __snake_case      :    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:
                  __snake_case      :    Optional[Any]  			=vae_state_dict.pop(
                      f'''encoder.down.{i}.downsample.conv.weight''' )
                  __snake_case      :    Optional[int]  			=vae_state_dict.pop(
                      f'''encoder.down.{i}.downsample.conv.bias''' )

            __snake_case      :    Optional[Any]  			=renew_vae_resnet_paths(a )
            __snake_case      :    Union[str, Any]  			={'''old''': f'''down.{i}.block''', '''new''': f'''down_blocks.{i}.resnets'''}
            assign_to_checkpoint(a       ,		a       ,		a       ,		additional_replacements=[meta_path]       ,		config=a )

      __snake_case      :    Tuple  			=[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      :    Tuple  			=[key for key in mid_resnets if f'''encoder.mid.block_{i}''' in key]

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

      __snake_case      :    int  			=[key for key in vae_state_dict if '''encoder.mid.attn''' in key]
      __snake_case      :    List[Any]  			=renew_vae_attention_paths(a )
      __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 )

      for i in range(a ):
            __snake_case      :    List[Any]  			=num_up_blocks - 1 - i
            __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:
                  __snake_case      :    int  			=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      :    List[str]  			=renew_vae_resnet_paths(a )
            __snake_case      :    int  			={'''old''': f'''up.{block_id}.block''', '''new''': f'''up_blocks.{i}.resnets'''}
            assign_to_checkpoint(a       ,		a       ,		a       ,		additional_replacements=[meta_path]       ,		config=a )

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

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

      __snake_case      :    Optional[Any]  			=[key for key in vae_state_dict if '''decoder.mid.attn''' in key]
      __snake_case      :    Dict  			=renew_vae_attention_paths(a )
      __snake_case      :    List[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 )
      return new_checkpoint



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

      __snake_case      :    Any  			=OmegaConf.load(a )
      __snake_case      :    Union[str, Any]  			=512
      __snake_case      :    List[str]  			='''cuda''' if torch.cuda.is_available() else '''cpu'''
      if checkpoint_path.endswith('''safetensors''' ):
            from safetensors import safe_open

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

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

      __snake_case      :    str  			=AutoencoderKL(**a )
      vae.load_state_dict(a )
      vae.save_pretrained(a )


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

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

    vae_pt_to_vae_diffuser(args.vae_pt_path, args.dump_path)


