


'''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


__A	=get_tests_dir('fixtures/dummy_feature_extractor_config.json')
__A	=get_tests_dir('fixtures/vocab.json')
__A	=get_tests_dir('fixtures')


class 		_snake_case					(     unittest.TestCase	):
					lowerCAmelCase      :List[str]   =					['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']




					def 				snake_case__						(      self):
								UpperCAmelCase__  :					Dict        =     0




					def 				snake_case__						(      self):
								UpperCAmelCase__  :					List[str]        =     AutoProcessor.from_pretrained("""facebook/wav2vec2-base-960h""")
								self.assertIsInstance(_lowerCamelCase							,							_lowerCamelCase)




					def 				snake_case__						(      self):
								with tempfile.TemporaryDirectory() as tmpdirname:
											UpperCAmelCase__  :					Tuple        =     WavaVecaConfig()
											UpperCAmelCase__  :					str        =     AutoProcessor.from_pretrained("""facebook/wav2vec2-base-960h""")

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

											UpperCAmelCase__  :					List[Any]        =     AutoProcessor.from_pretrained(_lowerCamelCase)

								self.assertIsInstance(_lowerCamelCase							,							_lowerCamelCase)




					def 				snake_case__						(      self):
								with tempfile.TemporaryDirectory() as tmpdirname:
											# copy relevant files
											copyfile(_lowerCamelCase							,							os.path.join(_lowerCamelCase							,							_lowerCamelCase))
											copyfile(_lowerCamelCase							,							os.path.join(_lowerCamelCase							,							"""vocab.json"""))

											UpperCAmelCase__  :					List[str]        =     AutoProcessor.from_pretrained(_lowerCamelCase)

								self.assertIsInstance(_lowerCamelCase							,							_lowerCamelCase)




					def 				snake_case__						(      self):
								with tempfile.TemporaryDirectory() as tmpdirname:
											UpperCAmelCase__  :					Tuple        =     WavaVecaFeatureExtractor()
											UpperCAmelCase__  :					int        =     AutoTokenizer.from_pretrained("""facebook/wav2vec2-base-960h""")

											UpperCAmelCase__  :					str        =     WavaVecaProcessor(_lowerCamelCase							,							_lowerCamelCase)

											# save in new folder
											processor.save_pretrained(_lowerCamelCase)

											# drop `processor_class` in tokenizer
											with open(os.path.join(_lowerCamelCase							,							_lowerCamelCase)							,							"""r""") as f:
														UpperCAmelCase__  :					Union[str, Any]        =     json.load(_lowerCamelCase)
														config_dict.pop("""processor_class""")

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

											UpperCAmelCase__  :					Dict        =     AutoProcessor.from_pretrained(_lowerCamelCase)

								self.assertIsInstance(_lowerCamelCase							,							_lowerCamelCase)




					def 				snake_case__						(      self):
								with tempfile.TemporaryDirectory() as tmpdirname:
											UpperCAmelCase__  :					str        =     WavaVecaFeatureExtractor()
											UpperCAmelCase__  :					List[Any]        =     AutoTokenizer.from_pretrained("""facebook/wav2vec2-base-960h""")

											UpperCAmelCase__  :					Dict        =     WavaVecaProcessor(_lowerCamelCase							,							_lowerCamelCase)

											# save in new folder
											processor.save_pretrained(_lowerCamelCase)

											# drop `processor_class` in feature extractor
											with open(os.path.join(_lowerCamelCase							,							_lowerCamelCase)							,							"""r""") as f:
														UpperCAmelCase__  :					Optional[Any]        =     json.load(_lowerCamelCase)
														config_dict.pop("""processor_class""")

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

											UpperCAmelCase__  :					Dict        =     AutoProcessor.from_pretrained(_lowerCamelCase)

								self.assertIsInstance(_lowerCamelCase							,							_lowerCamelCase)




					def 				snake_case__						(      self):
								with tempfile.TemporaryDirectory() as tmpdirname:
											UpperCAmelCase__  :					str        =     WavaVecaConfig(processor_class="""Wav2Vec2Processor""")
											model_config.save_pretrained(_lowerCamelCase)
											# copy relevant files
											copyfile(_lowerCamelCase							,							os.path.join(_lowerCamelCase							,							"""vocab.json"""))
											# create emtpy sample processor
											with open(os.path.join(_lowerCamelCase							,							_lowerCamelCase)							,							"""w""") as f:
														f.write("""{}""")

											UpperCAmelCase__  :					Optional[Any]        =     AutoProcessor.from_pretrained(_lowerCamelCase)

								self.assertIsInstance(_lowerCamelCase							,							_lowerCamelCase)




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

								UpperCAmelCase__  :					Optional[Any]        =     AutoProcessor.from_pretrained("""hf-internal-testing/test_dynamic_processor"""							,							trust_remote_code=_lowerCamelCase)
								self.assertTrue(processor.special_attribute_present)
								self.assertEqual(processor.__class__.__name__							,							"""NewProcessor""")

								UpperCAmelCase__  :					str        =     processor.feature_extractor
								self.assertTrue(feature_extractor.special_attribute_present)
								self.assertEqual(feature_extractor.__class__.__name__							,							"""NewFeatureExtractor""")

								UpperCAmelCase__  :					int        =     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
											UpperCAmelCase__  :					List[str]        =     AutoProcessor.from_pretrained(
											    """hf-internal-testing/test_dynamic_processor"""							,							trust_remote_code=_lowerCamelCase							,							use_fast=_lowerCamelCase)
											UpperCAmelCase__  :					str        =     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):
								try:
											AutoConfig.register("""custom"""							,							_lowerCamelCase)
											AutoFeatureExtractor.register(_lowerCamelCase							,							_lowerCamelCase)
											AutoTokenizer.register(_lowerCamelCase							,							slow_tokenizer_class=_lowerCamelCase)
											AutoProcessor.register(_lowerCamelCase							,							_lowerCamelCase)
											# Trying to register something existing in the Transformers library will raise an error
											with self.assertRaises(_lowerCamelCase):
														AutoProcessor.register(_lowerCamelCase							,							_lowerCamelCase)

											# Now that the config is registered, it can be used as any other config with the auto-API
											UpperCAmelCase__  :					Union[str, Any]        =     CustomFeatureExtractor.from_pretrained(_lowerCamelCase)

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

											UpperCAmelCase__  :					List[str]        =     CustomProcessor(_lowerCamelCase							,							_lowerCamelCase)

											with tempfile.TemporaryDirectory() as tmp_dir:
														processor.save_pretrained(_lowerCamelCase)
														UpperCAmelCase__  :					Optional[int]        =     AutoProcessor.from_pretrained(_lowerCamelCase)
														self.assertIsInstance(_lowerCamelCase							,							_lowerCamelCase)

								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):


								class 		_snake_case					(     a__	):
													lowerCAmelCase      :int   =					False


								class 		_snake_case					(     a__	):
													lowerCAmelCase      :Any   =					False


								class 		_snake_case					(     a__	):
													lowerCAmelCase      :Tuple   =					'''AutoFeatureExtractor'''
													lowerCAmelCase      :Optional[int]   =					'''AutoTokenizer'''
													lowerCAmelCase      :Optional[int]   =					False

								try:
											AutoConfig.register("""custom"""							,							_lowerCamelCase)
											AutoFeatureExtractor.register(_lowerCamelCase							,							_lowerCamelCase)
											AutoTokenizer.register(_lowerCamelCase							,							slow_tokenizer_class=_lowerCamelCase)
											AutoProcessor.register(_lowerCamelCase							,							_lowerCamelCase)
											# If remote code is not set, the default is to use local classes.
											UpperCAmelCase__  :					List[str]        =     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.
											UpperCAmelCase__  :					Optional[int]        =     AutoProcessor.from_pretrained(
											    """hf-internal-testing/test_dynamic_processor"""							,							trust_remote_code=_lowerCamelCase)
											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.
											UpperCAmelCase__  :					Tuple        =     AutoProcessor.from_pretrained(
											    """hf-internal-testing/test_dynamic_processor"""							,							trust_remote_code=_lowerCamelCase)
											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):
								UpperCAmelCase__  :					Union[str, Any]        =     AutoProcessor.from_pretrained("""hf-internal-testing/tiny-random-bert""")
								self.assertEqual(processor.__class__.__name__							,							"""BertTokenizerFast""")




					def 				snake_case__						(      self):
								UpperCAmelCase__  :					Optional[Any]        =     AutoProcessor.from_pretrained("""hf-internal-testing/tiny-random-convnext""")
								self.assertEqual(processor.__class__.__name__							,							"""ConvNextImageProcessor""")


@is_staging_test
class 		_snake_case					(     unittest.TestCase	):
					lowerCAmelCase      :Dict   =					['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']




					@classmethod
					def 				snake_case__						(      cls):
								UpperCAmelCase__  :					Any        =     TOKEN
								HfFolder.save_token(_lowerCamelCase)




					@classmethod
					def 				snake_case__						(      cls):
								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):
								UpperCAmelCase__  :					Any        =     WavaVecaProcessor.from_pretrained(_lowerCamelCase)
								with tempfile.TemporaryDirectory() as tmp_dir:
											processor.save_pretrained(
											    os.path.join(_lowerCamelCase							,							"""test-processor""")							,							push_to_hub=_lowerCamelCase							,							use_auth_token=self._token)

											UpperCAmelCase__  :					List[Any]        =     WavaVecaProcessor.from_pretrained(f'''{USER}/test-processor''')
											for k, v in processor.feature_extractor.__dict__.items():
														self.assertEqual(_lowerCamelCase							,							getattr(new_processor.feature_extractor							,							_lowerCamelCase))
											self.assertDictEqual(new_processor.tokenizer.get_vocab()							,							processor.tokenizer.get_vocab())




					def 				snake_case__						(      self):
								UpperCAmelCase__  :					int        =     WavaVecaProcessor.from_pretrained(_lowerCamelCase)

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

											UpperCAmelCase__  :					Union[str, Any]        =     WavaVecaProcessor.from_pretrained("""valid_org/test-processor-org""")
											for k, v in processor.feature_extractor.__dict__.items():
														self.assertEqual(_lowerCamelCase							,							getattr(new_processor.feature_extractor							,							_lowerCamelCase))
											self.assertDictEqual(new_processor.tokenizer.get_vocab()							,							processor.tokenizer.get_vocab())




					def 				snake_case__						(      self):
								CustomFeatureExtractor.register_for_auto_class()
								CustomTokenizer.register_for_auto_class()
								CustomProcessor.register_for_auto_class()

								UpperCAmelCase__  :					Dict        =     CustomFeatureExtractor.from_pretrained(_lowerCamelCase)

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

								UpperCAmelCase__  :					Union[str, Any]        =     CustomProcessor(_lowerCamelCase							,							_lowerCamelCase)

								with tempfile.TemporaryDirectory() as tmp_dir:
											create_repo(f'''{USER}/test-dynamic-processor'''							,							token=self._token)
											UpperCAmelCase__  :					List[str]        =     Repository(_lowerCamelCase							,							clone_from=f'''{USER}/test-dynamic-processor'''							,							token=self._token)
											processor.save_pretrained(_lowerCamelCase)

											# 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(_lowerCamelCase							,							"""tokenizer_config.json""")) as f:
														UpperCAmelCase__  :					Optional[Any]        =     json.load(_lowerCamelCase)
											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(_lowerCamelCase							,							"""custom_feature_extraction.py""")))
											self.assertTrue(os.path.isfile(os.path.join(_lowerCamelCase							,							"""custom_tokenization.py""")))
											self.assertTrue(os.path.isfile(os.path.join(_lowerCamelCase							,							"""custom_processing.py""")))

											repo.push_to_hub()

								UpperCAmelCase__  :					Union[str, Any]        =     AutoProcessor.from_pretrained(f'''{USER}/test-dynamic-processor'''							,							trust_remote_code=_lowerCamelCase)
								# 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""")



'''simple docstring'''



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

import torch

from ..utils import BaseOutput


__A	='scheduler_config.json'


class 		_snake_case					(     a__	):
					lowerCAmelCase      :Union[str, Any]   =					1
					lowerCAmelCase      :Union[str, Any]   =					2
					lowerCAmelCase      :Optional[int]   =					3
					lowerCAmelCase      :Optional[int]   =					4
					lowerCAmelCase      :Any   =					5
					lowerCAmelCase      :Tuple   =					6
					lowerCAmelCase      :List[Any]   =					7
					lowerCAmelCase      :str   =					8
					lowerCAmelCase      :List[Any]   =					9
					lowerCAmelCase      :List[str]   =					10
					lowerCAmelCase      :Union[str, Any]   =					11
					lowerCAmelCase      :Optional[int]   =					12
					lowerCAmelCase      :str   =					13
					lowerCAmelCase      :Dict   =					14


@dataclass
class 		_snake_case					(     a__	):
					lowerCAmelCase      :torch.FloatTensor


class 		_snake_case					:
					lowerCAmelCase      :str   =					SCHEDULER_CONFIG_NAME
					lowerCAmelCase      :Union[str, Any]   =					[]
					lowerCAmelCase      :List[str]   =					True




					@classmethod
					def 				snake_case__						(      cls							,							_lowerCamelCase = None							,							_lowerCamelCase = None							,							_lowerCamelCase=False							,							**_lowerCamelCase							,							):
								UpperCAmelCase__   ,       UpperCAmelCase__   ,       UpperCAmelCase__  :					Optional[int]        =     cls.load_config(
								    pretrained_model_name_or_path=_lowerCamelCase							,							subfolder=_lowerCamelCase							,							return_unused_kwargs=_lowerCamelCase							,							return_commit_hash=_lowerCamelCase							,							**_lowerCamelCase							,							)
								return cls.from_config(_lowerCamelCase							,							return_unused_kwargs=_lowerCamelCase							,							**_lowerCamelCase)




					def 				snake_case__						(      self							,							_lowerCamelCase							,							_lowerCamelCase = False							,							**_lowerCamelCase):
								self.save_config(save_directory=_lowerCamelCase							,							push_to_hub=_lowerCamelCase							,							**_lowerCamelCase)




					@property
					def 				snake_case__						(      self):
								return self._get_compatibles()




					@classmethod
					def 				snake_case__						(      cls):
								UpperCAmelCase__  :					List[Any]        =     list(set([cls.__name__] + cls._compatibles))
								UpperCAmelCase__  :					int        =     importlib.import_module(__name__.split(""".""")[0])
								UpperCAmelCase__  :					Dict        =     [
								    getattr(_lowerCamelCase							,							_lowerCamelCase) for c in compatible_classes_str if hasattr(_lowerCamelCase							,							_lowerCamelCase)
								]
								return compatible_classes