





"""simple docstring"""


from dataclasses import dataclass
from typing import Dict, Optional, Tuple, Union

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, apply_forward_hook
from .attention_processor import AttentionProcessor, AttnProcessor
from .modeling_utils import ModelMixin
from .vae import Decoder, DecoderOutput, DiagonalGaussianDistribution, Encoder



@dataclass
class   __SCREAMING_SNAKE_CASE     (    lowerCamelCase__ ):







  '''simple docstring'''


  SCREAMING_SNAKE_CASE__				:str						  =       42





class   __SCREAMING_SNAKE_CASE     (    lowerCamelCase__		,	lowerCamelCase__ ):







  '''simple docstring'''


  SCREAMING_SNAKE_CASE__				:Tuple						  =       True

  @register_to_config
  def __init__(		self :							Dict       , __a :							Optional[Any] = 3       , __a :							Tuple = 3       , __a :							Any = ("DownEncoderBlock2D",)       , __a :							List[Any] = ("UpDecoderBlock2D",)       , __a :							str = (64,)       , __a :							int = 1       , __a :							List[Any] = "silu"       , __a :							Optional[int] = 4       , __a :							Optional[int] = 32       , __a :							Tuple = 32       , __a :							Optional[int] = 0.1_82_15       , )		->					Optional[Any]:
        super().__init__()

        # pass init params to Encoder
        _UpperCamelCase   :						List[str]    =							Encoder(
            in_channels=SCREAMING_SNAKE_CASE__       , out_channels=SCREAMING_SNAKE_CASE__       , down_block_types=SCREAMING_SNAKE_CASE__       , block_out_channels=SCREAMING_SNAKE_CASE__       , layers_per_block=SCREAMING_SNAKE_CASE__       , act_fn=SCREAMING_SNAKE_CASE__       , norm_num_groups=SCREAMING_SNAKE_CASE__       , double_z=SCREAMING_SNAKE_CASE__       , )

        # pass init params to Decoder
        _UpperCamelCase   :						int    =							Decoder(
            in_channels=SCREAMING_SNAKE_CASE__       , out_channels=SCREAMING_SNAKE_CASE__       , up_block_types=SCREAMING_SNAKE_CASE__       , block_out_channels=SCREAMING_SNAKE_CASE__       , layers_per_block=SCREAMING_SNAKE_CASE__       , norm_num_groups=SCREAMING_SNAKE_CASE__       , act_fn=SCREAMING_SNAKE_CASE__       , )

        _UpperCamelCase   :						Any    =							nn.Convad(2 * latent_channels       , 2 * latent_channels       , 1						)
        _UpperCamelCase   :						Union[str, Any]    =							nn.Convad(SCREAMING_SNAKE_CASE__       , SCREAMING_SNAKE_CASE__       , 1						)

        _UpperCamelCase   :						List[Any]    =							False
        _UpperCamelCase   :						List[str]    =							False

        # only relevant if vae tiling is enabled
        _UpperCamelCase   :						Any    =							self.config.sample_size
        _UpperCamelCase   :						Optional[Any]    =							(
            self.config.sample_size[0]
            if isinstance(self.config.sample_size       , (list, tuple)						)
            else self.config.sample_size
        )
        _UpperCamelCase   :						Optional[Any]    =							int(sample_size / (2 ** (len(self.config.block_out_channels						) - 1))						)
        _UpperCamelCase   :						Tuple    =							0.25

  def  __SCREAMING_SNAKE_CASE					(		self :							List[str]       , __a :							Optional[int]       , __a :							Tuple=False						)		->					Optional[Any]:



        if isinstance(SCREAMING_SNAKE_CASE__       , (Encoder, Decoder)						):
              _UpperCamelCase   :						Optional[Any]    =							value

  def  __SCREAMING_SNAKE_CASE					(		self :							int       , __a :							List[Any] = True						)		->					int:
        _UpperCamelCase   :						int    =							use_tiling

  def  __SCREAMING_SNAKE_CASE					(		self :							Tuple						)		->					Any:
        self.enable_tiling(SCREAMING_SNAKE_CASE__						)

  def  __SCREAMING_SNAKE_CASE					(		self :							Optional[int]						)		->					Any:
        _UpperCamelCase   :						Optional[int]    =							True

  def  __SCREAMING_SNAKE_CASE					(		self :							Optional[int]						)		->					List[Any]:
        _UpperCamelCase   :						Tuple    =							False

  @property
  # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
  def  __SCREAMING_SNAKE_CASE					(		self :							Any						)		->					Dict[str, AttentionProcessor]:
        _UpperCamelCase   :						str    =							{}

        def fn_recursive_add_processors(__a :							Union[str, Any]       , __a :							int       , __a :							Dict						):
              if hasattr(SCREAMING_SNAKE_CASE__       , "set_processor"						):
                    _UpperCamelCase   :						Any    =							module.processor

              for sub_name, child in module.named_children():
                    fn_recursive_add_processors(F'''{name}.{sub_name}'''       , SCREAMING_SNAKE_CASE__       , SCREAMING_SNAKE_CASE__						)

              return processors

        for name, module in self.named_children():
              fn_recursive_add_processors(SCREAMING_SNAKE_CASE__       , SCREAMING_SNAKE_CASE__       , SCREAMING_SNAKE_CASE__						)

        return processors

  def  __SCREAMING_SNAKE_CASE					(		self :							Tuple       , __a :							Optional[int]						)		->					Union[str, Any]:
        _UpperCamelCase   :						List[Any]    =							len(self.attn_processors.keys()						)

        if isinstance(SCREAMING_SNAKE_CASE__       , SCREAMING_SNAKE_CASE__						) and len(SCREAMING_SNAKE_CASE__						) != count:
              raise ValueError(
                  F'''A dict of processors was passed, but the number of processors {len(SCREAMING_SNAKE_CASE__						)} does not match the'''
                  F''' number of attention layers: {count}. Please make sure to pass {count} processor classes.'''						)

        def fn_recursive_attn_processor(__a :							Any       , __a :							Optional[Any]       , __a :							Tuple						):
              if hasattr(SCREAMING_SNAKE_CASE__       , "set_processor"						):
                    if not isinstance(SCREAMING_SNAKE_CASE__       , SCREAMING_SNAKE_CASE__						):
                          module.set_processor(SCREAMING_SNAKE_CASE__						)
                    else:
                          module.set_processor(processor.pop(F'''{name}.processor'''						)						)

              for sub_name, child in module.named_children():
                    fn_recursive_attn_processor(F'''{name}.{sub_name}'''       , SCREAMING_SNAKE_CASE__       , SCREAMING_SNAKE_CASE__						)

        for name, module in self.named_children():
              fn_recursive_attn_processor(SCREAMING_SNAKE_CASE__       , SCREAMING_SNAKE_CASE__       , SCREAMING_SNAKE_CASE__						)

  def  __SCREAMING_SNAKE_CASE					(		self :							Tuple						)		->					Dict:
        self.set_attn_processor(AttnProcessor()						)

  @apply_forward_hook
  def  __SCREAMING_SNAKE_CASE					(		self :							List[Any]       , __a :							int       , __a :							Dict = True						)		->					AutoencoderKLOutput:



        if self.use_tiling and (x.shape[-1] > self.tile_sample_min_size or x.shape[-2] > self.tile_sample_min_size):
              return self.tiled_encode(SCREAMING_SNAKE_CASE__       , return_dict=SCREAMING_SNAKE_CASE__						)

        if self.use_slicing and x.shape[0] > 1:
              _UpperCamelCase   :						int    =							[self.encoder(SCREAMING_SNAKE_CASE__						) for x_slice in x.split(1						)]
              _UpperCamelCase   :						Tuple    =							torch.cat(SCREAMING_SNAKE_CASE__						)
        else:
              _UpperCamelCase   :						Optional[int]    =							self.encoder(SCREAMING_SNAKE_CASE__						)

        _UpperCamelCase   :						List[str]    =							self.quant_conv(SCREAMING_SNAKE_CASE__						)
        _UpperCamelCase   :						List[str]    =							DiagonalGaussianDistribution(SCREAMING_SNAKE_CASE__						)

        if not return_dict:
              return (posterior,)

        return AutoencoderKLOutput(latent_dist=SCREAMING_SNAKE_CASE__						)

  def  __SCREAMING_SNAKE_CASE					(		self :							Optional[Any]       , __a :							Optional[int]       , __a :							Tuple = True						)		->					Union[DecoderOutput, torch.FloatTensor]:



        if self.use_tiling and (z.shape[-1] > self.tile_latent_min_size or z.shape[-2] > self.tile_latent_min_size):
              return self.tiled_decode(SCREAMING_SNAKE_CASE__       , return_dict=SCREAMING_SNAKE_CASE__						)

        _UpperCamelCase   :						List[str]    =							self.post_quant_conv(SCREAMING_SNAKE_CASE__						)
        _UpperCamelCase   :						int    =							self.decoder(SCREAMING_SNAKE_CASE__						)

        if not return_dict:
              return (dec,)

        return DecoderOutput(sample=SCREAMING_SNAKE_CASE__						)

  @apply_forward_hook
  def  __SCREAMING_SNAKE_CASE					(		self :							Any       , __a :							Union[str, Any]       , __a :							Any = True						)		->					Union[DecoderOutput, torch.FloatTensor]:



        if self.use_slicing and z.shape[0] > 1:
              _UpperCamelCase   :						Dict    =							[self._decode(SCREAMING_SNAKE_CASE__						).sample for z_slice in z.split(1						)]
              _UpperCamelCase   :						Any    =							torch.cat(SCREAMING_SNAKE_CASE__						)
        else:
              _UpperCamelCase   :						Optional[int]    =							self._decode(SCREAMING_SNAKE_CASE__						).sample

        if not return_dict:
              return (decoded,)

        return DecoderOutput(sample=SCREAMING_SNAKE_CASE__						)

  def  __SCREAMING_SNAKE_CASE					(		self :							Any       , __a :							Union[str, Any]       , __a :							Union[str, Any]       , __a :							Optional[Any]						)		->					Union[str, Any]:
        _UpperCamelCase   :						Tuple    =							min(a.shape[2]       , b.shape[2]       , SCREAMING_SNAKE_CASE__						)
        for y in range(SCREAMING_SNAKE_CASE__						):
              _UpperCamelCase   :						int    =							a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
        return b

  def  __SCREAMING_SNAKE_CASE					(		self :							Optional[int]       , __a :							List[str]       , __a :							List[Any]       , __a :							int						)		->					List[str]:
        _UpperCamelCase   :						Dict    =							min(a.shape[3]       , b.shape[3]       , SCREAMING_SNAKE_CASE__						)
        for x in range(SCREAMING_SNAKE_CASE__						):
              _UpperCamelCase   :						Tuple    =							a[:, :, :, -blend_extent + x] * (1 - x / blend_extent) + b[:, :, :, x] * (x / blend_extent)
        return b

  def  __SCREAMING_SNAKE_CASE					(		self :							int       , __a :							Optional[Any]       , __a :							Optional[int] = True						)		->					AutoencoderKLOutput:
        _UpperCamelCase   :						Optional[Any]    =							int(self.tile_sample_min_size * (1 - self.tile_overlap_factor)						)
        _UpperCamelCase   :						int    =							int(self.tile_latent_min_size * self.tile_overlap_factor						)
        _UpperCamelCase   :						int    =							self.tile_latent_min_size - blend_extent

        # Split the image into 512x512 tiles and encode them separately.
        _UpperCamelCase   :						Any    =							[]
        for i in range(0       , x.shape[2]       , SCREAMING_SNAKE_CASE__						):
              _UpperCamelCase   :						Union[str, Any]    =							[]
              for j in range(0       , x.shape[3]       , SCREAMING_SNAKE_CASE__						):
                    _UpperCamelCase   :						str    =							x[:, :, i : i + self.tile_sample_min_size, j : j + self.tile_sample_min_size]
                    _UpperCamelCase   :						int    =							self.encoder(SCREAMING_SNAKE_CASE__						)
                    _UpperCamelCase   :						Any    =							self.quant_conv(SCREAMING_SNAKE_CASE__						)
                    row.append(SCREAMING_SNAKE_CASE__						)
              rows.append(SCREAMING_SNAKE_CASE__						)
        _UpperCamelCase   :						Optional[Any]    =							[]
        for i, row in enumerate(SCREAMING_SNAKE_CASE__						):
              _UpperCamelCase   :						str    =							[]
              for j, tile in enumerate(SCREAMING_SNAKE_CASE__						):
                    # blend the above tile and the left tile
                    # to the current tile and add the current tile to the result row
                    if i > 0:
                          _UpperCamelCase   :						Any    =							self.blend_v(rows[i - 1][j]       , SCREAMING_SNAKE_CASE__       , SCREAMING_SNAKE_CASE__						)
                    if j > 0:
                          _UpperCamelCase   :						Optional[Any]    =							self.blend_h(row[j - 1]       , SCREAMING_SNAKE_CASE__       , SCREAMING_SNAKE_CASE__						)
                    result_row.append(tile[:, :, :row_limit, :row_limit]						)
              result_rows.append(torch.cat(SCREAMING_SNAKE_CASE__       , dim=3						)						)

        _UpperCamelCase   :						Any    =							torch.cat(SCREAMING_SNAKE_CASE__       , dim=2						)
        _UpperCamelCase   :						Union[str, Any]    =							DiagonalGaussianDistribution(SCREAMING_SNAKE_CASE__						)

        if not return_dict:
              return (posterior,)

        return AutoencoderKLOutput(latent_dist=SCREAMING_SNAKE_CASE__						)

  def  __SCREAMING_SNAKE_CASE					(		self :							Dict       , __a :							List[Any]       , __a :							Dict = True						)		->					Union[DecoderOutput, torch.FloatTensor]:
        _UpperCamelCase   :						Union[str, Any]    =							int(self.tile_latent_min_size * (1 - self.tile_overlap_factor)						)
        _UpperCamelCase   :						Tuple    =							int(self.tile_sample_min_size * self.tile_overlap_factor						)
        _UpperCamelCase   :						List[str]    =							self.tile_sample_min_size - blend_extent

        # Split z into overlapping 64x64 tiles and decode them separately.
        # The tiles have an overlap to avoid seams between tiles.
        _UpperCamelCase   :						Optional[int]    =							[]
        for i in range(0       , z.shape[2]       , SCREAMING_SNAKE_CASE__						):
              _UpperCamelCase   :						List[str]    =							[]
              for j in range(0       , z.shape[3]       , SCREAMING_SNAKE_CASE__						):
                    _UpperCamelCase   :						List[str]    =							z[:, :, i : i + self.tile_latent_min_size, j : j + self.tile_latent_min_size]
                    _UpperCamelCase   :						Tuple    =							self.post_quant_conv(SCREAMING_SNAKE_CASE__						)
                    _UpperCamelCase   :						List[str]    =							self.decoder(SCREAMING_SNAKE_CASE__						)
                    row.append(SCREAMING_SNAKE_CASE__						)
              rows.append(SCREAMING_SNAKE_CASE__						)
        _UpperCamelCase   :						Optional[Any]    =							[]
        for i, row in enumerate(SCREAMING_SNAKE_CASE__						):
              _UpperCamelCase   :						Union[str, Any]    =							[]
              for j, tile in enumerate(SCREAMING_SNAKE_CASE__						):
                    # blend the above tile and the left tile
                    # to the current tile and add the current tile to the result row
                    if i > 0:
                          _UpperCamelCase   :						Dict    =							self.blend_v(rows[i - 1][j]       , SCREAMING_SNAKE_CASE__       , SCREAMING_SNAKE_CASE__						)
                    if j > 0:
                          _UpperCamelCase   :						Any    =							self.blend_h(row[j - 1]       , SCREAMING_SNAKE_CASE__       , SCREAMING_SNAKE_CASE__						)
                    result_row.append(tile[:, :, :row_limit, :row_limit]						)
              result_rows.append(torch.cat(SCREAMING_SNAKE_CASE__       , dim=3						)						)

        _UpperCamelCase   :						Union[str, Any]    =							torch.cat(SCREAMING_SNAKE_CASE__       , dim=2						)
        if not return_dict:
              return (dec,)

        return DecoderOutput(sample=SCREAMING_SNAKE_CASE__						)


  def  __SCREAMING_SNAKE_CASE					(		self :							List[Any]       , __a :							Optional[int]       , __a :							Tuple = False       , __a :							List[str] = True       , __a :							Optional[Any] = None       , )		->					Union[DecoderOutput, torch.FloatTensor]:
        _UpperCamelCase   :						int    =							sample
        _UpperCamelCase   :						Any    =							self.encode(SCREAMING_SNAKE_CASE__						).latent_dist
        if sample_posterior:
              _UpperCamelCase   :						Optional[int]    =							posterior.sample(generator=SCREAMING_SNAKE_CASE__						)
        else:
              _UpperCamelCase   :						Tuple    =							posterior.mode()
        _UpperCamelCase   :						Tuple    =							self.decode(SCREAMING_SNAKE_CASE__						).sample

        if not return_dict:
              return (dec,)

        return DecoderOutput(sample=SCREAMING_SNAKE_CASE__						)




"""simple docstring"""






import os
import unittest

from transformers.models.cpmant.tokenization_cpmant import VOCAB_FILES_NAMES, CpmAntTokenizer
from transformers.testing_utils import require_jieba, tooslow

from ...test_tokenization_common import TokenizerTesterMixin






@require_jieba
class      A(  lowerCamelCase__				,							unittest.TestCase		):

		"""simple docstring"""





		A    =							CpmAntTokenizer
		A    =							False
		def   _UpperCamelCase( self   )      ->	Union[str, Any]:

					"""simple docstring"""



					super().setUp()

					_UpperCamelCase					:str					     =	[
					    '''<d>''',
					    '''</d>''',
					    '''<s>''',
					    '''</s>''',
					    '''</_>''',
					    '''<unk>''',
					    '''<pad>''',
					    '''</n>''',
					    '''我''',
					    '''是''',
					    '''C''',
					    '''P''',
					    '''M''',
					    '''A''',
					    '''n''',
					    '''t''',
					]
					_UpperCamelCase					:List[str]					     =	os.path.join(self.tmpdirname					,				VOCAB_FILES_NAMES['''vocab_file''']   )
					with open(self.vocab_file					,				'''w'''					,				encoding='''utf-8'''   ) as vocab_writer:
								vocab_writer.write(''''''.join([x + '''\n''' for x in vocab_tokens]   )   )





		@tooslow
		def   _UpperCamelCase( self   )      ->	Tuple:

					"""simple docstring"""



					_UpperCamelCase					:Any					     =	CpmAntTokenizer.from_pretrained('''openbmb/cpm-ant-10b'''   )
					_UpperCamelCase					:List[str]					     =	'''今天天气真好！'''
					_UpperCamelCase					:Tuple					     =	['''今天''', '''天气''', '''真''', '''好''', '''！''']
					_UpperCamelCase					:Tuple					     =	tokenizer.tokenize(SCREAMING_SNAKE_CASE__   )
					self.assertListEqual(SCREAMING_SNAKE_CASE__					,				SCREAMING_SNAKE_CASE__   )
					_UpperCamelCase					:Dict					     =	'''今天天气真好！'''
					_UpperCamelCase					:Optional[int]					     =	[tokenizer.bos_token] + tokens

					_UpperCamelCase					:List[str]					     =	[6, 98_02, 1_49_62, 20_82, 8_31, 2_44]
					self.assertListEqual(tokenizer.convert_tokens_to_ids(SCREAMING_SNAKE_CASE__   )					,				SCREAMING_SNAKE_CASE__   )

					_UpperCamelCase					:str					     =	tokenizer.decode(SCREAMING_SNAKE_CASE__   )
					self.assertEqual(SCREAMING_SNAKE_CASE__					,				SCREAMING_SNAKE_CASE__   )

