






'''simple docstring'''




import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

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

from transformers import AutoFeatureExtractor, WavaVecaFeatureExtractor
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test


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

from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402


lowerCamelCase_												=					get_tests_dir('''fixtures''')

class    _UpperCAmelCase			(     unittest.TestCase						):




   """simple docstring"""

   def 						lowerCAmelCase (		self :	int		):





         '''simple docstring'''





         _A			     =	mock.Mock()
         _A			     =	500
         _A			     =	{}
         _A			     =	HTTPError
         _A			     =	{}

         # Download this model to make sure it's in the cache.
         _A			     =	WavaVecaFeatureExtractor.from_pretrained("hf-internal-testing/tiny-random-wav2vec2"		)
         # Under the mock environment we get a 500 error when trying to reach the model.
         with mock.patch("requests.Session.request"      ,      return_value=__UpperCAmelCase		) as mock_head:
               _A			     =	WavaVecaFeatureExtractor.from_pretrained("hf-internal-testing/tiny-random-wav2vec2"		)
               # This check we did call the fake head request
               mock_head.assert_called()

   def 						lowerCAmelCase (		self :	int		):





         '''simple docstring'''





         _A			     =	WavaVecaFeatureExtractor.from_pretrained(
             "https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json"		)





@is_staging_test
class    _UpperCAmelCase			(     unittest.TestCase						):




   """simple docstring"""

   @classmethod
   def 						lowerCAmelCase (		cls :	str		):





         '''simple docstring'''





         _A			     =	TOKEN
         HfFolder.save_token(__UpperCAmelCase		)

   @classmethod
   def 						lowerCAmelCase (		cls :	Any		):





         '''simple docstring'''





         try:
               delete_repo(token=cls._token      ,      repo_id="test-feature-extractor"		)
         except HTTPError:
               pass

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

         try:
               delete_repo(token=cls._token      ,      repo_id="test-dynamic-feature-extractor"		)
         except HTTPError:
               pass

   def 						lowerCAmelCase (		self :	int		):





         '''simple docstring'''





         _A			     =	WavaVecaFeatureExtractor.from_pretrained(__UpperCAmelCase		)
         feature_extractor.push_to_hub("test-feature-extractor"      ,      use_auth_token=self._token		)

         _A			     =	WavaVecaFeatureExtractor.from_pretrained(f'''{USER}/test-feature-extractor'''		)
         for k, v in feature_extractor.__dict__.items():
               self.assertEqual(__UpperCAmelCase      ,      getattr(__UpperCAmelCase      ,      __UpperCAmelCase		)		)

         # Reset repo
         delete_repo(token=self._token      ,      repo_id="test-feature-extractor"		)

         # Push to hub via save_pretrained
         with tempfile.TemporaryDirectory() as tmp_dir:
               feature_extractor.save_pretrained(
                   __UpperCAmelCase      ,      repo_id="test-feature-extractor"      ,      push_to_hub=__UpperCAmelCase      ,      use_auth_token=self._token		)

         _A			     =	WavaVecaFeatureExtractor.from_pretrained(f'''{USER}/test-feature-extractor'''		)
         for k, v in feature_extractor.__dict__.items():
               self.assertEqual(__UpperCAmelCase      ,      getattr(__UpperCAmelCase      ,      __UpperCAmelCase		)		)

   def 						lowerCAmelCase (		self :	Dict		):





         '''simple docstring'''





         _A			     =	WavaVecaFeatureExtractor.from_pretrained(__UpperCAmelCase		)
         feature_extractor.push_to_hub("valid_org/test-feature-extractor"      ,      use_auth_token=self._token		)

         _A			     =	WavaVecaFeatureExtractor.from_pretrained("valid_org/test-feature-extractor"		)
         for k, v in feature_extractor.__dict__.items():
               self.assertEqual(__UpperCAmelCase      ,      getattr(__UpperCAmelCase      ,      __UpperCAmelCase		)		)

         # Reset repo
         delete_repo(token=self._token      ,      repo_id="valid_org/test-feature-extractor"		)

         # Push to hub via save_pretrained
         with tempfile.TemporaryDirectory() as tmp_dir:
               feature_extractor.save_pretrained(
                   __UpperCAmelCase      ,      repo_id="valid_org/test-feature-extractor-org"      ,      push_to_hub=__UpperCAmelCase      ,      use_auth_token=self._token		)

         _A			     =	WavaVecaFeatureExtractor.from_pretrained("valid_org/test-feature-extractor-org"		)
         for k, v in feature_extractor.__dict__.items():
               self.assertEqual(__UpperCAmelCase      ,      getattr(__UpperCAmelCase      ,      __UpperCAmelCase		)		)

   def 						lowerCAmelCase (		self :	Optional[int]		):





         '''simple docstring'''





         CustomFeatureExtractor.register_for_auto_class()
         _A			     =	CustomFeatureExtractor.from_pretrained(__UpperCAmelCase		)

         feature_extractor.push_to_hub("test-dynamic-feature-extractor"      ,      use_auth_token=self._token		)

         # This has added the proper auto_map field to the config
         self.assertDictEqual(
             feature_extractor.auto_map      ,      {"AutoFeatureExtractor": "custom_feature_extraction.CustomFeatureExtractor"}      ,      )

         _A			     =	AutoFeatureExtractor.from_pretrained(
             f'''{USER}/test-dynamic-feature-extractor'''      ,      trust_remote_code=__UpperCAmelCase		)
         # Can't make an isinstance check because the new_feature_extractor is from the CustomFeatureExtractor class of a dynamic module
         self.assertEqual(new_feature_extractor.__class__.__name__      ,      "CustomFeatureExtractor"		)








import argparse
import os
import pickle
import sys

import torch

from transformers import TransfoXLConfig, TransfoXLLMHeadModel, load_tf_weights_in_transfo_xl
from transformers.models.transfo_xl import tokenization_transfo_xl as data_utils
from transformers.models.transfo_xl.tokenization_transfo_xl import CORPUS_NAME, VOCAB_FILES_NAMES
from transformers.utils import CONFIG_NAME, WEIGHTS_NAME, logging


logging.set_verbosity_info()

# We do this to be able to load python 2 datasets pickles
# See e.g. https://stackoverflow.com/questions/2121874/python-pickling-after-changing-a-modules-directory/2121918#2121918
lowercase_									=   data_utils.TransfoXLTokenizer
lowercase_									=   data_utils.TransfoXLCorpus
lowercase_									=   data_utils
lowercase_									=   data_utils
def      a      (			A__	:   int      ,   A__	:   Dict      ,   A__	:   Union[str, Any]      ,   A__	:   Union[str, Any]     )    ->      List[str]:





      """simple docstring"""



      if transfo_xl_dataset_file:
            # Convert a pre-processed corpus (see original TensorFlow repo)
            with open(A__      ,   'rb'     ) as fp:
                  _lowercase	       =pickle.load(A__      ,   encoding='latin1'     )
            # Save vocabulary and dataset cache as Dictionaries (should be better than pickles for the long-term)
            _lowercase	       =pytorch_dump_folder_path + '/' + VOCAB_FILES_NAMES['pretrained_vocab_file']
            print(F'''Save vocabulary to {pytorch_vocab_dump_path}'''     )
            _lowercase	       =corpus.vocab.__dict__
            torch.save(A__      ,   A__     )

            _lowercase	       =corpus.__dict__
            corpus_dict_no_vocab.pop('vocab'      ,   A__     )
            _lowercase	       =pytorch_dump_folder_path + '/' + CORPUS_NAME
            print(F'''Save dataset to {pytorch_dataset_dump_path}'''     )
            torch.save(A__      ,   A__     )

      if tf_checkpoint_path:
            # Convert a pre-trained TensorFlow model
            _lowercase	       =os.path.abspath(A__     )
            _lowercase	       =os.path.abspath(A__     )

            print(F'''Converting Transformer XL checkpoint from {tf_path} with config at {config_path}.'''     )
            # Initialise PyTorch model
            if transfo_xl_config_file == "":
                  _lowercase	       =TransfoXLConfig()
            else:
                  _lowercase	       =TransfoXLConfig.from_json_file(A__     )
            print(F'''Building PyTorch model from configuration: {config}'''     )
            _lowercase	       =TransfoXLLMHeadModel(A__     )

            _lowercase	       =load_tf_weights_in_transfo_xl(A__      ,   A__      ,   A__     )
            # Save pytorch-model
            _lowercase	       =os.path.join(A__      ,   A__     )
            _lowercase	       =os.path.join(A__      ,   A__     )
            print(F'''Save PyTorch model to {os.path.abspath(A__     )}'''     )
            torch.save(model.state_dict()      ,   A__     )
            print(F'''Save configuration file to {os.path.abspath(A__     )}'''     )
            with open(A__      ,   'w'      ,   encoding='utf-8'     ) as f:
                  f.write(config.to_json_string()     )


if __name__ == "__main__":
  lowercase_									=   argparse.ArgumentParser()
  parser.add_argument(
      '--pytorch_dump_folder_path',
      default=None,
      type=str,
      required=True,
      help='Path to the folder to store the PyTorch model or dataset/vocab.',
  )
  parser.add_argument(
      '--tf_checkpoint_path',
      default='',
      type=str,
      help='An optional path to a TensorFlow checkpoint path to be converted.',
  )
  parser.add_argument(
      '--transfo_xl_config_file',
      default='',
      type=str,
      help=(
          'An optional config json file corresponding to the pre-trained BERT model. \n'
          'This specifies the model architecture.'
      ),
  )
  parser.add_argument(
      '--transfo_xl_dataset_file',
      default='',
      type=str,
      help='An optional dataset file to be converted in a vocabulary.',
  )
  lowercase_									=   parser.parse_args()
  convert_transfo_xl_checkpoint_to_pytorch(
      args.tf_checkpoint_path,
      args.transfo_xl_config_file,
      args.pytorch_dump_folder_path,
      args.transfo_xl_dataset_file,
  )


