






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


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



class   lowerCamelCase				(unittest.TestCase      ):







		'''simple docstring'''



		_snake_case   :   int							=  ['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']



		def  __UpperCAmelCase      (							self       )     ->							List[Any]:
							UpperCAmelCase_   :				List[Any]						=						0



		def  __UpperCAmelCase      (							self       )     ->							Optional[Any]:
							UpperCAmelCase_   :				Optional[int]						=						AutoProcessor.from_pretrained('facebook/wav2vec2-base-960h'       )
							self.assertIsInstance(_UpperCamelCase    ,		_UpperCamelCase       )



		def  __UpperCAmelCase      (							self       )     ->							Dict:
							with tempfile.TemporaryDirectory() as tmpdirname:
												UpperCAmelCase_   :				Optional[int]						=						WavaVecaConfig()
												UpperCAmelCase_   :				List[Any]						=						AutoProcessor.from_pretrained('facebook/wav2vec2-base-960h'       )

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

												UpperCAmelCase_   :				Optional[Any]						=						AutoProcessor.from_pretrained(_UpperCamelCase       )

							self.assertIsInstance(_UpperCamelCase    ,		_UpperCamelCase       )



		def  __UpperCAmelCase      (							self       )     ->							List[Any]:
							with tempfile.TemporaryDirectory() as tmpdirname:
												# copy relevant files
												copyfile(_UpperCamelCase    ,		os.path.join(_UpperCamelCase    ,		_UpperCamelCase       )       )
												copyfile(_UpperCamelCase    ,		os.path.join(_UpperCamelCase    ,		'vocab.json'       )       )

												UpperCAmelCase_   :				str						=						AutoProcessor.from_pretrained(_UpperCamelCase       )

							self.assertIsInstance(_UpperCamelCase    ,		_UpperCamelCase       )



		def  __UpperCAmelCase      (							self       )     ->							Any:
							with tempfile.TemporaryDirectory() as tmpdirname:
												UpperCAmelCase_   :				Dict						=						WavaVecaFeatureExtractor()
												UpperCAmelCase_   :				int						=						AutoTokenizer.from_pretrained('facebook/wav2vec2-base-960h'       )

												UpperCAmelCase_   :				List[Any]						=						WavaVecaProcessor(_UpperCamelCase    ,		_UpperCamelCase       )

												# save in new folder
												processor.save_pretrained(_UpperCamelCase       )

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

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

												UpperCAmelCase_   :				Union[str, Any]						=						AutoProcessor.from_pretrained(_UpperCamelCase       )

							self.assertIsInstance(_UpperCamelCase    ,		_UpperCamelCase       )



		def  __UpperCAmelCase      (							self       )     ->							List[str]:
							with tempfile.TemporaryDirectory() as tmpdirname:
												UpperCAmelCase_   :				Tuple						=						WavaVecaFeatureExtractor()
												UpperCAmelCase_   :				Dict						=						AutoTokenizer.from_pretrained('facebook/wav2vec2-base-960h'       )

												UpperCAmelCase_   :				Any						=						WavaVecaProcessor(_UpperCamelCase    ,		_UpperCamelCase       )

												# save in new folder
												processor.save_pretrained(_UpperCamelCase       )

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

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

												UpperCAmelCase_   :				Union[str, Any]						=						AutoProcessor.from_pretrained(_UpperCamelCase       )

							self.assertIsInstance(_UpperCamelCase    ,		_UpperCamelCase       )



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

												UpperCAmelCase_   :				Dict						=						AutoProcessor.from_pretrained(_UpperCamelCase       )

							self.assertIsInstance(_UpperCamelCase    ,		_UpperCamelCase       )



		def  __UpperCAmelCase      (							self       )     ->							List[str]:
							# If remote code is not set, we will time out when asking whether to load the model.
							with self.assertRaises(_UpperCamelCase       ):
												UpperCAmelCase_   :				Optional[int]						=						AutoProcessor.from_pretrained('hf-internal-testing/test_dynamic_processor'       )
							# If remote code is disabled, we can't load this config.
							with self.assertRaises(_UpperCamelCase       ):
												UpperCAmelCase_   :				Optional[Any]						=						AutoProcessor.from_pretrained(
												    'hf-internal-testing/test_dynamic_processor'    ,		trust_remote_code=_UpperCamelCase       )

							UpperCAmelCase_   :				Optional[int]						=						AutoProcessor.from_pretrained('hf-internal-testing/test_dynamic_processor'    ,		trust_remote_code=_UpperCamelCase       )
							self.assertTrue(processor.special_attribute_present       )
							self.assertEqual(processor.__class__.__name__    ,		'NewProcessor'       )

							UpperCAmelCase_   :				Optional[Any]						=						processor.feature_extractor
							self.assertTrue(feature_extractor.special_attribute_present       )
							self.assertEqual(feature_extractor.__class__.__name__    ,		'NewFeatureExtractor'       )

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

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

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

												UpperCAmelCase_   :				Any						=						CustomProcessor(_UpperCamelCase    ,		_UpperCamelCase       )

												with tempfile.TemporaryDirectory() as tmp_dir:
																	processor.save_pretrained(_UpperCamelCase       )
																	UpperCAmelCase_   :				List[Any]						=						AutoProcessor.from_pretrained(_UpperCamelCase       )
																	self.assertIsInstance(_UpperCamelCase    ,		_UpperCamelCase       )

							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  __UpperCAmelCase      (							self       )     ->							Optional[int]:



							class   lowerCamelCase				(_snake_case      ):







									'''simple docstring'''



									_snake_case   :   Tuple							=  False





							class   lowerCamelCase				(_snake_case      ):







									'''simple docstring'''



									_snake_case   :   Optional[int]							=  False





							class   lowerCamelCase				(_snake_case      ):







									'''simple docstring'''



									_snake_case   :   int							=  '''AutoFeatureExtractor'''
									_snake_case   :   str							=  '''AutoTokenizer'''
									_snake_case   :   List[Any]							=  False

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



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





@is_staging_test
class   lowerCamelCase				(unittest.TestCase      ):







		'''simple docstring'''



		_snake_case   :   Union[str, Any]							=  ['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']



		@classmethod
		def  __UpperCAmelCase      (							cls       )     ->							Tuple:
							UpperCAmelCase_   :				Optional[Any]						=						TOKEN
							HfFolder.save_token(_UpperCamelCase       )



		@classmethod
		def  __UpperCAmelCase      (							cls       )     ->							Union[str, Any]:
							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  __UpperCAmelCase      (							self       )     ->							int:
							UpperCAmelCase_   :				List[str]						=						WavaVecaProcessor.from_pretrained(_UpperCamelCase       )
							with tempfile.TemporaryDirectory() as tmp_dir:
												processor.save_pretrained(
												    os.path.join(_UpperCamelCase    ,		'test-processor'       )    ,		push_to_hub=_UpperCamelCase    ,		use_auth_token=self._token       )

												UpperCAmelCase_   :				str						=						WavaVecaProcessor.from_pretrained(f"{USER}/test-processor"       )
												for k, v in processor.feature_extractor.__dict__.items():
																	self.assertEqual(_UpperCamelCase    ,		getattr(new_processor.feature_extractor    ,		_UpperCamelCase       )       )
												self.assertDictEqual(new_processor.tokenizer.get_vocab()    ,		processor.tokenizer.get_vocab()       )



		def  __UpperCAmelCase      (							self       )     ->							Dict:
							UpperCAmelCase_   :				Dict						=						WavaVecaProcessor.from_pretrained(_UpperCamelCase       )

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

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



		def  __UpperCAmelCase      (							self       )     ->							Optional[Any]:
							CustomFeatureExtractor.register_for_auto_class()
							CustomTokenizer.register_for_auto_class()
							CustomProcessor.register_for_auto_class()

							UpperCAmelCase_   :				List[str]						=						CustomFeatureExtractor.from_pretrained(_UpperCamelCase       )

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

							UpperCAmelCase_   :				str						=						CustomProcessor(_UpperCamelCase    ,		_UpperCamelCase       )

							with tempfile.TemporaryDirectory() as tmp_dir:
												create_repo(f"{USER}/test-dynamic-processor"    ,		token=self._token       )
												UpperCAmelCase_   :				Union[str, Any]						=						Repository(_UpperCamelCase    ,		clone_from=f"{USER}/test-dynamic-processor"    ,		token=self._token       )
												processor.save_pretrained(_UpperCamelCase       )

												# 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(_UpperCamelCase    ,		'tokenizer_config.json'       )       ) as f:
																	UpperCAmelCase_   :				Tuple						=						json.load(_UpperCamelCase       )
												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(_UpperCamelCase    ,		'custom_feature_extraction.py'       )       )       )
												self.assertTrue(os.path.isfile(os.path.join(_UpperCamelCase    ,		'custom_tokenization.py'       )       )       )
												self.assertTrue(os.path.isfile(os.path.join(_UpperCamelCase    ,		'custom_processing.py'       )       )       )

												repo.push_to_hub()

							UpperCAmelCase_   :				int						=						AutoProcessor.from_pretrained(f"{USER}/test-dynamic-processor"    ,		trust_remote_code=_UpperCamelCase       )
							# 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'       )














import argparse
import json
import os

from tensorflow.core.protobuf.saved_model_pba import SavedModel


# All paths are set with the intent you should run this script from the root of the repo with the command
# python utils/check_copies.py
__UpperCAmelCase     							=						'.'

# Internal TensorFlow ops that can be safely ignored (mostly specific to a saved model)
__UpperCAmelCase     							=						[
    'Assert',
    'AssignVariableOp',
    'EmptyTensorList',
    'MergeV2Checkpoints',
    'ReadVariableOp',
    'ResourceGather',
    'RestoreV2',
    'SaveV2',
    'ShardedFilename',
    'StatefulPartitionedCall',
    'StaticRegexFullMatch',
    'VarHandleOp',
]




def 					lowercase__				(					__snake_case   :							Union[str, Any]					,	__snake_case   :							Optional[int]					,	__snake_case   :							Union[str, Any]  ):


					'''simple docstring'''


					UpperCAmelCase_   :				Optional[Any]						=						SavedModel()
					UpperCAmelCase_   :				Optional[Any]						=						[]

					with open(os.path.join(__snake_case					,	'utils'					,	'tf_ops'					,	'onnx.json'  )  ) as f:
										UpperCAmelCase_   :				Optional[Any]						=						json.load(__snake_case  )['opsets']

					for i in range(1					,	opset + 1  ):
										onnx_ops.extend(onnx_opsets[str(__snake_case  )]  )

					with open(__snake_case					,	'rb'  ) as f:
										saved_model.ParseFromString(f.read()  )

					UpperCAmelCase_   :				List[Any]						=						set()

					# Iterate over every metagraph in case there is more than one (a saved model can contain multiple graphs)
					for meta_graph in saved_model.meta_graphs:
										# Add operations in the graph definition
										model_op_names.update(node.op for node in meta_graph.graph_def.node  )

										# Go through the functions in the graph definition
										for func in meta_graph.graph_def.library.function:
															# Add operations in each function
															model_op_names.update(node.op for node in func.node_def  )

    # Convert to list, sorted if you want
					UpperCAmelCase_   :				Optional[int]						=						sorted(__snake_case  )
					UpperCAmelCase_   :				int						=						[]

					for op in model_op_names:
										if op not in onnx_ops and op not in INTERNAL_OPS:
															incompatible_ops.append(__snake_case  )

					if strict and len(__snake_case  ) > 0:
										raise Exception(F"Found the following incompatible ops for the opset {opset}:\n" + incompatible_ops  )
					elif len(__snake_case  ) > 0:
										print(F"Found the following incompatible ops for the opset {opset}:"  )
										print(*__snake_case					,	sep='\n'  )
					else:
										print(F"The saved model {saved_model_path} can properly be converted with ONNX."  )


if __name__ == "__main__":
		__UpperCAmelCase     							=						argparse.ArgumentParser()
		parser.add_argument('--saved_model_path', help='Path of the saved model to check (the .pb file).')
		parser.add_argument(
		    '--opset', default=12, type=int, help='The ONNX opset against which the model has to be tested.'
		)
		parser.add_argument(
		    '--framework', choices=['onnx'], default='onnx', help='Frameworks against which to test the saved model.'
		)
		parser.add_argument(
		    '--strict', action='store_true', help='Whether make the checking strict (raise errors) or not (raise warnings)'
		)
		__UpperCAmelCase     							=						parser.parse_args()

		if args.framework == "onnx":
				onnx_compliancy(args.saved_model_path, args.strict, args.opset)






