






'''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        __SCREAMING_SNAKE_CASE (     lowerCamelCase		,			lowerCamelCase		,			unittest.TestCase			):
 snake_case_												=							CycleDiffusionPipeline
 snake_case_												=							TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
     """negative_prompt""",
     """height""",
     """width""",
     """negative_prompt_embeds""",
 }
 snake_case_												=							PipelineTesterMixin.required_optional_params - {"""latents"""}
 snake_case_												=							TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"""source_prompt"""}			)
 snake_case_												=							IMAGE_TO_IMAGE_IMAGE_PARAMS
 snake_case_												=							IMAGE_TO_IMAGE_IMAGE_PARAMS




 def        __magic_name__		( self		:							Optional[Any] )		->  Union[str, Any]:
       torch.manual_seed(0 )
       SCREAMING_SNAKE_CASE__       :			str													=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__       :			int													=DDIMScheduler(
           beta_start=0.00085   , beta_end=0.012   , beta_schedule='''scaled_linear'''   , num_train_timesteps=10_00   , clip_sample=__lowercase   , set_alpha_to_one=__lowercase   , )
       torch.manual_seed(0 )
       SCREAMING_SNAKE_CASE__       :			Optional[int]													=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__       :			str													=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   , )
       SCREAMING_SNAKE_CASE__       :			Tuple													=CLIPTextModel(__lowercase )
       SCREAMING_SNAKE_CASE__       :			Any													=CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip''' )

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




 def        __magic_name__		( self		:							List[Any]   , __lowercase		:							Dict   , __lowercase		:							Optional[int]=0 )		->  Tuple:
       SCREAMING_SNAKE_CASE__       :			List[Any]													=floats_tensor((1, 3, 32, 32)   , rng=random.Random(__lowercase ) ).to(__lowercase )
       SCREAMING_SNAKE_CASE__       :			Tuple													=image / 2 + 0.5
       if str(__lowercase ).startswith('''mps''' ):
             SCREAMING_SNAKE_CASE__       :			List[str]													=torch.manual_seed(__lowercase )
       else:
             SCREAMING_SNAKE_CASE__       :			int													=torch.Generator(device=__lowercase ).manual_seed(__lowercase )
       SCREAMING_SNAKE_CASE__       :			List[Any]													={
           '''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        __magic_name__		( self		:							Union[str, Any] )		->  List[str]:
       SCREAMING_SNAKE_CASE__       :			Any													='''cpu'''  # ensure determinism for the device-dependent torch.Generator

       SCREAMING_SNAKE_CASE__       :			List[str]													=self.get_dummy_components()
       SCREAMING_SNAKE_CASE__       :			Union[str, Any]													=CycleDiffusionPipeline(**__lowercase )
       SCREAMING_SNAKE_CASE__       :			Optional[Any]													=pipe.to(__lowercase )
       pipe.set_progress_bar_config(disable=__lowercase )

       SCREAMING_SNAKE_CASE__       :			Any													=self.get_dummy_inputs(__lowercase )
       SCREAMING_SNAKE_CASE__       :			Dict													=pipe(**__lowercase )
       SCREAMING_SNAKE_CASE__       :			List[str]													=output.images

       SCREAMING_SNAKE_CASE__       :			Optional[int]													=images[0, -3:, -3:, -1]

       assert images.shape == (1, 32, 32, 3)
       SCREAMING_SNAKE_CASE__       :			Union[str, Any]													=np.array([0.4459, 0.4943, 0.4544, 0.6643, 0.5474, 0.4327, 0.5701, 0.5959, 0.5179] )

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




 @unittest.skipIf(torch_device != '''cuda'''   , '''This test requires a GPU''' )
 def        __magic_name__		( self		:							Optional[int] )		->  int:
       SCREAMING_SNAKE_CASE__       :			int													=self.get_dummy_components()
       for name, module in components.items():
             if hasattr(__lowercase   , '''half''' ):
                   SCREAMING_SNAKE_CASE__       :			Union[str, Any]													=module.half()
       SCREAMING_SNAKE_CASE__       :			Dict													=CycleDiffusionPipeline(**__lowercase )
       SCREAMING_SNAKE_CASE__       :			Any													=pipe.to(__lowercase )
       pipe.set_progress_bar_config(disable=__lowercase )

       SCREAMING_SNAKE_CASE__       :			Dict													=self.get_dummy_inputs(__lowercase )
       SCREAMING_SNAKE_CASE__       :			Union[str, Any]													=pipe(**__lowercase )
       SCREAMING_SNAKE_CASE__       :			List[str]													=output.images

       SCREAMING_SNAKE_CASE__       :			Optional[Any]													=images[0, -3:, -3:, -1]

       assert images.shape == (1, 32, 32, 3)
       SCREAMING_SNAKE_CASE__       :			Any													=np.array([0.3506, 0.4543, 0.446, 0.4575, 0.5195, 0.4155, 0.5273, 0.518, 0.4116] )

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




 @skip_mps
 def        __magic_name__		( self		:							str )		->  Union[str, Any]:
       return super().test_save_load_local()




 @unittest.skip('''non-deterministic pipeline''' )
 def        __magic_name__		( self		:							int )		->  Union[str, Any]:
       return super().test_inference_batch_single_identical()




 @skip_mps
 def        __magic_name__		( self		:							int )		->  str:
       return super().test_dict_tuple_outputs_equivalent()




 @skip_mps
 def        __magic_name__		( self		:							Optional[int] )		->  Optional[int]:
       return super().test_save_load_optional_components()




 @skip_mps
 def        __magic_name__		( self		:							Dict )		->  Optional[int]:
       return super().test_attention_slicing_forward_pass()






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




 def        __magic_name__		( self		:							Optional[Any] )		->  str:
       # clean up the VRAM after each test
       super().tearDown()
       gc.collect()
       torch.cuda.empty_cache()




 def        __magic_name__		( self		:							Optional[int] )		->  int:
       SCREAMING_SNAKE_CASE__       :			Any													=load_image(
           '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
           '''/cycle-diffusion/black_colored_car.png''' )
       SCREAMING_SNAKE_CASE__       :			int													=load_numpy(
           '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car_fp16.npy''' )
       SCREAMING_SNAKE_CASE__       :			Dict													=init_image.resize((5_12, 5_12) )

       SCREAMING_SNAKE_CASE__       :			List[Any]													='''CompVis/stable-diffusion-v1-4'''
       SCREAMING_SNAKE_CASE__       :			Union[str, Any]													=DDIMScheduler.from_pretrained(__lowercase   , subfolder='''scheduler''' )
       SCREAMING_SNAKE_CASE__       :			Dict													=CycleDiffusionPipeline.from_pretrained(
           __lowercase   , scheduler=__lowercase   , safety_checker=__lowercase   , torch_dtype=torch.floataa   , revision='''fp16''' )

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

       SCREAMING_SNAKE_CASE__       :			List[Any]													='''A black colored car'''
       SCREAMING_SNAKE_CASE__       :			Optional[Any]													='''A blue colored car'''

       SCREAMING_SNAKE_CASE__       :			int													=torch.manual_seed(0 )
       SCREAMING_SNAKE_CASE__       :			Union[str, Any]													=pipe(
           prompt=__lowercase   , source_prompt=__lowercase   , image=__lowercase   , num_inference_steps=1_00   , eta=0.1   , strength=0.85   , guidance_scale=3   , source_guidance_scale=1   , generator=__lowercase   , output_type='''np'''   , )
       SCREAMING_SNAKE_CASE__       :			Optional[Any]													=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        __magic_name__		( self		:							Tuple )		->  Optional[Any]:
       SCREAMING_SNAKE_CASE__       :			Optional[int]													=load_image(
           '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
           '''/cycle-diffusion/black_colored_car.png''' )
       SCREAMING_SNAKE_CASE__       :			List[Any]													=load_numpy(
           '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car.npy''' )
       SCREAMING_SNAKE_CASE__       :			Optional[int]													=init_image.resize((5_12, 5_12) )

       SCREAMING_SNAKE_CASE__       :			Optional[Any]													='''CompVis/stable-diffusion-v1-4'''
       SCREAMING_SNAKE_CASE__       :			Any													=DDIMScheduler.from_pretrained(__lowercase   , subfolder='''scheduler''' )
       SCREAMING_SNAKE_CASE__       :			Optional[Any]													=CycleDiffusionPipeline.from_pretrained(__lowercase   , scheduler=__lowercase   , safety_checker=__lowercase )

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

       SCREAMING_SNAKE_CASE__       :			str													='''A black colored car'''
       SCREAMING_SNAKE_CASE__       :			str													='''A blue colored car'''

       SCREAMING_SNAKE_CASE__       :			str													=torch.manual_seed(0 )
       SCREAMING_SNAKE_CASE__       :			Optional[int]													=pipe(
           prompt=__lowercase   , source_prompt=__lowercase   , image=__lowercase   , num_inference_steps=1_00   , eta=0.1   , strength=0.85   , guidance_scale=3   , source_guidance_scale=1   , generator=__lowercase   , output_type='''np'''   , )
       SCREAMING_SNAKE_CASE__       :			Union[str, Any]													=output.images

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







'''simple docstring'''
import json
import os
from typing import Dict, List, Optional, Tuple

import regex as re

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


a_					=	logging.get_logger(__name__)


a_					=	{
    'vocab_file': 'vocab.json',
    'merges_file': 'merges.txt',
    'tokenizer_config_file': 'tokenizer_config.json',
}

a_					=	{
    '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'
        )
    },
}

a_					=	{'facebook/blenderbot_small-90M': 5_1_2}





def 	_a(					UpperCamelCase__			:  Dict							):




      '''simple docstring'''






      SCREAMING_SNAKE_CASE__       :			Union[str, Any]													=set()
      SCREAMING_SNAKE_CASE__       :			Optional[Any]													=word[0]
      for char in word[1:]:
            pairs.add((prev_char, char)							)
            SCREAMING_SNAKE_CASE__       :			Optional[Any]													=char

      SCREAMING_SNAKE_CASE__       :			Any													=set(UpperCamelCase__							)
      return pairs







class        __SCREAMING_SNAKE_CASE (     lowerCamelCase			):
 snake_case_												=							VOCAB_FILES_NAMES
 snake_case_												=							PRETRAINED_VOCAB_FILES_MAP
 snake_case_												=							PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 snake_case_												=							["""input_ids""", """attention_mask"""]




 def __init__( self		:							Tuple   , __lowercase		:							int   , __lowercase		:							Optional[int]   , __lowercase		:							List[str]="__start__"   , __lowercase		:							Union[str, Any]="__end__"   , __lowercase		:							str="__unk__"   , __lowercase		:							Union[str, Any]="__null__"   , **__lowercase		:							List[str]   , )		->  Optional[int]:
       super().__init__(unk_token=__lowercase   , bos_token=__lowercase   , eos_token=__lowercase   , pad_token=__lowercase   , **__lowercase )

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




 @property
 def        __magic_name__		( self		:							Any )		->  int:
       return len(self.encoder )




 def        __magic_name__		( self		:							Tuple )		->  Dict:
       return dict(self.encoder   , **self.added_tokens_encoder )




 def        __magic_name__		( self		:							Optional[Any]   , __lowercase		:							str )		->  str:
       if token in self.cache:
             return self.cache[token]
       SCREAMING_SNAKE_CASE__       :			Union[str, Any]													=re.sub('''([.,!?()])'''   , r''' \1'''   , __lowercase )
       SCREAMING_SNAKE_CASE__       :			int													=re.sub('''(\')'''   , r''' \1 '''   , __lowercase )
       SCREAMING_SNAKE_CASE__       :			Dict													=re.sub(r'''\s{2,}'''   , ''' '''   , __lowercase )
       if "\n" in token:
             SCREAMING_SNAKE_CASE__       :			List[str]													=token.replace('''\n'''   , ''' __newln__''' )

       SCREAMING_SNAKE_CASE__       :			Dict													=token.split(''' ''' )
       SCREAMING_SNAKE_CASE__       :			Dict													=[]
       for token in tokens:
             if not len(__lowercase ):
                   continue

             SCREAMING_SNAKE_CASE__       :			Union[str, Any]													=token.lower()
             SCREAMING_SNAKE_CASE__       :			Optional[Any]													=tuple(__lowercase )
             SCREAMING_SNAKE_CASE__       :			Optional[int]													=tuple(list(word[:-1] ) + [word[-1] + '''</w>'''] )
             SCREAMING_SNAKE_CASE__       :			Optional[Any]													=get_pairs(__lowercase )

             if not pairs:
                   words.append(__lowercase )
                   continue

             while True:
                   SCREAMING_SNAKE_CASE__       :			int													=min(__lowercase   , key=lambda __lowercase : self.bpe_ranks.get(__lowercase   , float('''inf''' ) ) )
                   if bigram not in self.bpe_ranks:
                         break
                   SCREAMING_SNAKE_CASE__						,		SCREAMING_SNAKE_CASE__       :			Any													=bigram
                   SCREAMING_SNAKE_CASE__       :			List[Any]													=[]
                   SCREAMING_SNAKE_CASE__       :			Tuple													=0

                   while i < len(__lowercase ):
                         try:
                               SCREAMING_SNAKE_CASE__       :			Optional[Any]													=word.index(__lowercase   , __lowercase )
                               new_word.extend(word[i:j] )
                               SCREAMING_SNAKE_CASE__       :			Optional[int]													=j
                         except ValueError:
                               new_word.extend(word[i:] )
                               break

                         if word[i] == first and i < len(__lowercase ) - 1 and word[i + 1] == second:
                               new_word.append(first + second )
                               i += 2
                         else:
                               new_word.append(word[i] )
                               i += 1
                   SCREAMING_SNAKE_CASE__       :			int													=tuple(__lowercase )
                   SCREAMING_SNAKE_CASE__       :			int													=new_word
                   if len(__lowercase ) == 1:
                         break
                   else:
                         SCREAMING_SNAKE_CASE__       :			Any													=get_pairs(__lowercase )
             SCREAMING_SNAKE_CASE__       :			Optional[int]													='''@@ '''.join(__lowercase )
             SCREAMING_SNAKE_CASE__       :			List[Any]													=word[:-4]

             SCREAMING_SNAKE_CASE__       :			str													=word
             words.append(__lowercase )
       return " ".join(__lowercase )




 def        __magic_name__		( self		:							str   , __lowercase		:							str )		->  List[str]:
       SCREAMING_SNAKE_CASE__       :			Dict													=[]

       SCREAMING_SNAKE_CASE__       :			List[Any]													=re.findall(r'''\S+\n?'''   , __lowercase )

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




 def        __magic_name__		( self		:							List[Any]   , __lowercase		:							str )		->  int:
       SCREAMING_SNAKE_CASE__       :			int													=token.lower()
       return self.encoder.get(__lowercase   , self.encoder.get(self.unk_token ) )




 def        __magic_name__		( self		:							int   , __lowercase		:							int )		->  str:
       return self.decoder.get(__lowercase   , self.unk_token )




 def        __magic_name__		( self		:							List[Any]   , __lowercase		:							List[str] )		->  str:
       SCREAMING_SNAKE_CASE__       :			Optional[int]													=''' '''.join(__lowercase ).replace('''@@ '''   , '''''' ).strip()
       return out_string




 def        __magic_name__		( self		:							Tuple   , __lowercase		:							str   , __lowercase		:							Optional[str] = None )		->  Tuple[str]:
       if not os.path.isdir(__lowercase ):
             logger.error(F"Vocabulary path ({save_directory}) should be a directory" )
             return
       SCREAMING_SNAKE_CASE__       :			str													=os.path.join(
           __lowercase   , (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file'''] )
       SCREAMING_SNAKE_CASE__       :			int													=os.path.join(
           __lowercase   , (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''merges_file'''] )

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

       SCREAMING_SNAKE_CASE__       :			List[str]													=0
       with open(__lowercase   , '''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 __lowercase : 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__       :			Any													=token_index
                   writer.write(''' '''.join(__lowercase ) + '''\n''' )
                   index += 1

       return vocab_file, merge_file