



from dataclasses import dataclass
from typing import Any, Dict, Optional

import torch
import torch.nn.functional as F
from torch import nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..models.embeddings import ImagePositionalEmbeddings
from ..utils import BaseOutput, deprecate
from .attention import BasicTransformerBlock
from .embeddings import PatchEmbed
from .modeling_utils import ModelMixin

@dataclass
class 						lowerCamelCase_     (			_A    ):






						'''simple docstring'''
						a__ =	42







class 						lowerCamelCase_     (			_A    ,_A    ):






						'''simple docstring'''







						@register_to_config
						def __init__(						self   :      int						,						__lowerCamelCase   :      int = 16						,						__lowerCamelCase   :      int = 88						,						__lowerCamelCase   :      Optional[int] = None						,						__lowerCamelCase   :      Optional[int] = None						,						__lowerCamelCase   :      int = 1						,						__lowerCamelCase   :      float = 0.0						,						__lowerCamelCase   :      int = 32						,						__lowerCamelCase   :      Optional[int] = None						,						__lowerCamelCase   :      bool = False						,						__lowerCamelCase   :      Optional[int] = None						,						__lowerCamelCase   :      Optional[int] = None						,						__lowerCamelCase   :      Optional[int] = None						,						__lowerCamelCase   :      str = "geglu"						,						__lowerCamelCase   :      Optional[int] = None						,						__lowerCamelCase   :      bool = False						,						__lowerCamelCase   :      bool = False						,						__lowerCamelCase   :      bool = False						,						__lowerCamelCase   :      str = "layer_norm"						,						__lowerCamelCase   :      bool = True						,						)   ->					Optional[Any]:
									super().__init__()
									A				:       Any														=    use_linear_projection
									A				:       List[Any]														=    num_attention_heads
									A				:       Tuple														=    attention_head_dim
									A				:       Any														=    num_attention_heads * attention_head_dim

									# 1. Transformer2DModel can process both standard continuous images of shape `(batch_size, num_channels, width, height)` as well as quantized image embeddings of shape `(batch_size, num_image_vectors)`
									# Define whether input is continuous or discrete depending on configuration
									A				:       Optional[Any]														=    (in_channels is not None) and (patch_size is None)
									A				:       Union[str, Any]														=    num_vector_embeds is not None
									A				:       str														=    in_channels is not None and patch_size is not None

									if norm_type == "layer_norm" and num_embeds_ada_norm is not None:
												A				:       List[str]														=    (
												    F"""The configuration file of this model: {self.__class__} is outdated. `norm_type` is either not set or"""
												    " incorrectly set to `'layer_norm'`.Make sure to set `norm_type` to `'ada_norm'` in the config."
												    " Please make sure to update the config accordingly as leaving `norm_type` might led to incorrect"
												    " results in future versions. If you have downloaded this checkpoint from the Hugging Face Hub, it"
												    " would be very nice if you could open a Pull request for the `transformer/config.json` file"
												)
												deprecate("norm_type!=num_embeds_ada_norm"						,						"1.0.0"						,						__lowerCamelCase						,						standard_warn=__lowerCamelCase			)
												A				:       List[Any]														=    "ada_norm"

									if self.is_input_continuous and self.is_input_vectorized:
												raise ValueError(
												    F"""Cannot define both `in_channels`: {in_channels} and `num_vector_embeds`: {num_vector_embeds}. Make"""
												    " sure that either `in_channels` or `num_vector_embeds` is None."			)
									elif self.is_input_vectorized and self.is_input_patches:
												raise ValueError(
												    F"""Cannot define both `num_vector_embeds`: {num_vector_embeds} and `patch_size`: {patch_size}. Make"""
												    " sure that either `num_vector_embeds` or `num_patches` is None."			)
									elif not self.is_input_continuous and not self.is_input_vectorized and not self.is_input_patches:
												raise ValueError(
												    F"""Has to define `in_channels`: {in_channels}, `num_vector_embeds`: {num_vector_embeds}, or patch_size:"""
												    F""" {patch_size}. Make sure that `in_channels`, `num_vector_embeds` or `num_patches` is not None."""			)

									# 2. Define input layers
									if self.is_input_continuous:
												A				:       Dict														=    in_channels

												A				:       List[Any]														=    torch.nn.GroupNorm(num_groups=__lowerCamelCase						,						num_channels=__lowerCamelCase						,						eps=1e-6						,						affine=__lowerCamelCase			)
												if use_linear_projection:
															A				:       Dict														=    nn.Linear(__lowerCamelCase						,						__lowerCamelCase			)
												else:
															A				:       int														=    nn.Convad(__lowerCamelCase						,						__lowerCamelCase						,						kernel_size=1						,						stride=1						,						padding=0			)
									elif self.is_input_vectorized:
												assert sample_size is not None, "Transformer2DModel over discrete input must provide sample_size"
												assert num_vector_embeds is not None, "Transformer2DModel over discrete input must provide num_embed"

												A				:       Tuple														=    sample_size
												A				:       int														=    sample_size
												A				:       Union[str, Any]														=    num_vector_embeds
												A				:       List[Any]														=    self.height * self.width

												A				:       Optional[Any]														=    ImagePositionalEmbeddings(
												    num_embed=__lowerCamelCase						,						embed_dim=__lowerCamelCase						,						height=self.height						,						width=self.width			)
									elif self.is_input_patches:
												assert sample_size is not None, "Transformer2DModel over patched input must provide sample_size"

												A				:       str														=    sample_size
												A				:       List[Any]														=    sample_size

												A				:       List[Any]														=    patch_size
												A				:       int														=    PatchEmbed(
												    height=__lowerCamelCase						,						width=__lowerCamelCase						,						patch_size=__lowerCamelCase						,						in_channels=__lowerCamelCase						,						embed_dim=__lowerCamelCase						,						)

									# 3. Define transformers blocks
									A				:       Tuple														=    nn.ModuleList(
									    [
									        BasicTransformerBlock(
									            __lowerCamelCase						,						__lowerCamelCase						,						__lowerCamelCase						,						dropout=__lowerCamelCase						,						cross_attention_dim=__lowerCamelCase						,						activation_fn=__lowerCamelCase						,						num_embeds_ada_norm=__lowerCamelCase						,						attention_bias=__lowerCamelCase						,						only_cross_attention=__lowerCamelCase						,						upcast_attention=__lowerCamelCase						,						norm_type=__lowerCamelCase						,						norm_elementwise_affine=__lowerCamelCase						,						)
									        for d in range(__lowerCamelCase			)
									    ]			)

									# 4. Define output layers
									A				:       int														=    in_channels if out_channels is None else out_channels
									if self.is_input_continuous:
												# TODO: should use out_channels for continuous projections
												if use_linear_projection:
															A				:       Optional[Any]														=    nn.Linear(__lowerCamelCase						,						__lowerCamelCase			)
												else:
															A				:       Tuple														=    nn.Convad(__lowerCamelCase						,						__lowerCamelCase						,						kernel_size=1						,						stride=1						,						padding=0			)
									elif self.is_input_vectorized:
												A				:       Optional[int]														=    nn.LayerNorm(__lowerCamelCase			)
												A				:       Optional[int]														=    nn.Linear(__lowerCamelCase						,						self.num_vector_embeds - 1			)
									elif self.is_input_patches:
												A				:       Optional[int]														=    nn.LayerNorm(__lowerCamelCase						,						elementwise_affine=__lowerCamelCase						,						eps=1e-6			)
												A				:       Optional[int]														=    nn.Linear(__lowerCamelCase						,						2 * inner_dim			)
												A				:       Optional[Any]														=    nn.Linear(__lowerCamelCase						,						patch_size * patch_size * self.out_channels			)







						def       SCREAMING_SNAKE_CASE__		(						self   :      Tuple						,						__lowerCamelCase   :      torch.Tensor						,						__lowerCamelCase   :      Optional[torch.Tensor] = None						,						__lowerCamelCase   :      Optional[torch.LongTensor] = None						,						__lowerCamelCase   :      Optional[torch.LongTensor] = None						,						__lowerCamelCase   :      Dict[str, Any] = None						,						__lowerCamelCase   :      Optional[torch.Tensor] = None						,						__lowerCamelCase   :      Optional[torch.Tensor] = None						,						__lowerCamelCase   :      bool = True						,						)   ->					Optional[Any]:
									# ensure attention_mask is a bias, and give it a singleton query_tokens dimension.
									#   we may have done this conversion already, e.g. if we came here via UNet2DConditionModel#forward.
									#   we can tell by counting dims; if ndim == 2: it's a mask rather than a bias.
									# expects mask of shape:
									#   [batch, key_tokens]
									# adds singleton query_tokens dimension:
									#   [batch,                    1, key_tokens]
									# this helps to broadcast it as a bias over attention scores, which will be in one of the following shapes:
									#   [batch,  heads, query_tokens, key_tokens] (e.g. torch sdp attn)
									#   [batch * heads, query_tokens, key_tokens] (e.g. xformers or classic attn)
									if attention_mask is not None and attention_mask.ndim == 2:
												# assume that mask is expressed as:
												#   (1 = keep,      0 = discard)
												# convert mask into a bias that can be added to attention scores:
												#       (keep = +0,     discard = -10000.0)
												A				:       Union[str, Any]														=    (1 - attention_mask.to(hidden_states.dtype			)) * -10000.0
												A				:       List[Any]														=    attention_mask.unsqueeze(1			)

									# convert encoder_attention_mask to a bias the same way we do for attention_mask
									if encoder_attention_mask is not None and encoder_attention_mask.ndim == 2:
												A				:       str														=    (1 - encoder_attention_mask.to(hidden_states.dtype			)) * -10000.0
												A				:       List[Any]														=    encoder_attention_mask.unsqueeze(1			)

									# 1. Input
									if self.is_input_continuous:
												A							,			A							,			A							,			A				:       Any														=    hidden_states.shape
												A				:       Any														=    hidden_states

												A				:       Any														=    self.norm(__lowerCamelCase			)
												if not self.use_linear_projection:
															A				:       Dict														=    self.proj_in(__lowerCamelCase			)
															A				:       List[str]														=    hidden_states.shape[1]
															A				:       str														=    hidden_states.permute(0						,						2						,						3						,						1			).reshape(__lowerCamelCase						,						height * width						,						__lowerCamelCase			)
												else:
															A				:       List[Any]														=    hidden_states.shape[1]
															A				:       Tuple														=    hidden_states.permute(0						,						2						,						3						,						1			).reshape(__lowerCamelCase						,						height * width						,						__lowerCamelCase			)
															A				:       Tuple														=    self.proj_in(__lowerCamelCase			)
									elif self.is_input_vectorized:
												A				:       List[str]														=    self.latent_image_embedding(__lowerCamelCase			)
									elif self.is_input_patches:
												A				:       Any														=    self.pos_embed(__lowerCamelCase			)

									# 2. Blocks
									for block in self.transformer_blocks:
												A				:       List[str]														=    block(
												    __lowerCamelCase						,						attention_mask=__lowerCamelCase						,						encoder_hidden_states=__lowerCamelCase						,						encoder_attention_mask=__lowerCamelCase						,						timestep=__lowerCamelCase						,						cross_attention_kwargs=__lowerCamelCase						,						class_labels=__lowerCamelCase						,						)

									# 3. Output
									if self.is_input_continuous:
												if not self.use_linear_projection:
															A				:       int														=    hidden_states.reshape(__lowerCamelCase						,						__lowerCamelCase						,						__lowerCamelCase						,						__lowerCamelCase			).permute(0						,						3						,						1						,						2			).contiguous()
															A				:       Dict														=    self.proj_out(__lowerCamelCase			)
												else:
															A				:       int														=    self.proj_out(__lowerCamelCase			)
															A				:       Dict														=    hidden_states.reshape(__lowerCamelCase						,						__lowerCamelCase						,						__lowerCamelCase						,						__lowerCamelCase			).permute(0						,						3						,						1						,						2			).contiguous()

												A				:       Optional[Any]														=    hidden_states + residual
									elif self.is_input_vectorized:
												A				:       List[str]														=    self.norm_out(__lowerCamelCase			)
												A				:       str														=    self.out(__lowerCamelCase			)
												# (batch, self.num_vector_embeds - 1, self.num_latent_pixels)
												A				:       Dict														=    logits.permute(0						,						2						,						1			)

												# log(p(x_0))
												A				:       Dict														=    F.log_softmax(logits.double()						,						dim=1			).float()
									elif self.is_input_patches:
												# TODO: cleanup!
												A				:       Union[str, Any]														=    self.transformer_blocks[0].norma.emb(
												    __lowerCamelCase						,						__lowerCamelCase						,						hidden_dtype=hidden_states.dtype			)
												A							,			A				:       int														=    self.proj_out_a(F.silu(__lowerCamelCase			)			).chunk(2						,						dim=1			)
												A				:       Tuple														=    self.norm_out(__lowerCamelCase			) * (1 + scale[:, None]) + shift[:, None]
												A				:       Union[str, Any]														=    self.proj_out_a(__lowerCamelCase			)

												# unpatchify
												A				:       Optional[Any]														=    int(hidden_states.shape[1] ** 0.5			)
												A				:       Any														=    hidden_states.reshape(
												    shape=(-1, height, width, self.patch_size, self.patch_size, self.out_channels)			)
												A				:       Optional[Any]														=    torch.einsum("nhwpqc->nchpwq"						,						__lowerCamelCase			)
												A				:       List[Any]														=    hidden_states.reshape(
												    shape=(-1, self.out_channels, height * self.patch_size, width * self.patch_size)			)

									if not return_dict:
												return (output,)

									return TransformeraDModelOutput(sample=__lowerCamelCase			)




import os
from shutil import copyfile
from typing import List, Optional, Tuple

from ...tokenization_utils_fast import AddedToken, PreTrainedTokenizerFast
from ...utils import logging


__SCREAMING_SNAKE_CASE     =	logging.get_logger(__name__)

__SCREAMING_SNAKE_CASE     =	{"""vocab_file""": """spiece.model""", """tokenizer_file""": """tokenizer.json"""}

__SCREAMING_SNAKE_CASE     =	{
    """vocab_file""": {
        """TsinghuaAI/CPM-Generate""": """https://huggingface.co/TsinghuaAI/CPM-Generate/resolve/main/spiece.model""",
    },
    """tokenizer_file""": {
        """TsinghuaAI/CPM-Generate""": """https://huggingface.co/TsinghuaAI/CPM-Generate/resolve/main/tokenizer.json""",
    },
}

class 						lowerCamelCase_     (			_A    ):






						'''simple docstring'''







						def __init__(						self   :      Dict						,						__lowerCamelCase   :      str=None						,						__lowerCamelCase   :      int=None						,						__lowerCamelCase   :      Optional[Any]=False						,						__lowerCamelCase   :      List[Any]=True						,						__lowerCamelCase   :      int=False						,						__lowerCamelCase   :      Optional[int]="<s>"						,						__lowerCamelCase   :      Union[str, Any]="</s>"						,						__lowerCamelCase   :      Dict="<unk>"						,						__lowerCamelCase   :      List[str]="<sep>"						,						__lowerCamelCase   :      Any="<pad>"						,						__lowerCamelCase   :      Optional[Any]="<cls>"						,						__lowerCamelCase   :      Union[str, Any]="<mask>"						,						__lowerCamelCase   :      List[str]=["<eop>", "<eod>"]						,						**__lowerCamelCase   :      Optional[Any]						,						)   ->					Optional[int]:
									A				:       Any														=    AddedToken(__lowerCamelCase						,						lstrip=__lowerCamelCase						,						rstrip=__lowerCamelCase			) if isinstance(__lowerCamelCase						,						__lowerCamelCase			) else mask_token

									super().__init__(
									    vocab_file=__lowerCamelCase						,						tokenizer_file=__lowerCamelCase						,						do_lower_case=__lowerCamelCase						,						remove_space=__lowerCamelCase						,						keep_accents=__lowerCamelCase						,						bos_token=__lowerCamelCase						,						eos_token=__lowerCamelCase						,						unk_token=__lowerCamelCase						,						sep_token=__lowerCamelCase						,						pad_token=__lowerCamelCase						,						cls_token=__lowerCamelCase						,						mask_token=__lowerCamelCase						,						additional_special_tokens=__lowerCamelCase						,						**__lowerCamelCase						,						)

									A				:       str														=    3
									A				:       str														=    do_lower_case
									A				:       int														=    remove_space
									A				:       Optional[int]														=    keep_accents
									A				:       Tuple														=    vocab_file
									A				:       int														=    False if not self.vocab_file else True

									try:
												import jieba
									except ModuleNotFoundError as error:
												raise error.__class__(
												    "You need to install jieba to use CpmTokenizer or CpmTokenizerFast. "
												    "See https://pypi.org/project/jieba/ for installation."			)
									A				:       str														=    jieba
									A				:       Optional[Any]														=    str.maketrans(" \n"						,						"\u2582\u2583"			)







						def       SCREAMING_SNAKE_CASE__		(						self   :      List[str]						,						__lowerCamelCase   :      List[int]						,						__lowerCamelCase   :      Optional[List[int]] = None			)   ->					List[int]:
									A				:       Dict														=    [self.sep_token_id]
									A				:       Tuple														=    [self.cls_token_id]
									if token_ids_a is None:
												return token_ids_a + sep + cls
									return token_ids_a + sep + token_ids_a + sep + cls







						def       SCREAMING_SNAKE_CASE__		(						self   :      int						,						__lowerCamelCase   :      List[int]						,						__lowerCamelCase   :      Optional[List[int]] = None			)   ->					List[int]:
									A				:       Tuple														=    [self.sep_token_id]
									A				:       Dict														=    [2]

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







						def       SCREAMING_SNAKE_CASE__		(						self   :      int						,						__lowerCamelCase   :      str						,						__lowerCamelCase   :      Optional[str] = None			)   ->					Tuple[str]:
									if not self.can_save_slow_tokenizer:
												raise ValueError(
												    "Your fast tokenizer does not have the necessary information to save the vocabulary for a slow "
												    "tokenizer."			)

									if not os.path.isdir(__lowerCamelCase			):
												logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""			)
												return
									A				:       List[Any]														=    os.path.join(
									    __lowerCamelCase						,						(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]			)

									if os.path.abspath(self.vocab_file			) != os.path.abspath(__lowerCamelCase			):
												copyfile(self.vocab_file						,						__lowerCamelCase			)

									return (out_vocab_file,)







						def       SCREAMING_SNAKE_CASE__		(						self   :      Any						,						__lowerCamelCase   :      Dict						,						*__lowerCamelCase   :      List[str]						,						**__lowerCamelCase   :      Tuple			)   ->					str:
									A				:       Union[str, Any]														=    [
									    " ".join([x.translate(self.translator			) for x in self.jieba.cut(__lowerCamelCase						,						cut_all=__lowerCamelCase			)]			)
									    for text in batch_text_or_text_pairs
									]
									return super()._batch_encode_plus(__lowerCamelCase						,						*__lowerCamelCase						,						**__lowerCamelCase			)







						def       SCREAMING_SNAKE_CASE__		(						self   :      Any						,						*__lowerCamelCase   :      Optional[Any]						,						**__lowerCamelCase   :      Optional[int]			)   ->					Tuple:
									A				:       Optional[Any]														=    super()._decode(*__lowerCamelCase						,						**__lowerCamelCase			)
									A				:       Tuple														=    text.replace(" "						,						""			).replace("\u2582"						,						" "			).replace("\u2583"						,						"\n"			)
									return text