



import gc
import os
import tempfile
import unittest

import torch
import torch.nn as nn
import torch.nn.functional as F
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import AutoencoderKL, DDIMScheduler, StableDiffusionPipeline, UNetaDConditionModel
from diffusers.loaders import AttnProcsLayers, LoraLoaderMixin
from diffusers.models.attention_processor import (
    Attention,
    AttnProcessor,
    AttnProcessor2_0,
    LoRAAttnProcessor,
    LoRAAttnProcessor2_0,
    LoRAXFormersAttnProcessor,
    XFormersAttnProcessor,
)
from diffusers.utils import TEXT_ENCODER_ATTN_MODULE, floats_tensor, torch_device




def 						UpperCAmelCase (    _lowerCamelCase     ):
			A				:       Optional[Any]														=    {}
			for name in unet.attn_processors.keys():
						A				:       Optional[int]														=    None if name.endswith("attn1.processor"     ) else unet.config.cross_attention_dim
						if name.startswith("mid_block"     ):
									A				:       Any														=    unet.config.block_out_channels[-1]
						elif name.startswith("up_blocks"     ):
									A				:       Dict														=    int(name[len("up_blocks."     )]     )
									A				:       Union[str, Any]														=    list(reversed(unet.config.block_out_channels     )     )[block_id]
						elif name.startswith("down_blocks"     ):
									A				:       Dict														=    int(name[len("down_blocks."     )]     )
									A				:       Tuple														=    unet.config.block_out_channels[block_id]
						A				:       List[str]														=    (
						    LoRAAttnProcessor2_0 if hasattr(_lowerCamelCase			,     "scaled_dot_product_attention"     ) else LoRAAttnProcessor
						)
						A				:       Dict														=    lora_attn_processor_class(
						    hidden_size=_lowerCamelCase			,     cross_attention_dim=_lowerCamelCase     )
			A				:       Optional[int]														=    AttnProcsLayers(_lowerCamelCase     )
			return lora_attn_procs, unet_lora_layers




def 						UpperCAmelCase (    _lowerCamelCase     ):
			A				:       Tuple														=    {}
			A				:       Optional[int]														=    (
			    LoRAAttnProcessor2_0 if hasattr(_lowerCamelCase			,     "scaled_dot_product_attention"     ) else LoRAAttnProcessor
			)
			for name, module in text_encoder.named_modules():
						if name.endswith(_lowerCamelCase     ):
									A				:       List[str]														=    lora_attn_processor_class(
									    hidden_size=module.out_proj.out_features			,     cross_attention_dim=_lowerCamelCase     )
			return text_lora_attn_procs




def 						UpperCAmelCase (    _lowerCamelCase     ):
			A				:       int														=    create_text_encoder_lora_attn_procs(_lowerCamelCase     )
			A				:       Any														=    AttnProcsLayers(_lowerCamelCase     )
			return text_encoder_lora_layers




def 						UpperCAmelCase (    _lowerCamelCase			,     _lowerCamelCase=False     ):
			for _, attn_proc in text_lora_attn_procs.items():
						# set up.weights
						for layer_name, layer_module in attn_proc.named_modules():
									if layer_name.endswith("_lora"     ):
												A				:       Optional[Any]														=    (
												    torch.randn_like(layer_module.up.weight     )
												    if randn_weight
												    else torch.zeros_like(layer_module.up.weight     )
												)
												A				:       Union[str, Any]														=    torch.nn.Parameter(_lowerCamelCase     )






class 						lowerCamelCase_     (			unittest.TestCase    ):






						'''simple docstring'''







						def       SCREAMING_SNAKE_CASE__		(						self   :      str			)   ->					List[Any]:
									torch.manual_seed(0			)
									A				:       Any														=    UNetaDConditionModel(
									    block_out_channels=(32, 64)						,						layers_per_block=2						,						sample_size=32						,						in_channels=4						,						out_channels=4						,						down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")						,						up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")						,						cross_attention_dim=32						,						)
									A				:       str														=    DDIMScheduler(
									    beta_start=0.00085						,						beta_end=0.012						,						beta_schedule="scaled_linear"						,						clip_sample=__lowerCamelCase						,						set_alpha_to_one=__lowerCamelCase						,						steps_offset=1						,						)
									torch.manual_seed(0			)
									A				:       List[Any]														=    AutoencoderKL(
									    block_out_channels=[32, 64]						,						in_channels=3						,						out_channels=3						,						down_block_types=["DownEncoderBlock2D", "DownEncoderBlock2D"]						,						up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D"]						,						latent_channels=4						,						)
									A				:       int														=    CLIPTextConfig(
									    bos_token_id=0						,						eos_token_id=2						,						hidden_size=32						,						intermediate_size=37						,						layer_norm_eps=1e-05						,						num_attention_heads=4						,						num_hidden_layers=5						,						pad_token_id=1						,						vocab_size=10_00						,						)
									A				:       Tuple														=    CLIPTextModel(__lowerCamelCase			)
									A				:       List[Any]														=    CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"			)

									A							,			A				:       Union[str, Any]														=    create_unet_lora_layers(__lowerCamelCase			)
									A				:       Optional[int]														=    create_text_encoder_lora_layers(__lowerCamelCase			)

									A				:       Optional[int]														=    {
									    "unet": unet,
									    "scheduler": scheduler,
									    "vae": vae,
									    "text_encoder": text_encoder,
									    "tokenizer": tokenizer,
									    "safety_checker": None,
									    "feature_extractor": None,
									}
									A				:       int														=    {
									    "unet_lora_layers": unet_lora_layers,
									    "text_encoder_lora_layers": text_encoder_lora_layers,
									    "unet_lora_attn_procs": unet_lora_attn_procs,
									}
									return pipeline_components, lora_components







						def       SCREAMING_SNAKE_CASE__		(						self   :      Any			)   ->					Tuple:
									A				:       List[str]														=    1
									A				:       str														=    10
									A				:       Any														=    4
									A				:       Tuple														=    (32, 32)

									A				:       List[str]														=    torch.manual_seed(0			)
									A				:       Optional[Any]														=    floats_tensor((batch_size, num_channels) + sizes			)
									A				:       Optional[Any]														=    torch.randint(1						,						__lowerCamelCase						,						size=(batch_size, sequence_length)						,						generator=__lowerCamelCase			)

									A				:       Tuple														=    {
									    "prompt": "A painting of a squirrel eating a burger",
									    "generator": generator,
									    "num_inference_steps": 2,
									    "guidance_scale": 6.0,
									    "output_type": "numpy",
									}

									return noise, input_ids, pipeline_inputs

									# copied from: https://colab.research.google.com/gist/sayakpaul/df2ef6e1ae6d8c10a49d859883b10860/scratchpad.ipynb







						def       SCREAMING_SNAKE_CASE__		(						self   :      List[str]			)   ->					Dict:
									A				:       Dict														=    77

									A				:       List[str]														=    torch.randint(2						,						56						,						size=(1, max_seq_length)						,						generator=torch.manual_seed(0			)			)

									A				:       str														=    {}
									A				:       Optional[int]														=    inputs
									return prepared_inputs







						def       SCREAMING_SNAKE_CASE__		(						self   :      Tuple						,						__lowerCamelCase   :      Tuple			)   ->					Dict:
									A							,			A				:       List[str]														=    self.get_dummy_components()
									LoraLoaderMixin.save_lora_weights(
									    save_directory=__lowerCamelCase						,						unet_lora_layers=lora_components["unet_lora_layers"]						,						text_encoder_lora_layers=lora_components["text_encoder_lora_layers"]						,						)
									self.assertTrue(os.path.isfile(os.path.join(__lowerCamelCase						,						"pytorch_lora_weights.bin"			)			)			)







						def       SCREAMING_SNAKE_CASE__		(						self   :      Dict			)   ->					Tuple:
									A							,			A				:       Any														=    self.get_dummy_components()
									A				:       List[str]														=    StableDiffusionPipeline(**__lowerCamelCase			)
									A				:       Dict														=    sd_pipe.to(__lowerCamelCase			)
									sd_pipe.set_progress_bar_config(disable=__lowerCamelCase			)

									A							,			A							,			A				:       Union[str, Any]														=    self.get_dummy_inputs()

									A				:       Optional[Any]														=    sd_pipe(**__lowerCamelCase			).images
									A				:       List[str]														=    original_images[0, -3:, -3:, -1]

									with tempfile.TemporaryDirectory() as tmpdirname:
												LoraLoaderMixin.save_lora_weights(
												    save_directory=__lowerCamelCase						,						unet_lora_layers=lora_components["unet_lora_layers"]						,						text_encoder_lora_layers=lora_components["text_encoder_lora_layers"]						,						)
												self.assertTrue(os.path.isfile(os.path.join(__lowerCamelCase						,						"pytorch_lora_weights.bin"			)			)			)
												sd_pipe.load_lora_weights(__lowerCamelCase			)

									A				:       Optional[int]														=    sd_pipe(**__lowerCamelCase			).images
									A				:       Optional[Any]														=    lora_images[0, -3:, -3:, -1]

									# Outputs shouldn't match.
									self.assertFalse(torch.allclose(torch.from_numpy(__lowerCamelCase			)						,						torch.from_numpy(__lowerCamelCase			)			)			)







						def       SCREAMING_SNAKE_CASE__		(						self   :      Tuple			)   ->					Tuple:
									A							,			A				:       List[Any]														=    self.get_dummy_components()
									A				:       List[str]														=    StableDiffusionPipeline(**__lowerCamelCase			)
									A				:       List[str]														=    sd_pipe.to(__lowerCamelCase			)
									sd_pipe.set_progress_bar_config(disable=__lowerCamelCase			)

									A							,			A							,			A				:       Optional[int]														=    self.get_dummy_inputs()

									A				:       Union[str, Any]														=    sd_pipe(**__lowerCamelCase			).images
									A				:       List[str]														=    original_images[0, -3:, -3:, -1]

									with tempfile.TemporaryDirectory() as tmpdirname:
												LoraLoaderMixin.save_lora_weights(
												    save_directory=__lowerCamelCase						,						unet_lora_layers=lora_components["unet_lora_layers"]						,						text_encoder_lora_layers=lora_components["text_encoder_lora_layers"]						,						safe_serialization=__lowerCamelCase						,						)
												self.assertTrue(os.path.isfile(os.path.join(__lowerCamelCase						,						"pytorch_lora_weights.safetensors"			)			)			)
												sd_pipe.load_lora_weights(__lowerCamelCase			)

									A				:       Tuple														=    sd_pipe(**__lowerCamelCase			).images
									A				:       Any														=    lora_images[0, -3:, -3:, -1]

									# Outputs shouldn't match.
									self.assertFalse(torch.allclose(torch.from_numpy(__lowerCamelCase			)						,						torch.from_numpy(__lowerCamelCase			)			)			)







						def       SCREAMING_SNAKE_CASE__		(						self   :      Dict			)   ->					Any:
									A							,			A				:       List[str]														=    self.get_dummy_components()
									A				:       Optional[int]														=    lora_components["unet_lora_attn_procs"]
									A				:       List[str]														=    StableDiffusionPipeline(**__lowerCamelCase			)
									A				:       Optional[int]														=    sd_pipe.to(__lowerCamelCase			)
									sd_pipe.set_progress_bar_config(disable=__lowerCamelCase			)

									A							,			A							,			A				:       Any														=    self.get_dummy_inputs()

									A				:       Optional[int]														=    sd_pipe(**__lowerCamelCase			).images
									A				:       Dict														=    original_images[0, -3:, -3:, -1]

									with tempfile.TemporaryDirectory() as tmpdirname:
												A				:       Tuple														=    sd_pipe.unet
												unet.set_attn_processor(__lowerCamelCase			)
												unet.save_attn_procs(__lowerCamelCase			)
												self.assertTrue(os.path.isfile(os.path.join(__lowerCamelCase						,						"pytorch_lora_weights.bin"			)			)			)
												sd_pipe.load_lora_weights(__lowerCamelCase			)

									A				:       str														=    sd_pipe(**__lowerCamelCase			).images
									A				:       Optional[int]														=    lora_images[0, -3:, -3:, -1]

									# Outputs shouldn't match.
									self.assertFalse(torch.allclose(torch.from_numpy(__lowerCamelCase			)						,						torch.from_numpy(__lowerCamelCase			)			)			)







						def       SCREAMING_SNAKE_CASE__		(						self   :      Optional[int]			)   ->					List[Any]:
									A							,			A				:       int														=    self.get_dummy_components()
									A				:       List[Any]														=    StableDiffusionPipeline(**__lowerCamelCase			)

									A				:       str														=    self.get_dummy_tokens()

									# inference without lora
									A				:       Optional[Any]														=    pipe.text_encoder(**__lowerCamelCase			)[0]
									assert outputs_without_lora.shape == (1, 77, 32)

									# create lora_attn_procs with zeroed out up.weights
									A				:       List[Any]														=    create_text_encoder_lora_attn_procs(pipe.text_encoder			)
									set_lora_up_weights(__lowerCamelCase						,						randn_weight=__lowerCamelCase			)

									# monkey patch
									pipe._modify_text_encoder(__lowerCamelCase			)

									# verify that it's okay to release the text_attn_procs which holds the LoRAAttnProcessor.
									del text_attn_procs
									gc.collect()

									# inference with lora
									A				:       Tuple														=    pipe.text_encoder(**__lowerCamelCase			)[0]
									assert outputs_with_lora.shape == (1, 77, 32)

									assert torch.allclose(
									    __lowerCamelCase						,						__lowerCamelCase			), "lora_up_weight are all zero, so the lora outputs should be the same to without lora outputs"

									# create lora_attn_procs with randn up.weights
									A				:       List[str]														=    create_text_encoder_lora_attn_procs(pipe.text_encoder			)
									set_lora_up_weights(__lowerCamelCase						,						randn_weight=__lowerCamelCase			)

									# monkey patch
									pipe._modify_text_encoder(__lowerCamelCase			)

									# verify that it's okay to release the text_attn_procs which holds the LoRAAttnProcessor.
									del text_attn_procs
									gc.collect()

									# inference with lora
									A				:       str														=    pipe.text_encoder(**__lowerCamelCase			)[0]
									assert outputs_with_lora.shape == (1, 77, 32)

									assert not torch.allclose(
									    __lowerCamelCase						,						__lowerCamelCase			), "lora_up_weight are not zero, so the lora outputs should be different to without lora outputs"







						def       SCREAMING_SNAKE_CASE__		(						self   :      Any			)   ->					Union[str, Any]:
									A							,			A				:       Any														=    self.get_dummy_components()
									A				:       Tuple														=    StableDiffusionPipeline(**__lowerCamelCase			)

									A				:       List[str]														=    self.get_dummy_tokens()

									# inference without lora
									A				:       List[str]														=    pipe.text_encoder(**__lowerCamelCase			)[0]
									assert outputs_without_lora.shape == (1, 77, 32)

									# create lora_attn_procs with randn up.weights
									A				:       Tuple														=    create_text_encoder_lora_attn_procs(pipe.text_encoder			)
									set_lora_up_weights(__lowerCamelCase						,						randn_weight=__lowerCamelCase			)

									# monkey patch
									pipe._modify_text_encoder(__lowerCamelCase			)

									# verify that it's okay to release the text_attn_procs which holds the LoRAAttnProcessor.
									del text_attn_procs
									gc.collect()

									# inference with lora
									A				:       Optional[int]														=    pipe.text_encoder(**__lowerCamelCase			)[0]
									assert outputs_with_lora.shape == (1, 77, 32)

									assert not torch.allclose(
									    __lowerCamelCase						,						__lowerCamelCase			), "lora outputs should be different to without lora outputs"

									# remove monkey patch
									pipe._remove_text_encoder_monkey_patch()

									# inference with removed lora
									A				:       List[Any]														=    pipe.text_encoder(**__lowerCamelCase			)[0]
									assert outputs_without_lora_removed.shape == (1, 77, 32)

									assert torch.allclose(
									    __lowerCamelCase						,						__lowerCamelCase			), "remove lora monkey patch should restore the original outputs"







						def       SCREAMING_SNAKE_CASE__		(						self   :      str			)   ->					Optional[Any]:
									A							,			A				:       List[str]														=    self.get_dummy_components()
									A				:       Optional[int]														=    StableDiffusionPipeline(**__lowerCamelCase			)
									A				:       Tuple														=    sd_pipe.to(__lowerCamelCase			)
									sd_pipe.set_progress_bar_config(disable=__lowerCamelCase			)

									A							,			A							,			A				:       List[Any]														=    self.get_dummy_inputs()

									with tempfile.TemporaryDirectory() as tmpdirname:
												LoraLoaderMixin.save_lora_weights(
												    save_directory=__lowerCamelCase						,						unet_lora_layers=lora_components["unet_lora_layers"]						,						text_encoder_lora_layers=lora_components["text_encoder_lora_layers"]						,						)
												self.assertTrue(os.path.isfile(os.path.join(__lowerCamelCase						,						"pytorch_lora_weights.bin"			)			)			)
												sd_pipe.load_lora_weights(__lowerCamelCase			)

									A				:       Union[str, Any]														=    sd_pipe(**__lowerCamelCase			).images
									A				:       str														=    lora_images[0, -3:, -3:, -1]

									A				:       Any														=    sd_pipe(**__lowerCamelCase						,						cross_attention_kwargs={"scale": 0.5}			).images
									A				:       str														=    lora_images_with_scale[0, -3:, -3:, -1]

									# Outputs shouldn't match.
									self.assertFalse(
									    torch.allclose(torch.from_numpy(__lowerCamelCase			)						,						torch.from_numpy(__lowerCamelCase			)			)			)







						def       SCREAMING_SNAKE_CASE__		(						self   :      Any			)   ->					Dict:
									with tempfile.TemporaryDirectory() as tmpdirname:
												self.create_lora_weight_file(__lowerCamelCase			)

												A							,			A				:       Tuple														=    self.get_dummy_components()
												A				:       List[str]														=    StableDiffusionPipeline(**__lowerCamelCase			)
												A				:       Optional[Any]														=    sd_pipe.to(__lowerCamelCase			)
												sd_pipe.set_progress_bar_config(disable=__lowerCamelCase			)

												# check if vanilla attention processors are used
												for _, module in sd_pipe.unet.named_modules():
															if isinstance(__lowerCamelCase						,						__lowerCamelCase			):
																		self.assertIsInstance(module.processor						,						(AttnProcessor, AttnProcessor2_0)			)

            # load LoRA weight file
												sd_pipe.load_lora_weights(__lowerCamelCase			)

												# check if lora attention processors are used
												for _, module in sd_pipe.unet.named_modules():
															if isinstance(__lowerCamelCase						,						__lowerCamelCase			):
																		A				:       str														=    (
																		    LoRAAttnProcessor2_0 if hasattr(__lowerCamelCase						,						"scaled_dot_product_attention"			) else LoRAAttnProcessor
																		)
																		self.assertIsInstance(module.processor						,						__lowerCamelCase			)







						@unittest.skipIf(torch_device != "cuda"						,						"This test is supposed to run on GPU"			)
						def       SCREAMING_SNAKE_CASE__		(						self   :      List[str]			)   ->					Union[str, Any]:
									with tempfile.TemporaryDirectory() as tmpdirname:
												self.create_lora_weight_file(__lowerCamelCase			)

												A							,			A				:       Tuple														=    self.get_dummy_components()
												A				:       int														=    StableDiffusionPipeline(**__lowerCamelCase			)
												A				:       List[Any]														=    sd_pipe.to(__lowerCamelCase			)
												sd_pipe.set_progress_bar_config(disable=__lowerCamelCase			)

												# enable XFormers
												sd_pipe.enable_xformers_memory_efficient_attention()

												# check if xFormers attention processors are used
												for _, module in sd_pipe.unet.named_modules():
															if isinstance(__lowerCamelCase						,						__lowerCamelCase			):
																		self.assertIsInstance(module.processor						,						__lowerCamelCase			)

            # load LoRA weight file
												sd_pipe.load_lora_weights(__lowerCamelCase			)

												# check if lora attention processors are used
												for _, module in sd_pipe.unet.named_modules():
															if isinstance(__lowerCamelCase						,						__lowerCamelCase			):
																		self.assertIsInstance(module.processor						,						__lowerCamelCase			)







						@unittest.skipIf(torch_device != "cuda"						,						"This test is supposed to run on GPU"			)
						def       SCREAMING_SNAKE_CASE__		(						self   :      Optional[int]			)   ->					List[str]:
									A							,			A				:       Union[str, Any]														=    self.get_dummy_components()
									A				:       Tuple														=    StableDiffusionPipeline(**__lowerCamelCase			)
									A				:       Union[str, Any]														=    sd_pipe.to(__lowerCamelCase			)
									sd_pipe.set_progress_bar_config(disable=__lowerCamelCase			)

									A							,			A							,			A				:       Dict														=    self.get_dummy_inputs()

									# enable XFormers
									sd_pipe.enable_xformers_memory_efficient_attention()

									A				:       str														=    sd_pipe(**__lowerCamelCase			).images
									A				:       List[Any]														=    original_images[0, -3:, -3:, -1]

									with tempfile.TemporaryDirectory() as tmpdirname:
												LoraLoaderMixin.save_lora_weights(
												    save_directory=__lowerCamelCase						,						unet_lora_layers=lora_components["unet_lora_layers"]						,						text_encoder_lora_layers=lora_components["text_encoder_lora_layers"]						,						)
												self.assertTrue(os.path.isfile(os.path.join(__lowerCamelCase						,						"pytorch_lora_weights.bin"			)			)			)
												sd_pipe.load_lora_weights(__lowerCamelCase			)

									A				:       str														=    sd_pipe(**__lowerCamelCase			).images
									A				:       List[str]														=    lora_images[0, -3:, -3:, -1]

									# Outputs shouldn't match.
									self.assertFalse(torch.allclose(torch.from_numpy(__lowerCamelCase			)						,						torch.from_numpy(__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