"""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 				snake_case__     (   snake_case_   ):
 _snake_case   :			torch.FloatTensor






class 				snake_case__     (   nn.Module   ):


 def __init__(				self  ,						lowerCamelCase=3  ,						lowerCamelCase=3  ,						lowerCamelCase=("DownEncoderBlock2D",)  ,						lowerCamelCase=(64,)  ,						lowerCamelCase=2  ,						lowerCamelCase=32  ,						lowerCamelCase="silu"  ,						lowerCamelCase=True  ,						):
        super().__init__()
        __a     				=  layers_per_block

        __a     				=  torch.nn.Convad(
            lowerCamelCase  ,						block_out_channels[0]  ,						kernel_size=3  ,						stride=1  ,						padding=1  ,						)

        __a     				=  None
        __a     				=  nn.ModuleList([]		)

        # down
        __a     				=  block_out_channels[0]
        for i, down_block_type in enumerate(lowerCamelCase		):
               __a     				=  output_channel
               __a     				=  block_out_channels[i]
               __a     				=  i == len(lowerCamelCase		) - 1

               __a     				=  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
        __a     				=  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
        __a     				=  nn.GroupNorm(num_channels=block_out_channels[-1]  ,						num_groups=lowerCamelCase  ,						eps=1E-6		)
        __a     				=  nn.SiLU()

        __a     				=  2 * out_channels if double_z else out_channels
        __a     				=  nn.Convad(block_out_channels[-1]  ,						lowerCamelCase  ,						3  ,						padding=1		)

        __a     				=  False






 def 		a__ (				self  ,						lowerCamelCase		):
        __a     				=  x
        __a     				=  self.conv_in(lowerCamelCase		)

        if self.training and self.gradient_checkpointing:

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

                      return custom_forward

               # down
               if is_torch_version(">="  ,						"1.11.0"		):
                      for down_block in self.down_blocks:
                             __a     				=  torch.utils.checkpoint.checkpoint(
                                 create_custom_forward(lowerCamelCase		)  ,						lowerCamelCase  ,						use_reentrant=lowerCamelCase		)
                      # middle
                      __a     				=  torch.utils.checkpoint.checkpoint(
                          create_custom_forward(self.mid_block		)  ,						lowerCamelCase  ,						use_reentrant=lowerCamelCase		)
               else:
                      for down_block in self.down_blocks:
                             __a     				=  torch.utils.checkpoint.checkpoint(create_custom_forward(lowerCamelCase		)  ,						lowerCamelCase		)
                      # middle
                      __a     				=  torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block		)  ,						lowerCamelCase		)

        else:
               # down
               for down_block in self.down_blocks:
                      __a     				=  down_block(lowerCamelCase		)

               # middle
               __a     				=  self.mid_block(lowerCamelCase		)

        # post-process
        __a     				=  self.conv_norm_out(lowerCamelCase		)
        __a     				=  self.conv_act(lowerCamelCase		)
        __a     				=  self.conv_out(lowerCamelCase		)

        return sample






class 				snake_case__     (   nn.Module   ):


 def __init__(				self  ,						lowerCamelCase=3  ,						lowerCamelCase=3  ,						lowerCamelCase=("UpDecoderBlock2D",)  ,						lowerCamelCase=(64,)  ,						lowerCamelCase=2  ,						lowerCamelCase=32  ,						lowerCamelCase="silu"  ,						lowerCamelCase="group"  ,						):
        super().__init__()
        __a     				=  layers_per_block

        __a     				=  nn.Convad(
            lowerCamelCase  ,						block_out_channels[-1]  ,						kernel_size=3  ,						stride=1  ,						padding=1  ,						)

        __a     				=  None
        __a     				=  nn.ModuleList([]		)

        __a     				=  in_channels if norm_type == "spatial" else None

        # mid
        __a     				=  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
        __a     				=  list(reversed(lowerCamelCase		)		)
        __a     				=  reversed_block_out_channels[0]
        for i, up_block_type in enumerate(lowerCamelCase		):
               __a     				=  output_channel
               __a     				=  reversed_block_out_channels[i]

               __a     				=  i == len(lowerCamelCase		) - 1

               __a     				=  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		)
               __a     				=  output_channel

        # out
        if norm_type == "spatial":
               __a     				=  SpatialNorm(block_out_channels[0]  ,						lowerCamelCase		)
        else:
               __a     				=  nn.GroupNorm(num_channels=block_out_channels[0]  ,						num_groups=lowerCamelCase  ,						eps=1E-6		)
        __a     				=  nn.SiLU()
        __a     				=  nn.Convad(block_out_channels[0]  ,						lowerCamelCase  ,						3  ,						padding=1		)

        __a     				=  False






 def 		a__ (				self  ,						lowerCamelCase  ,						lowerCamelCase=None		):
        __a     				=  z
        __a     				=  self.conv_in(lowerCamelCase		)

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

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

                      return custom_forward

               if is_torch_version(">="  ,						"1.11.0"		):
                      # middle
                      __a     				=  torch.utils.checkpoint.checkpoint(
                          create_custom_forward(self.mid_block		)  ,						lowerCamelCase  ,						lowerCamelCase  ,						use_reentrant=lowerCamelCase		)
                      __a     				=  sample.to(lowerCamelCase		)

                      # up
                      for up_block in self.up_blocks:
                             __a     				=  torch.utils.checkpoint.checkpoint(
                                 create_custom_forward(lowerCamelCase		)  ,						lowerCamelCase  ,						lowerCamelCase  ,						use_reentrant=lowerCamelCase		)
               else:
                      # middle
                      __a     				=  torch.utils.checkpoint.checkpoint(
                          create_custom_forward(self.mid_block		)  ,						lowerCamelCase  ,						lowerCamelCase		)
                      __a     				=  sample.to(lowerCamelCase		)

                      # up
                      for up_block in self.up_blocks:
                             __a     				=  torch.utils.checkpoint.checkpoint(create_custom_forward(lowerCamelCase		)  ,						lowerCamelCase  ,						lowerCamelCase		)
        else:
               # middle
               __a     				=  self.mid_block(lowerCamelCase  ,						lowerCamelCase		)
               __a     				=  sample.to(lowerCamelCase		)

               # up
               for up_block in self.up_blocks:
                      __a     				=  up_block(lowerCamelCase  ,						lowerCamelCase		)

        # post-process
        if latent_embeds is None:
               __a     				=  self.conv_norm_out(lowerCamelCase		)
        else:
               __a     				=  self.conv_norm_out(lowerCamelCase  ,						lowerCamelCase		)
        __a     				=  self.conv_act(lowerCamelCase		)
        __a     				=  self.conv_out(lowerCamelCase		)

        return sample






class 				snake_case__     (   nn.Module   ):


 def __init__(				self  ,						lowerCamelCase  ,						lowerCamelCase  ,						lowerCamelCase  ,						lowerCamelCase=None  ,						lowerCamelCase="random"  ,						lowerCamelCase=False  ,						lowerCamelCase=True		):
        super().__init__()
        __a     				=  n_e
        __a     				=  vq_embed_dim
        __a     				=  beta
        __a     				=  legacy

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

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

        __a     				=  sane_index_shape


 def 		a__ (				self  ,						lowerCamelCase		):
        __a     				=  inds.shape
        assert len(lowerCamelCase		) > 1
        __a     				=  inds.reshape(ishape[0]  ,						-1		)
        __a     				=  self.used.to(lowerCamelCase		)
        __a     				=  (inds[:, :, None] == used[None, None, ...]).long()
        __a     				=  match.argmax(-1		)
        __a     				=  match.sum(2		) < 1
        if self.unknown_index == "random":
               __a     				=  torch.randint(0  ,						self.re_embed  ,						size=new[unknown].shape		).to(device=new.device		)
        else:
               __a     				=  self.unknown_index
        return new.reshape(lowerCamelCase		)


 def 		a__ (				self  ,						lowerCamelCase		):
        __a     				=  inds.shape
        assert len(lowerCamelCase		) > 1
        __a     				=  inds.reshape(ishape[0]  ,						-1		)
        __a     				=  self.used.to(lowerCamelCase		)
        if self.re_embed > self.used.shape[0]:  # extra token
               __a     				=  0  # simply set to zero
        __a     				=  torch.gather(used[None, :][inds.shape[0] * [0], :]  ,						1  ,						lowerCamelCase		)
        return back.reshape(lowerCamelCase		)


 def 		a__ (				self  ,						lowerCamelCase		):
        # reshape z -> (batch, height, width, channel) and flatten
        __a     				=  z.permute(0  ,						2  ,						3  ,						1		).contiguous()
        __a     				=  z.view(-1  ,						self.vq_embed_dim		)

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

        __a     				=  self.embedding(lowerCamelCase		).view(z.shape		)
        __a     				=  None
        __a     				=  None

        # compute loss for embedding
        if not self.legacy:
               __a     				=  self.beta * torch.mean((z_q.detach() - z) ** 2		) + torch.mean((z_q - z.detach()) ** 2		)
        else:
               __a     				=  torch.mean((z_q.detach() - z) ** 2		) + self.beta * torch.mean((z_q - z.detach()) ** 2		)

        # preserve gradients
        __a     				=  z + (z_q - z).detach()

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

        if self.remap is not None:
               __a     				=  min_encoding_indices.reshape(z.shape[0]  ,						-1		)  # add batch axis
               __a     				=  self.remap_to_used(lowerCamelCase		)
               __a     				=  min_encoding_indices.reshape(-1  ,						1		)  # flatten

        if self.sane_index_shape:
               __a     				=  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 		a__ (				self  ,						lowerCamelCase  ,						lowerCamelCase		):
        # shape specifying (batch, height, width, channel)
        if self.remap is not None:
               __a     				=  indices.reshape(shape[0]  ,						-1		)  # add batch axis
               __a     				=  self.unmap_to_all(lowerCamelCase		)
               __a     				=  indices.reshape(-1		)  # flatten again

        # get quantized latent vectors
        __a     				=  self.embedding(lowerCamelCase		)

        if shape is not None:
               __a     				=  z_q.view(lowerCamelCase		)
               # reshape back to match original input shape
               __a     				=  z_q.permute(0  ,						3  ,						1  ,						2		).contiguous()

        return z_q






class 				snake_case__     (   snake_case_   ):


 def __init__(				self  ,						lowerCamelCase  ,						lowerCamelCase=False		):
        __a     				=  parameters
        __a				, __a     				=  torch.chunk(lowerCamelCase  ,						2  ,						dim=1		)
        __a     				=  torch.clamp(self.logvar  ,						-30.0  ,						20.0		)
        __a     				=  deterministic
        __a     				=  torch.exp(0.5 * self.logvar		)
        __a     				=  torch.exp(self.logvar		)
        if self.deterministic:
               __a     				=  __a     				=  torch.zeros_like(
                   self.mean  ,						device=self.parameters.device  ,						dtype=self.parameters.dtype		)


 def 		a__ (				self  ,						lowerCamelCase = None		):
        # make sure sample is on the same device as the parameters and has same dtype
        __a     				=  randn_tensor(
            self.mean.shape  ,						generator=lowerCamelCase  ,						device=self.parameters.device  ,						dtype=self.parameters.dtype		)
        __a     				=  self.mean + self.std * sample
        return x


 def 		a__ (				self  ,						lowerCamelCase=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 		a__ (				self  ,						lowerCamelCase  ,						lowerCamelCase=[1, 2, 3]		):
        if self.deterministic:
               return torch.Tensor([0.0]		)
        __a     				=  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 		a__ (				self		):
        return self.mean

"""simple docstring"""



import json
import os
from typing import Optional, Tuple

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


SCREAMING_SNAKE_CASE__:int			=	logging.get_logger(__name__)

SCREAMING_SNAKE_CASE__:List[Any]			=	{"""vocab_file""": """vocab.json"""}

SCREAMING_SNAKE_CASE__:Optional[int]			=	{
    """vocab_file""": {
        """mgp-str""": """https://huggingface.co/alibaba-damo/mgp-str-base/blob/main/vocab.json""",
    }
}

SCREAMING_SNAKE_CASE__:Optional[Any]			=	{"""mgp-str""": 27}






class 				snake_case__     (   snake_case_   ):
 _snake_case   :			Tuple           =		VOCAB_FILES_NAMES
 _snake_case   :			Tuple           =		PRETRAINED_VOCAB_FILES_MAP
 _snake_case   :			Optional[Any]           =		PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES


 def __init__(				self  ,						lowerCamelCase  ,						lowerCamelCase="[GO]"  ,						lowerCamelCase="[GO]"  ,						lowerCamelCase="[s]"  ,						lowerCamelCase="[GO]"  ,						**lowerCamelCase		):
        super().__init__(
            unk_token=lowerCamelCase  ,						bos_token=lowerCamelCase  ,						eos_token=lowerCamelCase  ,						pad_token=lowerCamelCase  ,						**lowerCamelCase  ,						)

        with open(lowerCamelCase  ,						encoding="utf-8"		) as vocab_handle:
               __a     				=  json.load(lowerCamelCase		)
        __a     				=  {v: k for k, v in self.vocab.items()}


 @property
 def 		a__ (				self		):
        return len(self.vocab		)


 def 		a__ (				self		):
        return dict(self.vocab  ,						**self.added_tokens_encoder		)


 def 		a__ (				self  ,						lowerCamelCase		):
        __a     				=  []
        for s in text:
               char_tokens.extend(lowerCamelCase		)
        return char_tokens


 def 		a__ (				self  ,						lowerCamelCase		):
        return self.vocab.get(lowerCamelCase  ,						self.vocab.get(self.unk_token		)		)


 def 		a__ (				self  ,						lowerCamelCase		):
        return self.decoder.get(lowerCamelCase		)






 def 		a__ (				self  ,						lowerCamelCase  ,						lowerCamelCase = None		):
        if not os.path.isdir(lowerCamelCase		):
               logger.error("Vocabulary path ({}) should be a directory".format(lowerCamelCase		)		)
               return
        __a     				=  os.path.join(
            lowerCamelCase  ,						(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]		)

        with open(lowerCamelCase  ,						"w"  ,						encoding="utf-8"		) as f:
               f.write(json.dumps(self.vocab  ,						indent=2  ,						sort_keys=lowerCamelCase  ,						ensure_ascii=lowerCamelCase		) + "\n"		)

        return (vocab_file,)
