


"""simple docstring"""


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 __a						(							self   :    List[str]		):
								# A mock response for an HTTP head request to emulate server down
								A					       = mock.Mock()
								A					       = 500
								A					       = {}
								A					       = HTTPError
								A					       = {}

								# Download this model to make sure it's in the cache.
								A					       = 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					       = 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 __a						(							self   :    List[str]		):
								# A mock response for an HTTP head request to emulate server down
								A					       = mock.Mock()
								A					       = 500
								A					       = {}
								A					       = HTTPError
								A					       = {}

								# Download this model to make sure it's in the cache.
								A					       = 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					       = GPTaTokenizerFast.from_pretrained('gpt2'		)
										# This check we did call the fake head request
										mock_head.assert_called()







						def __a						(							self   :    str		):
								# This test is for deprecated behavior and can be removed in v5
								try:
										A					       = tempfile.mktemp()
										with open(_lowercase			,			'wb'		) as f:
												http_get('https://huggingface.co/albert-base-v1/resolve/main/spiece.model'			,			_lowercase		)

										A					       = 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					       = 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			,			1_000		)
										# Tokenizer should depend on the remote checkpoint, not the local tokenizer.json file.

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







						def __a						(							self   :    Dict		):
								# This test is for deprecated behavior and can be removed in v5
								A					       = 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 __a						(							cls   :    Union[str, Any]		):
								A					       = TOKEN
								HfFolder.save_token(_lowercase		)







						@classmethod
						def __a						(							cls   :    List[Any]		):
								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 __a						(							self   :    Any		):
								with tempfile.TemporaryDirectory() as tmp_dir:
										A					       = 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					       = BertTokenizer(_lowercase		)

								tokenizer.push_to_hub('test-tokenizer'			,			use_auth_token=self._token		)
								A					       = 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					       = BertTokenizer.from_pretrained(f'{USER}/test-tokenizer'		)
								self.assertDictEqual(new_tokenizer.vocab			,			tokenizer.vocab		)







						def __a						(							self   :    List[Any]		):
								with tempfile.TemporaryDirectory() as tmp_dir:
										A					       = 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					       = BertTokenizer(_lowercase		)

								tokenizer.push_to_hub('valid_org/test-tokenizer-org'			,			use_auth_token=self._token		)
								A					       = 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					       = BertTokenizer.from_pretrained('valid_org/test-tokenizer-org'		)
								self.assertDictEqual(new_tokenizer.vocab			,			tokenizer.vocab		)







						@require_tokenizers
						def __a						(							self   :    int		):
								CustomTokenizer.register_for_auto_class()
								with tempfile.TemporaryDirectory() as tmp_dir:
										A					       = 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					       = CustomTokenizer(_lowercase		)

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

								A					       = 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					       = 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					       = BertTokenizerFast.from_pretrained(_lowercase		)
										bert_tokenizer.save_pretrained(_lowercase		)
										A					       = CustomTokenizerFast.from_pretrained(_lowercase		)

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

								A					       = 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					       = 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 __a						(							self   :    Union[str, Any]		):
								A					       = 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 __a						(							self   :    str		):
								A					       = 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 __a						(							self   :    Dict		):
								A					       = Trie()
								trie.add('A'		)
								self.assertEqual(trie.split('ABC'		)			,			['A', 'BC']		)
								self.assertEqual(trie.split('BCA'		)			,			['BC', 'A']		)







						def __a						(							self   :    List[str]		):
								A					       = Trie()
								trie.add('TOKEN]'		)
								trie.add('[SPECIAL_TOKEN]'		)
								self.assertEqual(trie.split('This is something [SPECIAL_TOKEN]'		)			,			['This is something ', '[SPECIAL_TOKEN]']		)







						def __a						(							self   :    Dict		):
								A					       = 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 __a						(							self   :    List[str]		):
								A					       = Trie()
								trie.add('AB'		)
								trie.add('B'		)
								trie.add('C'		)
								self.assertEqual(trie.split('ABC'		)			,			['AB', 'C']		)







						def __a						(							self   :    List[Any]		):
								A					       = Trie()
								trie.add('ABC'		)
								trie.add('B'		)
								trie.add('CD'		)
								self.assertEqual(trie.split('ABCD'		)			,			['ABC', 'D']		)







						def __a						(							self   :    Union[str, Any]		):
								# Even if the offsets are wrong, we necessarily output correct string
								# parts.
								A					       = Trie()
								A					       = trie.cut_text('ABC'			,			[0, 0, 2, 1, 2, 3]		)
								self.assertEqual(_lowercase			,			['AB', 'C']		)









"""simple docstring"""


from dataclasses import dataclass, field
from typing import Tuple

from ..utils import cached_property, is_tf_available, logging, requires_backends
from .benchmark_args_utils import BenchmarkArguments


if is_tf_available():
	import tensorflow as tf


UpperCamelCase						:				Optional[Any]             =      logging.get_logger(__name__)



@dataclass
class 			lowerCamelCase__						(   UpperCAmelCase_			):
						lowerCAmelCase       =     [
						    """no_inference""",
						    """no_cuda""",
						    """no_tpu""",
						    """no_speed""",
						    """no_memory""",
						    """no_env_print""",
						    """no_multi_process""",
						]







						def __init__(							self   :    str			,			**_lowercase   :    Union[str, Any]		):
								for deprecated_arg in self.deprecated_args:
										if deprecated_arg in kwargs:
												A					       = deprecated_arg[3:]
												A					       = not kwargs.pop(_lowercase		)
												logger.warning(
												    f'{deprecated_arg} is depreciated. Please use --no-{positive_arg} or'
												    f' {positive_arg}={kwargs[positive_arg]}'		)
								A					       = kwargs.pop('tpu_name'			,			self.tpu_name		)
								A					       = kwargs.pop('device_idx'			,			self.device_idx		)
								A					       = kwargs.pop('eager_mode'			,			self.eager_mode		)
								A					       = kwargs.pop('use_xla'			,			self.use_xla		)
								super().__init__(**_lowercase		)

						lowerCAmelCase       =     field(
						    default=UpperCAmelCase_							,    metadata={"""help""": """Name of TPU"""}							,    )
						lowerCAmelCase       =     field(
						    default=0							,    metadata={"""help""": """CPU / GPU device index. Defaults to 0."""}							,    )
						lowerCAmelCase       =     field(default=UpperCAmelCase_							,    metadata={"""help""": """Benchmark models in eager model."""}			)
						lowerCAmelCase       =     field(
						    default=UpperCAmelCase_							,    metadata={
						        """help""": """Benchmark models using XLA JIT compilation. Note that `eager_model` has to be set to `False`."""
						    }							,    )







						@cached_property
						def __a						(							self   :    Optional[Any]		):
								requires_backends(self			,			['tf']		)
								A					       = None
								if self.tpu:
										try:
												if self.tpu_name:
														A					       = tf.distribute.cluster_resolver.TPUClusterResolver(self.tpu_name		)
												else:
														A					       = tf.distribute.cluster_resolver.TPUClusterResolver()
										except ValueError:
												A					       = None
								return tpu







						@cached_property
						def __a						(							self   :    Dict		):
								requires_backends(self			,			['tf']		)
								if self.is_tpu:
										tf.config.experimental_connect_to_cluster(self._setup_tpu		)
										tf.tpu.experimental.initialize_tpu_system(self._setup_tpu		)

										A					       = tf.distribute.TPUStrategy(self._setup_tpu		)
								else:
										# currently no multi gpu is allowed
										if self.is_gpu:
												# TODO: Currently only single GPU is supported
												tf.config.set_visible_devices(self.gpu_list[self.device_idx]			,			'GPU'		)
												A					       = tf.distribute.OneDeviceStrategy(device=f'/gpu:{self.device_idx}'		)
										else:
												tf.config.set_visible_devices([]			,			'GPU'		)  # disable GPU
												A					       = tf.distribute.OneDeviceStrategy(device=f'/cpu:{self.device_idx}'		)

								return strategy







						@property
						def __a						(							self   :    List[Any]		):
								requires_backends(self			,			['tf']		)
								return self._setup_tpu is not None







						@property
						def __a						(							self   :    Optional[Any]		):
								requires_backends(self			,			['tf']		)
								return self._setup_strategy







						@property
						def __a						(							self   :    str		):
								requires_backends(self			,			['tf']		)
								return tf.config.list_physical_devices('GPU'		)







						@property
						def __a						(							self   :    Any		):
								requires_backends(self			,			['tf']		)
								if self.cuda:
										return len(self.gpu_list		)
								return 0







						@property
						def __a						(							self   :    Dict		):
								return self.n_gpu > 0





