



import json
import os
from typing import Dict, List, Optional, Tuple

import regex as re

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


lowerCAmelCase__		:					Tuple							=logging.get_logger(__name__)


lowerCAmelCase__		:					Union[str, Any]							={
    '''vocab_file''': '''vocab.json''',
    '''merges_file''': '''merges.txt''',
    '''tokenizer_config_file''': '''tokenizer_config.json''',
}

lowerCAmelCase__		:					Any							={
    '''vocab_file''': {
        '''facebook/blenderbot_small-90M''': '''https://huggingface.co/facebook/blenderbot_small-90M/resolve/main/vocab.json'''
    },
    '''merges_file''': {
        '''facebook/blenderbot_small-90M''': '''https://huggingface.co/facebook/blenderbot_small-90M/resolve/main/merges.txt'''
    },
    '''tokenizer_config_file''': {
        '''facebook/blenderbot_small-90M''': (
            '''https://huggingface.co/facebook/blenderbot_small-90M/resolve/main/tokenizer_config.json'''
        )
    },
}

lowerCAmelCase__		:					Union[str, Any]							={'''facebook/blenderbot_small-90M''': 512}






def      __lowercase       (			a__       )    ->			Any:
							__SCREAMING_SNAKE_CASE									= set()
							__SCREAMING_SNAKE_CASE									= word[0]
							for char in word[1:]:
														pairs.add((prev_char, char)       )
														__SCREAMING_SNAKE_CASE									= char

							__SCREAMING_SNAKE_CASE									= set(a__       )
							return pairs

class        UpperCAmelCase_  (       UpperCamelCase_	):

						'''simple docstring'''




						UpperCamelCase__      :			str      =  VOCAB_FILES_NAMES
						UpperCamelCase__      :			str      =  PRETRAINED_VOCAB_FILES_MAP
						UpperCamelCase__      :			Optional[int]      =  PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
						UpperCamelCase__      :			Union[str, Any]      =  ['''input_ids''', '''attention_mask''']






						def __init__(   self				,							_A				,							_A				,							_A="__start__"				,							_A="__end__"				,							_A="__unk__"				,							_A="__null__"				,							**_A				,							):







													'''simple docstring'''

													super().__init__(unk_token=_A				,							bos_token=_A				,							eos_token=_A				,							pad_token=_A				,							**_A		)

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






						@property
						def 	_A     (   self		):







													'''simple docstring'''

													return len(self.encoder		)






						def 	_A     (   self		):







													'''simple docstring'''

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






						def 	_A     (   self				,							_A		):







													'''simple docstring'''

													if token in self.cache:
																				return self.cache[token]
													__SCREAMING_SNAKE_CASE									= re.sub('([.,!?()])'				,							r' \1'				,							_A		)
													__SCREAMING_SNAKE_CASE									= re.sub('(\')'				,							r' \1 '				,							_A		)
													__SCREAMING_SNAKE_CASE									= re.sub(r'\s{2,}'				,							' '				,							_A		)
													if "\n" in token:
																				__SCREAMING_SNAKE_CASE									= token.replace('\n'				,							' __newln__'		)

													__SCREAMING_SNAKE_CASE									= token.split(' '		)
													__SCREAMING_SNAKE_CASE									= []
													for token in tokens:
																				if not len(_A		):
																											continue

																				__SCREAMING_SNAKE_CASE									= token.lower()
																				__SCREAMING_SNAKE_CASE									= tuple(_A		)
																				__SCREAMING_SNAKE_CASE									= tuple(list(word[:-1]		) + [word[-1] + '</w>']		)
																				__SCREAMING_SNAKE_CASE									= get_pairs(_A		)

																				if not pairs:
																											words.append(_A		)
																											continue

																				while True:
																											__SCREAMING_SNAKE_CASE									= min(_A				,							key=lambda _A		: self.bpe_ranks.get(_A				,							float('inf'		)		)		)
																											if bigram not in self.bpe_ranks:
																																		break
																											__SCREAMING_SNAKE_CASE							,	__SCREAMING_SNAKE_CASE									= bigram
																											__SCREAMING_SNAKE_CASE									= []
																											__SCREAMING_SNAKE_CASE									= 0

																											while i < len(_A		):
																																		try:
																																									__SCREAMING_SNAKE_CASE									= word.index(_A				,							_A		)
																																									new_word.extend(word[i:j]		)
																																									__SCREAMING_SNAKE_CASE									= j
																																		except ValueError:
																																									new_word.extend(word[i:]		)
																																									break

																																		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
																											__SCREAMING_SNAKE_CASE									= tuple(_A		)
																											__SCREAMING_SNAKE_CASE									= new_word
																											if len(_A		) == 1:
																																		break
																											else:
																																		__SCREAMING_SNAKE_CASE									= get_pairs(_A		)
																				__SCREAMING_SNAKE_CASE									= '@@ '.join(_A		)
																				__SCREAMING_SNAKE_CASE									= word[:-4]

																				__SCREAMING_SNAKE_CASE									= word
																				words.append(_A		)
													return " ".join(_A		)






						def 	_A     (   self				,							_A		):







													'''simple docstring'''

													__SCREAMING_SNAKE_CASE									= []

													__SCREAMING_SNAKE_CASE									= re.findall(r'\S+\n?'				,							_A		)

													for token in words:
																				split_tokens.extend(list(self.bpe(_A		).split(' '		)		)		)
													return split_tokens






						def 	_A     (   self				,							_A		):







													'''simple docstring'''

													__SCREAMING_SNAKE_CASE									= token.lower()
													return self.encoder.get(_A				,							self.encoder.get(self.unk_token		)		)






						def 	_A     (   self				,							_A		):







													'''simple docstring'''

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






						def 	_A     (   self				,							_A		):







													'''simple docstring'''

													__SCREAMING_SNAKE_CASE									= ' '.join(_A		).replace('@@ '				,							''		).strip()
													return out_string






						def 	_A     (   self				,							_A				,							_A = None		):







													'''simple docstring'''

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

													__SCREAMING_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 _A		: 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!'		)
																																		__SCREAMING_SNAKE_CASE									= token_index
																											writer.write(' '.join(_A		) + '\n'		)
																											index += 1

													return vocab_file, merge_file











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        UpperCAmelCase_  (       UpperCamelCase_   ,   UpperCamelCase_   ,   unittest.TestCase	):

						'''simple docstring'''




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






						def 	_A     (   self		):







													'''simple docstring'''

													torch.manual_seed(0		)
													__SCREAMING_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				,							)
													__SCREAMING_SNAKE_CASE									= DDIMScheduler(
													    beta_start=0.0_0_0_8_5				,							beta_end=0.0_1_2				,							beta_schedule='scaled_linear'				,							num_train_timesteps=1_000				,							clip_sample=_A				,							set_alpha_to_one=_A				,							)
													torch.manual_seed(0		)
													__SCREAMING_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		)
													__SCREAMING_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				,							)
													__SCREAMING_SNAKE_CASE									= CLIPTextModel(_A		)
													__SCREAMING_SNAKE_CASE									= CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'		)

													__SCREAMING_SNAKE_CASE									= {
													    'unet': unet,
													    'scheduler': scheduler,
													    'vae': vae,
													    'text_encoder': text_encoder,
													    'tokenizer': tokenizer,
													    'safety_checker': None,
													    'feature_extractor': None,
													}
													return components






						def 	_A     (   self				,							_A				,							_A=0		):







													'''simple docstring'''

													__SCREAMING_SNAKE_CASE									= floats_tensor((1, 3, 32, 32)				,							rng=random.Random(_A		)		).to(_A		)
													__SCREAMING_SNAKE_CASE									= image / 2 + 0.5
													if str(_A		).startswith('mps'		):
																				__SCREAMING_SNAKE_CASE									= torch.manual_seed(_A		)
													else:
																				__SCREAMING_SNAKE_CASE									= torch.Generator(device=_A		).manual_seed(_A		)
													__SCREAMING_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 	_A     (   self		):







													'''simple docstring'''

													__SCREAMING_SNAKE_CASE									= 'cpu'  # ensure determinism for the device-dependent torch.Generator

													__SCREAMING_SNAKE_CASE									= self.get_dummy_components()
													__SCREAMING_SNAKE_CASE									= CycleDiffusionPipeline(**_A		)
													__SCREAMING_SNAKE_CASE									= pipe.to(_A		)
													pipe.set_progress_bar_config(disable=_A		)

													__SCREAMING_SNAKE_CASE									= self.get_dummy_inputs(_A		)
													__SCREAMING_SNAKE_CASE									= pipe(**_A		)
													__SCREAMING_SNAKE_CASE									= output.images

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

													assert images.shape == (1, 32, 32, 3)
													__SCREAMING_SNAKE_CASE									= np.array([0.4_4_5_9, 0.4_9_4_3, 0.4_5_4_4, 0.6_6_4_3, 0.5_4_7_4, 0.4_3_2_7, 0.5_7_0_1, 0.5_9_5_9, 0.5_1_7_9]		)

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






						@unittest.skipIf(torch_device != 'cuda'				,							'This test requires a GPU'		)
						def 	_A     (   self		):







													'''simple docstring'''

													__SCREAMING_SNAKE_CASE									= self.get_dummy_components()
													for name, module in components.items():
																				if hasattr(_A				,							'half'		):
																											__SCREAMING_SNAKE_CASE									= module.half()
													__SCREAMING_SNAKE_CASE									= CycleDiffusionPipeline(**_A		)
													__SCREAMING_SNAKE_CASE									= pipe.to(_A		)
													pipe.set_progress_bar_config(disable=_A		)

													__SCREAMING_SNAKE_CASE									= self.get_dummy_inputs(_A		)
													__SCREAMING_SNAKE_CASE									= pipe(**_A		)
													__SCREAMING_SNAKE_CASE									= output.images

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

													assert images.shape == (1, 32, 32, 3)
													__SCREAMING_SNAKE_CASE									= np.array([0.3_5_0_6, 0.4_5_4_3, 0.4_4_6, 0.4_5_7_5, 0.5_1_9_5, 0.4_1_5_5, 0.5_2_7_3, 0.5_1_8, 0.4_1_1_6]		)

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






						@skip_mps
						def 	_A     (   self		):







													'''simple docstring'''

													return super().test_save_load_local()






						@unittest.skip('non-deterministic pipeline'		)
						def 	_A     (   self		):







													'''simple docstring'''

													return super().test_inference_batch_single_identical()






						@skip_mps
						def 	_A     (   self		):







													'''simple docstring'''

													return super().test_dict_tuple_outputs_equivalent()






						@skip_mps
						def 	_A     (   self		):







													'''simple docstring'''

													return super().test_save_load_optional_components()






						@skip_mps
						def 	_A     (   self		):







													'''simple docstring'''

													return super().test_attention_slicing_forward_pass()







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

						'''simple docstring'''




						def 	_A     (   self		):







													'''simple docstring'''

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






						def 	_A     (   self		):







													'''simple docstring'''

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

													__SCREAMING_SNAKE_CASE									= 'CompVis/stable-diffusion-v1-4'
													__SCREAMING_SNAKE_CASE									= DDIMScheduler.from_pretrained(_A				,							subfolder='scheduler'		)
													__SCREAMING_SNAKE_CASE									= CycleDiffusionPipeline.from_pretrained(
													    _A				,							scheduler=_A				,							safety_checker=_A				,							torch_dtype=torch.floataa				,							revision='fp16'		)

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

													__SCREAMING_SNAKE_CASE									= 'A black colored car'
													__SCREAMING_SNAKE_CASE									= 'A blue colored car'

													__SCREAMING_SNAKE_CASE									= torch.manual_seed(0		)
													__SCREAMING_SNAKE_CASE									= pipe(
													    prompt=_A				,							source_prompt=_A				,							image=_A				,							num_inference_steps=100				,							eta=0.1				,							strength=0.8_5				,							guidance_scale=3				,							source_guidance_scale=1				,							generator=_A				,							output_type='np'				,							)
													__SCREAMING_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 	_A     (   self		):







													'''simple docstring'''

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

													__SCREAMING_SNAKE_CASE									= 'CompVis/stable-diffusion-v1-4'
													__SCREAMING_SNAKE_CASE									= DDIMScheduler.from_pretrained(_A				,							subfolder='scheduler'		)
													__SCREAMING_SNAKE_CASE									= CycleDiffusionPipeline.from_pretrained(_A				,							scheduler=_A				,							safety_checker=_A		)

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

													__SCREAMING_SNAKE_CASE									= 'A black colored car'
													__SCREAMING_SNAKE_CASE									= 'A blue colored car'

													__SCREAMING_SNAKE_CASE									= torch.manual_seed(0		)
													__SCREAMING_SNAKE_CASE									= pipe(
													    prompt=_A				,							source_prompt=_A				,							image=_A				,							num_inference_steps=100				,							eta=0.1				,							strength=0.8_5				,							guidance_scale=3				,							source_guidance_scale=1				,							generator=_A				,							output_type='np'				,							)
													__SCREAMING_SNAKE_CASE									= output.images

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






