


"""simple docstring"""
import json
import os
from typing import Optional, Tuple

import regex as re

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


__SCREAMING_SNAKE_CASE							:      int						=      logging.get_logger(__name__)

__SCREAMING_SNAKE_CASE							:      str						=      {
    'vocab_file': 'vocab.json',
    'merges_file': 'merges.txt',
}

__SCREAMING_SNAKE_CASE							:      Optional[Any]						=      {
    'vocab_file': {'ctrl': 'https://raw.githubusercontent.com/salesforce/ctrl/master/ctrl-vocab.json'},
    'merges_file': {'ctrl': 'https://raw.githubusercontent.com/salesforce/ctrl/master/ctrl-merges.txt'},
}

__SCREAMING_SNAKE_CASE							:      Any						=      {
    'ctrl': 256,
}

__SCREAMING_SNAKE_CASE							:      Any						=      {
    'Pregnancy': 168_629,
    'Christianity': 7_675,
    'Explain': 106_423,
    'Fitness': 63_440,
    'Saving': 63_163,
    'Ask': 27_171,
    'Ass': 95_985,
    'Joke': 163_509,
    'Questions': 45_622,
    'Thoughts': 49_605,
    'Retail': 52_342,
    'Feminism': 164_338,
    'Writing': 11_992,
    'Atheism': 192_263,
    'Netflix': 48_616,
    'Computing': 39_639,
    'Opinion': 43_213,
    'Alone': 44_967,
    'Funny': 58_917,
    'Gaming': 40_358,
    'Human': 4_088,
    'India': 1_331,
    'Joker': 77_138,
    'Diet': 36_206,
    'Legal': 11_859,
    'Norman': 4_939,
    'Tip': 72_689,
    'Weight': 52_343,
    'Movies': 46_273,
    'Running': 23_425,
    'Science': 2_090,
    'Horror': 37_793,
    'Confession': 60_572,
    'Finance': 12_250,
    'Politics': 16_360,
    'Scary': 191_985,
    'Support': 12_654,
    'Technologies': 32_516,
    'Teenage': 66_160,
    'Event': 32_769,
    'Learned': 67_460,
    'Notion': 182_770,
    'Wikipedia': 37_583,
    'Books': 6_665,
    'Extract': 76_050,
    'Confessions': 102_701,
    'Conspiracy': 75_932,
    'Links': 63_674,
    'Narcissus': 150_425,
    'Relationship': 54_766,
    'Relationships': 134_796,
    'Reviews': 41_671,
    'News': 4_256,
    'Translation': 26_820,
    'multilingual': 128_406,
}







def        _a			( _SCREAMING_SNAKE_CASE			)      ->			str:
							snake_case_          =						set()
							snake_case_          =						word[0]
							for char in word[1:]:
														pairs.add((prev_char, char)			)
														snake_case_          =						char

							snake_case_          =						set(__a			)
							return pairs





class        __A       (__lowercase):





							'''simple docstring'''

							__lowercase:						Tuple							 =      VOCAB_FILES_NAMES
							__lowercase:						Optional[Any]							 =      PRETRAINED_VOCAB_FILES_MAP
							__lowercase:						Tuple							 =      PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							__lowercase:						List[Any]							 =      CONTROL_CODES



							def __init__(			self							: Dict						,    UpperCAmelCase_							: Dict						,    UpperCAmelCase_							: Optional[int]						,    UpperCAmelCase_							: Union[str, Any]="<unk>"						,    **UpperCAmelCase_							: str    )   ->Optional[Any]:



														"""simple docstring"""



														super().__init__(unk_token=_a						,    **_a    )

														with open(_a						,    encoding="""utf-8"""    ) as vocab_handle:
																					snake_case_          =						json.load(_a    )
														snake_case_          =						{v: k for k, v in self.encoder.items()}
														with open(_a						,    encoding="""utf-8"""    ) as merges_handle:
																					snake_case_          =						merges_handle.read().split("""\n"""    )[1:-1]
														snake_case_          =						[tuple(merge.split()    ) for merge in merges]
														snake_case_          =						dict(zip(_a						,    range(len(_a    )    )    )    )
														snake_case_          =						{}



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



														"""simple docstring"""



														return len(self.encoder    )



							def 						lowerCAmelCase    (			self							: Tuple    )   ->Any:



														"""simple docstring"""



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



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



														"""simple docstring"""



														if token in self.cache:
																					return self.cache[token]
														snake_case_          =						tuple(_a    )
														snake_case_          =						tuple(list(word[:-1]    ) + [word[-1] + """</w>"""]    )
														snake_case_          =						get_pairs(_a    )

														if not pairs:
																					return token

														while True:
																					snake_case_          =						min(_a						,    key=lambda UpperCAmelCase_    : self.bpe_ranks.get(_a						,    float("""inf"""    )    )    )
																					if bigram not in self.bpe_ranks:
																												break
																					snake_case_          =						bigram
																					snake_case_          =						[]
																					snake_case_          =						0
																					while i < len(_a    ):
																												try:
																																			snake_case_          =						word.index(_a						,    _a    )
																												except ValueError:
																																			new_word.extend(word[i:]    )
																																			break
																												else:
																																			new_word.extend(word[i:j]    )
																																			snake_case_          =						j

																												if word[i] == first and i < len(_a    ) - 1 and word[i + 1] == second:
																																			new_word.append(first + second    )
																																			i += 2
																												else:
																																			new_word.append(word[i]    )
																																			i += 1
																					snake_case_          =						tuple(_a    )
																					snake_case_          =						new_word
																					if len(_a    ) == 1:
																												break
																					else:
																												snake_case_          =						get_pairs(_a    )
														snake_case_          =						'''@@ '''.join(_a    )
														snake_case_          =						word[:-4]
														snake_case_          =						word
														return word



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



														"""simple docstring"""



														snake_case_          =						[]

														snake_case_          =						re.findall(R"""\S+\n?"""						,    _a    )

														for token in words:
																					split_tokens.extend(list(self.bpe(_a    ).split(""" """    )    )    )
														return split_tokens



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



														"""simple docstring"""



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



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



														"""simple docstring"""



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



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



														"""simple docstring"""



														snake_case_          =						''' '''.join(_a    ).replace("""@@ """						,    """"""    ).strip()
														return out_string



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



														"""simple docstring"""



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

														with open(_a						,    """w"""						,    encoding="""utf-8"""    ) as f:
																					f.write(json.dumps(self.encoder						,    indent=2						,    sort_keys=_a						,    ensure_ascii=_a    ) + """\n"""    )

														snake_case_          =						0
														with open(_a						,    """w"""						,    encoding="""utf-8"""    ) as writer:
																					writer.write("""#version: 0.2\n"""    )
																					for bpe_tokens, token_index in sorted(self.bpe_ranks.items()						,    key=lambda UpperCAmelCase_    : kv[1]    ):
																												if index != token_index:
																																			logger.warning(
																																			    F"""Saving vocabulary to {merge_file}: BPE merge indices are not consecutive."""
																																			    """ Please check that the tokenizer is not corrupted!"""    )
																																			snake_case_          =						token_index
																												writer.write(""" """.join(_a    ) + """\n"""    )
																												index += 1

														return vocab_file, merge_file

							# def decode(self, token_ids, skip_special_tokens=False, clean_up_tokenization_spaces=True):
							#     filtered_tokens = ' '.join(self.convert_ids_to_tokens(token_ids, skip_special_tokens=skip_special_tokens))
							#     tokens_generated_so_far = re.sub('(@@ )', '', string=filtered_tokens)
							#     tokens_generated_so_far = re.sub('(@@ ?$)', '', string=tokens_generated_so_far)
							#     return ''.join(tokens_generated_so_far)







"""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



