


"""simple docstring"""





import os
import shutil
import sys
import tempfile
import unittest
from pathlib import Path

import pytest

import transformers
from transformers import (
    BERT_PRETRAINED_CONFIG_ARCHIVE_MAP,
    GPT2_PRETRAINED_CONFIG_ARCHIVE_MAP,
    AutoTokenizer,
    BertConfig,
    BertTokenizer,
    BertTokenizerFast,
    CTRLTokenizer,
    GPTaTokenizer,
    GPTaTokenizerFast,
    PreTrainedTokenizerFast,
    RobertaTokenizer,
    RobertaTokenizerFast,
    is_tokenizers_available,
)
from transformers.models.auto.configuration_auto import CONFIG_MAPPING, AutoConfig
from transformers.models.auto.tokenization_auto import (
    TOKENIZER_MAPPING,
    get_tokenizer_config,
    tokenizer_class_from_name,
)
from transformers.models.roberta.configuration_roberta import RobertaConfig
from transformers.testing_utils import (
    DUMMY_DIFF_TOKENIZER_IDENTIFIER,
    DUMMY_UNKNOWN_IDENTIFIER,
    SMALL_MODEL_IDENTIFIER,
    RequestCounter,
    require_tokenizers,
    slow,
)


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

from test_module.custom_configuration import CustomConfig  # noqa E402
from test_module.custom_tokenization import CustomTokenizer  # noqa E402


if is_tokenizers_available():
 from test_module.custom_tokenization_fast import CustomTokenizerFast



class 						snake_case_(				unittest.TestCase	):






     def    lowerCamelCase__   (		self     :    Any ):
           lowerCAmelCase    :		List[Any]				=      0






     @slow
     def    lowerCamelCase__   (		self     :    Optional[int] ):
           for model_name in (x for x in BERT_PRETRAINED_CONFIG_ARCHIVE_MAP.keys() if "japanese" not in x):
                 lowerCAmelCase    :		Union[str, Any]				=      AutoTokenizer.from_pretrained(UpperCamelCase_ )
                 self.assertIsNotNone(UpperCamelCase_ )
                 self.assertIsInstance(UpperCamelCase_    , (BertTokenizer, BertTokenizerFast) )
                 self.assertGreater(len(UpperCamelCase_ )    , 0 )

           for model_name in GPT2_PRETRAINED_CONFIG_ARCHIVE_MAP.keys():
                 lowerCAmelCase    :		Union[str, Any]				=      AutoTokenizer.from_pretrained(UpperCamelCase_ )
                 self.assertIsNotNone(UpperCamelCase_ )
                 self.assertIsInstance(UpperCamelCase_    , (GPTaTokenizer, GPTaTokenizerFast) )
                 self.assertGreater(len(UpperCamelCase_ )    , 0 )






     def    lowerCamelCase__   (		self     :    List[Any] ):
           lowerCAmelCase    :		str				=      AutoTokenizer.from_pretrained(UpperCamelCase_ )
           self.assertIsInstance(UpperCamelCase_    , (BertTokenizer, BertTokenizerFast) )
           self.assertEqual(tokenizer.vocab_size    , 1_2 )






     def    lowerCamelCase__   (		self     :    str ):
           lowerCAmelCase    :		Any				=      AutoTokenizer.from_pretrained(UpperCamelCase_ )
           self.assertIsInstance(UpperCamelCase_    , (RobertaTokenizer, RobertaTokenizerFast) )
           self.assertEqual(tokenizer.vocab_size    , 2_0 )






     def    lowerCamelCase__   (		self     :    Optional[int] ):
           lowerCAmelCase    :		List[Any]				=      AutoConfig.from_pretrained(UpperCamelCase_ )
           self.assertIsInstance(UpperCamelCase_    , UpperCamelCase_ )
           # Check that tokenizer_type ≠ model_type
           lowerCAmelCase    :		List[str]				=      AutoTokenizer.from_pretrained(UpperCamelCase_    , config=UpperCamelCase_ )
           self.assertIsInstance(UpperCamelCase_    , (BertTokenizer, BertTokenizerFast) )
           self.assertEqual(tokenizer.vocab_size    , 1_2 )






     def    lowerCamelCase__   (		self     :    Optional[int] ):
           with tempfile.TemporaryDirectory() as tmp_dir:
                 shutil.copy('''./tests/fixtures/vocab.txt'''    , os.path.join(UpperCamelCase_    , '''vocab.txt''' ) )

                 lowerCAmelCase    :		int				=      AutoTokenizer.from_pretrained(UpperCamelCase_    , tokenizer_type='''bert'''    , use_fast=UpperCamelCase_ )
                 self.assertIsInstance(UpperCamelCase_    , UpperCamelCase_ )

           with tempfile.TemporaryDirectory() as tmp_dir:
                 shutil.copy('''./tests/fixtures/vocab.json'''    , os.path.join(UpperCamelCase_    , '''vocab.json''' ) )
                 shutil.copy('''./tests/fixtures/merges.txt'''    , os.path.join(UpperCamelCase_    , '''merges.txt''' ) )

                 lowerCAmelCase    :		Optional[Any]				=      AutoTokenizer.from_pretrained(UpperCamelCase_    , tokenizer_type='''gpt2'''    , use_fast=UpperCamelCase_ )
                 self.assertIsInstance(UpperCamelCase_    , UpperCamelCase_ )






     @require_tokenizers
     def    lowerCamelCase__   (		self     :    List[str] ):
           with tempfile.TemporaryDirectory() as tmp_dir:
                 shutil.copy('''./tests/fixtures/vocab.txt'''    , os.path.join(UpperCamelCase_    , '''vocab.txt''' ) )

                 lowerCAmelCase    :		List[str]				=      AutoTokenizer.from_pretrained(UpperCamelCase_    , tokenizer_type='''bert''' )
                 self.assertIsInstance(UpperCamelCase_    , UpperCamelCase_ )

           with tempfile.TemporaryDirectory() as tmp_dir:
                 shutil.copy('''./tests/fixtures/vocab.json'''    , os.path.join(UpperCamelCase_    , '''vocab.json''' ) )
                 shutil.copy('''./tests/fixtures/merges.txt'''    , os.path.join(UpperCamelCase_    , '''merges.txt''' ) )

                 lowerCAmelCase    :		Optional[Any]				=      AutoTokenizer.from_pretrained(UpperCamelCase_    , tokenizer_type='''gpt2''' )
                 self.assertIsInstance(UpperCamelCase_    , UpperCamelCase_ )






     def    lowerCamelCase__   (		self     :    Optional[Any] ):
           with pytest.raises(UpperCamelCase_ ):
                 AutoTokenizer.from_pretrained('''./'''    , tokenizer_type='''xxx''' )






     @require_tokenizers
     def    lowerCamelCase__   (		self     :    Optional[Any] ):
           for tokenizer_class in [BertTokenizer, BertTokenizerFast, AutoTokenizer]:
                 lowerCAmelCase    :		List[str]				=      tokenizer_class.from_pretrained('''wietsedv/bert-base-dutch-cased''' )
                 self.assertIsInstance(UpperCamelCase_    , (BertTokenizer, BertTokenizerFast) )

                 if isinstance(UpperCamelCase_    , UpperCamelCase_ ):
                       self.assertEqual(tokenizer.basic_tokenizer.do_lower_case    , UpperCamelCase_ )
                 else:
                       self.assertEqual(tokenizer.do_lower_case    , UpperCamelCase_ )

                 self.assertEqual(tokenizer.model_max_length    , 5_1_2 )






     @require_tokenizers
     def    lowerCamelCase__   (		self     :    Any ):
           for tokenizer_class in [BertTokenizer, BertTokenizerFast, AutoTokenizer]:
                 with self.assertRaisesRegex(
                     UpperCamelCase_    , '''julien-c/herlolip-not-exists is not a local folder and is not a valid model identifier'''    , ):
                       lowerCAmelCase    :		Optional[Any]				=      tokenizer_class.from_pretrained('''julien-c/herlolip-not-exists''' )






     def    lowerCamelCase__   (		self     :    Any ):
           # tests: https://github.com/huggingface/transformers/pull/13251
           # 1. models with `-`, e.g. xlm-roberta -> xlm_roberta
           # 2. models that don't remap 1-1 from model-name to model file, e.g., openai-gpt -> openai
           lowerCAmelCase    :		List[Any]				=      TOKENIZER_MAPPING.values()
           lowerCAmelCase    :		Tuple				=      []

           for slow_tok, fast_tok in tokenizers:
                 if slow_tok is not None:
                       tokenizer_names.append(slow_tok.__name__ )

                 if fast_tok is not None:
                       tokenizer_names.append(fast_tok.__name__ )

           for tokenizer_name in tokenizer_names:
                 # must find the right class
                 tokenizer_class_from_name(UpperCamelCase_ )






     @require_tokenizers
     def    lowerCamelCase__   (		self     :    Dict ):
           self.assertIsInstance(AutoTokenizer.from_pretrained('''bert-base-cased'''    , use_fast=UpperCamelCase_ )    , UpperCamelCase_ )
           self.assertIsInstance(AutoTokenizer.from_pretrained('''bert-base-cased''' )    , UpperCamelCase_ )






     @require_tokenizers
     def    lowerCamelCase__   (		self     :    int ):
           lowerCAmelCase    :		Union[str, Any]				=      AutoTokenizer.from_pretrained('''distilbert-base-uncased'''    , do_lower_case=UpperCamelCase_ )
           lowerCAmelCase    :		List[str]				=      '''Hello, world. How are you?'''
           lowerCAmelCase    :		Dict				=      tokenizer.tokenize(UpperCamelCase_ )
           self.assertEqual('''[UNK]'''    , tokens[0] )

           lowerCAmelCase    :		str				=      AutoTokenizer.from_pretrained('''microsoft/mpnet-base'''    , do_lower_case=UpperCamelCase_ )
           lowerCAmelCase    :		int				=      tokenizer.tokenize(UpperCamelCase_ )
           self.assertEqual('''[UNK]'''    , tokens[0] )






     @require_tokenizers
     def    lowerCamelCase__   (		self     :    Union[str, Any] ):
           lowerCAmelCase    :		List[Any]				=      AutoTokenizer.from_pretrained('''robot-test/dummy-tokenizer-fast-with-model-config''' )
           self.assertEqual(type(UpperCamelCase_ )    , UpperCamelCase_ )
           self.assertEqual(tokenizer.model_max_length    , 5_1_2 )
           self.assertEqual(tokenizer.vocab_size    , 3_0_0_0_0 )
           self.assertEqual(tokenizer.unk_token    , '''[UNK]''' )
           self.assertEqual(tokenizer.padding_side    , '''right''' )
           self.assertEqual(tokenizer.truncation_side    , '''right''' )






     def    lowerCamelCase__   (		self     :    Dict ):
           lowerCAmelCase    :		Dict				=      AutoTokenizer.from_pretrained(UpperCamelCase_ )
           self.assertIsInstance(UpperCamelCase_    , (BertTokenizer, BertTokenizerFast) )
           with tempfile.TemporaryDirectory() as tmp_dir:
                 tokenizer.save_pretrained(UpperCamelCase_ )
                 lowerCAmelCase    :		Union[str, Any]				=      AutoTokenizer.from_pretrained(UpperCamelCase_ )

           self.assertIsInstance(UpperCamelCase_    , tokenizer.__class__ )
           self.assertEqual(tokenizera.vocab_size    , 1_2 )






     def    lowerCamelCase__   (		self     :    Tuple ):
           lowerCAmelCase    :		List[str]				=      AutoTokenizer.from_pretrained('''ctrl''' )
           # There is no fast CTRL so this always gives us a slow tokenizer.
           self.assertIsInstance(UpperCamelCase_    , UpperCamelCase_ )






     def    lowerCamelCase__   (		self     :    str ):
           # Check we can load the tokenizer config of an online model.
           lowerCAmelCase    :		Union[str, Any]				=      get_tokenizer_config('''bert-base-cased''' )
           lowerCAmelCase    :		Optional[int]				=      config.pop('''_commit_hash'''    , UpperCamelCase_ )
           # If we ever update bert-base-cased tokenizer config, this dict here will need to be updated.
           self.assertEqual(UpperCamelCase_    , {'''do_lower_case''': False} )

           # This model does not have a tokenizer_config so we get back an empty dict.
           lowerCAmelCase    :		str				=      get_tokenizer_config(UpperCamelCase_ )
           self.assertDictEqual(UpperCamelCase_    , {} )

           # A tokenizer saved with `save_pretrained` always creates a tokenizer config.
           lowerCAmelCase    :		Dict				=      AutoTokenizer.from_pretrained(UpperCamelCase_ )
           with tempfile.TemporaryDirectory() as tmp_dir:
                 tokenizer.save_pretrained(UpperCamelCase_ )
                 lowerCAmelCase    :		Tuple				=      get_tokenizer_config(UpperCamelCase_ )

           # Check the class of the tokenizer was properly saved (note that it always saves the slow class).
           self.assertEqual(config['''tokenizer_class''']    , '''BertTokenizer''' )






     def    lowerCamelCase__   (		self     :    Union[str, Any] ):
           try:
                 AutoConfig.register('''custom'''    , UpperCamelCase_ )

                 AutoTokenizer.register(UpperCamelCase_    , slow_tokenizer_class=UpperCamelCase_ )
                 # Trying to register something existing in the Transformers library will raise an error
                 with self.assertRaises(UpperCamelCase_ ):
                       AutoTokenizer.register(UpperCamelCase_    , slow_tokenizer_class=UpperCamelCase_ )

                 lowerCAmelCase    :		Dict				=      CustomTokenizer.from_pretrained(UpperCamelCase_ )
                 with tempfile.TemporaryDirectory() as tmp_dir:
                       tokenizer.save_pretrained(UpperCamelCase_ )

                       lowerCAmelCase    :		Any				=      AutoTokenizer.from_pretrained(UpperCamelCase_ )
                       self.assertIsInstance(UpperCamelCase_    , UpperCamelCase_ )

           finally:
                 if "custom" in CONFIG_MAPPING._extra_content:
                       del CONFIG_MAPPING._extra_content["custom"]
                 if CustomConfig in TOKENIZER_MAPPING._extra_content:
                       del TOKENIZER_MAPPING._extra_content[CustomConfig]






     @require_tokenizers
     def    lowerCamelCase__   (		self     :    int ):
           try:
                 AutoConfig.register('''custom'''    , UpperCamelCase_ )

                 # Can register in two steps
                 AutoTokenizer.register(UpperCamelCase_    , slow_tokenizer_class=UpperCamelCase_ )
                 self.assertEqual(TOKENIZER_MAPPING[CustomConfig]    , (CustomTokenizer, None) )
                 AutoTokenizer.register(UpperCamelCase_    , fast_tokenizer_class=UpperCamelCase_ )
                 self.assertEqual(TOKENIZER_MAPPING[CustomConfig]    , (CustomTokenizer, CustomTokenizerFast) )

                 del TOKENIZER_MAPPING._extra_content[CustomConfig]
                 # Can register in one step
                 AutoTokenizer.register(
                     UpperCamelCase_    , slow_tokenizer_class=UpperCamelCase_    , fast_tokenizer_class=UpperCamelCase_ )
                 self.assertEqual(TOKENIZER_MAPPING[CustomConfig]    , (CustomTokenizer, CustomTokenizerFast) )

                 # Trying to register something existing in the Transformers library will raise an error
                 with self.assertRaises(UpperCamelCase_ ):
                       AutoTokenizer.register(UpperCamelCase_    , fast_tokenizer_class=UpperCamelCase_ )

                 # We pass through a bert tokenizer fast cause there is no converter slow to fast for our new toknizer
                 # and that model does not have a tokenizer.json
                 with tempfile.TemporaryDirectory() as tmp_dir:
                       lowerCAmelCase    :		Any				=      BertTokenizerFast.from_pretrained(UpperCamelCase_ )
                       bert_tokenizer.save_pretrained(UpperCamelCase_ )
                       lowerCAmelCase    :		int				=      CustomTokenizerFast.from_pretrained(UpperCamelCase_ )

                 with tempfile.TemporaryDirectory() as tmp_dir:
                       tokenizer.save_pretrained(UpperCamelCase_ )

                       lowerCAmelCase    :		Optional[Any]				=      AutoTokenizer.from_pretrained(UpperCamelCase_ )
                       self.assertIsInstance(UpperCamelCase_    , UpperCamelCase_ )

                       lowerCAmelCase    :		Optional[int]				=      AutoTokenizer.from_pretrained(UpperCamelCase_    , use_fast=UpperCamelCase_ )
                       self.assertIsInstance(UpperCamelCase_    , UpperCamelCase_ )

           finally:
                 if "custom" in CONFIG_MAPPING._extra_content:
                       del CONFIG_MAPPING._extra_content["custom"]
                 if CustomConfig in TOKENIZER_MAPPING._extra_content:
                       del TOKENIZER_MAPPING._extra_content[CustomConfig]






     def    lowerCamelCase__   (		self     :    Dict ):
           # If remote code is not set, we will time out when asking whether to load the model.
           with self.assertRaises(UpperCamelCase_ ):
                 lowerCAmelCase    :		Union[str, Any]				=      AutoTokenizer.from_pretrained('''hf-internal-testing/test_dynamic_tokenizer''' )
           # If remote code is disabled, we can't load this config.
           with self.assertRaises(UpperCamelCase_ ):
                 lowerCAmelCase    :		str				=      AutoTokenizer.from_pretrained(
                     '''hf-internal-testing/test_dynamic_tokenizer'''    , trust_remote_code=UpperCamelCase_ )

           lowerCAmelCase    :		Optional[int]				=      AutoTokenizer.from_pretrained('''hf-internal-testing/test_dynamic_tokenizer'''    , trust_remote_code=UpperCamelCase_ )
           self.assertTrue(tokenizer.special_attribute_present )
           # Test tokenizer can be reloaded.
           with tempfile.TemporaryDirectory() as tmp_dir:
                 tokenizer.save_pretrained(UpperCamelCase_ )
                 lowerCAmelCase    :		Any				=      AutoTokenizer.from_pretrained(UpperCamelCase_    , trust_remote_code=UpperCamelCase_ )
           self.assertTrue(reloaded_tokenizer.special_attribute_present )

           if is_tokenizers_available():
                 self.assertEqual(tokenizer.__class__.__name__    , '''NewTokenizerFast''' )
                 self.assertEqual(reloaded_tokenizer.__class__.__name__    , '''NewTokenizerFast''' )

                 # Test we can also load the slow version
                 lowerCAmelCase    :		str				=      AutoTokenizer.from_pretrained(
                     '''hf-internal-testing/test_dynamic_tokenizer'''    , trust_remote_code=UpperCamelCase_    , use_fast=UpperCamelCase_ )
                 self.assertTrue(tokenizer.special_attribute_present )
                 self.assertEqual(tokenizer.__class__.__name__    , '''NewTokenizer''' )
                 # Test tokenizer can be reloaded.
                 with tempfile.TemporaryDirectory() as tmp_dir:
                       tokenizer.save_pretrained(UpperCamelCase_ )
                       lowerCAmelCase    :		Any				=      AutoTokenizer.from_pretrained(UpperCamelCase_    , trust_remote_code=UpperCamelCase_    , use_fast=UpperCamelCase_ )
                 self.assertEqual(reloaded_tokenizer.__class__.__name__    , '''NewTokenizer''' )
                 self.assertTrue(reloaded_tokenizer.special_attribute_present )
           else:
                 self.assertEqual(tokenizer.__class__.__name__    , '''NewTokenizer''' )
                 self.assertEqual(reloaded_tokenizer.__class__.__name__    , '''NewTokenizer''' )






     @require_tokenizers
     def    lowerCamelCase__   (		self     :    int ):



           class 						snake_case_(				a__	):
                __UpperCamelCase     =    False



           class 						snake_case_(				a__	):
                __UpperCamelCase     =    NewTokenizer
                __UpperCamelCase     =    False

           try:
                 AutoConfig.register('''custom'''    , UpperCamelCase_ )
                 AutoTokenizer.register(UpperCamelCase_    , slow_tokenizer_class=UpperCamelCase_ )
                 AutoTokenizer.register(UpperCamelCase_    , fast_tokenizer_class=UpperCamelCase_ )
                 # If remote code is not set, the default is to use local
                 lowerCAmelCase    :		Tuple				=      AutoTokenizer.from_pretrained('''hf-internal-testing/test_dynamic_tokenizer''' )
                 self.assertEqual(tokenizer.__class__.__name__    , '''NewTokenizerFast''' )
                 self.assertFalse(tokenizer.special_attribute_present )
                 lowerCAmelCase    :		Dict				=      AutoTokenizer.from_pretrained('''hf-internal-testing/test_dynamic_tokenizer'''    , use_fast=UpperCamelCase_ )
                 self.assertEqual(tokenizer.__class__.__name__    , '''NewTokenizer''' )
                 self.assertFalse(tokenizer.special_attribute_present )

                 # If remote code is disabled, we load the local one.
                 lowerCAmelCase    :		Any				=      AutoTokenizer.from_pretrained(
                     '''hf-internal-testing/test_dynamic_tokenizer'''    , trust_remote_code=UpperCamelCase_ )
                 self.assertEqual(tokenizer.__class__.__name__    , '''NewTokenizerFast''' )
                 self.assertFalse(tokenizer.special_attribute_present )
                 lowerCAmelCase    :		str				=      AutoTokenizer.from_pretrained(
                     '''hf-internal-testing/test_dynamic_tokenizer'''    , trust_remote_code=UpperCamelCase_    , use_fast=UpperCamelCase_ )
                 self.assertEqual(tokenizer.__class__.__name__    , '''NewTokenizer''' )
                 self.assertFalse(tokenizer.special_attribute_present )

                 # If remote is enabled, we load from the Hub
                 lowerCAmelCase    :		Optional[int]				=      AutoTokenizer.from_pretrained(
                     '''hf-internal-testing/test_dynamic_tokenizer'''    , trust_remote_code=UpperCamelCase_ )
                 self.assertEqual(tokenizer.__class__.__name__    , '''NewTokenizerFast''' )
                 self.assertTrue(tokenizer.special_attribute_present )
                 lowerCAmelCase    :		Dict				=      AutoTokenizer.from_pretrained(
                     '''hf-internal-testing/test_dynamic_tokenizer'''    , trust_remote_code=UpperCamelCase_    , use_fast=UpperCamelCase_ )
                 self.assertEqual(tokenizer.__class__.__name__    , '''NewTokenizer''' )
                 self.assertTrue(tokenizer.special_attribute_present )

           finally:
                 if "custom" in CONFIG_MAPPING._extra_content:
                       del CONFIG_MAPPING._extra_content["custom"]
                 if CustomConfig in TOKENIZER_MAPPING._extra_content:
                       del TOKENIZER_MAPPING._extra_content[CustomConfig]






     def    lowerCamelCase__   (		self     :    Dict ):
           lowerCAmelCase    :		List[str]				=      AutoTokenizer.from_pretrained(
               '''hf-internal-testing/test_dynamic_tokenizer_legacy'''    , trust_remote_code=UpperCamelCase_ )
           self.assertTrue(tokenizer.special_attribute_present )
           if is_tokenizers_available():
                 self.assertEqual(tokenizer.__class__.__name__    , '''NewTokenizerFast''' )

                 # Test we can also load the slow version
                 lowerCAmelCase    :		List[Any]				=      AutoTokenizer.from_pretrained(
                     '''hf-internal-testing/test_dynamic_tokenizer_legacy'''    , trust_remote_code=UpperCamelCase_    , use_fast=UpperCamelCase_ )
                 self.assertTrue(tokenizer.special_attribute_present )
                 self.assertEqual(tokenizer.__class__.__name__    , '''NewTokenizer''' )
           else:
                 self.assertEqual(tokenizer.__class__.__name__    , '''NewTokenizer''' )






     def    lowerCamelCase__   (		self     :    Optional[Any] ):
           with self.assertRaisesRegex(
               UpperCamelCase_    , '''bert-base is not a local folder and is not a valid model identifier''' ):
                 lowerCAmelCase    :		str				=      AutoTokenizer.from_pretrained('''bert-base''' )






     def    lowerCamelCase__   (		self     :    Any ):
           with self.assertRaisesRegex(
               UpperCamelCase_    , r'''aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)''' ):
                 lowerCAmelCase    :		Optional[Any]				=      AutoTokenizer.from_pretrained(UpperCamelCase_    , revision='''aaaaaa''' )






     def    lowerCamelCase__   (		self     :    Any ):
           # Make sure we have cached the tokenizer.
           lowerCAmelCase    :		str				=      AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-bert''' )
           with RequestCounter() as counter:
                 lowerCAmelCase    :		Optional[Any]				=      AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-bert''' )
                 self.assertEqual(counter.get_request_count    , 0 )
                 self.assertEqual(counter.head_request_count    , 1 )
                 self.assertEqual(counter.other_request_count    , 0 )




import importlib.util
import json
import os
import warnings
from dataclasses import dataclass, field

import torch

from ..training_args import TrainingArguments
from ..utils import cached_property, is_sagemaker_dp_enabled, logging


__A				=logging.get_logger(__name__)





def 			lowerCamelCase_ (				):
				# Get the sagemaker specific mp parameters from smp_options variable.
				lowerCamelCase_						=			os.getenv("SM_HP_MP_PARAMETERS"	, "{}"    )
				try:
								# Parse it and check the field "partitions" is included, it is required for model parallel.
								lowerCamelCase_						=			json.loads(lowerCamelCase__    )
								if "partitions" not in smp_options:
												return False
				except json.JSONDecodeError:
								return False

				# Get the sagemaker specific framework parameters from mpi_options variable.
				lowerCamelCase_						=			os.getenv("SM_FRAMEWORK_PARAMS"	, "{}"    )
				try:
								# Parse it and check the field "sagemaker_distributed_dataparallel_enabled".
								lowerCamelCase_						=			json.loads(lowerCamelCase__    )
								if not mpi_options.get("sagemaker_mpi_enabled"	, lowerCamelCase__    ):
												return False
				except json.JSONDecodeError:
								return False
				# Lastly, check if the `smdistributed` module is present.
				return importlib.util.find_spec("smdistributed"    ) is not None


if is_sagemaker_model_parallel_available():
			import smdistributed.modelparallel.torch as smp

			smp.init()





@dataclass
class     _SCREAMING_SNAKE_CASE     (      snake_case_						):
	lowerCAmelCase__         = field(
	    default=''			,       metadata={'help': 'Used by the SageMaker launcher to send mp-specific args. Ignored in SageMakerTrainer'}			,       )



	def     SCREAMING_SNAKE_CASE_(    self )				->       Tuple:
					super().__post_init__()
					warnings.warn(
					    "`SageMakerTrainingArguments` is deprecated and will be removed in v5 of Transformers. You can use "
					    "`TrainingArguments` instead."						,  lowercase						,  )



	@cached_property
	def     SCREAMING_SNAKE_CASE_(    self )				->       "torch.device":
					logger.info("PyTorch: setting up devices" )
					if torch.distributed.is_available() and torch.distributed.is_initialized() and self.local_rank == -1:
									logger.warning(
									    "torch.distributed process group is initialized, but local_rank == -1. "
									    "In order to use Torch DDP, launch your script with `python -m torch.distributed.launch" )
					if self.no_cuda:
									lowerCamelCase_						=			torch.device("cpu" )
									lowerCamelCase_						=			0
					elif is_sagemaker_model_parallel_available():
									lowerCamelCase_						=			smp.local_rank()
									lowerCamelCase_						=			torch.device("cuda"						,  lowercase )
									lowerCamelCase_						=			1
					elif is_sagemaker_dp_enabled():
									import smdistributed.dataparallel.torch.torch_smddp  # noqa: F401

									torch.distributed.init_process_group(backend="smddp"						,  timeout=self.ddp_timeout_delta )
									lowerCamelCase_						=			int(os.getenv("SMDATAPARALLEL_LOCAL_RANK" ) )
									lowerCamelCase_						=			torch.device("cuda"						,  self.local_rank )
									lowerCamelCase_						=			1
					elif self.local_rank == -1:
									# if n_gpu is > 1 we'll use nn.DataParallel.
									# If you only want to use a specific subset of GPUs use `CUDA_VISIBLE_DEVICES=0`
									# Explicitly set CUDA to the first (index 0) CUDA device, otherwise `set_device` will
									# trigger an error that a device index is missing. Index 0 takes into account the
									# GPUs available in the environment, so `CUDA_VISIBLE_DEVICES=1,2` with `cuda:0`
									# will use the first GPU in that env, i.e. GPU#1
									lowerCamelCase_						=			torch.device("cuda:0" if torch.cuda.is_available() else "cpu" )
									# Sometimes the line in the postinit has not been run before we end up here, so just checking we're not at
									# the default value.
									lowerCamelCase_						=			torch.cuda.device_count()
					else:
									# Here, we'll use torch.distributed.
									# Initializes the distributed backend which will take care of synchronizing nodes/GPUs
									if not torch.distributed.is_initialized():
													torch.distributed.init_process_group(backend="nccl"						,  timeout=self.ddp_timeout_delta )
									lowerCamelCase_						=			torch.device("cuda"						,  self.local_rank )
									lowerCamelCase_						=			1

					if device.type == "cuda":
									torch.cuda.set_device(lowercase )

					return device



	@property
	def     SCREAMING_SNAKE_CASE_(    self )				->       Tuple:
					if is_sagemaker_model_parallel_available():
									return smp.dp_size()

					return super().world_size



	@property
	def     SCREAMING_SNAKE_CASE_(    self )				->       List[str]:
					return not is_sagemaker_model_parallel_available()



	@property
	def     SCREAMING_SNAKE_CASE_(    self )				->       Dict:
					return False


