


import os
import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from huggingface_hub.file_download import http_get
from requests.exceptions import HTTPError

from transformers import (
    AlbertTokenizer,
    AutoTokenizer,
    BertTokenizer,
    BertTokenizerFast,
    GPTaTokenizerFast,
    is_tokenizers_available,
)
from transformers.testing_utils import TOKEN, USER, is_staging_test, require_tokenizers
from transformers.tokenization_utils import Trie


sys.path.append(str(Path(__file__).parent.parent / '''utils'''))

from test_module.custom_tokenization import CustomTokenizer  # noqa E402


if is_tokenizers_available():
		from test_module.custom_tokenization_fast import CustomTokenizerFast


class        lowerCAmelCase_			(							unittest.TestCase):




						def 			_snake_case       (						self    : List[str]    )   ->Tuple:






									"""simple docstring"""






									a__			:Tuple    	=     mock.Mock()
									a__			:Dict    	=     500
									a__			:Optional[Any]    	=     {}
									a__			:str    	=     HTTPError
									a__			:Any    	=     {}

									# Download this model to make sure it's in the cache.
									a__			:Tuple    	=     BertTokenizer.from_pretrained("hf-internal-testing/tiny-random-bert"    )

									# Under the mock environment we get a 500 error when trying to reach the tokenizer.
									with mock.patch("requests.Session.request" ,  return_value=lowercase_    ) as mock_head:
												a__			:Dict    	=     BertTokenizer.from_pretrained("hf-internal-testing/tiny-random-bert"    )
												# This check we did call the fake head request
												mock_head.assert_called()




						@require_tokenizers
						def 			_snake_case       (						self    : Any    )   ->Optional[Any]:






									"""simple docstring"""






									a__			:List[Any]    	=     mock.Mock()
									a__			:Optional[int]    	=     500
									a__			:Union[str, Any]    	=     {}
									a__			:Optional[Any]    	=     HTTPError
									a__			:List[Any]    	=     {}

									# Download this model to make sure it's in the cache.
									a__			:str    	=     GPTaTokenizerFast.from_pretrained("gpt2"    )

									# Under the mock environment we get a 500 error when trying to reach the tokenizer.
									with mock.patch("requests.Session.request" ,  return_value=lowercase_    ) as mock_head:
												a__			:int    	=     GPTaTokenizerFast.from_pretrained("gpt2"    )
												# This check we did call the fake head request
												mock_head.assert_called()




						def 			_snake_case       (						self    : List[Any]    )   ->Any:






									"""simple docstring"""






									try:
												a__			:List[str]    	=     tempfile.mktemp()
												with open(lowercase_ ,  "wb"    ) as f:
															http_get("https://huggingface.co/albert-base-v1/resolve/main/spiece.model" ,  lowercase_    )

												a__			:List[str]    	=     AlbertTokenizer.from_pretrained(lowercase_    )
									finally:
												os.remove(lowercase_    )

									# Supporting this legacy load introduced a weird bug where the tokenizer would load local files if they are in
									# the current folder and have the right name.
									if os.path.isfile("tokenizer.json"    ):
												# We skip the test if the user has a `tokenizer.json` in this folder to avoid deleting it.
												return
									try:
												with open("tokenizer.json" ,  "wb"    ) as f:
															http_get("https://huggingface.co/hf-internal-testing/tiny-random-bert/blob/main/tokenizer.json" ,  lowercase_    )
												a__			:Tuple    	=     AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-gpt2"    )
												# The tiny random BERT has a vocab size of 1024, tiny gpt2 as a vocab size of 1000
												self.assertEqual(tokenizer.vocab_size ,  1000    )
												# Tokenizer should depend on the remote checkpoint, not the local tokenizer.json file.

									finally:
												os.remove("tokenizer.json"    )






						def 			_snake_case       (						self    : List[str]    )   ->Dict:






									"""simple docstring"""






									a__			:str    	=     AlbertTokenizer.from_pretrained("https://huggingface.co/albert-base-v1/resolve/main/spiece.model"    )






@is_staging_test
class        lowerCAmelCase_			(							unittest.TestCase):
						lowerCamelCase_    =  ["[UNK]", "[CLS]", "[SEP]", "[PAD]", "[MASK]", "bla", "blou"]




						@classmethod
						def 			_snake_case       (						cls    : str    )   ->Dict:






									"""simple docstring"""






									a__			:Union[str, Any]    	=     TOKEN
									HfFolder.save_token(lowercase_    )




						@classmethod
						def 			_snake_case       (						cls    : Union[str, Any]    )   ->Tuple:






									"""simple docstring"""






									try:
												delete_repo(token=cls._token ,  repo_id="test-tokenizer"    )
									except HTTPError:
												pass

									try:
												delete_repo(token=cls._token ,  repo_id="valid_org/test-tokenizer-org"    )
									except HTTPError:
												pass

									try:
												delete_repo(token=cls._token ,  repo_id="test-dynamic-tokenizer"    )
									except HTTPError:
												pass




						def 			_snake_case       (						self    : Tuple    )   ->str:






									"""simple docstring"""






									with tempfile.TemporaryDirectory() as tmp_dir:
												a__			:Dict    	=     os.path.join(lowercase_ ,  "vocab.txt"    )
												with open(lowercase_ ,  "w" ,  encoding="utf-8"    ) as vocab_writer:
															vocab_writer.write("".join([x + "\n" for x in self.vocab_tokens]    )    )
												a__			:Dict    	=     BertTokenizer(lowercase_    )

									tokenizer.push_to_hub("test-tokenizer" ,  use_auth_token=self._token    )
									a__			:Union[str, Any]    	=     BertTokenizer.from_pretrained(F'''{USER}/test-tokenizer'''    )
									self.assertDictEqual(new_tokenizer.vocab ,  tokenizer.vocab    )

									# Reset repo
									delete_repo(token=self._token ,  repo_id="test-tokenizer"    )

									# Push to hub via save_pretrained
									with tempfile.TemporaryDirectory() as tmp_dir:
												tokenizer.save_pretrained(lowercase_ ,  repo_id="test-tokenizer" ,  push_to_hub=lowercase_ ,  use_auth_token=self._token    )

									a__			:Union[str, Any]    	=     BertTokenizer.from_pretrained(F'''{USER}/test-tokenizer'''    )
									self.assertDictEqual(new_tokenizer.vocab ,  tokenizer.vocab    )




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






									"""simple docstring"""






									with tempfile.TemporaryDirectory() as tmp_dir:
												a__			:Any    	=     os.path.join(lowercase_ ,  "vocab.txt"    )
												with open(lowercase_ ,  "w" ,  encoding="utf-8"    ) as vocab_writer:
															vocab_writer.write("".join([x + "\n" for x in self.vocab_tokens]    )    )
												a__			:Dict    	=     BertTokenizer(lowercase_    )

									tokenizer.push_to_hub("valid_org/test-tokenizer-org" ,  use_auth_token=self._token    )
									a__			:str    	=     BertTokenizer.from_pretrained("valid_org/test-tokenizer-org"    )
									self.assertDictEqual(new_tokenizer.vocab ,  tokenizer.vocab    )

									# Reset repo
									delete_repo(token=self._token ,  repo_id="valid_org/test-tokenizer-org"    )

									# Push to hub via save_pretrained
									with tempfile.TemporaryDirectory() as tmp_dir:
												tokenizer.save_pretrained(
												    lowercase_ ,  repo_id="valid_org/test-tokenizer-org" ,  push_to_hub=lowercase_ ,  use_auth_token=self._token    )

									a__			:Optional[int]    	=     BertTokenizer.from_pretrained("valid_org/test-tokenizer-org"    )
									self.assertDictEqual(new_tokenizer.vocab ,  tokenizer.vocab    )






						@require_tokenizers
						def 			_snake_case       (						self    : Optional[int]    )   ->Dict:






									"""simple docstring"""






									CustomTokenizer.register_for_auto_class()
									with tempfile.TemporaryDirectory() as tmp_dir:
												a__			:List[Any]    	=     os.path.join(lowercase_ ,  "vocab.txt"    )
												with open(lowercase_ ,  "w" ,  encoding="utf-8"    ) as vocab_writer:
															vocab_writer.write("".join([x + "\n" for x in self.vocab_tokens]    )    )
												a__			:int    	=     CustomTokenizer(lowercase_    )

									# No fast custom tokenizer
									tokenizer.push_to_hub("test-dynamic-tokenizer" ,  use_auth_token=self._token    )

									a__			:Any    	=     AutoTokenizer.from_pretrained(F'''{USER}/test-dynamic-tokenizer''' ,  trust_remote_code=lowercase_    )
									# Can't make an isinstance check because the new_model.config is from the CustomTokenizer class of a dynamic module
									self.assertEqual(tokenizer.__class__.__name__ ,  "CustomTokenizer"    )

									# Fast and slow custom tokenizer
									CustomTokenizerFast.register_for_auto_class()
									with tempfile.TemporaryDirectory() as tmp_dir:
												a__			:Any    	=     os.path.join(lowercase_ ,  "vocab.txt"    )
												with open(lowercase_ ,  "w" ,  encoding="utf-8"    ) as vocab_writer:
															vocab_writer.write("".join([x + "\n" for x in self.vocab_tokens]    )    )

												a__			:List[Any]    	=     BertTokenizerFast.from_pretrained(lowercase_    )
												bert_tokenizer.save_pretrained(lowercase_    )
												a__			:Union[str, Any]    	=     CustomTokenizerFast.from_pretrained(lowercase_    )

									tokenizer.push_to_hub("test-dynamic-tokenizer" ,  use_auth_token=self._token    )

									a__			:List[str]    	=     AutoTokenizer.from_pretrained(F'''{USER}/test-dynamic-tokenizer''' ,  trust_remote_code=lowercase_    )
									# Can't make an isinstance check because the new_model.config is from the FakeConfig class of a dynamic module
									self.assertEqual(tokenizer.__class__.__name__ ,  "CustomTokenizerFast"    )
									a__			:List[Any]    	=     AutoTokenizer.from_pretrained(
									    F'''{USER}/test-dynamic-tokenizer''' ,  use_fast=lowercase_ ,  trust_remote_code=lowercase_    )
									# Can't make an isinstance check because the new_model.config is from the FakeConfig class of a dynamic module
									self.assertEqual(tokenizer.__class__.__name__ ,  "CustomTokenizer"    )






class        lowerCAmelCase_			(							unittest.TestCase):




						def 			_snake_case       (						self    : int    )   ->Dict:






									"""simple docstring"""






									a__			:Tuple    	=     Trie()
									trie.add("Hello 友達"    )
									self.assertEqual(trie.data ,  {"H": {"e": {"l": {"l": {"o": {" ": {"友": {"達": {"": 1}}}}}}}}}    )
									trie.add("Hello"    )
									trie.data
									self.assertEqual(trie.data ,  {"H": {"e": {"l": {"l": {"o": {"": 1, " ": {"友": {"達": {"": 1}}}}}}}}}    )




						def 			_snake_case       (						self    : List[str]    )   ->str:






									"""simple docstring"""






									a__			:Any    	=     Trie()
									self.assertEqual(trie.split("[CLS] This is a extra_id_100"    ) ,  ["[CLS] This is a extra_id_100"]    )
									trie.add("[CLS]"    )
									trie.add("extra_id_1"    )
									trie.add("extra_id_100"    )
									self.assertEqual(trie.split("[CLS] This is a extra_id_100"    ) ,  ["[CLS]", " This is a ", "extra_id_100"]    )




						def 			_snake_case       (						self    : List[str]    )   ->str:






									"""simple docstring"""






									a__			:int    	=     Trie()
									trie.add("A"    )
									self.assertEqual(trie.split("ABC"    ) ,  ["A", "BC"]    )
									self.assertEqual(trie.split("BCA"    ) ,  ["BC", "A"]    )




						def 			_snake_case       (						self    : str    )   ->Optional[int]:






									"""simple docstring"""






									a__			:Dict    	=     Trie()
									trie.add("TOKEN]"    )
									trie.add("[SPECIAL_TOKEN]"    )
									self.assertEqual(trie.split("This is something [SPECIAL_TOKEN]"    ) ,  ["This is something ", "[SPECIAL_TOKEN]"]    )




						def 			_snake_case       (						self    : Tuple    )   ->Any:






									"""simple docstring"""






									a__			:Tuple    	=     Trie()
									trie.add("A"    )
									trie.add("P"    )
									trie.add("[SPECIAL_TOKEN]"    )
									self.assertEqual(trie.split("This is something [SPECIAL_TOKEN]"    ) ,  ["This is something ", "[SPECIAL_TOKEN]"]    )




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






									"""simple docstring"""






									a__			:Any    	=     Trie()
									trie.add("AB"    )
									trie.add("B"    )
									trie.add("C"    )
									self.assertEqual(trie.split("ABC"    ) ,  ["AB", "C"]    )




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






									"""simple docstring"""






									a__			:int    	=     Trie()
									trie.add("ABC"    )
									trie.add("B"    )
									trie.add("CD"    )
									self.assertEqual(trie.split("ABCD"    ) ,  ["ABC", "D"]    )






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






									"""simple docstring"""






									a__			:str    	=     Trie()
									a__			:Tuple    	=     trie.cut_text("ABC" ,  [0, 0, 2, 1, 2, 3]    )
									self.assertEqual(lowercase_ ,  ["AB", "C"]    )




"""simple docstring"""

import importlib
import os
from dataclasses import dataclass
from enum import Enum
from typing import Any, Dict, Optional, Union

import torch

from ..utils import BaseOutput


SCREAMING_SNAKE_CASE_            =    '''scheduler_config.json'''
class   _UpperCAmelCase				(		SCREAMING_SNAKE_CASE_				):
      __SCREAMING_SNAKE_CASE :				List[Any]				   =						1
      __SCREAMING_SNAKE_CASE :				Dict				   =						2
      __SCREAMING_SNAKE_CASE :				List[Any]				   =						3
      __SCREAMING_SNAKE_CASE :				Any				   =						4
      __SCREAMING_SNAKE_CASE :				Any				   =						5
      __SCREAMING_SNAKE_CASE :				Union[str, Any]				   =						6
      __SCREAMING_SNAKE_CASE :				str				   =						7
      __SCREAMING_SNAKE_CASE :				Any				   =						8
      __SCREAMING_SNAKE_CASE :				Tuple				   =						9
      __SCREAMING_SNAKE_CASE :				int				   =						1_0
      __SCREAMING_SNAKE_CASE :				Union[str, Any]				   =						1_1
      __SCREAMING_SNAKE_CASE :				Union[str, Any]				   =						1_2
      __SCREAMING_SNAKE_CASE :				Dict				   =						1_3
      __SCREAMING_SNAKE_CASE :				Optional[Any]				   =						1_4

@dataclass
class   _UpperCAmelCase				(		SCREAMING_SNAKE_CASE_				):
      __SCREAMING_SNAKE_CASE :				torch.FloatTensor

class   _UpperCAmelCase				:
      __SCREAMING_SNAKE_CASE :				str				   =						SCHEDULER_CONFIG_NAME
      __SCREAMING_SNAKE_CASE :				Dict				   =						[]
      __SCREAMING_SNAKE_CASE :				Union[str, Any]				   =						True




      @classmethod
      def      a_   (				cls ,							lowercase_ = None ,							lowercase_ = None ,							lowercase_=False ,							**lowercase_ ,							)    ->       List[Any]:
          UpperCAmelCase      ,				UpperCAmelCase      ,				UpperCAmelCase          =       cls.load_config(
              pretrained_model_name_or_path=lowercase_ ,							subfolder=lowercase_ ,							return_unused_kwargs=lowercase_ ,							return_commit_hash=lowercase_ ,							**lowercase_ ,							)
          return cls.from_config(lowercase_ ,							return_unused_kwargs=lowercase_ ,							**lowercase_				)




      def      a_   (				self ,							lowercase_ ,							lowercase_ = False ,							**lowercase_				)    ->       str:
          self.save_config(save_directory=lowercase_ ,							push_to_hub=lowercase_ ,							**lowercase_				)




      @property
      def      a_   (				self				)    ->       Union[str, Any]:
          return self._get_compatibles()




      @classmethod
      def      a_   (				cls				)    ->       Any:
          UpperCAmelCase          =       list(set([cls.__name__] + cls._compatibles				)				)
          UpperCAmelCase          =       importlib.import_module(__name__.split('.'				)[0]				)
          UpperCAmelCase          =       [
              getattr(lowercase_ ,							lowercase_				) for c in compatible_classes_str if hasattr(lowercase_ ,							lowercase_				)
          ]
          return compatible_classes



