
"""simple docstring"""
from dataclasses import dataclass
from typing import Optional

import numpy as np
import torch
import torch.nn as nn

from ..utils import BaseOutput, is_torch_version, randn_tensor
from .attention_processor import SpatialNorm
from .unet_ad_blocks import UNetMidBlockaD, get_down_block, get_up_block






@dataclass
class      _UpperCAmelCase				(					a__):
    __lowercase							: torch.FloatTensor






class      _UpperCAmelCase				(					nn.Module):
    def __init__(    self						,    snake_case_=3						,    snake_case_=3						,    snake_case_=("DownEncoderBlock2D",)						,    snake_case_=(64,)						,    snake_case_=2						,    snake_case_=32						,    snake_case_="silu"						,    snake_case_=True						,    ):
      super().__init__()
      _snake_case     :						Tuple						=   layers_per_block

      _snake_case     :						List[str]						=   torch.nn.Convad(
          lowerCAmelCase__						,    block_out_channels[0]						,    kernel_size=3						,    stride=1						,    padding=1						,    )

      _snake_case     :						str						=   None
      _snake_case     :						Any						=   nn.ModuleList([]     )

      # down
      _snake_case     :						Tuple						=   block_out_channels[0]
      for i, down_block_type in enumerate(lowerCAmelCase__     ):
        _snake_case     :						Dict						=   output_channel
        _snake_case     :						str						=   block_out_channels[i]
        _snake_case     :						int						=   i == len(lowerCAmelCase__     ) - 1

        _snake_case     :						List[Any]						=   get_down_block(
            lowerCAmelCase__						,    num_layers=self.layers_per_block						,    in_channels=lowerCAmelCase__						,    out_channels=lowerCAmelCase__						,    add_downsample=not is_final_block						,    resnet_eps=1E-6						,    downsample_padding=0						,    resnet_act_fn=lowerCAmelCase__						,    resnet_groups=lowerCAmelCase__						,    attention_head_dim=lowerCAmelCase__						,    temb_channels=lowerCAmelCase__						,    )
        self.down_blocks.append(lowerCAmelCase__     )

      # mid
      _snake_case     :						str						=   UNetMidBlockaD(
          in_channels=block_out_channels[-1]						,    resnet_eps=1E-6						,    resnet_act_fn=lowerCAmelCase__						,    output_scale_factor=1						,    resnet_time_scale_shift="default"						,    attention_head_dim=block_out_channels[-1]						,    resnet_groups=lowerCAmelCase__						,    temb_channels=lowerCAmelCase__						,    )

      # out
      _snake_case     :						Tuple						=   nn.GroupNorm(num_channels=block_out_channels[-1]						,    num_groups=lowerCAmelCase__						,    eps=1E-6     )
      _snake_case     :						Dict						=   nn.SiLU()

      _snake_case     :						Union[str, Any]						=   2 * out_channels if double_z else out_channels
      _snake_case     :						List[Any]						=   nn.Convad(block_out_channels[-1]						,    lowerCAmelCase__						,    3						,    padding=1     )

      _snake_case     :						List[str]						=   False


    def 					lowerCamelCase__	(    self						,    snake_case_     ):
      _snake_case     :						Optional[Any]						=   x
      _snake_case     :						Dict						=   self.conv_in(lowerCAmelCase__     )

      if self.training and self.gradient_checkpointing:

        def create_custom_forward(snake_case_     ):
          def custom_forward(*snake_case_     ):
            return module(*lowerCAmelCase__     )

          return custom_forward

        # down
        if is_torch_version(">="						,    "1.11.0"     ):
          for down_block in self.down_blocks:
            _snake_case     :						List[Any]						=   torch.utils.checkpoint.checkpoint(
                create_custom_forward(lowerCAmelCase__     )						,    lowerCAmelCase__						,    use_reentrant=lowerCAmelCase__     )
          # middle
          _snake_case     :						List[str]						=   torch.utils.checkpoint.checkpoint(
              create_custom_forward(self.mid_block     )						,    lowerCAmelCase__						,    use_reentrant=lowerCAmelCase__     )
        else:
          for down_block in self.down_blocks:
            _snake_case     :						List[Any]						=   torch.utils.checkpoint.checkpoint(create_custom_forward(lowerCAmelCase__     )						,    lowerCAmelCase__     )
          # middle
          _snake_case     :						Optional[int]						=   torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block     )						,    lowerCAmelCase__     )

      else:
        # down
        for down_block in self.down_blocks:
          _snake_case     :						Optional[int]						=   down_block(lowerCAmelCase__     )

        # middle
        _snake_case     :						int						=   self.mid_block(lowerCAmelCase__     )

      # post-process
      _snake_case     :						Tuple						=   self.conv_norm_out(lowerCAmelCase__     )
      _snake_case     :						int						=   self.conv_act(lowerCAmelCase__     )
      _snake_case     :						str						=   self.conv_out(lowerCAmelCase__     )

      return sample






class      _UpperCAmelCase				(					nn.Module):
    def __init__(    self						,    snake_case_=3						,    snake_case_=3						,    snake_case_=("UpDecoderBlock2D",)						,    snake_case_=(64,)						,    snake_case_=2						,    snake_case_=32						,    snake_case_="silu"						,    snake_case_="group"						,    ):
      super().__init__()
      _snake_case     :						List[str]						=   layers_per_block

      _snake_case     :						Optional[int]						=   nn.Convad(
          lowerCAmelCase__						,    block_out_channels[-1]						,    kernel_size=3						,    stride=1						,    padding=1						,    )

      _snake_case     :						int						=   None
      _snake_case     :						Union[str, Any]						=   nn.ModuleList([]     )

      _snake_case     :						Optional[int]						=   in_channels if norm_type == "spatial" else None

      # mid
      _snake_case     :						Any						=   UNetMidBlockaD(
          in_channels=block_out_channels[-1]						,    resnet_eps=1E-6						,    resnet_act_fn=lowerCAmelCase__						,    output_scale_factor=1						,    resnet_time_scale_shift="default" if norm_type == "group" else norm_type						,    attention_head_dim=block_out_channels[-1]						,    resnet_groups=lowerCAmelCase__						,    temb_channels=lowerCAmelCase__						,    )

      # up
      _snake_case     :						List[str]						=   list(reversed(lowerCAmelCase__     )     )
      _snake_case     :						List[Any]						=   reversed_block_out_channels[0]
      for i, up_block_type in enumerate(lowerCAmelCase__     ):
        _snake_case     :						str						=   output_channel
        _snake_case     :						Tuple						=   reversed_block_out_channels[i]

        _snake_case     :						Any						=   i == len(lowerCAmelCase__     ) - 1

        _snake_case     :						str						=   get_up_block(
            lowerCAmelCase__						,    num_layers=self.layers_per_block + 1						,    in_channels=lowerCAmelCase__						,    out_channels=lowerCAmelCase__						,    prev_output_channel=lowerCAmelCase__						,    add_upsample=not is_final_block						,    resnet_eps=1E-6						,    resnet_act_fn=lowerCAmelCase__						,    resnet_groups=lowerCAmelCase__						,    attention_head_dim=lowerCAmelCase__						,    temb_channels=lowerCAmelCase__						,    resnet_time_scale_shift=lowerCAmelCase__						,    )
        self.up_blocks.append(lowerCAmelCase__     )
        _snake_case     :						List[str]						=   output_channel

      # out
      if norm_type == "spatial":
        _snake_case     :						Optional[Any]						=   SpatialNorm(block_out_channels[0]						,    lowerCAmelCase__     )
      else:
        _snake_case     :						Optional[Any]						=   nn.GroupNorm(num_channels=block_out_channels[0]						,    num_groups=lowerCAmelCase__						,    eps=1E-6     )
      _snake_case     :						int						=   nn.SiLU()
      _snake_case     :						Union[str, Any]						=   nn.Convad(block_out_channels[0]						,    lowerCAmelCase__						,    3						,    padding=1     )

      _snake_case     :						int						=   False


    def 					lowerCamelCase__	(    self						,    snake_case_						,    snake_case_=None     ):
      _snake_case     :						Any						=   z
      _snake_case     :						Union[str, Any]						=   self.conv_in(lowerCAmelCase__     )

      _snake_case     :						Any						=   next(iter(self.up_blocks.parameters()     )     ).dtype
      if self.training and self.gradient_checkpointing:

        def create_custom_forward(snake_case_     ):
          def custom_forward(*snake_case_     ):
            return module(*lowerCAmelCase__     )

          return custom_forward

        if is_torch_version(">="						,    "1.11.0"     ):
          # middle
          _snake_case     :						Optional[Any]						=   torch.utils.checkpoint.checkpoint(
              create_custom_forward(self.mid_block     )						,    lowerCAmelCase__						,    lowerCAmelCase__						,    use_reentrant=lowerCAmelCase__     )
          _snake_case     :						List[Any]						=   sample.to(lowerCAmelCase__     )

          # up
          for up_block in self.up_blocks:
            _snake_case     :						str						=   torch.utils.checkpoint.checkpoint(
                create_custom_forward(lowerCAmelCase__     )						,    lowerCAmelCase__						,    lowerCAmelCase__						,    use_reentrant=lowerCAmelCase__     )
        else:
          # middle
          _snake_case     :						Optional[Any]						=   torch.utils.checkpoint.checkpoint(
              create_custom_forward(self.mid_block     )						,    lowerCAmelCase__						,    lowerCAmelCase__     )
          _snake_case     :						int						=   sample.to(lowerCAmelCase__     )

          # up
          for up_block in self.up_blocks:
            _snake_case     :						Dict						=   torch.utils.checkpoint.checkpoint(create_custom_forward(lowerCAmelCase__     )						,    lowerCAmelCase__						,    lowerCAmelCase__     )
      else:
        # middle
        _snake_case     :						List[Any]						=   self.mid_block(lowerCAmelCase__						,    lowerCAmelCase__     )
        _snake_case     :						int						=   sample.to(lowerCAmelCase__     )

        # up
        for up_block in self.up_blocks:
          _snake_case     :						Optional[Any]						=   up_block(lowerCAmelCase__						,    lowerCAmelCase__     )

        # post-process
      if latent_embeds is None:
        _snake_case     :						str						=   self.conv_norm_out(lowerCAmelCase__     )
      else:
        _snake_case     :						Optional[int]						=   self.conv_norm_out(lowerCAmelCase__						,    lowerCAmelCase__     )
      _snake_case     :						Any						=   self.conv_act(lowerCAmelCase__     )
      _snake_case     :						Tuple						=   self.conv_out(lowerCAmelCase__     )

      return sample






class      _UpperCAmelCase				(					nn.Module):
    def __init__(    self						,    snake_case_						,    snake_case_						,    snake_case_						,    snake_case_=None						,    snake_case_="random"						,    snake_case_=False						,    snake_case_=True     ):
      super().__init__()
      _snake_case     :						Union[str, Any]						=   n_e
      _snake_case     :						Optional[Any]						=   vq_embed_dim
      _snake_case     :						int						=   beta
      _snake_case     :						Any						=   legacy

      _snake_case     :						str						=   nn.Embedding(self.n_e						,    self.vq_embed_dim     )
      self.embedding.weight.data.uniform_(-1.0 / self.n_e						,    1.0 / self.n_e     )

      _snake_case     :						str						=   remap
      if self.remap is not None:
        self.register_buffer("used"						,    torch.tensor(np.load(self.remap     )     )     )
        _snake_case     :						List[Any]						=   self.used.shape[0]
        _snake_case     :						Optional[int]						=   unknown_index  # "random" or "extra" or integer
        if self.unknown_index == "extra":
          _snake_case     :						List[str]						=   self.re_embed
          _snake_case     :						List[str]						=   self.re_embed + 1
        print(
            F'Remapping {self.n_e} indices to {self.re_embed} indices. '
            F'Using {self.unknown_index} for unknown indices.'     )
      else:
        _snake_case     :						str						=   n_e

      _snake_case     :						Union[str, Any]						=   sane_index_shape
    def 					lowerCamelCase__	(    self						,    snake_case_     ):
      _snake_case     :						List[Any]						=   inds.shape
      assert len(lowerCAmelCase__     ) > 1
      _snake_case     :						Union[str, Any]						=   inds.reshape(ishape[0]						,    -1     )
      _snake_case     :						Optional[Any]						=   self.used.to(lowerCAmelCase__     )
      _snake_case     :						List[str]						=   (inds[:, :, None] == used[None, None, ...]).long()
      _snake_case     :						str						=   match.argmax(-1     )
      _snake_case     :						int						=   match.sum(2     ) < 1
      if self.unknown_index == "random":
        _snake_case     :						Optional[int]						=   torch.randint(0						,    self.re_embed						,    size=new[unknown].shape     ).to(device=new.device     )
      else:
        _snake_case     :						Optional[Any]						=   self.unknown_index
      return new.reshape(lowerCAmelCase__     )
    def 					lowerCamelCase__	(    self						,    snake_case_     ):
      _snake_case     :						str						=   inds.shape
      assert len(lowerCAmelCase__     ) > 1
      _snake_case     :						Tuple						=   inds.reshape(ishape[0]						,    -1     )
      _snake_case     :						Dict						=   self.used.to(lowerCAmelCase__     )
      if self.re_embed > self.used.shape[0]:  # extra token
        _snake_case     :						Optional[Any]						=   0  # simply set to zero
      _snake_case     :						Optional[Any]						=   torch.gather(used[None, :][inds.shape[0] * [0], :]						,    1						,    lowerCAmelCase__     )
      return back.reshape(lowerCAmelCase__     )
    def 					lowerCamelCase__	(    self						,    snake_case_     ):
      # reshape z -> (batch, height, width, channel) and flatten
      _snake_case     :						Tuple						=   z.permute(0						,    2						,    3						,    1     ).contiguous()
      _snake_case     :						Optional[int]						=   z.view(-1						,    self.vq_embed_dim     )

      # distances from z to embeddings e_j (z - e)^2 = z^2 + e^2 - 2 e * z
      _snake_case     :						Tuple						=   torch.argmin(torch.cdist(lowerCAmelCase__						,    self.embedding.weight     )						,    dim=1     )

      _snake_case     :						Dict						=   self.embedding(lowerCAmelCase__     ).view(z.shape     )
      _snake_case     :						Dict						=   None
      _snake_case     :						Tuple						=   None

      # compute loss for embedding
      if not self.legacy:
        _snake_case     :						Optional[int]						=   self.beta * torch.mean((z_q.detach() - z) ** 2     ) + torch.mean((z_q - z.detach()) ** 2     )
      else:
        _snake_case     :						Union[str, Any]						=   torch.mean((z_q.detach() - z) ** 2     ) + self.beta * torch.mean((z_q - z.detach()) ** 2     )

      # preserve gradients
      _snake_case     :						Optional[int]						=   z + (z_q - z).detach()

      # reshape back to match original input shape
      _snake_case     :						Dict						=   z_q.permute(0						,    3						,    1						,    2     ).contiguous()

      if self.remap is not None:
        _snake_case     :						Optional[Any]						=   min_encoding_indices.reshape(z.shape[0]						,    -1     )  # add batch axis
        _snake_case     :						Optional[Any]						=   self.remap_to_used(lowerCAmelCase__     )
        _snake_case     :						List[Any]						=   min_encoding_indices.reshape(-1						,    1     )  # flatten

      if self.sane_index_shape:
        _snake_case     :						List[Any]						=   min_encoding_indices.reshape(z_q.shape[0]						,    z_q.shape[2]						,    z_q.shape[3]     )

      return z_q, loss, (perplexity, min_encodings, min_encoding_indices)


    def 					lowerCamelCase__	(    self						,    snake_case_						,    snake_case_     ):
      # shape specifying (batch, height, width, channel)
      if self.remap is not None:
        _snake_case     :						List[str]						=   indices.reshape(shape[0]						,    -1     )  # add batch axis
        _snake_case     :						Optional[Any]						=   self.unmap_to_all(lowerCAmelCase__     )
        _snake_case     :						Any						=   indices.reshape(-1     )  # flatten again

      # get quantized latent vectors
      _snake_case     :						str						=   self.embedding(lowerCAmelCase__     )

      if shape is not None:
        _snake_case     :						List[str]						=   z_q.view(lowerCAmelCase__     )
        # reshape back to match original input shape
        _snake_case     :						Optional[int]						=   z_q.permute(0						,    3						,    1						,    2     ).contiguous()

      return z_q






class      _UpperCAmelCase				(					a__):
    def __init__(    self						,    snake_case_						,    snake_case_=False     ):
      _snake_case     :						Any						=   parameters
      _snake_case     :						Tuple						=   torch.chunk(lowerCAmelCase__						,    2						,    dim=1     )
      _snake_case     :						Optional[Any]						=   torch.clamp(self.logvar						,    -30.0						,    20.0     )
      _snake_case     :						int						=   deterministic
      _snake_case     :						Optional[Any]						=   torch.exp(0.5 * self.logvar     )
      _snake_case     :						Optional[int]						=   torch.exp(self.logvar     )
      if self.deterministic:
        _snake_case     :						int						=   torch.zeros_like(
            self.mean						,    device=self.parameters.device						,    dtype=self.parameters.dtype     )
    def 					lowerCamelCase__	(    self						,    snake_case_ = None     ):
      # make sure sample is on the same device as the parameters and has same dtype
      _snake_case     :						List[str]						=   randn_tensor(
          self.mean.shape						,    generator=lowerCAmelCase__						,    device=self.parameters.device						,    dtype=self.parameters.dtype     )
      _snake_case     :						List[str]						=   self.mean + self.std * sample
      return x
    def 					lowerCamelCase__	(    self						,    snake_case_=None     ):
      if self.deterministic:
        return torch.Tensor([0.0]     )
      else:
        if other is None:
          return 0.5 * torch.sum(torch.pow(self.mean						,    2     ) + self.var - 1.0 - self.logvar						,    dim=[1, 2, 3]     )
        else:
          return 0.5 * torch.sum(
              torch.pow(self.mean - other.mean						,    2     ) / other.var
              + self.var / other.var
              - 1.0
              - self.logvar
              + other.logvar						,    dim=[1, 2, 3]						,    )
    def 					lowerCamelCase__	(    self						,    snake_case_						,    snake_case_=[1, 2, 3]     ):
      if self.deterministic:
        return torch.Tensor([0.0]     )
      _snake_case     :						List[str]						=   np.log(2.0 * np.pi     )
      return 0.5 * torch.sum(logtwopi + self.logvar + torch.pow(sample - self.mean						,    2     ) / self.var						,    dim=lowerCAmelCase__     )


    def 					lowerCamelCase__	(    self     ):
      return self.mean



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


_a      :			Optional[int]					=							logging.get_logger(__name__)

_a      :			str					=							{"""openai-gpt""": """https://huggingface.co/openai-gpt/resolve/main/config.json"""}






class      _UpperCAmelCase				(					_snake_case):
    __lowercase							: Optional[Any]  =					"""openai-gpt"""
    __lowercase							: Dict  =					{
        """max_position_embeddings""": """n_positions""",
        """hidden_size""": """n_embd""",
        """num_attention_heads""": """n_head""",
        """num_hidden_layers""": """n_layer""",
    }
    def __init__(    self						,    snake_case_=4_04_78						,    snake_case_=5_12						,    snake_case_=7_68						,    snake_case_=12						,    snake_case_=12						,    snake_case_="gelu"						,    snake_case_=0.1						,    snake_case_=0.1						,    snake_case_=0.1						,    snake_case_=1E-5						,    snake_case_=0.02						,    snake_case_="cls_index"						,    snake_case_=True						,    snake_case_=None						,    snake_case_=True						,    snake_case_=0.1						,    **snake_case_						,    ):
      _snake_case     :						Tuple						=   vocab_size
      _snake_case     :						Dict						=   n_positions
      _snake_case     :						Any						=   n_embd
      _snake_case     :						Any						=   n_layer
      _snake_case     :						Optional[int]						=   n_head
      _snake_case     :						Union[str, Any]						=   afn
      _snake_case     :						Dict						=   resid_pdrop
      _snake_case     :						str						=   embd_pdrop
      _snake_case     :						Union[str, Any]						=   attn_pdrop
      _snake_case     :						str						=   layer_norm_epsilon
      _snake_case     :						Union[str, Any]						=   initializer_range
      _snake_case     :						Any						=   summary_type
      _snake_case     :						List[str]						=   summary_use_proj
      _snake_case     :						Optional[int]						=   summary_activation
      _snake_case     :						Union[str, Any]						=   summary_first_dropout
      _snake_case     :						Optional[int]						=   summary_proj_to_labels
      super().__init__(**snake_case_     )

