






"""simple docstring"""





import torch
from torch import nn


class      _lowercase (     nn.Module						):


      def __init__(    self				:	Optional[Any]							,   a				:	Union[str, Any]							,   a				:	Any							,   a				:	int							,   a				:	str							,   a				:	Any=1							,   a				:	Dict=False						):





            """simple docstring"""







            super().__init__()

            __snake_case      :    Optional[Any]  			=n_token
            __snake_case      :    List[str]  			=d_embed
            __snake_case      :    int  			=d_proj

            __snake_case      :    str  			=cutoffs + [n_token]
            __snake_case      :    Optional[Any]  			=[0] + self.cutoffs
            __snake_case      :    int  			=div_val

            __snake_case      :    Optional[int]  			=self.cutoffs[0]
            __snake_case      :    Optional[int]  			=len(self.cutoffs						) - 1
            __snake_case      :    Optional[Any]  			=self.shortlist_size + self.n_clusters

            if self.n_clusters > 0:
                  __snake_case      :    Dict  			=nn.Parameter(torch.zeros(self.n_clusters							,   self.d_embed						)						)
                  __snake_case      :    Dict  			=nn.Parameter(torch.zeros(self.n_clusters						)						)

            __snake_case      :    Any  			=nn.ModuleList()
            __snake_case      :    Tuple  			=nn.ParameterList()

            if div_val == 1:
                  for i in range(len(self.cutoffs						)						):
                        if d_proj != d_embed:
                              self.out_projs.append(nn.Parameter(torch.FloatTensor(a							,   a						)						)						)
                        else:
                              self.out_projs.append(a						)

                  self.out_layers.append(nn.Linear(a							,   a						)						)
            else:
                  for i in range(len(self.cutoffs						)						):
                        __snake_case      :    Optional[Any]  			=self.cutoff_ends[i], self.cutoff_ends[i + 1]
                        __snake_case      :    str  			=d_embed // (div_val**i)

                        self.out_projs.append(nn.Parameter(torch.FloatTensor(a							,   a						)						)						)

                        self.out_layers.append(nn.Linear(a							,   r_idx - l_idx						)						)

            __snake_case      :    List[Any]  			=keep_order


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





            """simple docstring"""







            if proj is None:
                  __snake_case      :    Optional[Any]  			=nn.functional.linear(a							,   a							,   bias=a						)
            else:
                  # if CUDA_MAJOR <= 9 and CUDA_MINOR <= 1:
                  __snake_case      :    Tuple  			=nn.functional.linear(a							,   proj.t().contiguous()						)
                  __snake_case      :    Tuple  			=nn.functional.linear(a							,   a							,   bias=a						)
                  # else:
                  #     logit = torch.einsum('bd,de,ev->bv', (hidden, proj, weight.t()))
                  #     if bias is not None:
                  #         logit = logit + bias

            return logit


      def _UpperCamelCase						(    self				:	Optional[Any]							,   a				:	Optional[Any]							,   a				:	Union[str, Any]=None							,   a				:	Optional[int]=False						):





            """simple docstring"""







            if labels is not None:
                  # Shift so that tokens < n predict n
                  __snake_case      :    str  			=hidden[..., :-1, :].contiguous()
                  __snake_case      :    int  			=labels[..., 1:].contiguous()
                  __snake_case      :    List[str]  			=hidden.view(-1							,   hidden.size(-1						)						)
                  __snake_case      :    Optional[Any]  			=labels.view(-1						)
                  if hidden.size(0						) != labels.size(0						):
                        raise RuntimeError('''Input and labels should have the same size in the batch dimension.'''						)
            else:
                  __snake_case      :    Dict  			=hidden.view(-1							,   hidden.size(-1						)						)

            if self.n_clusters == 0:
                  __snake_case      :    str  			=self._compute_logit(a							,   self.out_layers[0].weight							,   self.out_layers[0].bias							,   self.out_projs[0]						)
                  if labels is not None:
                        __snake_case      :    Dict  			=labels != -1_0_0
                        __snake_case      :    List[Any]  			=torch.zeros_like(a							,   dtype=hidden.dtype							,   device=hidden.device						)
                        __snake_case      :    int  			=(
                            -nn.functional.log_softmax(a							,   dim=-1						)[mask].gather(1							,   labels[mask].unsqueeze(1						)						).squeeze(1						)
                        )
                  else:
                        __snake_case      :    Tuple  			=nn.functional.log_softmax(a							,   dim=-1						)
            else:
                  # construct weights and biases
                  __snake_case      :    List[str]  			=[], []
                  for i in range(len(self.cutoffs						)						):
                        if self.div_val == 1:
                              __snake_case      :    int  			=self.cutoff_ends[i], self.cutoff_ends[i + 1]
                              __snake_case      :    Union[str, Any]  			=self.out_layers[0].weight[l_idx:r_idx]
                              __snake_case      :    List[str]  			=self.out_layers[0].bias[l_idx:r_idx]
                        else:
                              __snake_case      :    Optional[Any]  			=self.out_layers[i].weight
                              __snake_case      :    int  			=self.out_layers[i].bias

                        if i == 0:
                              __snake_case      :    Optional[int]  			=torch.cat([weight_i, self.cluster_weight]							,   dim=0						)
                              __snake_case      :    List[str]  			=torch.cat([bias_i, self.cluster_bias]							,   dim=0						)

                        weights.append(a						)
                        biases.append(a						)

                  __snake_case      :    int  			=weights[0], biases[0], self.out_projs[0]

                  __snake_case      :    Dict  			=self._compute_logit(a							,   a							,   a							,   a						)
                  __snake_case      :    int  			=nn.functional.log_softmax(a							,   dim=1						)

                  if labels is None:
                        __snake_case      :    List[Any]  			=hidden.new_empty((head_logit.size(0						), self.n_token)						)
                  else:
                        __snake_case      :    List[str]  			=torch.zeros_like(a							,   dtype=hidden.dtype							,   device=hidden.device						)

                  __snake_case      :    str  			=0
                  __snake_case      :    Dict  			=[0] + self.cutoffs
                  for i in range(len(a						) - 1						):
                        __snake_case      :    Any  			=cutoff_values[i], cutoff_values[i + 1]

                        if labels is not None:
                              __snake_case      :    int  			=(labels >= l_idx) & (labels < r_idx)
                              __snake_case      :    Union[str, Any]  			=mask_i.nonzero().squeeze()

                              if indices_i.numel() == 0:
                                    continue

                              __snake_case      :    Union[str, Any]  			=labels.index_select(0							,   a						) - l_idx
                              __snake_case      :    List[str]  			=head_logprob.index_select(0							,   a						)
                              __snake_case      :    Union[str, Any]  			=hidden.index_select(0							,   a						)
                        else:
                              __snake_case      :    Any  			=hidden

                        if i == 0:
                              if labels is not None:
                                    __snake_case      :    Dict  			=head_logprob_i.gather(1							,   target_i[:, None]						).squeeze(1						)
                              else:
                                    __snake_case      :    Optional[Any]  			=head_logprob[:, : self.cutoffs[0]]
                        else:
                              __snake_case      :    Optional[int]  			=weights[i], biases[i], self.out_projs[i]

                              __snake_case      :    Optional[int]  			=self._compute_logit(a							,   a							,   a							,   a						)
                              __snake_case      :    Optional[Any]  			=nn.functional.log_softmax(a							,   dim=1						)
                              __snake_case      :    List[str]  			=self.cutoffs[0] + i - 1  # No probability for the head cluster
                              if labels is not None:
                                    __snake_case      :    Dict  			=head_logprob_i[:, cluster_prob_idx] + tail_logprob_i.gather(
                                        1							,   target_i[:, None]						).squeeze(1						)
                              else:
                                    __snake_case      :    List[Any]  			=head_logprob[:, cluster_prob_idx, None] + tail_logprob_i
                                    __snake_case      :    Tuple  			=logprob_i

                        if labels is not None:
                              if (hasattr(self							,   '''keep_order'''						) and self.keep_order) or keep_order:
                                    out.index_copy_(0							,   a							,   -logprob_i						)
                              else:
                                    out[offset : offset + logprob_i.size(0						)].copy_(-logprob_i						)
                              offset += logprob_i.size(0						)

            return out







      def _UpperCamelCase						(    self				:	List[Any]							,   a				:	Optional[int]						):





            """simple docstring"""







            if self.n_clusters == 0:
                  __snake_case      :    Union[str, Any]  			=self._compute_logit(a							,   self.out_layers[0].weight							,   self.out_layers[0].bias							,   self.out_projs[0]						)
                  return nn.functional.log_softmax(a							,   dim=-1						)
            else:
                  # construct weights and biases
                  __snake_case      :    Optional[Any]  			=[], []
                  for i in range(len(self.cutoffs						)						):
                        if self.div_val == 1:
                              __snake_case      :    Union[str, Any]  			=self.cutoff_ends[i], self.cutoff_ends[i + 1]
                              __snake_case      :    int  			=self.out_layers[0].weight[l_idx:r_idx]
                              __snake_case      :    Tuple  			=self.out_layers[0].bias[l_idx:r_idx]
                        else:
                              __snake_case      :    Optional[int]  			=self.out_layers[i].weight
                              __snake_case      :    Dict  			=self.out_layers[i].bias

                        if i == 0:
                              __snake_case      :    Union[str, Any]  			=torch.cat([weight_i, self.cluster_weight]							,   dim=0						)
                              __snake_case      :    Union[str, Any]  			=torch.cat([bias_i, self.cluster_bias]							,   dim=0						)

                        weights.append(a						)
                        biases.append(a						)

                  __snake_case      :    Dict  			=weights[0], biases[0], self.out_projs[0]
                  __snake_case      :    Optional[int]  			=self._compute_logit(a							,   a							,   a							,   a						)

                  __snake_case      :    Any  			=hidden.new_empty((head_logit.size(0						), self.n_token)						)
                  __snake_case      :    Optional[Any]  			=nn.functional.log_softmax(a							,   dim=1						)

                  __snake_case      :    Tuple  			=[0] + self.cutoffs
                  for i in range(len(a						) - 1						):
                        __snake_case      :    Any  			=cutoff_values[i], cutoff_values[i + 1]

                        if i == 0:
                              __snake_case      :    str  			=head_logprob[:, : self.cutoffs[0]]
                        else:
                              __snake_case      :    Optional[Any]  			=weights[i], biases[i], self.out_projs[i]

                              __snake_case      :    Union[str, Any]  			=self._compute_logit(a							,   a							,   a							,   a						)
                              __snake_case      :    int  			=nn.functional.log_softmax(a							,   dim=1						)

                              __snake_case      :    Dict  			=head_logprob[:, -i] + tail_logprob_i
                              __snake_case      :    Optional[Any]  			=logprob_i

                  return out








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


