



import json
import os
import re
import unittest

from transformers import CodeGenTokenizer, CodeGenTokenizerFast
from transformers.models.codegen.tokenization_codegen import VOCAB_FILES_NAMES
from transformers.testing_utils import require_tokenizers, slow

from ...test_tokenization_common import TokenizerTesterMixin

@require_tokenizers
class 							a__( lowerCamelCase__		,			unittest.TestCase     ):
    lowercase__			=   CodeGenTokenizer
    lowercase__			=   CodeGenTokenizerFast
    lowercase__			=   True
    lowercase__			=   {"""add_prefix_space""": True}
    lowercase__			=   False







    def       lowercase_	(				self			:  Optional[int]     ):
      super().setUp()

      # Adapted from Sennrich et al. 2015 and https://github.com/rsennrich/subword-nmt
      a						: Optional[Any]					       =					[
          'l',
          'o',
          'w',
          'e',
          'r',
          's',
          't',
          'i',
          'd',
          'n',
          '\u0120',
          '\u0120l',
          '\u0120n',
          '\u0120lo',
          '\u0120low',
          'er',
          '\u0120lowest',
          '\u0120newer',
          '\u0120wider',
          '<unk>',
          '<|endoftext|>',
      ]
      a						: Any					       =					dict(zip(__snake_case						,      range(len(__snake_case     )     )     )     )
      a						: int					       =					['#version: 0.2', '\u0120 l', '\u0120l o', '\u0120lo w', 'e r', '']
      a						: Tuple					       =					{'unk_token': '<unk>'}

      a						: Any					       =					os.path.join(self.tmpdirname						,      VOCAB_FILES_NAMES['vocab_file']     )
      a						: List[Any]					       =					os.path.join(self.tmpdirname						,      VOCAB_FILES_NAMES['merges_file']     )
      with open(self.vocab_file						,      'w'						,      encoding='utf-8'     ) as fp:
        fp.write(json.dumps(__snake_case     ) + '\n'     )
      with open(self.merges_file						,      'w'						,      encoding='utf-8'     ) as fp:
        fp.write('\n'.join(__snake_case     )     )







    def       lowercase_	(				self			:  str						,      **__snake_case			:  str     ):
      kwargs.update(self.special_tokens_map     )
      return CodeGenTokenizer.from_pretrained(self.tmpdirname						,      **__snake_case     )







    def       lowercase_	(				self			:  List[Any]						,      **__snake_case			:  str     ):
      kwargs.update(self.special_tokens_map     )
      return CodeGenTokenizerFast.from_pretrained(self.tmpdirname						,      **__snake_case     )







    def       lowercase_	(				self			:  Dict						,      __snake_case			:  List[str]     ):
      a						: Any					       =					'lower newer'
      a						: Any					       =					'lower newer'
      return input_text, output_text







    def       lowercase_	(				self			:  Tuple     ):
      a						: Optional[int]					       =					CodeGenTokenizer(self.vocab_file						,      self.merges_file						,      **self.special_tokens_map     )
      a						: str					       =					'lower newer'
      a						: Optional[int]					       =					['\u0120low', 'er', '\u0120', 'n', 'e', 'w', 'er']
      a						: str					       =					tokenizer.tokenize(__snake_case						,      add_prefix_space=__snake_case     )
      self.assertListEqual(__snake_case						,      __snake_case     )

      a						: Optional[int]					       =					tokens + [tokenizer.unk_token]
      a						: Union[str, Any]					       =					[14, 15, 10, 9, 3, 2, 15, 19]
      self.assertListEqual(tokenizer.convert_tokens_to_ids(__snake_case     )						,      __snake_case     )







    def       lowercase_	(				self			:  Dict     ):
      if not self.test_rust_tokenizer:
        return

      a						: int					       =					self.get_tokenizer()
      a						: Optional[int]					       =					self.get_rust_tokenizer(add_prefix_space=__snake_case     )

      a						: List[Any]					       =					'lower newer'

      # Testing tokenization
      a						: List[Any]					       =					tokenizer.tokenize(__snake_case						,      add_prefix_space=__snake_case     )
      a						: Union[str, Any]					       =					rust_tokenizer.tokenize(__snake_case     )
      self.assertListEqual(__snake_case						,      __snake_case     )

      # Testing conversion to ids without special tokens
      a						: List[str]					       =					tokenizer.encode(__snake_case						,      add_special_tokens=__snake_case						,      add_prefix_space=__snake_case     )
      a						: Dict					       =					rust_tokenizer.encode(__snake_case						,      add_special_tokens=__snake_case     )
      self.assertListEqual(__snake_case						,      __snake_case     )

      # Testing conversion to ids with special tokens
      a						: Optional[Any]					       =					self.get_rust_tokenizer(add_prefix_space=__snake_case     )
      a						: str					       =					tokenizer.encode(__snake_case						,      add_prefix_space=__snake_case     )
      a						: int					       =					rust_tokenizer.encode(__snake_case     )
      self.assertListEqual(__snake_case						,      __snake_case     )

      # Testing the unknown token
      a						: Optional[int]					       =					tokens + [rust_tokenizer.unk_token]
      a						: int					       =					[14, 15, 10, 9, 3, 2, 15, 19]
      self.assertListEqual(rust_tokenizer.convert_tokens_to_ids(__snake_case     )						,      __snake_case     )







    def       lowercase_	(				self			:  int						,      *__snake_case			:  Dict						,      **__snake_case			:  List[str]     ):
      # It's very difficult to mix/test pretokenization with byte-level
      # And get both CodeGen and Roberta to work at the same time (mostly an issue of adding a space before the string)
      pass







    def       lowercase_	(				self			:  Optional[Any]						,      __snake_case			:  Union[str, Any]=15     ):
      for tokenizer, pretrained_name, kwargs in self.tokenizers_list:
        with self.subTest(F"""{tokenizer.__class__.__name__} ({pretrained_name})"""     ):
          a						: str					       =					self.rust_tokenizer_class.from_pretrained(__snake_case						,      **__snake_case     )

          # Simple input
          a						: int					       =					'This is a simple input'
          a						: Dict					       =					['This is a simple input 1', 'This is a simple input 2']
          a						: Optional[Any]					       =					('This is a simple input', 'This is a pair')
          a						: List[Any]					       =					[
              ('This is a simple input 1', 'This is a simple input 2'),
              ('This is a simple pair 1', 'This is a simple pair 2'),
          ]

          # Simple input tests
          self.assertRaises(__snake_case						,      tokenizer_r.encode						,      __snake_case						,      max_length=__snake_case						,      padding='max_length'     )

          # Simple input
          self.assertRaises(__snake_case						,      tokenizer_r.encode_plus						,      __snake_case						,      max_length=__snake_case						,      padding='max_length'     )

          # Simple input
          self.assertRaises(
              __snake_case						,      tokenizer_r.batch_encode_plus						,      __snake_case						,      max_length=__snake_case						,      padding='max_length'						,      )

          # Pair input
          self.assertRaises(__snake_case						,      tokenizer_r.encode						,      __snake_case						,      max_length=__snake_case						,      padding='max_length'     )

          # Pair input
          self.assertRaises(__snake_case						,      tokenizer_r.encode_plus						,      __snake_case						,      max_length=__snake_case						,      padding='max_length'     )

          # Pair input
          self.assertRaises(
              __snake_case						,      tokenizer_r.batch_encode_plus						,      __snake_case						,      max_length=__snake_case						,      padding='max_length'						,      )







    def       lowercase_	(				self			:  int     ):
      a						: List[Any]					       =					CodeGenTokenizer.from_pretrained(self.tmpdirname						,      pad_token='<pad>'     )

      # Simple input
      a						: Dict					       =					'This is a simple input'
      a						: Optional[Any]					       =					['This is a simple input looooooooong', 'This is a simple input']
      a						: Dict					       =					('This is a simple input', 'This is a pair')
      a						: Optional[Any]					       =					[
          ('This is a simple input loooooong', 'This is a simple input'),
          ('This is a simple pair loooooong', 'This is a simple pair'),
      ]

      a						: Union[str, Any]					       =					tokenizer.pad_token_id

      a						: Dict					       =					tokenizer(__snake_case						,      padding='max_length'						,      max_length=30						,      return_tensors='np'     )
      a						: int					       =					tokenizer(__snake_case						,      padding=__snake_case						,      truncate=__snake_case						,      return_tensors='np'     )
      a						: Dict					       =					tokenizer(*__snake_case						,      padding='max_length'						,      max_length=60						,      return_tensors='np'     )
      a						: List[Any]					       =					tokenizer(__snake_case						,      padding=__snake_case						,      truncate=__snake_case						,      return_tensors='np'     )

      # s
      # test single string max_length padding
      self.assertEqual(out_s['input_ids'].shape[-1]						,      30     )
      self.assertTrue(pad_token_id in out_s['input_ids']     )
      self.assertTrue(0 in out_s['attention_mask']     )

      # s2
      # test automatic padding
      self.assertEqual(out_sa['input_ids'].shape[-1]						,      33     )
      # long slice doesn't have padding
      self.assertFalse(pad_token_id in out_sa['input_ids'][0]     )
      self.assertFalse(0 in out_sa['attention_mask'][0]     )
      # short slice does have padding
      self.assertTrue(pad_token_id in out_sa['input_ids'][1]     )
      self.assertTrue(0 in out_sa['attention_mask'][1]     )

      # p
      # test single pair max_length padding
      self.assertEqual(out_p['input_ids'].shape[-1]						,      60     )
      self.assertTrue(pad_token_id in out_p['input_ids']     )
      self.assertTrue(0 in out_p['attention_mask']     )

      # p2
      # test automatic padding pair
      self.assertEqual(out_pa['input_ids'].shape[-1]						,      52     )
      # long slice pair doesn't have padding
      self.assertFalse(pad_token_id in out_pa['input_ids'][0]     )
      self.assertFalse(0 in out_pa['attention_mask'][0]     )
      # short slice pair does have padding
      self.assertTrue(pad_token_id in out_pa['input_ids'][1]     )
      self.assertTrue(0 in out_pa['attention_mask'][1]     )







    def       lowercase_	(				self			:  int     ):
      a						: int					       =					'$$$'
      a						: List[str]					       =					CodeGenTokenizer.from_pretrained(self.tmpdirname						,      bos_token=__snake_case						,      add_bos_token=__snake_case     )

      a						: Optional[int]					       =					'This is a simple input'
      a						: Optional[Any]					       =					['This is a simple input 1', 'This is a simple input 2']

      a						: Tuple					       =					tokenizer.bos_token_id

      a						: int					       =					tokenizer(__snake_case     )
      a						: Any					       =					tokenizer(__snake_case     )

      self.assertEqual(out_s.input_ids[0]						,      __snake_case     )
      self.assertTrue(all(o[0] == bos_token_id for o in out_sa.input_ids     )     )

      a						: Optional[Any]					       =					tokenizer.decode(out_s.input_ids     )
      a						: str					       =					tokenizer.batch_decode(out_sa.input_ids     )

      self.assertEqual(decode_s.split()[0]						,      __snake_case     )
      self.assertTrue(all(d.split()[0] == bos_token for d in decode_sa     )     )







    @slow
    def       lowercase_	(				self			:  Union[str, Any]     ):
      a						: Optional[Any]					       =					CodeGenTokenizer.from_pretrained('Salesforce/codegen-350M-mono'     )

      a						: str					       =					'\nif len_a > len_b:\n    result = a\nelse:\n    result = b\n\n\n\n#'
      a						: Optional[int]					       =					'\nif len_a > len_b:      result = a\nelse:      result = b'

      a						: Dict					       =					tokenizer.encode(__snake_case     )
      a						: Optional[int]					       =					['^#', re.escape('<|endoftext|>'     ), '^\'\'\'', '^"""', '\n\n\n']
      a						: Any					       =					tokenizer.decode(__snake_case						,      truncate_before_pattern=__snake_case     )
      self.assertEqual(__snake_case						,      __snake_case     )







    def       lowercase_	(				self			:  List[str]     ):
      pass




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







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

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







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

      a						: int					       =					self.get_dummy_components()
      a						: str					       =					CycleDiffusionPipeline(**__snake_case     )
      a						: List[str]					       =					pipe.to(__snake_case     )
      pipe.set_progress_bar_config(disable=__snake_case     )

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

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

      assert images.shape == (1, 32, 32, 3)
      a						: Tuple					       =					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       lowercase_	(				self			:  int     ):
      a						: List[Any]					       =					self.get_dummy_components()
      for name, module in components.items():
        if hasattr(__snake_case						,      'half'     ):
          a						: Any					       =					module.half()
      a						: Tuple					       =					CycleDiffusionPipeline(**__snake_case     )
      a						: Any					       =					pipe.to(__snake_case     )
      pipe.set_progress_bar_config(disable=__snake_case     )

      a						: str					       =					self.get_dummy_inputs(__snake_case     )
      a						: int					       =					pipe(**__snake_case     )
      a						: Optional[int]					       =					output.images

      a						: Tuple					       =					images[0, -3:, -3:, -1]

      assert images.shape == (1, 32, 32, 3)
      a						: 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       lowercase_	(				self			:  List[Any]     ):
      return super().test_save_load_local()







    @unittest.skip('non-deterministic pipeline'     )
    def       lowercase_	(				self			:  Dict     ):
      return super().test_inference_batch_single_identical()







    @skip_mps
    def       lowercase_	(				self			:  int     ):
      return super().test_dict_tuple_outputs_equivalent()







    @skip_mps
    def       lowercase_	(				self			:  Dict     ):
      return super().test_save_load_optional_components()







    @skip_mps
    def       lowercase_	(				self			:  List[Any]     ):
      return super().test_attention_slicing_forward_pass()

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







    def       lowercase_	(				self			:  Tuple     ):
      # clean up the VRAM after each test
      super().tearDown()
      gc.collect()
      torch.cuda.empty_cache()







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

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

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

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

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

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

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

      a						: Tuple					       =					'A black colored car'
      a						: Tuple					       =					'A blue colored car'

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

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