


'''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 	a__       (							unittest.TestCase						):





							'''simple docstring'''






							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							List[str]:
														lowerCAmelCase__      =			0







							@slow
							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							Union[str, Any]:
														for model_name in (x for x in BERT_PRETRAINED_CONFIG_ARCHIVE_MAP.keys() if "japanese" not in x):
																					lowerCAmelCase__      =			AutoTokenizer.from_pretrained(lowerCamelCase_     )
																					self.assertIsNotNone(lowerCamelCase_     )
																					self.assertIsInstance(lowerCamelCase_      ,					(BertTokenizer, BertTokenizerFast)     )
																					self.assertGreater(len(lowerCamelCase_     )      ,					0     )

														for model_name in GPT2_PRETRAINED_CONFIG_ARCHIVE_MAP.keys():
																					lowerCAmelCase__      =			AutoTokenizer.from_pretrained(lowerCamelCase_     )
																					self.assertIsNotNone(lowerCamelCase_     )
																					self.assertIsInstance(lowerCamelCase_      ,					(GPTaTokenizer, GPTaTokenizerFast)     )
																					self.assertGreater(len(lowerCamelCase_     )      ,					0     )







							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							Union[str, Any]:
														lowerCAmelCase__      =			AutoTokenizer.from_pretrained(lowerCamelCase_     )
														self.assertIsInstance(lowerCamelCase_      ,					(BertTokenizer, BertTokenizerFast)     )
														self.assertEqual(tokenizer.vocab_size      ,					12     )







							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							Dict:
														lowerCAmelCase__      =			AutoTokenizer.from_pretrained(lowerCamelCase_     )
														self.assertIsInstance(lowerCamelCase_      ,					(RobertaTokenizer, RobertaTokenizerFast)     )
														self.assertEqual(tokenizer.vocab_size      ,					20     )







							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							Union[str, Any]:
														lowerCAmelCase__      =			AutoConfig.from_pretrained(lowerCamelCase_     )
														self.assertIsInstance(lowerCamelCase_      ,					lowerCamelCase_     )
														# Check that tokenizer_type ≠ model_type
														lowerCAmelCase__      =			AutoTokenizer.from_pretrained(lowerCamelCase_      ,					config=lowerCamelCase_     )
														self.assertIsInstance(lowerCamelCase_      ,					(BertTokenizer, BertTokenizerFast)     )
														self.assertEqual(tokenizer.vocab_size      ,					12     )







							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							Tuple:
														with tempfile.TemporaryDirectory() as tmp_dir:
																					shutil.copy('''./tests/fixtures/vocab.txt'''      ,					os.path.join(lowerCamelCase_      ,					'''vocab.txt'''     )     )

																					lowerCAmelCase__      =			AutoTokenizer.from_pretrained(lowerCamelCase_      ,					tokenizer_type='''bert'''      ,					use_fast=lowerCamelCase_     )
																					self.assertIsInstance(lowerCamelCase_      ,					lowerCamelCase_     )

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

																					lowerCAmelCase__      =			AutoTokenizer.from_pretrained(lowerCamelCase_      ,					tokenizer_type='''gpt2'''      ,					use_fast=lowerCamelCase_     )
																					self.assertIsInstance(lowerCamelCase_      ,					lowerCamelCase_     )







							@require_tokenizers
							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							Optional[int]:
														with tempfile.TemporaryDirectory() as tmp_dir:
																					shutil.copy('''./tests/fixtures/vocab.txt'''      ,					os.path.join(lowerCamelCase_      ,					'''vocab.txt'''     )     )

																					lowerCAmelCase__      =			AutoTokenizer.from_pretrained(lowerCamelCase_      ,					tokenizer_type='''bert'''     )
																					self.assertIsInstance(lowerCamelCase_      ,					lowerCamelCase_     )

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

																					lowerCAmelCase__      =			AutoTokenizer.from_pretrained(lowerCamelCase_      ,					tokenizer_type='''gpt2'''     )
																					self.assertIsInstance(lowerCamelCase_      ,					lowerCamelCase_     )







							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							Union[str, Any]:
														with pytest.raises(lowerCamelCase_     ):
																					AutoTokenizer.from_pretrained('''./'''      ,					tokenizer_type='''xxx'''     )







							@require_tokenizers
							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							Optional[int]:
														for tokenizer_class in [BertTokenizer, BertTokenizerFast, AutoTokenizer]:
																					lowerCAmelCase__      =			tokenizer_class.from_pretrained('''wietsedv/bert-base-dutch-cased'''     )
																					self.assertIsInstance(lowerCamelCase_      ,					(BertTokenizer, BertTokenizerFast)     )

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

																					self.assertEqual(tokenizer.model_max_length      ,					5_12     )







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







							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							int:
														# 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__      =			TOKENIZER_MAPPING.values()
														lowerCAmelCase__      =			[]

														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(lowerCamelCase_     )







							@require_tokenizers
							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							List[Any]:
														self.assertIsInstance(AutoTokenizer.from_pretrained('''bert-base-cased'''      ,					use_fast=lowerCamelCase_     )      ,					lowerCamelCase_     )
														self.assertIsInstance(AutoTokenizer.from_pretrained('''bert-base-cased'''     )      ,					lowerCamelCase_     )







							@require_tokenizers
							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							Union[str, Any]:
														lowerCAmelCase__      =			AutoTokenizer.from_pretrained('''distilbert-base-uncased'''      ,					do_lower_case=lowerCamelCase_     )
														lowerCAmelCase__      =			'''Hello, world. How are you?'''
														lowerCAmelCase__      =			tokenizer.tokenize(lowerCamelCase_     )
														self.assertEqual('''[UNK]'''      ,					tokens[0]     )

														lowerCAmelCase__      =			AutoTokenizer.from_pretrained('''microsoft/mpnet-base'''      ,					do_lower_case=lowerCamelCase_     )
														lowerCAmelCase__      =			tokenizer.tokenize(lowerCamelCase_     )
														self.assertEqual('''[UNK]'''      ,					tokens[0]     )







							@require_tokenizers
							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							int:
														lowerCAmelCase__      =			AutoTokenizer.from_pretrained('''robot-test/dummy-tokenizer-fast-with-model-config'''     )
														self.assertEqual(type(lowerCamelCase_     )      ,					lowerCamelCase_     )
														self.assertEqual(tokenizer.model_max_length      ,					5_12     )
														self.assertEqual(tokenizer.vocab_size      ,					3_00_00     )
														self.assertEqual(tokenizer.unk_token      ,					'''[UNK]'''     )
														self.assertEqual(tokenizer.padding_side      ,					'''right'''     )
														self.assertEqual(tokenizer.truncation_side      ,					'''right'''     )







							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							Dict:
														lowerCAmelCase__      =			AutoTokenizer.from_pretrained(lowerCamelCase_     )
														self.assertIsInstance(lowerCamelCase_      ,					(BertTokenizer, BertTokenizerFast)     )
														with tempfile.TemporaryDirectory() as tmp_dir:
																					tokenizer.save_pretrained(lowerCamelCase_     )
																					lowerCAmelCase__      =			AutoTokenizer.from_pretrained(lowerCamelCase_     )

														self.assertIsInstance(lowerCamelCase_      ,					tokenizer.__class__     )
														self.assertEqual(tokenizera.vocab_size      ,					12     )







							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							List[Any]:
														lowerCAmelCase__      =			AutoTokenizer.from_pretrained('''ctrl'''     )
														# There is no fast CTRL so this always gives us a slow tokenizer.
														self.assertIsInstance(lowerCamelCase_      ,					lowerCamelCase_     )







							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							Dict:
														# Check we can load the tokenizer config of an online model.
														lowerCAmelCase__      =			get_tokenizer_config('''bert-base-cased'''     )
														lowerCAmelCase__      =			config.pop('''_commit_hash'''      ,					lowerCamelCase_     )
														# If we ever update bert-base-cased tokenizer config, this dict here will need to be updated.
														self.assertEqual(lowerCamelCase_      ,					{'''do_lower_case''': False}     )

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

														# A tokenizer saved with `save_pretrained` always creates a tokenizer config.
														lowerCAmelCase__      =			AutoTokenizer.from_pretrained(lowerCamelCase_     )
														with tempfile.TemporaryDirectory() as tmp_dir:
																					tokenizer.save_pretrained(lowerCamelCase_     )
																					lowerCAmelCase__      =			get_tokenizer_config(lowerCamelCase_     )

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







							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							Any:
														try:
																					AutoConfig.register('''custom'''      ,					lowerCamelCase_     )

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

																					lowerCAmelCase__      =			CustomTokenizer.from_pretrained(lowerCamelCase_     )
																					with tempfile.TemporaryDirectory() as tmp_dir:
																												tokenizer.save_pretrained(lowerCamelCase_     )

																												lowerCAmelCase__      =			AutoTokenizer.from_pretrained(lowerCamelCase_     )
																												self.assertIsInstance(lowerCamelCase_      ,					lowerCamelCase_     )

														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 						__SCREAMING_SNAKE_CASE							(	self     ) ->							Any:
														try:
																					AutoConfig.register('''custom'''      ,					lowerCamelCase_     )

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

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

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

																					# 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__      =			BertTokenizerFast.from_pretrained(lowerCamelCase_     )
																												bert_tokenizer.save_pretrained(lowerCamelCase_     )
																												lowerCAmelCase__      =			CustomTokenizerFast.from_pretrained(lowerCamelCase_     )

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

																												lowerCAmelCase__      =			AutoTokenizer.from_pretrained(lowerCamelCase_     )
																												self.assertIsInstance(lowerCamelCase_      ,					lowerCamelCase_     )

																												lowerCAmelCase__      =			AutoTokenizer.from_pretrained(lowerCamelCase_      ,					use_fast=lowerCamelCase_     )
																												self.assertIsInstance(lowerCamelCase_      ,					lowerCamelCase_     )

														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 						__SCREAMING_SNAKE_CASE							(	self     ) ->							str:
														# If remote code is not set, we will time out when asking whether to load the model.
														with self.assertRaises(lowerCamelCase_     ):
																					lowerCAmelCase__      =			AutoTokenizer.from_pretrained('''hf-internal-testing/test_dynamic_tokenizer'''     )
														# If remote code is disabled, we can't load this config.
														with self.assertRaises(lowerCamelCase_     ):
																					lowerCAmelCase__      =			AutoTokenizer.from_pretrained(
																					    '''hf-internal-testing/test_dynamic_tokenizer'''      ,					trust_remote_code=lowerCamelCase_     )

														lowerCAmelCase__      =			AutoTokenizer.from_pretrained('''hf-internal-testing/test_dynamic_tokenizer'''      ,					trust_remote_code=lowerCamelCase_     )
														self.assertTrue(tokenizer.special_attribute_present     )
														# Test tokenizer can be reloaded.
														with tempfile.TemporaryDirectory() as tmp_dir:
																					tokenizer.save_pretrained(lowerCamelCase_     )
																					lowerCAmelCase__      =			AutoTokenizer.from_pretrained(lowerCamelCase_      ,					trust_remote_code=lowerCamelCase_     )
														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__      =			AutoTokenizer.from_pretrained(
																					    '''hf-internal-testing/test_dynamic_tokenizer'''      ,					trust_remote_code=lowerCamelCase_      ,					use_fast=lowerCamelCase_     )
																					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(lowerCamelCase_     )
																												lowerCAmelCase__      =			AutoTokenizer.from_pretrained(lowerCamelCase_      ,					trust_remote_code=lowerCamelCase_      ,					use_fast=lowerCamelCase_     )
																					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 						__SCREAMING_SNAKE_CASE							(	self     ) ->							Optional[Any]:




														class 	a__       (							a__						):





																					'''simple docstring'''






																					lowercase__    :     List[str]					=   False




														class 	a__       (							a__						):





																					'''simple docstring'''






																					lowercase__    :     Any					=   NewTokenizer
																					lowercase__    :     Optional[int]					=   False

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

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

																					# If remote is enabled, we load from the Hub
																					lowerCAmelCase__      =			AutoTokenizer.from_pretrained(
																					    '''hf-internal-testing/test_dynamic_tokenizer'''      ,					trust_remote_code=lowerCamelCase_     )
																					self.assertEqual(tokenizer.__class__.__name__      ,					'''NewTokenizerFast'''     )
																					self.assertTrue(tokenizer.special_attribute_present     )
																					lowerCAmelCase__      =			AutoTokenizer.from_pretrained(
																					    '''hf-internal-testing/test_dynamic_tokenizer'''      ,					trust_remote_code=lowerCamelCase_      ,					use_fast=lowerCamelCase_     )
																					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 						__SCREAMING_SNAKE_CASE							(	self     ) ->							int:
														lowerCAmelCase__      =			AutoTokenizer.from_pretrained(
														    '''hf-internal-testing/test_dynamic_tokenizer_legacy'''      ,					trust_remote_code=lowerCamelCase_     )
														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__      =			AutoTokenizer.from_pretrained(
																					    '''hf-internal-testing/test_dynamic_tokenizer_legacy'''      ,					trust_remote_code=lowerCamelCase_      ,					use_fast=lowerCamelCase_     )
																					self.assertTrue(tokenizer.special_attribute_present     )
																					self.assertEqual(tokenizer.__class__.__name__      ,					'''NewTokenizer'''     )
														else:
																					self.assertEqual(tokenizer.__class__.__name__      ,					'''NewTokenizer'''     )







							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							str:
														with self.assertRaisesRegex(
														    lowerCamelCase_      ,					'''bert-base is not a local folder and is not a valid model identifier'''     ):
																					lowerCAmelCase__      =			AutoTokenizer.from_pretrained('''bert-base'''     )







							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							Optional[int]:
														with self.assertRaisesRegex(
														    lowerCamelCase_      ,					r'''aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)'''     ):
																					lowerCAmelCase__      =			AutoTokenizer.from_pretrained(lowerCamelCase_      ,					revision='''aaaaaa'''     )







							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							Tuple:
														# Make sure we have cached the tokenizer.
														lowerCAmelCase__      =			AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-bert'''     )
														with RequestCounter() as counter:
																					lowerCAmelCase__      =			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     )



'''simple docstring'''






import logging
import os
from typing import List, Tuple

import numpy as np
import psutil
import torch
import torch.distributed as dist

from transformers import RagRetriever


__UpperCAmelCase					       =							logging.getLogger(__name__)




class 	a__       (							a__						):





							'''simple docstring'''






							def __init__(	self      ,					lowerCamelCase_      ,					lowerCamelCase_      ,					lowerCamelCase_      ,					lowerCamelCase_=None     ) ->							Dict:
														super().__init__(
														    lowerCamelCase_      ,					question_encoder_tokenizer=lowerCamelCase_      ,					generator_tokenizer=lowerCamelCase_      ,					index=lowerCamelCase_      ,					init_retrieval=lowerCamelCase_      ,					)
														lowerCAmelCase__      =			None







							def 						__SCREAMING_SNAKE_CASE							(	self      ,					lowerCamelCase_     ) ->							str:
														logger.info('''initializing retrieval'''     )

														# initializing a separate process group for retrieval as the default
														# nccl backend doesn't support gather/scatter operations while gloo
														# is too slow to replace nccl for the core gpu communication
														if dist.is_initialized():
																					logger.info('''dist initialized'''     )
																					# needs to be set manually
																					lowerCAmelCase__      =			self._infer_socket_ifname()
																					# avoid clash with the NCCL port
																					lowerCAmelCase__      =			str(distributed_port + 1     )
																					lowerCAmelCase__      =			dist.new_group(ranks=lowerCamelCase_      ,					backend='''gloo'''     )

														# initialize retriever only on the main worker
														if not dist.is_initialized() or self._is_main():
																					logger.info('''dist not initialized / main'''     )
																					self.index.init_index()

														# all processes wait untill the retriever is initialized by the main process
														if dist.is_initialized():
																					torch.distributed.barrier(group=self.process_group     )







							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							Optional[Any]:
														return dist.get_rank(group=self.process_group     ) == 0







							def 						__SCREAMING_SNAKE_CASE							(	self      ,					lowerCamelCase_      ,					lowerCamelCase_      ,					lowerCamelCase_=torch.floataa     ) ->							Union[str, Any]:
														lowerCAmelCase__      =			torch.empty(lowerCamelCase_      ,					dtype=lowerCamelCase_     )
														dist.scatter(lowerCamelCase_      ,					src=0      ,					scatter_list=lowerCamelCase_      ,					group=self.process_group     )
														return target_tensor







							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							str:
														lowerCAmelCase__      =			psutil.net_if_addrs()
														# a hacky way to deal with varying network interface names
														lowerCAmelCase__      =			next((addr for addr in addrs if addr.startswith('''e'''     ))      ,					lowerCamelCase_     )
														return ifname







							def 						__SCREAMING_SNAKE_CASE							(	self      ,					lowerCamelCase_      ,					lowerCamelCase_     ) ->							Tuple[np.ndarray, List[dict]]:

														# single GPU training
														if not dist.is_initialized():
																					lowerCAmelCase__     ,      lowerCAmelCase__      =			self._main_retrieve(lowerCamelCase_      ,					lowerCamelCase_     )
																					return retrieved_doc_embeds, doc_ids, self.index.get_doc_dicts(lowerCamelCase_     )

														# distributed training
														lowerCAmelCase__      =			dist.get_world_size(group=self.process_group     )

														# gather logic
														lowerCAmelCase__      =			None
														if self._is_main():
																					lowerCAmelCase__      =			[torch.empty(question_hidden_states.shape      ,					dtype=torch.floataa     ) for _ in range(lowerCamelCase_     )]
														dist.gather(torch.tensor(lowerCamelCase_     )      ,					dst=0      ,					gather_list=lowerCamelCase_      ,					group=self.process_group     )

														# scatter logic
														lowerCAmelCase__      =			question_hidden_states.shape[0]
														lowerCAmelCase__      =			[]
														lowerCAmelCase__      =			[]
														if self._is_main():
																					assert len(lowerCamelCase_     ) == world_size
																					lowerCAmelCase__     ,      lowerCAmelCase__      =			self._main_retrieve(torch.cat(lowerCamelCase_     ).numpy()      ,					lowerCamelCase_     )
																					lowerCAmelCase__     ,      lowerCAmelCase__      =			torch.tensor(lowerCamelCase_     ), torch.tensor(lowerCamelCase_     )
																					lowerCAmelCase__      =			self._chunk_tensor(lowerCamelCase_      ,					lowerCamelCase_     )
																					lowerCAmelCase__      =			self._chunk_tensor(lowerCamelCase_      ,					lowerCamelCase_     )
														lowerCAmelCase__      =			self._scattered(lowerCamelCase_      ,					[n_queries, n_docs]      ,					target_type=torch.intaa     )
														lowerCAmelCase__      =			self._scattered(lowerCamelCase_      ,					[n_queries, n_docs, question_hidden_states.shape[1]]     )

														return retrieved_doc_embeds.numpy(), doc_ids.numpy(), self.index.get_doc_dicts(lowerCamelCase_     )