





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       A__     ( __snake_case    ):
      _UpperCAmelCase			:torch.FloatTensor



class       A__     ( nn.Module    ):





      def __init__(				self ,							A_=3 ,							A_=3 ,							A_=("DownEncoderBlock2D",) ,							A_=(64,) ,							A_=2 ,							A_=32 ,							A_="silu" ,							A_=True ,							):
       '''simple docstring'''




       super().__init__()
       UpperCamelCase			:      int 			=     layers_per_block

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

       UpperCamelCase			:      Dict 			=     None
       UpperCamelCase			:      List[str] 			=     nn.ModuleList([]     )

       # down
       UpperCamelCase			:      Union[str, Any] 			=     block_out_channels[0]
       for i, down_block_type in enumerate(A_     ):
        UpperCamelCase			:      List[str] 			=     output_channel
        UpperCamelCase			:      Optional[int] 			=     block_out_channels[i]
        UpperCamelCase			:      Union[str, Any] 			=     i == len(A_     ) - 1

        UpperCamelCase			:      List[str] 			=     get_down_block(
            A_ ,							num_layers=self.layers_per_block ,							in_channels=A_ ,							out_channels=A_ ,							add_downsample=not is_final_block ,							resnet_eps=1e-6 ,							downsample_padding=0 ,							resnet_act_fn=A_ ,							resnet_groups=A_ ,							attention_head_dim=A_ ,							temb_channels=A_ ,							)
        self.down_blocks.append(A_     )

       # mid
       UpperCamelCase			:      Dict 			=     UNetMidBlockaD(
           in_channels=block_out_channels[-1] ,							resnet_eps=1e-6 ,							resnet_act_fn=A_ ,							output_scale_factor=1 ,							resnet_time_scale_shift="default" ,							attention_head_dim=block_out_channels[-1] ,							resnet_groups=A_ ,							temb_channels=A_ ,							)

       # out
       UpperCamelCase			:      List[str] 			=     nn.GroupNorm(num_channels=block_out_channels[-1] ,							num_groups=A_ ,							eps=1e-6     )
       UpperCamelCase			:      Any 			=     nn.SiLU()

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

       UpperCamelCase			:      Optional[int] 			=     False





      def  __UpperCamelCase(				self ,							A_     ):
       '''simple docstring'''




       UpperCamelCase			:      int 			=     x
       UpperCamelCase			:      List[Any] 			=     self.conv_in(A_     )

       if self.training and self.gradient_checkpointing:

        def create_custom_forward(A_     ):
         def custom_forward(*A_     ):
          return module(*A_     )

         return custom_forward

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

       else:
        # down
        for down_block in self.down_blocks:
         UpperCamelCase			:      Union[str, Any] 			=     down_block(A_     )

        # middle
        UpperCamelCase			:      Optional[Any] 			=     self.mid_block(A_     )

       # post-process
       UpperCamelCase			:      Dict 			=     self.conv_norm_out(A_     )
       UpperCamelCase			:      Union[str, Any] 			=     self.conv_act(A_     )
       UpperCamelCase			:      Optional[Any] 			=     self.conv_out(A_     )

       return sample



class       A__     ( nn.Module    ):





      def __init__(				self ,							A_=3 ,							A_=3 ,							A_=("UpDecoderBlock2D",) ,							A_=(64,) ,							A_=2 ,							A_=32 ,							A_="silu" ,							A_="group" ,							):
       '''simple docstring'''




       super().__init__()
       UpperCamelCase			:      List[str] 			=     layers_per_block

       UpperCamelCase			:      Optional[Any] 			=     nn.Convad(
           A_ ,							block_out_channels[-1] ,							kernel_size=3 ,							stride=1 ,							padding=1 ,							)

       UpperCamelCase			:      Dict 			=     None
       UpperCamelCase			:      Tuple 			=     nn.ModuleList([]     )

       UpperCamelCase			:      Dict 			=     in_channels if norm_type == "spatial" else None

       # mid
       UpperCamelCase			:      Dict 			=     UNetMidBlockaD(
           in_channels=block_out_channels[-1] ,							resnet_eps=1e-6 ,							resnet_act_fn=A_ ,							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=A_ ,							temb_channels=A_ ,							)

       # up
       UpperCamelCase			:      Optional[int] 			=     list(reversed(A_     )     )
       UpperCamelCase			:      Union[str, Any] 			=     reversed_block_out_channels[0]
       for i, up_block_type in enumerate(A_     ):
        UpperCamelCase			:      Tuple 			=     output_channel
        UpperCamelCase			:      Optional[int] 			=     reversed_block_out_channels[i]

        UpperCamelCase			:      str 			=     i == len(A_     ) - 1

        UpperCamelCase			:      int 			=     get_up_block(
            A_ ,							num_layers=self.layers_per_block + 1 ,							in_channels=A_ ,							out_channels=A_ ,							prev_output_channel=A_ ,							add_upsample=not is_final_block ,							resnet_eps=1e-6 ,							resnet_act_fn=A_ ,							resnet_groups=A_ ,							attention_head_dim=A_ ,							temb_channels=A_ ,							resnet_time_scale_shift=A_ ,							)
        self.up_blocks.append(A_     )
        UpperCamelCase			:      Any 			=     output_channel

       # out
       if norm_type == "spatial":
        UpperCamelCase			:      Tuple 			=     SpatialNorm(block_out_channels[0] ,							A_     )
       else:
        UpperCamelCase			:      str 			=     nn.GroupNorm(num_channels=block_out_channels[0] ,							num_groups=A_ ,							eps=1e-6     )
       UpperCamelCase			:      str 			=     nn.SiLU()
       UpperCamelCase			:      str 			=     nn.Convad(block_out_channels[0] ,							A_ ,							3 ,							padding=1     )

       UpperCamelCase			:      Optional[int] 			=     False





      def  __UpperCamelCase(				self ,							A_ ,							A_=None     ):
       '''simple docstring'''




       UpperCamelCase			:      Tuple 			=     z
       UpperCamelCase			:      int 			=     self.conv_in(A_     )

       UpperCamelCase			:      List[str] 			=     next(iter(self.up_blocks.parameters()     )     ).dtype
       if self.training and self.gradient_checkpointing:

        def create_custom_forward(A_     ):
         def custom_forward(*A_     ):
          return module(*A_     )

         return custom_forward

        if is_torch_version(">=" ,							"1.11.0"     ):
         # middle
         UpperCamelCase			:      Any 			=     torch.utils.checkpoint.checkpoint(
             create_custom_forward(self.mid_block     ) ,							A_ ,							A_ ,							use_reentrant=A_     )
         UpperCamelCase			:      Union[str, Any] 			=     sample.to(A_     )

         # up
         for up_block in self.up_blocks:
          UpperCamelCase			:      Union[str, Any] 			=     torch.utils.checkpoint.checkpoint(
              create_custom_forward(A_     ) ,							A_ ,							A_ ,							use_reentrant=A_     )
        else:
         # middle
         UpperCamelCase			:      List[Any] 			=     torch.utils.checkpoint.checkpoint(
             create_custom_forward(self.mid_block     ) ,							A_ ,							A_     )
         UpperCamelCase			:      List[Any] 			=     sample.to(A_     )

         # up
         for up_block in self.up_blocks:
          UpperCamelCase			:      int 			=     torch.utils.checkpoint.checkpoint(create_custom_forward(A_     ) ,							A_ ,							A_     )
       else:
        # middle
        UpperCamelCase			:      str 			=     self.mid_block(A_ ,							A_     )
        UpperCamelCase			:      List[str] 			=     sample.to(A_     )

        # up
        for up_block in self.up_blocks:
         UpperCamelCase			:      Optional[int] 			=     up_block(A_ ,							A_     )

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

       return sample



class       A__     ( nn.Module    ):





      def __init__(				self ,							A_ ,							A_ ,							A_ ,							A_=None ,							A_="random" ,							A_=False ,							A_=True     ):
       '''simple docstring'''




       super().__init__()
       UpperCamelCase			:      List[Any] 			=     n_e
       UpperCamelCase			:      Dict 			=     vq_embed_dim
       UpperCamelCase			:      Optional[Any] 			=     beta
       UpperCamelCase			:      int 			=     legacy

       UpperCamelCase			:      Optional[int] 			=     nn.Embedding(self.n_e ,							self.vq_embed_dim     )
       self.embedding.weight.data.uniform_(-1.0 / self.n_e ,							1.0 / self.n_e     )

       UpperCamelCase			:      str 			=     remap
       if self.remap is not None:
        self.register_buffer("used" ,							torch.tensor(np.load(self.remap     )     )     )
        UpperCamelCase			:      str 			=     self.used.shape[0]
        UpperCamelCase			:      List[Any] 			=     unknown_index  # "random" or "extra" or integer
        if self.unknown_index == "extra":
         UpperCamelCase			:      Tuple 			=     self.re_embed
         UpperCamelCase			:      Optional[Any] 			=     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:
        UpperCamelCase			:      Optional[Any] 			=     n_e

       UpperCamelCase			:      Tuple 			=     sane_index_shape





      def  __UpperCamelCase(				self ,							A_     ):
       '''simple docstring'''




       UpperCamelCase			:      Optional[Any] 			=     inds.shape
       assert len(A_     ) > 1
       UpperCamelCase			:      Dict 			=     inds.reshape(ishape[0] ,							-1     )
       UpperCamelCase			:      Any 			=     self.used.to(A_     )
       UpperCamelCase			:      Optional[int] 			=     (inds[:, :, None] == used[None, None, ...]).long()
       UpperCamelCase			:      Optional[Any] 			=     match.argmax(-1     )
       UpperCamelCase			:      Union[str, Any] 			=     match.sum(2     ) < 1
       if self.unknown_index == "random":
        UpperCamelCase			:      Dict 			=     torch.randint(0 ,							self.re_embed ,							size=new[unknown].shape     ).to(device=new.device     )
       else:
        UpperCamelCase			:      List[Any] 			=     self.unknown_index
       return new.reshape(A_     )





      def  __UpperCamelCase(				self ,							A_     ):
       '''simple docstring'''




       UpperCamelCase			:      Tuple 			=     inds.shape
       assert len(A_     ) > 1
       UpperCamelCase			:      int 			=     inds.reshape(ishape[0] ,							-1     )
       UpperCamelCase			:      Dict 			=     self.used.to(A_     )
       if self.re_embed > self.used.shape[0]:  # extra token
        UpperCamelCase			:      List[Any] 			=     0  # simply set to zero
       UpperCamelCase			:      str 			=     torch.gather(used[None, :][inds.shape[0] * [0], :] ,							1 ,							A_     )
       return back.reshape(A_     )





      def  __UpperCamelCase(				self ,							A_     ):
       '''simple docstring'''




       UpperCamelCase			:      Dict 			=     z.permute(0 ,							2 ,							3 ,							1     ).contiguous()
       UpperCamelCase			:      Any 			=     z.view(-1 ,							self.vq_embed_dim     )

       # distances from z to embeddings e_j (z - e)^2 = z^2 + e^2 - 2 e * z
       UpperCamelCase			:      Optional[int] 			=     torch.argmin(torch.cdist(A_ ,							self.embedding.weight     ) ,							dim=1     )

       UpperCamelCase			:      Optional[int] 			=     self.embedding(A_     ).view(z.shape     )
       UpperCamelCase			:      Dict 			=     None
       UpperCamelCase			:      Optional[Any] 			=     None

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

       # preserve gradients
       UpperCamelCase			:      Dict 			=     z + (z_q - z).detach()

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

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

       if self.sane_index_shape:
        UpperCamelCase			:      Tuple 			=     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  __UpperCamelCase(				self ,							A_ ,							A_     ):
       '''simple docstring'''




       if self.remap is not None:
        UpperCamelCase			:      Dict 			=     indices.reshape(shape[0] ,							-1     )  # add batch axis
        UpperCamelCase			:      List[str] 			=     self.unmap_to_all(A_     )
        UpperCamelCase			:      List[str] 			=     indices.reshape(-1     )  # flatten again

       # get quantized latent vectors
       UpperCamelCase			:      Union[str, Any] 			=     self.embedding(A_     )

       if shape is not None:
        UpperCamelCase			:      int 			=     z_q.view(A_     )
        # reshape back to match original input shape
        UpperCamelCase			:      Union[str, Any] 			=     z_q.permute(0 ,							3 ,							1 ,							2     ).contiguous()

       return z_q



class       A__     ( __snake_case    ):





      def __init__(				self ,							A_ ,							A_=False     ):
       '''simple docstring'''




       UpperCamelCase			:      Optional[Any] 			=     parameters
       UpperCamelCase			,					UpperCamelCase			:      Union[str, Any] 			=     torch.chunk(A_ ,							2 ,							dim=1     )
       UpperCamelCase			:      List[Any] 			=     torch.clamp(self.logvar ,							-30.0 ,							20.0     )
       UpperCamelCase			:      Dict 			=     deterministic
       UpperCamelCase			:      str 			=     torch.exp(0.5 * self.logvar     )
       UpperCamelCase			:      Any 			=     torch.exp(self.logvar     )
       if self.deterministic:
        UpperCamelCase			:      List[Any] 			=     torch.zeros_like(
            self.mean ,							device=self.parameters.device ,							dtype=self.parameters.dtype     )





      def  __UpperCamelCase(				self ,							A_ = None     ):
       '''simple docstring'''




       UpperCamelCase			:      Union[str, Any] 			=     randn_tensor(
           self.mean.shape ,							generator=A_ ,							device=self.parameters.device ,							dtype=self.parameters.dtype     )
       UpperCamelCase			:      Optional[int] 			=     self.mean + self.std * sample
       return x





      def  __UpperCamelCase(				self ,							A_=None     ):
       '''simple docstring'''




       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  __UpperCamelCase(				self ,							A_ ,							A_=[1, 2, 3]     ):
       '''simple docstring'''




       if self.deterministic:
        return torch.Tensor([0.0]     )
       UpperCamelCase			:      str 			=     np.log(2.0 * np.pi     )
       return 0.5 * torch.sum(logtwopi + self.logvar + torch.pow(sample - self.mean ,							2     ) / self.var ,							dim=A_     )





      def  __UpperCamelCase(				self     ):
       '''simple docstring'''




       return self.mean







from typing import List, Optional

from tokenizers import ByteLevelBPETokenizer

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import logging
from .tokenization_blenderbot_small import BlenderbotSmallTokenizer


__lowerCamelCase :  Dict										=  logging.get_logger(__name__)

__lowerCamelCase :  Union[str, Any]										=  {
    """vocab_file""": """vocab.json""",
    """merges_file""": """merges.txt""",
    """tokenizer_config_file""": """tokenizer_config.json""",
}

__lowerCamelCase :  Dict										=  {
    """vocab_file""": {
        """facebook/blenderbot_small-90M""": """https://huggingface.co/facebook/blenderbot_small-90M/resolve/main/vocab.json"""
    },
    """merges_file""": {
        """facebook/blenderbot_small-90M""": """https://huggingface.co/facebook/blenderbot_small-90M/resolve/main/merges.txt"""
    },
    """tokenizer_config_file""": {
        """facebook/blenderbot_small-90M""": (
            """https://huggingface.co/facebook/blenderbot_small-90M/resolve/main/tokenizer_config.json"""
        )
    },
}

__lowerCamelCase :  Tuple										=  {
    """facebook/blenderbot_small-90M""": 512,
}



class       A__     ( __snake_case    ):
      _UpperCAmelCase			:Union[str, Any]							=     VOCAB_FILES_NAMES
      _UpperCAmelCase			:Dict							=     PRETRAINED_VOCAB_FILES_MAP
      _UpperCAmelCase			:List[Any]							=     PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      _UpperCAmelCase			:Optional[Any]							=     BlenderbotSmallTokenizer





      def __init__(				self ,							A_=None ,							A_=None ,							A_="<|endoftext|>" ,							A_="<|endoftext|>" ,							A_="<|endoftext|>" ,							A_=False ,							A_=True ,							**A_ ,							):
       '''simple docstring'''




       super().__init__(
           ByteLevelBPETokenizer(
               vocab=A_ ,							merges=A_ ,							add_prefix_space=A_ ,							trim_offsets=A_ ,							) ,							bos_token=A_ ,							eos_token=A_ ,							unk_token=A_ ,							**A_ ,							)
       UpperCamelCase			:      Union[str, Any] 			=     add_prefix_space





      def  __UpperCamelCase(				self ,							A_ ,							A_=None     ):
       '''simple docstring'''




       UpperCamelCase			:      Dict 			=     [self.bos_token_id] + token_ids_a + [self.eos_token_id]
       if token_ids_a is None:
        return output

       return output + [self.eos_token_id] + token_ids_a + [self.eos_token_id]





      def  __UpperCamelCase(				self ,							A_ ,							A_ = None     ):
       '''simple docstring'''




       UpperCamelCase			:      Tuple 			=     [self.sep_token_id]
       UpperCamelCase			:      int 			=     [self.cls_token_id]

       if token_ids_a is None:
        return len(cls + token_ids_a + sep     ) * [0]
       return len(cls + token_ids_a + sep + sep + token_ids_a + sep     ) * [0]
