






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 						UpperCAmelCase_  (		SCREAMING_SNAKE_CASE__    ):


						'''simple docstring'''




						A   :      torch.FloatTensor




class 						UpperCAmelCase_  (		SCREAMING_SNAKE_CASE__    ,     SCREAMING_SNAKE_CASE__    ):


						'''simple docstring'''




						@register_to_config
						def __init__(		self						, _SCREAMING_SNAKE_CASE = 16						, _SCREAMING_SNAKE_CASE = 88						, _SCREAMING_SNAKE_CASE = None						, _SCREAMING_SNAKE_CASE = None						, _SCREAMING_SNAKE_CASE = 1						, _SCREAMING_SNAKE_CASE = 0.0						, _SCREAMING_SNAKE_CASE = 32						, _SCREAMING_SNAKE_CASE = None						, _SCREAMING_SNAKE_CASE = False						, _SCREAMING_SNAKE_CASE = None						, _SCREAMING_SNAKE_CASE = None						, _SCREAMING_SNAKE_CASE = None						, _SCREAMING_SNAKE_CASE = "geglu"						, _SCREAMING_SNAKE_CASE = None						, _SCREAMING_SNAKE_CASE = False						, _SCREAMING_SNAKE_CASE = False						, _SCREAMING_SNAKE_CASE = False						, _SCREAMING_SNAKE_CASE = "layer_norm"						, _SCREAMING_SNAKE_CASE = True						, )       -> Dict:
								super().__init__()
								snake_case_ :							Dict         =				use_linear_projection
								snake_case_ :							List[Any]         =				num_attention_heads
								snake_case_ :							List[Any]         =				attention_head_dim
								snake_case_ :							Optional[int]         =				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
								snake_case_ :							Tuple         =				(in_channels is not None) and (patch_size is None)
								snake_case_ :							List[Any]         =				num_vector_embeds is not None
								snake_case_ :							Union[str, Any]         =				in_channels is not None and patch_size is not None

								if norm_type == "layer_norm" and num_embeds_ada_norm is not None:
										snake_case_ :							Any         =				(
										    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"						, lowercase__						, standard_warn=lowercase__ )
										snake_case_ :							Union[str, 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:
										snake_case_ :							Tuple         =				in_channels

										snake_case_ :							Dict         =				torch.nn.GroupNorm(num_groups=lowercase__						, num_channels=lowercase__						, eps=1e-6						, affine=lowercase__ )
										if use_linear_projection:
												snake_case_ :							int         =				nn.Linear(lowercase__						, lowercase__ )
										else:
												snake_case_ :							int         =				nn.Convad(lowercase__						, lowercase__						, 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"

										snake_case_ :							Tuple         =				sample_size
										snake_case_ :							List[Any]         =				sample_size
										snake_case_ :							Dict         =				num_vector_embeds
										snake_case_ :							Any         =				self.height * self.width

										snake_case_ :							List[str]         =				ImagePositionalEmbeddings(
										    num_embed=lowercase__						, embed_dim=lowercase__						, 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"

										snake_case_ :							Optional[int]         =				sample_size
										snake_case_ :							List[str]         =				sample_size

										snake_case_ :							Optional[int]         =				patch_size
										snake_case_ :							Any         =				PatchEmbed(
										    height=lowercase__						, width=lowercase__						, patch_size=lowercase__						, in_channels=lowercase__						, embed_dim=lowercase__						, )

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

								# 4. Define output layers
								snake_case_ :							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:
												snake_case_ :							Dict         =				nn.Linear(lowercase__						, lowercase__ )
										else:
												snake_case_ :							List[str]         =				nn.Convad(lowercase__						, lowercase__						, kernel_size=1						, stride=1						, padding=0 )
								elif self.is_input_vectorized:
										snake_case_ :							int         =				nn.LayerNorm(lowercase__ )
										snake_case_ :							int         =				nn.Linear(lowercase__						, self.num_vector_embeds - 1 )
								elif self.is_input_patches:
										snake_case_ :							Optional[int]         =				nn.LayerNorm(lowercase__						, elementwise_affine=lowercase__						, eps=1e-6 )
										snake_case_ :							List[Any]         =				nn.Linear(lowercase__						, 2 * inner_dim )
										snake_case_ :							Tuple         =				nn.Linear(lowercase__						, patch_size * patch_size * self.out_channels )






						def 					_lowerCAmelCase				(		self						, _SCREAMING_SNAKE_CASE						, _SCREAMING_SNAKE_CASE = None						, _SCREAMING_SNAKE_CASE = None						, _SCREAMING_SNAKE_CASE = None						, _SCREAMING_SNAKE_CASE = None						, _SCREAMING_SNAKE_CASE = None						, _SCREAMING_SNAKE_CASE = None						, _SCREAMING_SNAKE_CASE = True						, )       -> List[Any]:







								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)
										snake_case_ :							List[str]         =				(1 - attention_mask.to(hidden_states.dtype )) * -1_0000.0
										snake_case_ :							Tuple         =				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:
										snake_case_ :							Union[str, Any]         =				(1 - encoder_attention_mask.to(hidden_states.dtype )) * -1_0000.0
										snake_case_ :							Any         =				encoder_attention_mask.unsqueeze(1 )

								# 1. Input
								if self.is_input_continuous:
										snake_case_       ,				snake_case_       ,				snake_case_       ,				snake_case_ :							List[Any]         =				hidden_states.shape
										snake_case_ :							List[Any]         =				hidden_states

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

								# 2. Blocks
								for block in self.transformer_blocks:
										snake_case_ :							Optional[int]         =				block(
										    lowercase__						, attention_mask=lowercase__						, encoder_hidden_states=lowercase__						, encoder_attention_mask=lowercase__						, timestep=lowercase__						, cross_attention_kwargs=lowercase__						, class_labels=lowercase__						, )

								# 3. Output
								if self.is_input_continuous:
										if not self.use_linear_projection:
												snake_case_ :							Tuple         =				hidden_states.reshape(lowercase__						, lowercase__						, lowercase__						, lowercase__ ).permute(0						, 3						, 1						, 2 ).contiguous()
												snake_case_ :							List[str]         =				self.proj_out(lowercase__ )
										else:
												snake_case_ :							List[str]         =				self.proj_out(lowercase__ )
												snake_case_ :							Optional[int]         =				hidden_states.reshape(lowercase__						, lowercase__						, lowercase__						, lowercase__ ).permute(0						, 3						, 1						, 2 ).contiguous()

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

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

										# unpatchify
										snake_case_ :							Any         =				int(hidden_states.shape[1] ** 0.5 )
										snake_case_ :							Dict         =				hidden_states.reshape(
										    shape=(-1, height, width, self.patch_size, self.patch_size, self.out_channels) )
										snake_case_ :							Optional[int]         =				torch.einsum("nhwpqc->nchpwq"						, lowercase__ )
										snake_case_ :							List[str]         =				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=lowercase__ )


"""simple docstring"""




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

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


snake_case    =		logging.get_logger(__name__)

snake_case    =		{'vocab_file': 'spiece.model', 'tokenizer_file': 'tokenizer.json'}

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      UpperCamelCase	(       __magic_name__      ):

				"""simple docstring"""



				def __init__(				self		,	lowercase__=None		,	lowercase__=None		,	lowercase__=False		,	lowercase__=True		,	lowercase__=False		,	lowercase__="<s>"		,	lowercase__="</s>"		,	lowercase__="<unk>"		,	lowercase__="<sep>"		,	lowercase__="<pad>"		,	lowercase__="<cls>"		,	lowercase__="<mask>"		,	lowercase__=["<eop>", "<eod>"]		,	**lowercase__		,	)    ->    int:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE				=	AddedToken(lowercase__		,	lstrip=lowercase__		,	rstrip=lowercase__			) if isinstance(lowercase__		,	lowercase__			) else mask_token

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

									SCREAMING_SNAKE_CASE				=	3
									SCREAMING_SNAKE_CASE				=	do_lower_case
									SCREAMING_SNAKE_CASE				=	remove_space
									SCREAMING_SNAKE_CASE				=	keep_accents
									SCREAMING_SNAKE_CASE				=	vocab_file
									SCREAMING_SNAKE_CASE				=	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.'			)
									SCREAMING_SNAKE_CASE				=	jieba
									SCREAMING_SNAKE_CASE				=	str.maketrans(' \n'		,	'\u2582\u2583'			)


				def 							A					(				self		,	lowercase__		,	lowercase__ = None			)    ->    List[int]:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE				=	[self.sep_token_id]
									SCREAMING_SNAKE_CASE				=	[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 							A					(				self		,	lowercase__		,	lowercase__ = None			)    ->    List[int]:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE				=	[self.sep_token_id]
									SCREAMING_SNAKE_CASE				=	[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 							A					(				self		,	lowercase__		,	lowercase__ = None			)    ->    Tuple[str]:
									"""simple docstring"""







									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(lowercase__			):
														logger.error(f'''Vocabulary path ({save_directory}) should be a directory'''			)
														return
									SCREAMING_SNAKE_CASE				=	os.path.join(
									    lowercase__		,	(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']			)

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

									return (out_vocab_file,)


				def 							A					(				self		,	lowercase__		,	*lowercase__		,	**lowercase__			)    ->    Any:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE				=	[
									    ' '.join([x.translate(self.translator			) for x in self.jieba.cut(lowercase__		,	cut_all=lowercase__			)]			)
									    for text in batch_text_or_text_pairs
									]
									return super()._batch_encode_plus(lowercase__		,	*lowercase__		,	**lowercase__			)




				def 							A					(				self		,	*lowercase__		,	**lowercase__			)    ->    Optional[int]:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE				=	super()._decode(*lowercase__		,	**lowercase__			)
									SCREAMING_SNAKE_CASE				=	text.replace(' '		,	''			).replace('\u2582'		,	' '			).replace('\u2583'		,	'\n'			)
									return text
