
import subprocess
import sys

from transformers import BertConfig, BertModel, BertTokenizer, pipeline
from transformers.testing_utils import TestCasePlus, require_torch

class        snake_case__    (A__		):







						"""simple docstring"""




						@require_torch
						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				Optional[int]:
									"""simple docstring"""




									a__       :						Tuple							=  """
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        """

									a__       :						int							=  """
mname = \"hf-internal-testing/tiny-random-bert\"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
BertTokenizer.from_pretrained(mname)
pipe = pipeline(task=\"fill-mask\", model=mname)
print(\"success\")
        """

									a__       :						Any							=  """
import socket
def offline_socket(*args, **kwargs): raise RuntimeError(\"Offline mode is enabled, we shouldn\'t access internet\")
socket.socket = offline_socket
        """

									# Force fetching the files so that we can use the cache
									a__       :						Union[str, Any]							=  """hf-internal-testing/tiny-random-bert"""
									BertConfig.from_pretrained(__A  )
									BertModel.from_pretrained(__A  )
									BertTokenizer.from_pretrained(__A  )
									pipeline(task="""fill-mask""" ,       model=__A  )

									# baseline - just load from_pretrained with normal network
									a__       :						int							=  [sys.executable, """-c""", """\n""".join([load, run, mock]  )]

									# should succeed
									a__       :						Dict							=  self.get_env()
									# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
									a__       :						str							=  """1"""
									a__       :						List[str]							=  subprocess.run(__A ,       env=__A ,       check=__A ,       capture_output=__A  )
									self.assertEqual(result.returncode ,       0 ,       result.stderr  )
									self.assertIn("""success""" ,       result.stdout.decode()  )

						@require_torch
						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				Tuple:
									"""simple docstring"""




									a__       :						Dict							=  """
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        """

									a__       :						Union[str, Any]							=  """
mname = \"hf-internal-testing/tiny-random-bert\"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
BertTokenizer.from_pretrained(mname)
pipe = pipeline(task=\"fill-mask\", model=mname)
print(\"success\")
        """

									a__       :						List[Any]							=  """
import socket
def offline_socket(*args, **kwargs): raise socket.error(\"Faking flaky internet\")
socket.socket = offline_socket
        """

									# Force fetching the files so that we can use the cache
									a__       :						List[str]							=  """hf-internal-testing/tiny-random-bert"""
									BertConfig.from_pretrained(__A  )
									BertModel.from_pretrained(__A  )
									BertTokenizer.from_pretrained(__A  )
									pipeline(task="""fill-mask""" ,       model=__A  )

									# baseline - just load from_pretrained with normal network
									a__       :						Dict							=  [sys.executable, """-c""", """\n""".join([load, run, mock]  )]

									# should succeed
									a__       :						int							=  self.get_env()
									a__       :						Any							=  subprocess.run(__A ,       env=__A ,       check=__A ,       capture_output=__A  )
									self.assertEqual(result.returncode ,       0 ,       result.stderr  )
									self.assertIn("""success""" ,       result.stdout.decode()  )

						@require_torch
						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				Optional[Any]:
									"""simple docstring"""




									a__       :						List[Any]							=  """
from transformers import BertConfig, BertModel, BertTokenizer
        """

									a__       :						Optional[int]							=  """
mname = \"hf-internal-testing/tiny-random-bert-sharded\"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
print(\"success\")
        """

									a__       :						str							=  """
import socket
def offline_socket(*args, **kwargs): raise ValueError(\"Offline mode is enabled\")
socket.socket = offline_socket
        """

									# baseline - just load from_pretrained with normal network
									a__       :						Optional[Any]							=  [sys.executable, """-c""", """\n""".join([load, run]  )]

									# should succeed
									a__       :						Optional[Any]							=  self.get_env()
									a__       :						Union[str, Any]							=  subprocess.run(__A ,       env=__A ,       check=__A ,       capture_output=__A  )
									self.assertEqual(result.returncode ,       0 ,       result.stderr  )
									self.assertIn("""success""" ,       result.stdout.decode()  )

									# next emulate no network
									a__       :						int							=  [sys.executable, """-c""", """\n""".join([load, mock, run]  )]

									# Doesn't fail anymore since the model is in the cache due to other tests, so commenting this.
									# env["TRANSFORMERS_OFFLINE"] = "0"
									# result = subprocess.run(cmd, env=env, check=False, capture_output=True)
									# self.assertEqual(result.returncode, 1, result.stderr)

									# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
									a__       :						Any							=  """1"""
									a__       :						Union[str, Any]							=  subprocess.run(__A ,       env=__A ,       check=__A ,       capture_output=__A  )
									self.assertEqual(result.returncode ,       0 ,       result.stderr  )
									self.assertIn("""success""" ,       result.stdout.decode()  )

						@require_torch
						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				Dict:
									"""simple docstring"""




									a__       :						List[Any]							=  """
from transformers import pipeline
        """
									a__       :						List[str]							=  """
mname = \"hf-internal-testing/tiny-random-bert\"
pipe = pipeline(model=mname)
        """

									a__       :						Union[str, Any]							=  """
import socket
def offline_socket(*args, **kwargs): raise socket.error(\"Offline mode is enabled\")
socket.socket = offline_socket
        """
									a__       :						Dict							=  self.get_env()
									a__       :						Union[str, Any]							=  """1"""
									a__       :						Tuple							=  [sys.executable, """-c""", """\n""".join([load, mock, run]  )]
									a__       :						Tuple							=  subprocess.run(__A ,       env=__A ,       check=__A ,       capture_output=__A  )
									self.assertEqual(result.returncode ,       1 ,       result.stderr  )
									self.assertIn(
									    """You cannot infer task automatically within `pipeline` when using offline mode""" ,       result.stderr.decode().replace("""\n""" ,       """"""  ) ,       )

						@require_torch
						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				int:
									"""simple docstring"""




									a__       :						Optional[Any]							=  """
from transformers import AutoModel
        """
									a__       :						List[str]							=  """
mname = \"hf-internal-testing/test_dynamic_model\"
AutoModel.from_pretrained(mname, trust_remote_code=True)
print(\"success\")
        """

									# baseline - just load from_pretrained with normal network
									a__       :						Optional[int]							=  [sys.executable, """-c""", """\n""".join([load, run]  )]

									# should succeed
									a__       :						Optional[int]							=  self.get_env()
									a__       :						Any							=  subprocess.run(__A ,       env=__A ,       check=__A ,       capture_output=__A  )
									self.assertEqual(result.returncode ,       0 ,       result.stderr  )
									self.assertIn("""success""" ,       result.stdout.decode()  )

									# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
									a__       :						Optional[int]							=  """1"""
									a__       :						int							=  subprocess.run(__A ,       env=__A ,       check=__A ,       capture_output=__A  )
									self.assertEqual(result.returncode ,       0 ,       result.stderr  )
									self.assertIn("""success""" ,       result.stdout.decode()  )









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		:				Union[str, Any]      		=data_utils.TransfoXLTokenizer
_lowercase		:				Tuple      		=data_utils.TransfoXLCorpus
_lowercase		:				Optional[int]      		=data_utils
_lowercase		:				int      		=data_utils

def      lowerCAmelCase_					(  _lowercase    :       Optional[Any]						,       _lowercase    :       List[Any]						,       _lowercase    :       Optional[Any]						,       _lowercase    :       Optional[Any])							->	List[Any]:







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

						a__       :						Optional[int]							=  corpus.__dict__
						corpus_dict_no_vocab.pop("""vocab"""						,       _lowercase)
						a__       :						Tuple							=  pytorch_dump_folder_path + """/""" + CORPUS_NAME
						print(F'''Save dataset to {pytorch_dataset_dump_path}''')
						torch.save(_lowercase						,       _lowercase)

			if tf_checkpoint_path:
						# Convert a pre-trained TensorFlow model
						a__       :						Optional[Any]							=  os.path.abspath(_lowercase)
						a__       :						str							=  os.path.abspath(_lowercase)

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

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


if __name__ == "__main__":
		_lowercase		:				Tuple      		=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		:				Dict      		=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,
		)




