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



     def  lowerCamelCase_						(							self :Any   ):







      '''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_     :					Optional[int]   	=DDIMScheduler(
          beta_start=0.00085      ,  beta_end=0.012      ,  beta_schedule='scaled_linear'      ,  num_train_timesteps=1_000      ,  clip_sample=_lowerCamelCase      ,  set_alpha_to_one=_lowerCamelCase      ,  )
      torch.manual_seed(0   )
      UpperCamelCase_     :					List[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_     :					Tuple   	=CLIPTextConfig(
          bos_token_id=0      ,  eos_token_id=2      ,  hidden_size=32      ,  intermediate_size=37      ,  layer_norm_eps=1E-0_5      ,  num_attention_heads=4      ,  num_hidden_layers=5      ,  pad_token_id=1      ,  vocab_size=1_000      ,  )
      UpperCamelCase_     :					List[str]   	=CLIPTextModel(_lowerCamelCase   )
      UpperCamelCase_     :					int   	=CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'   )

      UpperCamelCase_     :					Optional[Any]   	={
          'unet': unet,
          'scheduler': scheduler,
          'vae': vae,
          'text_encoder': text_encoder,
          'tokenizer': tokenizer,
          'safety_checker': None,
          'feature_extractor': None,
      }
      return components



     def  lowerCamelCase_						(							self :Union[str, Any]      ,  _lowerCamelCase :str      ,  _lowerCamelCase :Dict=0   ):







      '''simple docstring'''



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







      '''simple docstring'''



      UpperCamelCase_     :					Any   	='cpu'  # ensure determinism for the device-dependent torch.Generator

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

      UpperCamelCase_     :					Tuple   	=self.get_dummy_inputs(_lowerCamelCase   )
      UpperCamelCase_     :					List[Any]   	=pipe(**_lowerCamelCase   )
      UpperCamelCase_     :					Any   	=output.images

      UpperCamelCase_     :					List[str]   	=images[0, -3:, -3:, -1]

      assert images.shape == (1, 32, 32, 3)
      UpperCamelCase_     :					int   	=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  lowerCamelCase_						(							self :Dict   ):







      '''simple docstring'''



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

      UpperCamelCase_     :					int   	=self.get_dummy_inputs(_lowerCamelCase   )
      UpperCamelCase_     :					Any   	=pipe(**_lowerCamelCase   )
      UpperCamelCase_     :					List[Any]   	=output.images

      UpperCamelCase_     :					Any   	=images[0, -3:, -3:, -1]

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







      '''simple docstring'''



      return super().test_save_load_local()



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







      '''simple docstring'''



      return super().test_inference_batch_single_identical()



     @skip_mps
     def  lowerCamelCase_						(							self :Tuple   ):







      '''simple docstring'''



      return super().test_dict_tuple_outputs_equivalent()



     @skip_mps
     def  lowerCamelCase_						(							self :Dict   ):







      '''simple docstring'''



      return super().test_save_load_optional_components()




     @skip_mps
     def  lowerCamelCase_						(							self :Union[str, Any]   ):







      '''simple docstring'''



      return super().test_attention_slicing_forward_pass()





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



     def  lowerCamelCase_						(							self :int   ):







      '''simple docstring'''



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



     def  lowerCamelCase_						(							self :Union[str, Any]   ):







      '''simple docstring'''



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

      UpperCamelCase_     :					Optional[Any]   	='CompVis/stable-diffusion-v1-4'
      UpperCamelCase_     :					Optional[Any]   	=DDIMScheduler.from_pretrained(_lowerCamelCase      ,  subfolder='scheduler'   )
      UpperCamelCase_     :					Dict   	=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_     :					Dict   	='A black colored car'
      UpperCamelCase_     :					int   	='A blue colored car'

      UpperCamelCase_     :					Any   	=torch.manual_seed(0   )
      UpperCamelCase_     :					Optional[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_     :					Dict   	=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 :str   ):







      '''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_     :					Optional[Any]   	=load_numpy(
          'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car.npy'   )
      UpperCamelCase_     :					Any   	=init_image.resize((512, 512)   )

      UpperCamelCase_     :					str   	='CompVis/stable-diffusion-v1-4'
      UpperCamelCase_     :					Dict   	=DDIMScheduler.from_pretrained(_lowerCamelCase      ,  subfolder='scheduler'   )
      UpperCamelCase_     :					Optional[int]   	=CycleDiffusionPipeline.from_pretrained(_lowerCamelCase      ,  scheduler=_lowerCamelCase      ,  safety_checker=_lowerCamelCase   )

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

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

      UpperCamelCase_     :					List[str]   	=torch.manual_seed(0   )
      UpperCamelCase_     :					Optional[int]   	=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_     :					List[str]   	=output.images

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


"""simple docstring"""






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

import sentencepiece as spm

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


__SCREAMING_SNAKE_CASE							=     logging.get_logger(__name__)

__SCREAMING_SNAKE_CASE							=     {'vocab_file': 'sentencepiece.bpe.model'}

__SCREAMING_SNAKE_CASE							=     {
    'vocab_file': {
        'camembert-base': 'https://huggingface.co/camembert-base/resolve/main/sentencepiece.bpe.model',
    }
}

__SCREAMING_SNAKE_CASE							=     {
    'camembert-base': 512,
}

__SCREAMING_SNAKE_CASE							=     '▁'





class 				a__					(						A__       ):
     UpperCAmelCase__         =      VOCAB_FILES_NAMES
     UpperCAmelCase__         =      PRETRAINED_VOCAB_FILES_MAP
     UpperCAmelCase__         =      PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     UpperCAmelCase__         =      ['''input_ids''', '''attention_mask''']



     def __init__(							self :Optional[Any]      ,  _lowerCamelCase :int      ,  _lowerCamelCase :List[Any]="<s>"      ,  _lowerCamelCase :List[Any]="</s>"      ,  _lowerCamelCase :Optional[int]="</s>"      ,  _lowerCamelCase :List[Any]="<s>"      ,  _lowerCamelCase :List[Any]="<unk>"      ,  _lowerCamelCase :Union[str, Any]="<pad>"      ,  _lowerCamelCase :int="<mask>"      ,  _lowerCamelCase :Union[str, Any]=["<s>NOTUSED", "</s>NOTUSED"]      ,  _lowerCamelCase :Optional[Dict[str, Any]] = None      ,  **_lowerCamelCase :Optional[int]      ,  ):







      '''simple docstring'''



      UpperCamelCase_     :					Optional[Any]   	=AddedToken(_lowerCamelCase      ,  lstrip=_lowerCamelCase      ,  rstrip=_lowerCamelCase   ) if isinstance(_lowerCamelCase      ,  _lowerCamelCase   ) else mask_token

      UpperCamelCase_     :					List[str]   	={} if sp_model_kwargs is None else sp_model_kwargs

      super().__init__(
          bos_token=_lowerCamelCase      ,  eos_token=_lowerCamelCase      ,  unk_token=_lowerCamelCase      ,  sep_token=_lowerCamelCase      ,  cls_token=_lowerCamelCase      ,  pad_token=_lowerCamelCase      ,  mask_token=_lowerCamelCase      ,  additional_special_tokens=_lowerCamelCase      ,  sp_model_kwargs=self.sp_model_kwargs      ,  **_lowerCamelCase      ,  )
      UpperCamelCase_     :					Optional[int]   	=spm.SentencePieceProcessor(**self.sp_model_kwargs   )
      self.sp_model.Load(str(_lowerCamelCase   )   )
      UpperCamelCase_     :					str   	=vocab_file
      # HACK: These tokens were added by fairseq but don't seem to be actually used when duplicated in the actual
      # sentencepiece vocabulary (this is the case for <s> and </s>
      UpperCamelCase_     :					Tuple   	={'<s>NOTUSED': 0, '<pad>': 1, '</s>NOTUSED': 2, '<unk>': 3}
      UpperCamelCase_     :					List[str]   	=len(self.fairseq_tokens_to_ids   )
      UpperCamelCase_     :					List[str]   	=len(self.sp_model   ) + len(self.fairseq_tokens_to_ids   )
      UpperCamelCase_     :					Any   	={v: k for k, v in self.fairseq_tokens_to_ids.items()}



     def  lowerCamelCase_						(							self :Optional[Any]      ,  _lowerCamelCase :List[int]      ,  _lowerCamelCase :Optional[List[int]] = None   ):







      '''simple docstring'''



      if token_ids_a is None:
       return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
      UpperCamelCase_     :					Any   	=[self.cls_token_id]
      UpperCamelCase_     :					str   	=[self.sep_token_id]
      return cls + token_ids_a + sep + sep + token_ids_a + sep



     def  lowerCamelCase_						(							self :Optional[Any]      ,  _lowerCamelCase :List[int]      ,  _lowerCamelCase :Optional[List[int]] = None      ,  _lowerCamelCase :bool = False   ):







      '''simple docstring'''



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

      if token_ids_a is None:
       return [1] + ([0] * len(_lowerCamelCase   )) + [1]
      return [1] + ([0] * len(_lowerCamelCase   )) + [1, 1] + ([0] * len(_lowerCamelCase   )) + [1]



     def  lowerCamelCase_						(							self :Union[str, Any]      ,  _lowerCamelCase :List[int]      ,  _lowerCamelCase :Optional[List[int]] = None   ):







      '''simple docstring'''



      UpperCamelCase_     :					Union[str, Any]   	=[self.sep_token_id]
      UpperCamelCase_     :					Dict   	=[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 + sep + token_ids_a + sep   ) * [0]



     @property
     def  lowerCamelCase_						(							self :str   ):







      '''simple docstring'''



      return len(self.fairseq_tokens_to_ids   ) + len(self.sp_model   )



     def  lowerCamelCase_						(							self :Tuple   ):







      '''simple docstring'''



      UpperCamelCase_     :					Optional[Any]   	={self.convert_ids_to_tokens(_lowerCamelCase   ): i for i in range(self.vocab_size   )}
      vocab.update(self.added_tokens_encoder   )
      return vocab



     def  lowerCamelCase_						(							self :List[str]      ,  _lowerCamelCase :str   ):







      '''simple docstring'''



      return self.sp_model.encode(_lowerCamelCase      ,  out_type=_lowerCamelCase   )



     def  lowerCamelCase_						(							self :str      ,  _lowerCamelCase :Optional[Any]   ):







      '''simple docstring'''



      if token in self.fairseq_tokens_to_ids:
       return self.fairseq_tokens_to_ids[token]
      elif self.sp_model.PieceToId(_lowerCamelCase   ) == 0:
       # Convert sentence piece unk token to fairseq unk token index
       return self.unk_token_id
      return self.fairseq_offset + self.sp_model.PieceToId(_lowerCamelCase   )



     def  lowerCamelCase_						(							self :Tuple      ,  _lowerCamelCase :List[Any]   ):







      '''simple docstring'''



      if index in self.fairseq_ids_to_tokens:
       return self.fairseq_ids_to_tokens[index]
      return self.sp_model.IdToPiece(index - self.fairseq_offset   )



     def  lowerCamelCase_						(							self :int      ,  _lowerCamelCase :List[str]   ):







      '''simple docstring'''



      UpperCamelCase_     :					int   	=[]
      UpperCamelCase_     :					Optional[Any]   	=''
      UpperCamelCase_     :					Optional[Any]   	=False
      for token in tokens:
       # make sure that special tokens are not decoded using sentencepiece model
       if token in self.all_special_tokens:
        if not prev_is_special:
         out_string += " "
        out_string += self.sp_model.decode(_lowerCamelCase   ) + token
        UpperCamelCase_     :					Union[str, Any]   	=True
        UpperCamelCase_     :					Optional[Any]   	=[]
       else:
        current_sub_tokens.append(_lowerCamelCase   )
        UpperCamelCase_     :					Any   	=False
      out_string += self.sp_model.decode(_lowerCamelCase   )
      return out_string.strip()



     def __getstate__(							self :Optional[Any]   ):







      '''simple docstring'''



      UpperCamelCase_     :					List[Any]   	=self.__dict__.copy()
      UpperCamelCase_     :					str   	=None
      return state



     def __setstate__(							self :Optional[Any]      ,  _lowerCamelCase :List[Any]   ):







      '''simple docstring'''



      UpperCamelCase_     :					str   	=d

      # for backward compatibility
      if not hasattr(self      ,  'sp_model_kwargs'   ):
       UpperCamelCase_     :					Any   	={}

      UpperCamelCase_     :					List[Any]   	=spm.SentencePieceProcessor(**self.sp_model_kwargs   )
      self.sp_model.Load(self.vocab_file   )




     def  lowerCamelCase_						(							self :Tuple      ,  _lowerCamelCase :str      ,  _lowerCamelCase :Optional[str] = None   ):







      '''simple docstring'''



      if not os.path.isdir(_lowerCamelCase   ):
       logger.error(f'''Vocabulary path ({save_directory}) should be a directory'''   )
       return
      UpperCamelCase_     :					Any   	=os.path.join(
          _lowerCamelCase      ,  (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']   )

      if os.path.abspath(self.vocab_file   ) != os.path.abspath(_lowerCamelCase   ) and os.path.isfile(self.vocab_file   ):
       copyfile(self.vocab_file      ,  _lowerCamelCase   )
      elif not os.path.isfile(self.vocab_file   ):
       with open(_lowerCamelCase      ,  'wb'   ) as fi:
        UpperCamelCase_     :					int   	=self.sp_model.serialized_model_proto()
        fi.write(_lowerCamelCase   )

      return (out_vocab_file,)
