




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
lowerCAmelCase_						=			'.'

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






def  snake_case(    __magic_name__			,		__magic_name__			,		__magic_name__   )			->							Any:





     '''simple docstring'''


     lowercase :       List[str]       =			SavedModel()
     lowercase :       Tuple       =			[]

     with open(os.path.join(__magic_name__			,		'''utils'''			,		'''tf_ops'''			,		'''onnx.json'''   )   ) as f:
          lowercase :       Optional[int]       =			json.load(__magic_name__   )['''opsets''']

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

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

     lowercase :       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
     lowercase :       List[Any]       =			sorted(__magic_name__   )
     lowercase :       List[str]       =			[]

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

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


if __name__ == "__main__":
 lowerCAmelCase_						=			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)'
 )
 lowerCAmelCase_						=			parser.parse_args()

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





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


lowerCAmelCase_						=			get_tests_dir('fixtures/dummy_feature_extractor_config.json')
lowerCAmelCase_						=			get_tests_dir('fixtures/vocab.json')
lowerCAmelCase_						=			get_tests_dir('fixtures')
class 						_A (  unittest.TestCase				):
  _UpperCamelCase       :      int			=      ['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']






  def  __a      (  self		:     Union[str, Any]				)		->   Tuple:






       """simple docstring"""







       lowercase :       Union[str, Any]       =			0






  def  __a      (  self		:     Union[str, Any]				)		->   Union[str, Any]:






       """simple docstring"""







       lowercase :       Union[str, Any]       =			AutoProcessor.from_pretrained('''facebook/wav2vec2-base-960h'''				)
       self.assertIsInstance(_A				,						_A				)






  def  __a      (  self		:     Any				)		->   int:






       """simple docstring"""







       with tempfile.TemporaryDirectory() as tmpdirname:
            lowercase :       str       =			WavaVecaConfig()
            lowercase :       Tuple       =			AutoProcessor.from_pretrained('''facebook/wav2vec2-base-960h'''				)

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

            lowercase :       Optional[Any]       =			AutoProcessor.from_pretrained(_A				)

       self.assertIsInstance(_A				,						_A				)






  def  __a      (  self		:     Tuple				)		->   Union[str, Any]:






       """simple docstring"""







       with tempfile.TemporaryDirectory() as tmpdirname:
            # copy relevant files
            copyfile(_A				,						os.path.join(_A				,						_A				)				)
            copyfile(_A				,						os.path.join(_A				,						'''vocab.json'''				)				)

            lowercase :       Any       =			AutoProcessor.from_pretrained(_A				)

       self.assertIsInstance(_A				,						_A				)






  def  __a      (  self		:     Tuple				)		->   Any:






       """simple docstring"""







       with tempfile.TemporaryDirectory() as tmpdirname:
            lowercase :       List[Any]       =			WavaVecaFeatureExtractor()
            lowercase :       Optional[int]       =			AutoTokenizer.from_pretrained('''facebook/wav2vec2-base-960h'''				)

            lowercase :       Union[str, Any]       =			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:
                 lowercase :       Union[str, Any]       =			json.load(_A				)
                 config_dict.pop('''processor_class'''				)

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

            lowercase :       int       =			AutoProcessor.from_pretrained(_A				)

       self.assertIsInstance(_A				,						_A				)






  def  __a      (  self		:     str				)		->   List[str]:






       """simple docstring"""







       with tempfile.TemporaryDirectory() as tmpdirname:
            lowercase :       int       =			WavaVecaFeatureExtractor()
            lowercase :       Dict       =			AutoTokenizer.from_pretrained('''facebook/wav2vec2-base-960h'''				)

            lowercase :       Optional[int]       =			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:
                 lowercase :       int       =			json.load(_A				)
                 config_dict.pop('''processor_class'''				)

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

            lowercase :       Any       =			AutoProcessor.from_pretrained(_A				)

       self.assertIsInstance(_A				,						_A				)






  def  __a      (  self		:     List[Any]				)		->   Tuple:






       """simple docstring"""







       with tempfile.TemporaryDirectory() as tmpdirname:
            lowercase :       Dict       =			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('''{}'''				)

            lowercase :       Tuple       =			AutoProcessor.from_pretrained(_A				)

       self.assertIsInstance(_A				,						_A				)






  def  __a      (  self		:     Union[str, Any]				)		->   Tuple:






       """simple docstring"""







       with self.assertRaises(_A				):
            lowercase :       Optional[Any]       =			AutoProcessor.from_pretrained('''hf-internal-testing/test_dynamic_processor'''				)
       # If remote code is disabled, we can't load this config.
       with self.assertRaises(_A				):
            lowercase :       List[str]       =			AutoProcessor.from_pretrained(
                '''hf-internal-testing/test_dynamic_processor'''				,						trust_remote_code=_A				)

       lowercase :       str       =			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'''				)

       lowercase :       Union[str, Any]       =			processor.feature_extractor
       self.assertTrue(feature_extractor.special_attribute_present				)
       self.assertEqual(feature_extractor.__class__.__name__				,						'''NewFeatureExtractor'''				)

       lowercase :       Optional[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
            lowercase :       Any       =			AutoProcessor.from_pretrained(
                '''hf-internal-testing/test_dynamic_processor'''				,						trust_remote_code=_A				,						use_fast=_A				)
            lowercase :       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  __a      (  self		:     Dict				)		->   Optional[Any]:






       """simple docstring"""







       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
            lowercase :       Optional[Any]       =			CustomFeatureExtractor.from_pretrained(_A				)

            with tempfile.TemporaryDirectory() as tmp_dir:
                 lowercase :       List[Any]       =			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]				)				)
                 lowercase :       Optional[int]       =			CustomTokenizer(_A				)

            lowercase :       Union[str, Any]       =			CustomProcessor(_A				,						_A				)

            with tempfile.TemporaryDirectory() as tmp_dir:
                 processor.save_pretrained(_A				)
                 lowercase :       Dict       =			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  __a      (  self		:     str				)		->   Any:






       """simple docstring"""







       class 						_A (  _lowerCamelCase				):
         _UpperCamelCase       :      int			=      False


       class 						_A (  _lowerCamelCase				):
         _UpperCamelCase       :      Union[str, Any]			=      False


       class 						_A (  _lowerCamelCase				):
         _UpperCamelCase       :      Tuple			=      '''AutoFeatureExtractor'''
         _UpperCamelCase       :      Any			=      '''AutoTokenizer'''
         _UpperCamelCase       :      Any			=      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.
            lowercase :       List[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.
            lowercase :       Any       =			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.
            lowercase :       Union[str, Any]       =			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  __a      (  self		:     Tuple				)		->   Dict:






       """simple docstring"""







       lowercase :       Optional[int]       =			AutoProcessor.from_pretrained('''hf-internal-testing/tiny-random-bert'''				)
       self.assertEqual(processor.__class__.__name__				,						'''BertTokenizerFast'''				)






  def  __a      (  self		:     Any				)		->   int:






       """simple docstring"""







       lowercase :       Any       =			AutoProcessor.from_pretrained('''hf-internal-testing/tiny-random-convnext'''				)
       self.assertEqual(processor.__class__.__name__				,						'''ConvNextImageProcessor'''				)


@is_staging_test
class 						_A (  unittest.TestCase				):
  _UpperCamelCase       :      int			=      ['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']






  @classmethod
  def  __a      (  cls		:     Tuple				)		->   List[Any]:






       """simple docstring"""







       lowercase :       str       =			TOKEN
       HfFolder.save_token(_A				)






  @classmethod
  def  __a      (  cls		:     List[Any]				)		->   Dict:






       """simple docstring"""







       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  __a      (  self		:     Union[str, Any]				)		->   Tuple:






       """simple docstring"""







       lowercase :       Any       =			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				)

            lowercase :       Any       =			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  __a      (  self		:     Optional[Any]				)		->   str:






       """simple docstring"""







       lowercase :       int       =			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'''				,						)

            lowercase :       Dict       =			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  __a      (  self		:     str				)		->   Dict:






       """simple docstring"""







       CustomFeatureExtractor.register_for_auto_class()
       CustomTokenizer.register_for_auto_class()
       CustomProcessor.register_for_auto_class()

       lowercase :       List[str]       =			CustomFeatureExtractor.from_pretrained(_A				)

       with tempfile.TemporaryDirectory() as tmp_dir:
            lowercase :       Optional[int]       =			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]				)				)
            lowercase :       Optional[Any]       =			CustomTokenizer(_A				)

       lowercase :       Optional[Any]       =			CustomProcessor(_A				,						_A				)

       with tempfile.TemporaryDirectory() as tmp_dir:
            create_repo(f"""{USER}/test-dynamic-processor"""				,						token=self._token				)
            lowercase :       List[str]       =			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:
                 lowercase :       Optional[Any]       =			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()

       lowercase :       Tuple       =			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'''				)