




'''simple docstring'''
import json
import os
import sys
import tempfile
import unittest
from pathlib import Path
from shutil import copyfile

from huggingface_hub import HfFolder, Repository, create_repo, delete_repo
from requests.exceptions import HTTPError

import transformers
from transformers import (
    CONFIG_MAPPING,
    FEATURE_EXTRACTOR_MAPPING,
    PROCESSOR_MAPPING,
    TOKENIZER_MAPPING,
    AutoConfig,
    AutoFeatureExtractor,
    AutoProcessor,
    AutoTokenizer,
    BertTokenizer,
    ProcessorMixin,
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
    WavaVecaProcessor,
)
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test
from transformers.tokenization_utils import TOKENIZER_CONFIG_FILE
from transformers.utils import FEATURE_EXTRACTOR_NAME, is_tokenizers_available


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

from test_module.custom_configuration import CustomConfig  # noqa E402
from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402
from test_module.custom_processing import CustomProcessor  # noqa E402
from test_module.custom_tokenization import CustomTokenizer  # noqa E402


_SCREAMING_SNAKE_CASE			=						get_tests_dir("fixtures/dummy_feature_extractor_config.json")
_SCREAMING_SNAKE_CASE			=						get_tests_dir("fixtures/vocab.json")
_SCREAMING_SNAKE_CASE			=						get_tests_dir("fixtures")
class   __UpperCAmelCase				(     unittest.TestCase				):



					'''simple docstring'''





					_UpperCamelCase							=		["""[UNK]""", """[CLS]""", """[SEP]""", """[PAD]""", """[MASK]""", """bla""", """blou"""]



					def     __snake_case						(	self     :						List[str])    ->							Tuple:
								A_									= 0



					def     __snake_case						(	self     :						Dict)    ->							Union[str, Any]:
								A_									= AutoProcessor.from_pretrained('facebook/wav2vec2-base-960h')
								self.assertIsInstance(A_      ,       A_)



					def     __snake_case						(	self     :						Tuple)    ->							Optional[Any]:
								with tempfile.TemporaryDirectory() as tmpdirname:
											A_									= WavaVecaConfig()
											A_									= AutoProcessor.from_pretrained('facebook/wav2vec2-base-960h')

											# save in new folder
											model_config.save_pretrained(A_)
											processor.save_pretrained(A_)

											A_									= AutoProcessor.from_pretrained(A_)

								self.assertIsInstance(A_      ,       A_)



					def     __snake_case						(	self     :						Optional[int])    ->							Tuple:
								with tempfile.TemporaryDirectory() as tmpdirname:
											# copy relevant files
											copyfile(A_      ,       os.path.join(A_      ,       A_))
											copyfile(A_      ,       os.path.join(A_      ,       'vocab.json'))

											A_									= AutoProcessor.from_pretrained(A_)

								self.assertIsInstance(A_      ,       A_)



					def     __snake_case						(	self     :						List[Any])    ->							List[Any]:
								with tempfile.TemporaryDirectory() as tmpdirname:
											A_									= WavaVecaFeatureExtractor()
											A_									= AutoTokenizer.from_pretrained('facebook/wav2vec2-base-960h')

											A_									= WavaVecaProcessor(A_      ,       A_)

											# save in new folder
											processor.save_pretrained(A_)

											# drop `processor_class` in tokenizer
											with open(os.path.join(A_      ,       A_)      ,       'r') as f:
														A_									= json.load(A_)
														config_dict.pop('processor_class')

											with open(os.path.join(A_      ,       A_)      ,       'w') as f:
														f.write(json.dumps(A_))

											A_									= AutoProcessor.from_pretrained(A_)

								self.assertIsInstance(A_      ,       A_)



					def     __snake_case						(	self     :						str)    ->							Any:
								with tempfile.TemporaryDirectory() as tmpdirname:
											A_									= WavaVecaFeatureExtractor()
											A_									= AutoTokenizer.from_pretrained('facebook/wav2vec2-base-960h')

											A_									= WavaVecaProcessor(A_      ,       A_)

											# save in new folder
											processor.save_pretrained(A_)

											# drop `processor_class` in feature extractor
											with open(os.path.join(A_      ,       A_)      ,       'r') as f:
														A_									= json.load(A_)
														config_dict.pop('processor_class')

											with open(os.path.join(A_      ,       A_)      ,       'w') as f:
														f.write(json.dumps(A_))

											A_									= AutoProcessor.from_pretrained(A_)

								self.assertIsInstance(A_      ,       A_)



					def     __snake_case						(	self     :						List[str])    ->							List[str]:
								with tempfile.TemporaryDirectory() as tmpdirname:
											A_									= WavaVecaConfig(processor_class='Wav2Vec2Processor')
											model_config.save_pretrained(A_)
											# copy relevant files
											copyfile(A_      ,       os.path.join(A_      ,       'vocab.json'))
											# create emtpy sample processor
											with open(os.path.join(A_      ,       A_)      ,       'w') as f:
														f.write('{}')

											A_									= AutoProcessor.from_pretrained(A_)

								self.assertIsInstance(A_      ,       A_)



					def     __snake_case						(	self     :						Dict)    ->							int:
								# If remote code is not set, we will time out when asking whether to load the model.
								with self.assertRaises(A_):
											A_									= AutoProcessor.from_pretrained('hf-internal-testing/test_dynamic_processor')
								# If remote code is disabled, we can't load this config.
								with self.assertRaises(A_):
											A_									= AutoProcessor.from_pretrained(
											    'hf-internal-testing/test_dynamic_processor'      ,       trust_remote_code=A_)

								A_									= AutoProcessor.from_pretrained('hf-internal-testing/test_dynamic_processor'      ,       trust_remote_code=A_)
								self.assertTrue(processor.special_attribute_present)
								self.assertEqual(processor.__class__.__name__      ,       'NewProcessor')

								A_									= processor.feature_extractor
								self.assertTrue(feature_extractor.special_attribute_present)
								self.assertEqual(feature_extractor.__class__.__name__      ,       'NewFeatureExtractor')

								A_									= processor.tokenizer
								self.assertTrue(tokenizer.special_attribute_present)
								if is_tokenizers_available():
											self.assertEqual(tokenizer.__class__.__name__      ,       'NewTokenizerFast')

											# Test we can also load the slow version
											A_									= AutoProcessor.from_pretrained(
											    'hf-internal-testing/test_dynamic_processor'      ,       trust_remote_code=A_      ,       use_fast=A_)
											A_									= new_processor.tokenizer
											self.assertTrue(new_tokenizer.special_attribute_present)
											self.assertEqual(new_tokenizer.__class__.__name__      ,       'NewTokenizer')
								else:
											self.assertEqual(tokenizer.__class__.__name__      ,       'NewTokenizer')



					def     __snake_case						(	self     :						List[str])    ->							List[Any]:
								try:
											AutoConfig.register('custom'      ,       A_)
											AutoFeatureExtractor.register(A_      ,       A_)
											AutoTokenizer.register(A_      ,       slow_tokenizer_class=A_)
											AutoProcessor.register(A_      ,       A_)
											# Trying to register something existing in the Transformers library will raise an error
											with self.assertRaises(A_):
														AutoProcessor.register(A_      ,       A_)

											# Now that the config is registered, it can be used as any other config with the auto-API
											A_									= CustomFeatureExtractor.from_pretrained(A_)

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

											A_									= CustomProcessor(A_      ,       A_)

											with tempfile.TemporaryDirectory() as tmp_dir:
														processor.save_pretrained(A_)
														A_									= AutoProcessor.from_pretrained(A_)
														self.assertIsInstance(A_      ,       A_)

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



					def     __snake_case						(	self     :						List[str])    ->							List[Any]:
								class   __UpperCAmelCase				(     lowerCAmelCase				):



													'''simple docstring'''





													_UpperCamelCase							=		False





								class   __UpperCAmelCase				(     lowerCAmelCase				):



													'''simple docstring'''





													_UpperCamelCase							=		False





								class   __UpperCAmelCase				(     lowerCAmelCase				):



													'''simple docstring'''





													_UpperCamelCase							=		"""AutoFeatureExtractor"""
													_UpperCamelCase							=		"""AutoTokenizer"""
													_UpperCamelCase							=		False

								try:
											AutoConfig.register('custom'      ,       A_)
											AutoFeatureExtractor.register(A_      ,       A_)
											AutoTokenizer.register(A_      ,       slow_tokenizer_class=A_)
											AutoProcessor.register(A_      ,       A_)
											# If remote code is not set, the default is to use local classes.
											A_									= AutoProcessor.from_pretrained('hf-internal-testing/test_dynamic_processor')
											self.assertEqual(processor.__class__.__name__      ,       'NewProcessor')
											self.assertFalse(processor.special_attribute_present)
											self.assertFalse(processor.feature_extractor.special_attribute_present)
											self.assertFalse(processor.tokenizer.special_attribute_present)

											# If remote code is disabled, we load the local ones.
											A_									= AutoProcessor.from_pretrained(
											    'hf-internal-testing/test_dynamic_processor'      ,       trust_remote_code=A_)
											self.assertEqual(processor.__class__.__name__      ,       'NewProcessor')
											self.assertFalse(processor.special_attribute_present)
											self.assertFalse(processor.feature_extractor.special_attribute_present)
											self.assertFalse(processor.tokenizer.special_attribute_present)

											# If remote is enabled, we load from the Hub.
											A_									= AutoProcessor.from_pretrained(
											    'hf-internal-testing/test_dynamic_processor'      ,       trust_remote_code=A_)
											self.assertEqual(processor.__class__.__name__      ,       'NewProcessor')
											self.assertTrue(processor.special_attribute_present)
											self.assertTrue(processor.feature_extractor.special_attribute_present)
											self.assertTrue(processor.tokenizer.special_attribute_present)

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



					def     __snake_case						(	self     :						List[Any])    ->							Any:
								A_									= AutoProcessor.from_pretrained('hf-internal-testing/tiny-random-bert')
								self.assertEqual(processor.__class__.__name__      ,       'BertTokenizerFast')



					def     __snake_case						(	self     :						Tuple)    ->							List[Any]:
								A_									= AutoProcessor.from_pretrained('hf-internal-testing/tiny-random-convnext')
								self.assertEqual(processor.__class__.__name__      ,       'ConvNextImageProcessor')





@is_staging_test
class   __UpperCAmelCase				(     unittest.TestCase				):



					'''simple docstring'''





					_UpperCamelCase							=		["""[UNK]""", """[CLS]""", """[SEP]""", """[PAD]""", """[MASK]""", """bla""", """blou"""]



					@classmethod
					def     __snake_case						(	cls     :						Tuple)    ->							Union[str, Any]:
								A_									= TOKEN
								HfFolder.save_token(A_)



					@classmethod
					def     __snake_case						(	cls     :						int)    ->							List[str]:
								try:
											delete_repo(token=cls._token      ,       repo_id='test-processor')
								except HTTPError:
											pass

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

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



					def     __snake_case						(	self     :						int)    ->							Dict:
								A_									= WavaVecaProcessor.from_pretrained(A_)
								with tempfile.TemporaryDirectory() as tmp_dir:
											processor.save_pretrained(
											    os.path.join(A_      ,       'test-processor')      ,       push_to_hub=A_      ,       use_auth_token=self._token)

											A_									= WavaVecaProcessor.from_pretrained(F'{USER}/test-processor')
											for k, v in processor.feature_extractor.__dict__.items():
														self.assertEqual(A_      ,       getattr(new_processor.feature_extractor      ,       A_))
											self.assertDictEqual(new_processor.tokenizer.get_vocab()      ,       processor.tokenizer.get_vocab())



					def     __snake_case						(	self     :						Tuple)    ->							Optional[int]:
								A_									= WavaVecaProcessor.from_pretrained(A_)

								with tempfile.TemporaryDirectory() as tmp_dir:
											processor.save_pretrained(
											    os.path.join(A_      ,       'test-processor-org')      ,       push_to_hub=A_      ,       use_auth_token=self._token      ,       organization='valid_org'      ,       )

											A_									= WavaVecaProcessor.from_pretrained('valid_org/test-processor-org')
											for k, v in processor.feature_extractor.__dict__.items():
														self.assertEqual(A_      ,       getattr(new_processor.feature_extractor      ,       A_))
											self.assertDictEqual(new_processor.tokenizer.get_vocab()      ,       processor.tokenizer.get_vocab())



					def     __snake_case						(	self     :						Dict)    ->							Dict:
								CustomFeatureExtractor.register_for_auto_class()
								CustomTokenizer.register_for_auto_class()
								CustomProcessor.register_for_auto_class()

								A_									= CustomFeatureExtractor.from_pretrained(A_)

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

								A_									= CustomProcessor(A_      ,       A_)

								with tempfile.TemporaryDirectory() as tmp_dir:
											create_repo(F'{USER}/test-dynamic-processor'      ,       token=self._token)
											A_									= Repository(A_      ,       clone_from=F'{USER}/test-dynamic-processor'      ,       token=self._token)
											processor.save_pretrained(A_)

											# This has added the proper auto_map field to the feature extractor config
											self.assertDictEqual(
											    processor.feature_extractor.auto_map      ,       {
											        'AutoFeatureExtractor': 'custom_feature_extraction.CustomFeatureExtractor',
											        'AutoProcessor': 'custom_processing.CustomProcessor',
											    }      ,       )

											# This has added the proper auto_map field to the tokenizer config
											with open(os.path.join(A_      ,       'tokenizer_config.json')) as f:
														A_									= json.load(A_)
											self.assertDictEqual(
											    tokenizer_config['auto_map']      ,       {
											        'AutoTokenizer': ['custom_tokenization.CustomTokenizer', None],
											        'AutoProcessor': 'custom_processing.CustomProcessor',
											    }      ,       )

											# The code has been copied from fixtures
											self.assertTrue(os.path.isfile(os.path.join(A_      ,       'custom_feature_extraction.py')))
											self.assertTrue(os.path.isfile(os.path.join(A_      ,       'custom_tokenization.py')))
											self.assertTrue(os.path.isfile(os.path.join(A_      ,       'custom_processing.py')))

											repo.push_to_hub()

								A_									= AutoProcessor.from_pretrained(F'{USER}/test-dynamic-processor'      ,       trust_remote_code=A_)
								# Can't make an isinstance check because the new_processor is from the CustomProcessor class of a dynamic module
								self.assertEqual(new_processor.__class__.__name__      ,       'CustomProcessor')











from collections import OrderedDict
from typing import Mapping

from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfig
from ...utils import logging


UpperCamelCase_												=	logging.get_logger(__name__)

UpperCamelCase_												=	{
    'junnyu/roformer_chinese_small': 'https://huggingface.co/junnyu/roformer_chinese_small/resolve/main/config.json',
    'junnyu/roformer_chinese_base': 'https://huggingface.co/junnyu/roformer_chinese_base/resolve/main/config.json',
    'junnyu/roformer_chinese_char_small': (
        'https://huggingface.co/junnyu/roformer_chinese_char_small/resolve/main/config.json'
    ),
    'junnyu/roformer_chinese_char_base': (
        'https://huggingface.co/junnyu/roformer_chinese_char_base/resolve/main/config.json'
    ),
    'junnyu/roformer_small_discriminator': (
        'https://huggingface.co/junnyu/roformer_small_discriminator/resolve/main/config.json'
    ),
    'junnyu/roformer_small_generator': (
        'https://huggingface.co/junnyu/roformer_small_generator/resolve/main/config.json'
    ),
    # See all RoFormer models at https://huggingface.co/models?filter=roformer
}

class     snake_case_		(   a							):






 '''simple docstring'''





 __UpperCamelCase       =      'roformer'

 def __init__(  self,		A_=5_0000,		A_=None,		A_=768,		A_=12,		A_=12,		A_=3072,		A_="gelu",		A_=0.1,		A_=0.1,		A_=1536,		A_=2,		A_=0.02,		A_=1E-12,		A_=0,		A_=False,		A_=True,		**A_,		)	->       Dict:
    super().__init__(pad_token_id=A_,		**A_		)

    UpperCAmelCase__						=vocab_size
    UpperCAmelCase__						=hidden_size if embedding_size is None else embedding_size
    UpperCAmelCase__						=hidden_size
    UpperCAmelCase__						=num_hidden_layers
    UpperCAmelCase__						=num_attention_heads
    UpperCAmelCase__						=hidden_act
    UpperCAmelCase__						=intermediate_size
    UpperCAmelCase__						=hidden_dropout_prob
    UpperCAmelCase__						=attention_probs_dropout_prob
    UpperCAmelCase__						=max_position_embeddings
    UpperCAmelCase__						=type_vocab_size
    UpperCAmelCase__						=initializer_range
    UpperCAmelCase__						=layer_norm_eps
    UpperCAmelCase__						=rotary_value
    UpperCAmelCase__						=use_cache

class     snake_case_		(   a							):






 '''simple docstring'''





 @property
 def  __UpperCAmelCase						(  self		)	->       Mapping[str, Mapping[int, str]]:
    if self.task == "multiple-choice":
       UpperCAmelCase__						={0: "batch", 1: "choice", 2: "sequence"}
    else:
       UpperCAmelCase__						={0: "batch", 1: "sequence"}
    UpperCAmelCase__						={0: "batch", 1: "sequence"}
    return OrderedDict(
        [
            ("input_ids", dynamic_axis),
            ("attention_mask", dynamic_axis),
            ("token_type_ids", dynamic_axis),
        ]		)



