



import concurrent.futures
import json
import os
import shutil
import tempfile
import unittest

from transformers import AutoTokenizer, PreTrainedTokenizerFast
from transformers.testing_utils import require_tokenizers

from ..test_tokenization_common import TokenizerTesterMixin







@require_tokenizers
class 						snake_case  (		SCREAMING_SNAKE_CASE_   ,unittest.TestCase					):



   '''simple docstring'''






   snake_case_   :	List[Any]       =						PreTrainedTokenizerFast
   snake_case_   :	str       =						False
   snake_case_   :	Tuple       =						True
   snake_case_   :	Optional[Any]       =						"""tokenizer_file"""

   def        UpperCamelCase_					(   self							:     Optional[Any])			->			Optional[Any]:

    """simple docstring"""







    _snake_case      :					Optional[Any]					=   False  # because we don't have pretrained_vocab_files_map
    super().setUp()
    _snake_case      :					Any					=   True

    _snake_case      :					Union[str, Any]					=   ["""robot-test/dummy-tokenizer-fast""", """robot-test/dummy-tokenizer-wordlevel"""]
    _snake_case      :					List[Any]					=   """SaulLu/dummy-tokenizer-bytelevel-bpe"""

    # Inclusion of 2 tokenizers to test different types of models (Unigram and WordLevel for the moment)
    _snake_case      :					str					=   [(PreTrainedTokenizerFast, model_path, {}) for model_path in model_paths]

    _snake_case      :					List[str]					=   PreTrainedTokenizerFast.from_pretrained(model_paths[0])
    tokenizer.save_pretrained(self.tmpdirname)

   def        UpperCamelCase_					(   self							:     Optional[Any])			->			Optional[Any]:

    """simple docstring"""







    pass

   def        UpperCamelCase_					(   self							:     Optional[int])			->			Any:

    """simple docstring"""







    pass

   def        UpperCamelCase_					(   self							:     str)			->			Dict:

    """simple docstring"""







    pass

   def        UpperCamelCase_					(   self							:     Optional[Any])			->			Any:

    """simple docstring"""







    pass

   def        UpperCamelCase_					(   self							:     Dict)			->			Optional[Any]:

    """simple docstring"""







    _snake_case      :					List[Any]					=   self.tmpdirname
    # Here we want to test the 2 available tokenizers that use 2 different types of models: Unigram and WordLevel.
    for tokenizer, pretrained_name, kwargs in self.tokenizers_list:
     with self.subTest(F'''{tokenizer.__class__.__name__} ({pretrained_name})'''):
      try:
       _snake_case      :					List[str]					=   tempfile.mkdtemp()
       _snake_case      :					Optional[int]					=   self.rust_tokenizer_class.from_pretrained(lowerCAmelCase			,						**lowerCAmelCase)

       tokenizer.save_pretrained(self.tmpdirname)
       super().test_training_new_tokenizer()
      finally:
       # Even if the test fails, we must be sure that the folder is deleted and that the default tokenizer
       # is restored
       shutil.rmtree(self.tmpdirname)
       _snake_case      :					Dict					=   tmpdirname_orig

   def        UpperCamelCase_					(   self							:     List[str])			->			int:

    """simple docstring"""







    _snake_case      :					Optional[Any]					=   self.tmpdirname
    # Here we want to test the 2 available tokenizers that use 2 different types of models: Unigram and WordLevel.
    for tokenizer, pretrained_name, kwargs in self.tokenizers_list:
     with self.subTest(F'''{tokenizer.__class__.__name__} ({pretrained_name})'''):
      try:
       _snake_case      :					List[Any]					=   tempfile.mkdtemp()
       _snake_case      :					List[str]					=   self.rust_tokenizer_class.from_pretrained(lowerCAmelCase			,						**lowerCAmelCase)

       tokenizer.save_pretrained(self.tmpdirname)
       super().test_training_new_tokenizer_with_special_tokens_change()
      finally:
       # Even if the test fails, we must be sure that the folder is deleted and that the default tokenizer
       # is restored
       shutil.rmtree(self.tmpdirname)
       _snake_case      :					Optional[Any]					=   tmpdirname_orig


   def        UpperCamelCase_					(   self							:     Union[str, Any])			->			Tuple:

    """simple docstring"""







    _snake_case      :					List[Any]					=   self.rust_tokenizer_class.from_pretrained(self.bytelevel_bpe_model_name)

    _snake_case      :					List[Any]					=   ("""a""" for _ in range(1000))
    _snake_case      :					Any					=   tokenizer.train_new_from_iterator(text_iterator=lowerCAmelCase			,						length=1000			,						vocab_size=50)

    _snake_case      :					str					=   new_tokenizer.encode("""a🤗""")
    self.assertEqual(lowerCAmelCase			,						[64, 172, 253, 97, 245])







@require_tokenizers
class 						snake_case  (		unittest.TestCase					):



   '''simple docstring'''






   def        UpperCamelCase_					(   self							:     int)			->			Optional[int]:

    """simple docstring"""







    _snake_case      :					Optional[Any]					=   AutoTokenizer.from_pretrained("""bert-base-cased""")
    _snake_case      :					Optional[int]					=   json.loads(tokenizer._tokenizer.to_str())
    _snake_case      :					Optional[Any]					=   len(lowerCAmelCase)

    with tempfile.TemporaryDirectory() as tmp_dir:
     # Hack to save this in the tokenizer_config.json
     _snake_case      :					str					=   ["""tokenizer.4.0.0.json"""]
     tokenizer.save_pretrained(lowerCAmelCase)
     json.dump(lowerCAmelCase			,						open(os.path.join(lowerCAmelCase			,						"""tokenizer.4.0.0.json""")			,						"""w"""))

     # This should pick the new tokenizer file as the version of Transformers is > 4.0.0
     _snake_case      :					str					=   AutoTokenizer.from_pretrained(lowerCAmelCase)
     self.assertEqual(len(lowerCAmelCase)			,						len(lowerCAmelCase) + 1)
     _snake_case      :					Optional[int]					=   json.loads(new_tokenizer._tokenizer.to_str())
     self.assertIn("""huggingface"""			,						json_tokenizer["""model"""]["""vocab"""])

     # Will need to be adjusted if we reach v42 and this test is still here.
     # Should pick the old tokenizer file as the version of Transformers is < 4.0.0
     shutil.move(os.path.join(lowerCAmelCase			,						"""tokenizer.4.0.0.json""")			,						os.path.join(lowerCAmelCase			,						"""tokenizer.42.0.0.json"""))
     _snake_case      :					Tuple					=   ["""tokenizer.42.0.0.json"""]
     tokenizer.save_pretrained(lowerCAmelCase)
     _snake_case      :					Union[str, Any]					=   AutoTokenizer.from_pretrained(lowerCAmelCase)
     self.assertEqual(len(lowerCAmelCase)			,						len(lowerCAmelCase))
     _snake_case      :					Dict					=   json.loads(new_tokenizer._tokenizer.to_str())
     self.assertNotIn("""huggingface"""			,						json_tokenizer["""model"""]["""vocab"""])


   def        UpperCamelCase_					(   self							:     Dict)			->			int:

    """simple docstring"""







    _snake_case      :					List[Any]					=   """hf-internal-testing/test-two-tokenizers"""

    # This should pick the new tokenizer file as the version of Transformers is > 4.0.0
    _snake_case      :					List[str]					=   AutoTokenizer.from_pretrained(lowerCAmelCase)
    self.assertEqual(len(lowerCAmelCase)			,						2_8997)
    _snake_case      :					List[Any]					=   json.loads(tokenizer._tokenizer.to_str())
    self.assertIn("""huggingface"""			,						json_tokenizer["""model"""]["""vocab"""])

    # Testing an older version by monkey-patching the version in the module it's used.
    import transformers as old_transformers

    _snake_case      :					Any					=   """3.0.0"""
    _snake_case      :					Dict					=   old_transformers.models.auto.AutoTokenizer.from_pretrained(lowerCAmelCase)
    self.assertEqual(len(lowerCAmelCase)			,						2_8996)
    _snake_case      :					Any					=   json.loads(old_tokenizer._tokenizer.to_str())
    self.assertNotIn("""huggingface"""			,						json_tokenizer["""model"""]["""vocab"""])







@require_tokenizers
class 						snake_case  (		unittest.TestCase					):



   '''simple docstring'''






   def        UpperCamelCase_					(   self							:     Optional[Any])			->			int:

    """simple docstring"""







    _snake_case      :					str					=   PreTrainedTokenizerFast.from_pretrained("""robot-test/dummy-tokenizer-wordlevel""")
    _snake_case      :					Optional[int]					=   """The Matrix is a 1999 science fiction action film."""

    with concurrent.futures.ThreadPoolExecutor() as executor:
     _snake_case      :					Dict					=   [executor.submit(self.fetch			,						lowerCAmelCase			,						lowerCAmelCase) for i in range(10)]
     _snake_case      :					Union[str, Any]					=   [future.result() for future in futures]
     self.assertEqual(lowerCAmelCase			,						[[1, 10, 0, 8, 0, 18, 0, 0, 0, 2] for i in range(10)])


   def        UpperCamelCase_					(   self							:     Optional[Any]			,						lowerCAmelCase							:     List[Any]			,						lowerCAmelCase							:     Any)			->			Tuple:

    """simple docstring"""







    return tokenizer.encode(lowerCAmelCase			,						truncation="""longest_first"""			,						padding="""longest""")





import unittest

import numpy as np
import torch

from diffusers import DDPMPipeline, DDPMScheduler, UNetaDModel
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, slow, torch_device


enable_full_determinism()







class 						snake_case  (		unittest.TestCase					):



   '''simple docstring'''






   @property
   def        UpperCamelCase_					(   self							:     Union[str, Any])			->			int:

    """simple docstring"""







    torch.manual_seed(0)
    _snake_case      :					Any					=   UNetaDModel(
        block_out_channels=(32, 64)			,						layers_per_block=2			,						sample_size=32			,						in_channels=3			,						out_channels=3			,						down_block_types=("""DownBlock2D""", """AttnDownBlock2D""")			,						up_block_types=("""AttnUpBlock2D""", """UpBlock2D""")			,						)
    return model

   def        UpperCamelCase_					(   self							:     List[Any])			->			Optional[Any]:

    """simple docstring"""







    _snake_case      :					List[str]					=   """cpu"""
    _snake_case      :					Union[str, Any]					=   self.dummy_uncond_unet
    _snake_case      :					Union[str, Any]					=   DDPMScheduler()

    _snake_case      :					Dict					=   DDPMPipeline(unet=lowerCAmelCase			,						scheduler=lowerCAmelCase)
    ddpm.to(lowerCAmelCase)
    ddpm.set_progress_bar_config(disable=lowerCAmelCase)

    _snake_case      :					Optional[Any]					=   torch.Generator(device=lowerCAmelCase).manual_seed(0)
    _snake_case      :					Optional[int]					=   ddpm(generator=lowerCAmelCase			,						num_inference_steps=2			,						output_type="""numpy""").images

    _snake_case      :					Optional[int]					=   torch.Generator(device=lowerCAmelCase).manual_seed(0)
    _snake_case      :					Union[str, Any]					=   ddpm(generator=lowerCAmelCase			,						num_inference_steps=2			,						output_type="""numpy"""			,						return_dict=lowerCAmelCase)[0]

    _snake_case      :					Any					=   image[0, -3:, -3:, -1]
    _snake_case      :					str					=   image_from_tuple[0, -3:, -3:, -1]

    assert image.shape == (1, 32, 32, 3)
    _snake_case      :					Optional[int]					=   np.array(
        [9.956E-01, 5.785E-01, 4.675E-01, 9.930E-01, 0.0, 1.000, 1.199E-03, 2.648E-04, 5.101E-04])

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


   def        UpperCamelCase_					(   self							:     List[Any])			->			Optional[int]:

    """simple docstring"""







    _snake_case      :					int					=   self.dummy_uncond_unet
    _snake_case      :					int					=   DDPMScheduler(prediction_type="""sample""")

    _snake_case      :					Union[str, Any]					=   DDPMPipeline(unet=lowerCAmelCase			,						scheduler=lowerCAmelCase)
    ddpm.to(lowerCAmelCase)
    ddpm.set_progress_bar_config(disable=lowerCAmelCase)

    _snake_case      :					Tuple					=   torch.manual_seed(0)
    _snake_case      :					Union[str, Any]					=   ddpm(generator=lowerCAmelCase			,						num_inference_steps=2			,						output_type="""numpy""").images

    _snake_case      :					Any					=   torch.manual_seed(0)
    _snake_case      :					Dict					=   ddpm(generator=lowerCAmelCase			,						num_inference_steps=2			,						output_type="""numpy""")[0]

    _snake_case      :					str					=   image[0, -3:, -3:, -1]
    _snake_case      :					List[Any]					=   image_eps[0, -3:, -3:, -1]

    assert image.shape == (1, 32, 32, 3)
    _snake_case      :					Any					=   1E-2 if torch_device != """mps""" else 3E-2
    assert np.abs(image_slice.flatten() - image_eps_slice.flatten()).max() < tolerance







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



   '''simple docstring'''






   def        UpperCamelCase_					(   self							:     Dict)			->			Optional[Any]:

    """simple docstring"""







    _snake_case      :					Dict					=   """google/ddpm-cifar10-32"""

    _snake_case      :					List[Any]					=   UNetaDModel.from_pretrained(lowerCAmelCase)
    _snake_case      :					Optional[Any]					=   DDPMScheduler.from_pretrained(lowerCAmelCase)

    _snake_case      :					Dict					=   DDPMPipeline(unet=lowerCAmelCase			,						scheduler=lowerCAmelCase)
    ddpm.to(lowerCAmelCase)
    ddpm.set_progress_bar_config(disable=lowerCAmelCase)

    _snake_case      :					Optional[Any]					=   torch.manual_seed(0)
    _snake_case      :					Any					=   ddpm(generator=lowerCAmelCase			,						output_type="""numpy""").images

    _snake_case      :					Optional[Any]					=   image[0, -3:, -3:, -1]

    assert image.shape == (1, 32, 32, 3)
    _snake_case      :					Union[str, Any]					=   np.array([0.4_200, 0.3_588, 0.1_939, 0.3_847, 0.3_382, 0.2_647, 0.4_155, 0.3_582, 0.3_385])
    assert np.abs(image_slice.flatten() - expected_slice).max() < 1E-2
