

'''simple docstring'''






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 	SCREAMING_SNAKE_CASE__			(				_UpperCAmelCase,unittest.TestCase      ):
    _A			    =							PreTrainedTokenizerFast
    _A			    =							False
    _A			    =							True
    _A			    =							"tokenizer_file"




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






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

          SCREAMING_SNAKE_CASE_      :			Any     =       ["robot-test/dummy-tokenizer-fast", "robot-test/dummy-tokenizer-wordlevel"]
          SCREAMING_SNAKE_CASE_      :			Any     =       "SaulLu/dummy-tokenizer-bytelevel-bpe"

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

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




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          pass




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          pass




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          pass




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          pass




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          SCREAMING_SNAKE_CASE_      :			str     =       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:
                            SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       tempfile.mkdtemp()
                            SCREAMING_SNAKE_CASE_      :			Optional[int]     =       self.rust_tokenizer_class.from_pretrained(lowercase__				,     **lowercase__			)

                            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			)
                            SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       tmpdirname_orig




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          SCREAMING_SNAKE_CASE_      :			Tuple     =       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:
                            SCREAMING_SNAKE_CASE_      :			Optional[Any]     =       tempfile.mkdtemp()
                            SCREAMING_SNAKE_CASE_      :			Optional[int]     =       self.rust_tokenizer_class.from_pretrained(lowercase__				,     **lowercase__			)

                            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			)
                            SCREAMING_SNAKE_CASE_      :			str     =       tmpdirname_orig







    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          SCREAMING_SNAKE_CASE_      :			Optional[int]     =       self.rust_tokenizer_class.from_pretrained(self.bytelevel_bpe_model_name			)

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

          SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       new_tokenizer.encode("a🤗"			)
          self.assertEqual(lowercase__				,     [64, 172, 253, 97, 245]			)






@require_tokenizers
class 	SCREAMING_SNAKE_CASE__			(				unittest.TestCase      ):




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          SCREAMING_SNAKE_CASE_      :			Optional[Any]     =       AutoTokenizer.from_pretrained("bert-base-cased"			)
          SCREAMING_SNAKE_CASE_      :			List[str]     =       json.loads(tokenizer._tokenizer.to_str()			)
          SCREAMING_SNAKE_CASE_      :			List[Any]     =       len(lowercase__			)

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

                # This should pick the new tokenizer file as the version of Transformers is > 4.0.0
                SCREAMING_SNAKE_CASE_      :			Dict     =       AutoTokenizer.from_pretrained(lowercase__			)
                self.assertEqual(len(lowercase__			)				,     len(lowercase__			) + 1			)
                SCREAMING_SNAKE_CASE_      :			Tuple     =       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(lowercase__				,     "tokenizer.4.0.0.json"			)				,     os.path.join(lowercase__				,     "tokenizer.42.0.0.json"			)			)
                SCREAMING_SNAKE_CASE_      :			str     =       ["tokenizer.42.0.0.json"]
                tokenizer.save_pretrained(lowercase__			)
                SCREAMING_SNAKE_CASE_      :			Optional[int]     =       AutoTokenizer.from_pretrained(lowercase__			)
                self.assertEqual(len(lowercase__			)				,     len(lowercase__			)			)
                SCREAMING_SNAKE_CASE_      :			Dict     =       json.loads(new_tokenizer._tokenizer.to_str()			)
                self.assertNotIn("huggingface"				,     json_tokenizer["model"]["vocab"]			)







    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          SCREAMING_SNAKE_CASE_      :			Optional[Any]     =       "hf-internal-testing/test-two-tokenizers"

          # This should pick the new tokenizer file as the version of Transformers is > 4.0.0
          SCREAMING_SNAKE_CASE_      :			Optional[Any]     =       AutoTokenizer.from_pretrained(lowercase__			)
          self.assertEqual(len(lowercase__			)				,     2_8997			)
          SCREAMING_SNAKE_CASE_      :			Tuple     =       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

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






@require_tokenizers
class 	SCREAMING_SNAKE_CASE__			(				unittest.TestCase      ):




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       PreTrainedTokenizerFast.from_pretrained("robot-test/dummy-tokenizer-wordlevel"			)
          SCREAMING_SNAKE_CASE_      :			List[str]     =       "The Matrix is a 1999 science fiction action film."

          with concurrent.futures.ThreadPoolExecutor() as executor:
                SCREAMING_SNAKE_CASE_      :			Tuple     =       [executor.submit(self.fetch				,     lowercase__				,     lowercase__			) for i in range(10			)]
                SCREAMING_SNAKE_CASE_      :			Optional[Any]     =       [future.result() for future in futures]
                self.assertEqual(lowercase__				,     [[1, 10, 0, 8, 0, 18, 0, 0, 0, 2] for i in range(10			)]			)







    def   __lowerCamelCase      (					self				,     lowercase__				,     lowercase__			):






          """simple docstring"""






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




'''simple docstring'''






import torch

from diffusers import HeunDiscreteScheduler
from diffusers.utils import torch_device

from .test_schedulers import SchedulerCommonTest
class 	SCREAMING_SNAKE_CASE__			(				_UpperCAmelCase      ):
    _A			    =							(HeunDiscreteScheduler,)
    _A			    =							10




    def   __lowerCamelCase      (					self				,     **lowercase__			):






          """simple docstring"""






          SCREAMING_SNAKE_CASE_      :			List[Any]     =       {
              "num_train_timesteps": 1100,
              "beta_start": 0.0001,
              "beta_end": 0.02,
              "beta_schedule": "linear",
          }

          config.update(**lowercase__			)
          return config




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          for timesteps in [10, 50, 100, 1000]:
                self.check_over_configs(num_train_timesteps=lowercase__			)




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          for beta_start, beta_end in zip([0.00001, 0.0001, 0.001]				,     [0.0002, 0.002, 0.02]			):
                self.check_over_configs(beta_start=lowercase__				,     beta_end=lowercase__			)




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          for schedule in ["linear", "scaled_linear", "exp"]:
                self.check_over_configs(beta_schedule=lowercase__			)




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          for clip_sample_range in [1.0, 2.0, 3.0]:
                self.check_over_configs(clip_sample_range=lowercase__				,     clip_sample=lowercase__			)




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          for prediction_type in ["epsilon", "v_prediction", "sample"]:
                self.check_over_configs(prediction_type=lowercase__			)




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          SCREAMING_SNAKE_CASE_      :			Tuple     =       self.scheduler_classes[0]
          SCREAMING_SNAKE_CASE_      :			List[str]     =       self.get_scheduler_config()
          SCREAMING_SNAKE_CASE_      :			List[str]     =       scheduler_class(**lowercase__			)

          scheduler.set_timesteps(self.num_inference_steps			)

          SCREAMING_SNAKE_CASE_      :			int     =       self.dummy_model()
          SCREAMING_SNAKE_CASE_      :			Dict     =       self.dummy_sample_deter * scheduler.init_noise_sigma
          SCREAMING_SNAKE_CASE_      :			Optional[int]     =       sample.to(lowercase__			)

          for i, t in enumerate(scheduler.timesteps			):
                SCREAMING_SNAKE_CASE_      :			Optional[Any]     =       scheduler.scale_model_input(lowercase__				,     lowercase__			)

                SCREAMING_SNAKE_CASE_      :			Dict     =       model(lowercase__				,     lowercase__			)

                SCREAMING_SNAKE_CASE_      :			str     =       scheduler.step(lowercase__				,     lowercase__				,     lowercase__			)
                SCREAMING_SNAKE_CASE_      :			Dict     =       output.prev_sample

          SCREAMING_SNAKE_CASE_      :			List[Any]     =       torch.sum(torch.abs(lowercase__			)			)
          SCREAMING_SNAKE_CASE_      :			List[str]     =       torch.mean(torch.abs(lowercase__			)			)

          if torch_device in ["cpu", "mps"]:
                assert abs(result_sum.item() - 0.1233			) < 1e-2
                assert abs(result_mean.item() - 0.0002			) < 1e-3
          else:
                # CUDA
                assert abs(result_sum.item() - 0.1233			) < 1e-2
                assert abs(result_mean.item() - 0.0002			) < 1e-3




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          SCREAMING_SNAKE_CASE_      :			Optional[Any]     =       self.scheduler_classes[0]
          SCREAMING_SNAKE_CASE_      :			Optional[int]     =       self.get_scheduler_config(prediction_type="v_prediction"			)
          SCREAMING_SNAKE_CASE_      :			Any     =       scheduler_class(**lowercase__			)

          scheduler.set_timesteps(self.num_inference_steps			)

          SCREAMING_SNAKE_CASE_      :			Any     =       self.dummy_model()
          SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       self.dummy_sample_deter * scheduler.init_noise_sigma
          SCREAMING_SNAKE_CASE_      :			List[str]     =       sample.to(lowercase__			)

          for i, t in enumerate(scheduler.timesteps			):
                SCREAMING_SNAKE_CASE_      :			Tuple     =       scheduler.scale_model_input(lowercase__				,     lowercase__			)

                SCREAMING_SNAKE_CASE_      :			int     =       model(lowercase__				,     lowercase__			)

                SCREAMING_SNAKE_CASE_      :			Optional[Any]     =       scheduler.step(lowercase__				,     lowercase__				,     lowercase__			)
                SCREAMING_SNAKE_CASE_      :			Optional[int]     =       output.prev_sample

          SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       torch.sum(torch.abs(lowercase__			)			)
          SCREAMING_SNAKE_CASE_      :			int     =       torch.mean(torch.abs(lowercase__			)			)

          if torch_device in ["cpu", "mps"]:
                assert abs(result_sum.item() - 4.69_34e-07			) < 1e-2
                assert abs(result_mean.item() - 6.11_12e-10			) < 1e-3
          else:
                # CUDA
                assert abs(result_sum.item() - 4.6_93_42_86_50_17_09_72e-07			) < 1e-2
                assert abs(result_mean.item() - 0.0002			) < 1e-3




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          SCREAMING_SNAKE_CASE_      :			List[str]     =       self.scheduler_classes[0]
          SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       self.get_scheduler_config()
          SCREAMING_SNAKE_CASE_      :			List[Any]     =       scheduler_class(**lowercase__			)

          scheduler.set_timesteps(self.num_inference_steps				,     device=lowercase__			)

          SCREAMING_SNAKE_CASE_      :			Any     =       self.dummy_model()
          SCREAMING_SNAKE_CASE_      :			str     =       self.dummy_sample_deter.to(lowercase__			) * scheduler.init_noise_sigma

          for t in scheduler.timesteps:
                SCREAMING_SNAKE_CASE_      :			Optional[int]     =       scheduler.scale_model_input(lowercase__				,     lowercase__			)

                SCREAMING_SNAKE_CASE_      :			Dict     =       model(lowercase__				,     lowercase__			)

                SCREAMING_SNAKE_CASE_      :			Optional[Any]     =       scheduler.step(lowercase__				,     lowercase__				,     lowercase__			)
                SCREAMING_SNAKE_CASE_      :			List[Any]     =       output.prev_sample

          SCREAMING_SNAKE_CASE_      :			str     =       torch.sum(torch.abs(lowercase__			)			)
          SCREAMING_SNAKE_CASE_      :			List[Any]     =       torch.mean(torch.abs(lowercase__			)			)

          if str(lowercase__			).startswith("cpu"			):
                # The following sum varies between 148 and 156 on mps. Why?
                assert abs(result_sum.item() - 0.1233			) < 1e-2
                assert abs(result_mean.item() - 0.0002			) < 1e-3
          elif str(lowercase__			).startswith("mps"			):
                # Larger tolerance on mps
                assert abs(result_mean.item() - 0.0002			) < 1e-2
          else:
                # CUDA
                assert abs(result_sum.item() - 0.1233			) < 1e-2
                assert abs(result_mean.item() - 0.0002			) < 1e-3







    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       self.scheduler_classes[0]
          SCREAMING_SNAKE_CASE_      :			int     =       self.get_scheduler_config()
          SCREAMING_SNAKE_CASE_      :			Optional[int]     =       scheduler_class(**lowercase__				,     use_karras_sigmas=lowercase__			)

          scheduler.set_timesteps(self.num_inference_steps				,     device=lowercase__			)

          SCREAMING_SNAKE_CASE_      :			List[str]     =       self.dummy_model()
          SCREAMING_SNAKE_CASE_      :			Optional[int]     =       self.dummy_sample_deter.to(lowercase__			) * scheduler.init_noise_sigma
          SCREAMING_SNAKE_CASE_      :			Any     =       sample.to(lowercase__			)

          for t in scheduler.timesteps:
                SCREAMING_SNAKE_CASE_      :			Optional[int]     =       scheduler.scale_model_input(lowercase__				,     lowercase__			)

                SCREAMING_SNAKE_CASE_      :			List[str]     =       model(lowercase__				,     lowercase__			)

                SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       scheduler.step(lowercase__				,     lowercase__				,     lowercase__			)
                SCREAMING_SNAKE_CASE_      :			Tuple     =       output.prev_sample

          SCREAMING_SNAKE_CASE_      :			Any     =       torch.sum(torch.abs(lowercase__			)			)
          SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       torch.mean(torch.abs(lowercase__			)			)

          assert abs(result_sum.item() - 0.00015			) < 1e-2
          assert abs(result_mean.item() - 1.98_69_55_45_35_03_46_95e-07			) < 1e-2

