






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





				"""simple docstring"""







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


				def       __lowercase							( self    ):




									"""simple docstring"""

									torch.manual_seed(0    )
									__UpperCamelCase      :					List[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							,  )
									__UpperCamelCase      :					int       		=DDIMScheduler(
									    beta_start=0.00_085							,  beta_end=0.012							,  beta_schedule='scaled_linear'							,  num_train_timesteps=1000							,  clip_sample=lowerCamelCase__							,  set_alpha_to_one=lowerCamelCase__							,  )
									torch.manual_seed(0    )
									__UpperCamelCase      :					Union[str, 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							,  )
									torch.manual_seed(0    )
									__UpperCamelCase      :					List[Any]       		=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=1000							,  )
									__UpperCamelCase      :					Any       		=CLIPTextModel(lowerCamelCase__    )
									__UpperCamelCase      :					Any       		=CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'    )

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


				def       __lowercase							( self							,  lowerCamelCase__							,  lowerCamelCase__=0    ):




									"""simple docstring"""

									__UpperCamelCase      :					Optional[int]       		=floats_tensor((1, 3, 32, 32)							,  rng=random.Random(lowerCamelCase__    )    ).to(lowerCamelCase__    )
									__UpperCamelCase      :					List[str]       		=image / 2 + 0.5
									if str(lowerCamelCase__    ).startswith('mps'    ):
														__UpperCamelCase      :					int       		=torch.manual_seed(lowerCamelCase__    )
									else:
														__UpperCamelCase      :					Tuple       		=torch.Generator(device=lowerCamelCase__    ).manual_seed(lowerCamelCase__    )
									__UpperCamelCase      :					Optional[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       __lowercase							( self    ):




									"""simple docstring"""

									__UpperCamelCase      :					Tuple       		='cpu'  # ensure determinism for the device-dependent torch.Generator

									__UpperCamelCase      :					Tuple       		=self.get_dummy_components()
									__UpperCamelCase      :					Optional[Any]       		=CycleDiffusionPipeline(**lowerCamelCase__    )
									__UpperCamelCase      :					int       		=pipe.to(lowerCamelCase__    )
									pipe.set_progress_bar_config(disable=lowerCamelCase__    )

									__UpperCamelCase      :					Union[str, Any]       		=self.get_dummy_inputs(lowerCamelCase__    )
									__UpperCamelCase      :					Dict       		=pipe(**lowerCamelCase__    )
									__UpperCamelCase      :					int       		=output.images

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

									assert images.shape == (1, 32, 32, 3)
									__UpperCamelCase      :					Any       		=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       __lowercase							( self    ):




									"""simple docstring"""

									__UpperCamelCase      :					Tuple       		=self.get_dummy_components()
									for name, module in components.items():
														if hasattr(lowerCamelCase__							,  'half'    ):
																			__UpperCamelCase      :					List[str]       		=module.half()
									__UpperCamelCase      :					int       		=CycleDiffusionPipeline(**lowerCamelCase__    )
									__UpperCamelCase      :					Any       		=pipe.to(lowerCamelCase__    )
									pipe.set_progress_bar_config(disable=lowerCamelCase__    )

									__UpperCamelCase      :					int       		=self.get_dummy_inputs(lowerCamelCase__    )
									__UpperCamelCase      :					Union[str, Any]       		=pipe(**lowerCamelCase__    )
									__UpperCamelCase      :					Dict       		=output.images

									__UpperCamelCase      :					int       		=images[0, -3:, -3:, -1]

									assert images.shape == (1, 32, 32, 3)
									__UpperCamelCase      :					Optional[int]       		=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       __lowercase							( self    ):




									"""simple docstring"""

									return super().test_save_load_local()


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




									"""simple docstring"""

									return super().test_inference_batch_single_identical()


				@skip_mps
				def       __lowercase							( self    ):




									"""simple docstring"""

									return super().test_dict_tuple_outputs_equivalent()


				@skip_mps
				def       __lowercase							( self    ):




									"""simple docstring"""

									return super().test_save_load_optional_components()







				@skip_mps
				def       __lowercase							( self    ):




									"""simple docstring"""

									return super().test_attention_slicing_forward_pass()




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





				"""simple docstring"""







				def       __lowercase							( self    ):




									"""simple docstring"""

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


				def       __lowercase							( self    ):




									"""simple docstring"""

									__UpperCamelCase      :					List[str]       		=load_image(
									    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
									    '/cycle-diffusion/black_colored_car.png'    )
									__UpperCamelCase      :					Union[str, Any]       		=load_numpy(
									    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car_fp16.npy'    )
									__UpperCamelCase      :					Union[str, Any]       		=init_image.resize((512, 512)    )

									__UpperCamelCase      :					Dict       		='CompVis/stable-diffusion-v1-4'
									__UpperCamelCase      :					Optional[int]       		=DDIMScheduler.from_pretrained(lowerCamelCase__							,  subfolder='scheduler'    )
									__UpperCamelCase      :					List[str]       		=CycleDiffusionPipeline.from_pretrained(
									    lowerCamelCase__							,  scheduler=lowerCamelCase__							,  safety_checker=lowerCamelCase__							,  torch_dtype=torch.floataa							,  revision='fp16'    )

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

									__UpperCamelCase      :					str       		='A black colored car'
									__UpperCamelCase      :					Optional[int]       		='A blue colored car'

									__UpperCamelCase      :					Optional[int]       		=torch.manual_seed(0    )
									__UpperCamelCase      :					Dict       		=pipe(
									    prompt=lowerCamelCase__							,  source_prompt=lowerCamelCase__							,  image=lowerCamelCase__							,  num_inference_steps=100							,  eta=0.1							,  strength=0.85							,  guidance_scale=3							,  source_guidance_scale=1							,  generator=lowerCamelCase__							,  output_type='np'							,  )
									__UpperCamelCase      :					str       		=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       __lowercase							( self    ):




									"""simple docstring"""

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

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

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

									__UpperCamelCase      :					Union[str, Any]       		='A black colored car'
									__UpperCamelCase      :					Optional[Any]       		='A blue colored car'

									__UpperCamelCase      :					str       		=torch.manual_seed(0    )
									__UpperCamelCase      :					List[Any]       		=pipe(
									    prompt=lowerCamelCase__							,  source_prompt=lowerCamelCase__							,  image=lowerCamelCase__							,  num_inference_steps=100							,  eta=0.1							,  strength=0.85							,  guidance_scale=3							,  source_guidance_scale=1							,  generator=lowerCamelCase__							,  output_type='np'							,  )
									__UpperCamelCase      :					Any       		=output.images

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



"""simple docstring"""





import os
from shutil import copyfile
from typing import List, Optional, Tuple

from ...tokenization_utils import AddedToken
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import is_sentencepiece_available, logging


if is_sentencepiece_available():
    from .tokenization_fnet import FNetTokenizer
else:
    _a		       =							None

_a		       =							logging.get_logger(__name__)
_a		       =							{'vocab_file': 'spiece.model', 'tokenizer_file': 'tokenizer.json'}

_a		       =							{
    'vocab_file': {
        'google/fnet-base': 'https://huggingface.co/google/fnet-base/resolve/main/spiece.model',
        'google/fnet-large': 'https://huggingface.co/google/fnet-large/resolve/main/spiece.model',
    },
    'tokenizer_file': {
        'google/fnet-base': 'https://huggingface.co/google/fnet-base/resolve/main/tokenizer.json',
        'google/fnet-large': 'https://huggingface.co/google/fnet-large/resolve/main/tokenizer.json',
    },
}

_a		       =							{
    'google/fnet-base': 512,
    'google/fnet-large': 512,
}

_a		       =							'▁'


class A_	(lowercase__      ):




  '''simple docstring'''
  SCREAMING_SNAKE_CASE__			:						List[str]       =		VOCAB_FILES_NAMES
  SCREAMING_SNAKE_CASE__			:						Tuple       =		PRETRAINED_VOCAB_FILES_MAP
  SCREAMING_SNAKE_CASE__			:						Any       =		PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
  SCREAMING_SNAKE_CASE__			:						Union[str, Any]       =		["""input_ids""", """token_type_ids"""]
  SCREAMING_SNAKE_CASE__			:						Tuple       =		FNetTokenizer







  def __init__(		self				,   lowercase_=None				,   lowercase_=None				,   lowercase_=False				,   lowercase_=True				,   lowercase_=True				,   lowercase_="<unk>"				,   lowercase_="[SEP]"				,   lowercase_="<pad>"				,   lowercase_="[CLS]"				,   lowercase_="[MASK]"				,   **lowercase_				,   ):

     """simple docstring"""
     # Mask token behave like a normal word, i.e. include the space before it and
     # is included in the raw text, there should be a match in a non-normalized sentence.
     UpperCAmelCase_					:					int			     =    (
         AddedToken(lowercase_				,   lstrip=lowercase_				,   rstrip=lowercase_				,   normalized=lowercase_  )
         if isinstance(lowercase_				,   lowercase_  )
         else mask_token
     )

     super().__init__(
         lowercase_				,   tokenizer_file=lowercase_				,   do_lower_case=lowercase_				,   remove_space=lowercase_				,   keep_accents=lowercase_				,   unk_token=lowercase_				,   sep_token=lowercase_				,   pad_token=lowercase_				,   cls_token=lowercase_				,   mask_token=lowercase_				,   **lowercase_				,   )

     UpperCAmelCase_					:					Any			     =    do_lower_case
     UpperCAmelCase_					:					Tuple			     =    remove_space
     UpperCAmelCase_					:					str			     =    keep_accents
     UpperCAmelCase_					:					Any			     =    vocab_file
     UpperCAmelCase_					:					List[Any]			     =    False if not self.vocab_file else True







  def        UpperCamelCase__					(		self				,   lowercase_				,   lowercase_ = None  ):

     """simple docstring"""
     UpperCAmelCase_					:					Tuple			     =    [self.sep_token_id]
     UpperCAmelCase_					:					Any			     =    [self.cls_token_id]
     if token_ids_a is None:
        return cls + token_ids_a + sep
     return cls + token_ids_a + sep + token_ids_a + sep







  def        UpperCamelCase__					(		self				,   lowercase_				,   lowercase_ = None  ):

     """simple docstring"""
     UpperCAmelCase_					:					Any			     =    [self.sep_token_id]
     UpperCAmelCase_					:					Any			     =    [self.cls_token_id]

     if token_ids_a is None:
        return len(cls + token_ids_a + sep  ) * [0]
     return len(cls + token_ids_a + sep  ) * [0] + len(token_ids_a + sep  ) * [1]







  def        UpperCamelCase__					(		self				,   lowercase_				,   lowercase_ = None  ):

     """simple docstring"""
     if not os.path.isdir(lowercase_  ):
        logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""  )
        return
     UpperCAmelCase_					:					List[str]			     =    os.path.join(
         lowercase_				,   (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]  )

     if os.path.abspath(self.vocab_file  ) != os.path.abspath(lowercase_  ):
        copyfile(self.vocab_file				,   lowercase_  )

     return (out_vocab_file,)


