


"""simple docstring"""
import collections
import os
from typing import List, Optional, Tuple

from transformers.utils import is_jieba_available, requires_backends


if is_jieba_available():
				import jieba

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


__SCREAMING_SNAKE_CASE							:      Dict						=      logging.get_logger(__name__)

__SCREAMING_SNAKE_CASE							:      Tuple						=      {'vocab_file': 'vocab.txt'}

__SCREAMING_SNAKE_CASE							:      Any						=      {
    'vocab_file': {
        'openbmb/cpm-ant-10b': 'https://huggingface.co/openbmb/cpm-ant-10b/blob/main/vocab.txt',
    },
}

__SCREAMING_SNAKE_CASE							:      Any						=      {
    'openbmb/cpm-ant-10b': 1_024,
}







def        _a			( _SCREAMING_SNAKE_CASE			)      ->			Any:
							snake_case_          =						collections.OrderedDict()
							with open(_SCREAMING_SNAKE_CASE			,	"""r"""			,	encoding="""utf-8"""			) as reader:
														snake_case_          =						reader.readlines()
							for index, token in enumerate(_SCREAMING_SNAKE_CASE			):
														snake_case_          =						token.rstrip("""\n"""			)
														snake_case_          =						index
							return vocab





class        __A       (snake_case__):





							'''simple docstring'''

							def __init__(			self							: Tuple						,    UpperCAmelCase_							: Any						,    UpperCAmelCase_							: Dict="<unk>"						,    UpperCAmelCase_							: List[Any]=200    )   ->Any:



														"""simple docstring"""



														snake_case_          =						vocab
														snake_case_          =						unk_token
														snake_case_          =						max_input_chars_per_word



							def 						lowerCAmelCase    (			self							: Optional[Any]						,    UpperCAmelCase_							: Optional[Any]    )   ->Union[str, Any]:



														"""simple docstring"""



														snake_case_          =						list(UpperCAmelCase_    )
														if len(UpperCAmelCase_    ) > self.max_input_chars_per_word:
																					return [self.unk_token]

														snake_case_          =						0
														snake_case_          =						[]
														while start < len(UpperCAmelCase_    ):
																					snake_case_          =						len(UpperCAmelCase_    )
																					snake_case_          =						None
																					while start < end:
																												snake_case_          =						"""""".join(chars[start:end]    )
																												if substr in self.vocab:
																																			snake_case_          =						substr
																																			break
																												end -= 1
																					if cur_substr is None:
																												sub_tokens.append(self.unk_token    )
																												start += 1
																					else:
																												sub_tokens.append(UpperCAmelCase_    )
																												snake_case_          =						end

														return sub_tokens





class        __A       (snake_case__):





							'''simple docstring'''

							__lowercase:						List[str]							 =      VOCAB_FILES_NAMES
							__lowercase:						str							 =      PRETRAINED_VOCAB_FILES_MAP
							__lowercase:						Any							 =      PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							__lowercase:						str							 =      ["""input_ids""", """attention_mask"""]
							__lowercase:						Dict							 =      False



							def __init__(			self							: Tuple						,    UpperCAmelCase_							: Union[str, Any]						,    UpperCAmelCase_							: Any="<d>"						,    UpperCAmelCase_							: List[Any]="</d>"						,    UpperCAmelCase_							: Any="<s>"						,    UpperCAmelCase_							: Tuple="</s>"						,    UpperCAmelCase_							: Optional[int]="<pad>"						,    UpperCAmelCase_							: List[Any]="<unk>"						,    UpperCAmelCase_							: Tuple="</n>"						,    UpperCAmelCase_							: Tuple="</_>"						,    UpperCAmelCase_							: List[Any]="left"						,    **UpperCAmelCase_							: Optional[int]						,    )   ->int:



														"""simple docstring"""



														requires_backends(self						,    ["""jieba"""]    )
														super().__init__(
														    bod_token=UpperCAmelCase_						,    eod_token=UpperCAmelCase_						,    bos_token=UpperCAmelCase_						,    eos_token=UpperCAmelCase_						,    pad_token=UpperCAmelCase_						,    unk_token=UpperCAmelCase_						,    line_token=UpperCAmelCase_						,    space_token=UpperCAmelCase_						,    padding_side=UpperCAmelCase_						,    **UpperCAmelCase_						,    )
														snake_case_          =						bod_token
														snake_case_          =						eod_token
														snake_case_          =						load_vocab(UpperCAmelCase_    )
														snake_case_          =						self.encoder[space_token]
														snake_case_          =						self.encoder[line_token]

														del self.encoder[space_token]
														del self.encoder[line_token]

														snake_case_          =						collections.OrderedDict(sorted(self.encoder.items()						,    key=lambda UpperCAmelCase_    : x[1]    )    )
														snake_case_          =						{v: k for k, v in self.encoder.items()}

														snake_case_          =						WordpieceTokenizer(vocab=self.encoder						,    unk_token=self.unk_token    )



							@property
							def 						lowerCAmelCase    (			self							: Any    )   ->int:



														"""simple docstring"""



														return self.encoder[self.bod_token]



							@property
							def 						lowerCAmelCase    (			self							: List[Any]    )   ->Union[str, Any]:



														"""simple docstring"""



														return self.encoder[self.eod_token]



							@property
							def 						lowerCAmelCase    (			self							: str    )   ->Optional[Any]:



														"""simple docstring"""



														return self.encoder["\n"]



							@property
							def 						lowerCAmelCase    (			self							: Dict    )   ->int:



														"""simple docstring"""



														return len(self.encoder    )



							def 						lowerCAmelCase    (			self							: List[Any]    )   ->Union[str, Any]:



														"""simple docstring"""



														return dict(self.encoder						,    **self.added_tokens_encoder    )



							def 						lowerCAmelCase    (			self							: Union[str, Any]						,    UpperCAmelCase_							: int    )   ->int:



														"""simple docstring"""



														snake_case_          =						[]
														for x in jieba.cut(UpperCAmelCase_						,    cut_all=UpperCAmelCase_    ):
																					output_tokens.extend(self.wordpiece_tokenizer.tokenize(UpperCAmelCase_    )    )
														return output_tokens



							def 						lowerCAmelCase    (			self							: str						,    UpperCAmelCase_							: Union[str, Any]						,    **UpperCAmelCase_							: Tuple    )   ->Optional[int]:



														"""simple docstring"""



														snake_case_          =						[i for i in token_ids if i >= 0]
														snake_case_          =						[
														    x for x in token_ids if x != self.pad_token_id and x != self.eos_token_id and x != self.bos_token_id
														]
														return super()._decode(UpperCAmelCase_						,    **UpperCAmelCase_    )



							def 						lowerCAmelCase    (			self							: List[str]						,    UpperCAmelCase_							: str    )   ->Tuple:



														"""simple docstring"""



														return token in self.encoder



							def 						lowerCAmelCase    (			self							: int						,    UpperCAmelCase_							: List[str]    )   ->str:



														"""simple docstring"""



														return "".join(UpperCAmelCase_    )



							def 						lowerCAmelCase    (			self							: int						,    UpperCAmelCase_							: Tuple    )   ->int:



														"""simple docstring"""



														return self.encoder.get(UpperCAmelCase_						,    self.encoder.get(self.unk_token    )    )



							def 						lowerCAmelCase    (			self							: str						,    UpperCAmelCase_							: str    )   ->Dict:



														"""simple docstring"""



														return self.decoder.get(UpperCAmelCase_						,    self.unk_token    )



							def 						lowerCAmelCase    (			self							: Union[str, Any]						,    UpperCAmelCase_							: str						,    UpperCAmelCase_							: Optional[str] = None    )   ->Tuple[str]:



														"""simple docstring"""



														if os.path.isdir(UpperCAmelCase_    ):
																					snake_case_          =						os.path.join(
																					    UpperCAmelCase_						,    (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]    )
														else:
																					snake_case_          =						(filename_prefix + """-""" if filename_prefix else """""") + save_directory
														snake_case_          =						0
														if " " in self.encoder:
																					snake_case_          =						self.encoder[""" """]
																					del self.encoder[" "]
														if "\n" in self.encoder:
																					snake_case_          =						self.encoder["""\n"""]
																					del self.encoder["\n"]
														snake_case_          =						collections.OrderedDict(sorted(self.encoder.items()						,    key=lambda UpperCAmelCase_    : x[1]    )    )
														with open(UpperCAmelCase_						,    """w"""						,    encoding="""utf-8"""    ) as writer:
																					for token, token_index in self.encoder.items():
																												if index != token_index:
																																			logger.warning(
																																			    F"""Saving vocabulary to {vocab_file}: vocabulary indices are not consecutive."""
																																			    """ Please check that the vocabulary is not corrupted!"""    )
																																			snake_case_          =						token_index
																												writer.write(token + """\n"""    )
																												index += 1
														return (vocab_file,)



							def 						lowerCAmelCase    (			self							: str						,    UpperCAmelCase_							: List[int]						,    UpperCAmelCase_							: List[int] = None    )   ->List[int]:



														"""simple docstring"""



														if token_ids_a is None:
																					return [self.bos_token_id] + token_ids_a
														return [self.bos_token_id] + token_ids_a + [self.bos_token_id] + token_ids_a



							def 						lowerCAmelCase    (			self							: Optional[int]						,    UpperCAmelCase_							: List[int]						,    UpperCAmelCase_							: Optional[List[int]] = None						,    UpperCAmelCase_							: bool = False    )   ->List[int]:



														"""simple docstring"""



														if already_has_special_tokens:
																					return super().get_special_tokens_mask(
																					    token_ids_a=UpperCAmelCase_						,    token_ids_a=UpperCAmelCase_						,    already_has_special_tokens=UpperCAmelCase_    )

														if token_ids_a is not None:
																					return [1] + ([0] * len(UpperCAmelCase_    )) + [1] + ([0] * len(UpperCAmelCase_    ))
														return [1] + ([0] * len(UpperCAmelCase_    ))







"""simple docstring"""
import gc
import random
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import AutoencoderKL, CycleDiffusionPipeline, DDIMScheduler, UNetaDConditionModel
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

from ..pipeline_params import (
    IMAGE_TO_IMAGE_IMAGE_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()





class        __A       (snake_case__  ,						snake_case__  ,						unittest.TestCase):





							'''simple docstring'''

							__lowercase:						List[Any]							 =      CycleDiffusionPipeline
							__lowercase:						int							 =      TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
							    """negative_prompt""",
							    """height""",
							    """width""",
							    """negative_prompt_embeds""",
							}
							__lowercase:						str							 =      PipelineTesterMixin.required_optional_params - {"""latents"""}
							__lowercase:						Any							 =      TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"""source_prompt"""})
							__lowercase:						Any							 =      IMAGE_TO_IMAGE_IMAGE_PARAMS
							__lowercase:						Any							 =      IMAGE_TO_IMAGE_IMAGE_PARAMS



							def 						lowerCAmelCase    (			self							: Dict    )   ->Any:



														"""simple docstring"""



														torch.manual_seed(0    )
														snake_case_          =						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						,    )
														snake_case_          =						DDIMScheduler(
														    beta_start=0.00_085						,    beta_end=0.012						,    beta_schedule="""scaled_linear"""						,    num_train_timesteps=1_000						,    clip_sample=UpperCAmelCase_						,    set_alpha_to_one=UpperCAmelCase_						,    )
														torch.manual_seed(0    )
														snake_case_          =						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						,    )
														torch.manual_seed(0    )
														snake_case_          =						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=1_000						,    )
														snake_case_          =						CLIPTextModel(UpperCAmelCase_    )
														snake_case_          =						CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""    )

														snake_case_          =						{
														    """unet""": unet,
														    """scheduler""": scheduler,
														    """vae""": vae,
														    """text_encoder""": text_encoder,
														    """tokenizer""": tokenizer,
														    """safety_checker""": None,
														    """feature_extractor""": None,
														}
														return components



							def 						lowerCAmelCase    (			self							: int						,    UpperCAmelCase_							: str						,    UpperCAmelCase_							: List[str]=0    )   ->str:



														"""simple docstring"""



														snake_case_          =						floats_tensor((1, 3, 32, 32)						,    rng=random.Random(UpperCAmelCase_    )    ).to(UpperCAmelCase_    )
														snake_case_          =						image / 2 + 0.5
														if str(UpperCAmelCase_    ).startswith("""mps"""    ):
																					snake_case_          =						torch.manual_seed(UpperCAmelCase_    )
														else:
																					snake_case_          =						torch.Generator(device=UpperCAmelCase_    ).manual_seed(UpperCAmelCase_    )
														snake_case_          =						{
														    """prompt""": """An astronaut riding an elephant""",
														    """source_prompt""": """An astronaut riding a horse""",
														    """image""": image,
														    """generator""": generator,
														    """num_inference_steps""": 2,
														    """eta""": 0.1,
														    """strength""": 0.8,
														    """guidance_scale""": 3,
														    """source_guidance_scale""": 1,
														    """output_type""": """numpy""",
														}
														return inputs



							def 						lowerCAmelCase    (			self							: Union[str, Any]    )   ->Union[str, Any]:



														"""simple docstring"""



														snake_case_          =						"""cpu"""  # ensure determinism for the device-dependent torch.Generator

														snake_case_          =						self.get_dummy_components()
														snake_case_          =						CycleDiffusionPipeline(**UpperCAmelCase_    )
														snake_case_          =						pipe.to(UpperCAmelCase_    )
														pipe.set_progress_bar_config(disable=UpperCAmelCase_    )

														snake_case_          =						self.get_dummy_inputs(UpperCAmelCase_    )
														snake_case_          =						pipe(**UpperCAmelCase_    )
														snake_case_          =						output.images

														snake_case_          =						images[0, -3:, -3:, -1]

														assert images.shape == (1, 32, 32, 3)
														snake_case_          =						np.array([0.4_459, 0.4_943, 0.4_544, 0.6_643, 0.5_474, 0.4_327, 0.5_701, 0.5_959, 0.5_179]    )

														assert np.abs(image_slice.flatten() - expected_slice    ).max() < 1E-2



							@unittest.skipIf(torch_device != """cuda"""						,    """This test requires a GPU"""    )
							def 						lowerCAmelCase    (			self							: Union[str, Any]    )   ->str:



														"""simple docstring"""



														snake_case_          =						self.get_dummy_components()
														for name, module in components.items():
																					if hasattr(UpperCAmelCase_						,    """half"""    ):
																												snake_case_          =						module.half()
														snake_case_          =						CycleDiffusionPipeline(**UpperCAmelCase_    )
														snake_case_          =						pipe.to(UpperCAmelCase_    )
														pipe.set_progress_bar_config(disable=UpperCAmelCase_    )

														snake_case_          =						self.get_dummy_inputs(UpperCAmelCase_    )
														snake_case_          =						pipe(**UpperCAmelCase_    )
														snake_case_          =						output.images

														snake_case_          =						images[0, -3:, -3:, -1]

														assert images.shape == (1, 32, 32, 3)
														snake_case_          =						np.array([0.3_506, 0.4_543, 0.446, 0.4_575, 0.5_195, 0.4_155, 0.5_273, 0.518, 0.4_116]    )

														assert np.abs(image_slice.flatten() - expected_slice    ).max() < 1E-2



							@skip_mps
							def 						lowerCAmelCase    (			self							: Tuple    )   ->Optional[int]:



														"""simple docstring"""



														return super().test_save_load_local()



							@unittest.skip("""non-deterministic pipeline"""    )
							def 						lowerCAmelCase    (			self							: int    )   ->Optional[int]:



														"""simple docstring"""



														return super().test_inference_batch_single_identical()



							@skip_mps
							def 						lowerCAmelCase    (			self							: Tuple    )   ->List[Any]:



														"""simple docstring"""



														return super().test_dict_tuple_outputs_equivalent()



							@skip_mps
							def 						lowerCAmelCase    (			self							: Any    )   ->Tuple:



														"""simple docstring"""



														return super().test_save_load_optional_components()



							@skip_mps
							def 						lowerCAmelCase    (			self							: List[Any]    )   ->str:



														"""simple docstring"""



														return super().test_attention_slicing_forward_pass()





@slow
@require_torch_gpu
class        __A       (unittest.TestCase):





							'''simple docstring'''

							def 						lowerCAmelCase    (			self							: List[Any]    )   ->Optional[int]:



														"""simple docstring"""



														super().tearDown()
														gc.collect()
														torch.cuda.empty_cache()



							def 						lowerCAmelCase    (			self							: Optional[int]    )   ->str:



														"""simple docstring"""



														snake_case_          =						load_image(
														    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
														    """/cycle-diffusion/black_colored_car.png"""    )
														snake_case_          =						load_numpy(
														    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car_fp16.npy"""    )
														snake_case_          =						init_image.resize((512, 512)    )

														snake_case_          =						"""CompVis/stable-diffusion-v1-4"""
														snake_case_          =						DDIMScheduler.from_pretrained(UpperCAmelCase_						,    subfolder="""scheduler"""    )
														snake_case_          =						CycleDiffusionPipeline.from_pretrained(
														    UpperCAmelCase_						,    scheduler=UpperCAmelCase_						,    safety_checker=UpperCAmelCase_						,    torch_dtype=torch.floataa						,    revision="""fp16"""    )

														pipe.to(UpperCAmelCase_    )
														pipe.set_progress_bar_config(disable=UpperCAmelCase_    )
														pipe.enable_attention_slicing()

														snake_case_          =						"""A black colored car"""
														snake_case_          =						"""A blue colored car"""

														snake_case_          =						torch.manual_seed(0    )
														snake_case_          =						pipe(
														    prompt=UpperCAmelCase_						,    source_prompt=UpperCAmelCase_						,    image=UpperCAmelCase_						,    num_inference_steps=100						,    eta=0.1						,    strength=0.85						,    guidance_scale=3						,    source_guidance_scale=1						,    generator=UpperCAmelCase_						,    output_type="""np"""						,    )
														snake_case_          =						output.images

														# the values aren't exactly equal, but the images look the same visually
														assert np.abs(image - expected_image    ).max() < 5E-1



							def 						lowerCAmelCase    (			self							: Tuple    )   ->List[str]:



														"""simple docstring"""



														snake_case_          =						load_image(
														    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
														    """/cycle-diffusion/black_colored_car.png"""    )
														snake_case_          =						load_numpy(
														    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car.npy"""    )
														snake_case_          =						init_image.resize((512, 512)    )

														snake_case_          =						"""CompVis/stable-diffusion-v1-4"""
														snake_case_          =						DDIMScheduler.from_pretrained(UpperCAmelCase_						,    subfolder="""scheduler"""    )
														snake_case_          =						CycleDiffusionPipeline.from_pretrained(UpperCAmelCase_						,    scheduler=UpperCAmelCase_						,    safety_checker=UpperCAmelCase_    )

														pipe.to(UpperCAmelCase_    )
														pipe.set_progress_bar_config(disable=UpperCAmelCase_    )
														pipe.enable_attention_slicing()

														snake_case_          =						"""A black colored car"""
														snake_case_          =						"""A blue colored car"""

														snake_case_          =						torch.manual_seed(0    )
														snake_case_          =						pipe(
														    prompt=UpperCAmelCase_						,    source_prompt=UpperCAmelCase_						,    image=UpperCAmelCase_						,    num_inference_steps=100						,    eta=0.1						,    strength=0.85						,    guidance_scale=3						,    source_guidance_scale=1						,    generator=UpperCAmelCase_						,    output_type="""np"""						,    )
														snake_case_          =						output.images

														assert np.abs(image - expected_image    ).max() < 2E-2



