






"""simple docstring"""







import argparse
import logging
import math
import time

import torch

from transformers import TransfoXLCorpus, TransfoXLLMHeadModel


logging.basicConfig(
    format="""%(asctime)s - %(levelname)s - %(name)s - %(message)s""", datefmt="""%m/%d/%Y %H:%M:%S""", level=logging.INFO
)
a :				int      					=		logging.getLogger(__name__)





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




						"""simple docstring"""



						lowercase__				:      List[Any]=       argparse.ArgumentParser(description="PyTorch Transformer Language Model"						)
						parser.add_argument("--model_name"					,		type=A					,		default="transfo-xl-wt103"					,		help="pretrained model name"						)
						parser.add_argument(
						    "--split"					,		type=A					,		default="test"					,		choices=["all", "valid", "test"]					,		help="which split to evaluate"						)
						parser.add_argument("--batch_size"					,		type=A					,		default=10					,		help="batch size"						)
						parser.add_argument("--tgt_len"					,		type=A					,		default=128					,		help="number of tokens to predict"						)
						parser.add_argument("--ext_len"					,		type=A					,		default=0					,		help="length of the extended context"						)
						parser.add_argument("--mem_len"					,		type=A					,		default=1_600					,		help="length of the retained previous heads"						)
						parser.add_argument("--clamp_len"					,		type=A					,		default=1_000					,		help="max positional embedding index"						)
						parser.add_argument("--no_cuda"					,		action="store_true"					,		help="Do not use CUDA even though CUA is available"						)
						parser.add_argument("--work_dir"					,		type=A					,		required=A					,		help="path to the work_dir"						)
						parser.add_argument("--no_log"					,		action="store_true"					,		help="do not log the eval result"						)
						parser.add_argument("--same_length"					,		action="store_true"					,		help="set same length attention with masking"						)
						parser.add_argument("--server_ip"					,		type=A					,		default=""					,		help="Can be used for distant debugging."						)
						parser.add_argument("--server_port"					,		type=A					,		default=""					,		help="Can be used for distant debugging."						)
						lowercase__				:      Dict=       parser.parse_args()
						assert args.ext_len >= 0, "extended context length must be non-negative"

						if args.server_ip and args.server_port:
												# Distant debugging - see https://code.visualstudio.com/docs/python/debugging#_attach-to-a-local-script
												import ptvsd

												print("Waiting for debugger attach"						)
												ptvsd.enable_attach(address=(args.server_ip, args.server_port)					,		redirect_output=A						)
												ptvsd.wait_for_attach()

						lowercase__				:      Dict=       torch.device("cuda" if torch.cuda.is_available() and not args.no_cuda else "cpu"						)
						logger.info("device: {}".format(A						)						)

						# Load a pre-processed dataset
						# You can also build the corpus yourself using TransfoXLCorpus methods
						# The pre-processing involve computing word frequencies to prepare the Adaptive input and SoftMax
						# and tokenizing the dataset
						# The pre-processed corpus is a convertion (using the conversion script )
						lowercase__				:      List[Any]=       TransfoXLCorpus.from_pretrained(args.model_name						)

						lowercase__				:      Tuple=       corpus.get_iterator("valid"					,		args.batch_size					,		args.tgt_len					,		device=A					,		ext_len=args.ext_len						)
						lowercase__				:      int=       corpus.get_iterator("test"					,		args.batch_size					,		args.tgt_len					,		device=A					,		ext_len=args.ext_len						)

						# Load a pre-trained model
						lowercase__				:      Tuple=       TransfoXLLMHeadModel.from_pretrained(args.model_name						)
						model.to(A						)

						logger.info(
						    "Evaluating with bsz {} tgt_len {} ext_len {} mem_len {} clamp_len {}".format(
						        args.batch_size					,		args.tgt_len					,		args.ext_len					,		args.mem_len					,		args.clamp_len						)						)

						model.reset_memory_length(args.mem_len						)
						if args.clamp_len > 0:
												lowercase__				:      Tuple=       args.clamp_len
						if args.same_length:
												lowercase__				:      Optional[Any]=       True

						###############################################################################
						# Evaluation code
						###############################################################################
						def evaluate(A						):
												# Turn on evaluation mode which disables dropout.
												model.eval()
												lowercase__,	lowercase__				:      Tuple=       0, 0.0
												lowercase__				:      Optional[int]=       time.time()
												with torch.no_grad():
																		lowercase__				:      Optional[Any]=       None
																		for idx, (data, target, seq_len) in enumerate(A						):
																								lowercase__				:      Optional[Any]=       model(A					,		lm_labels=A					,		mems=A						)
																								lowercase__,	lowercase__,	lowercase__				:      List[str]=       ret
																								lowercase__				:      int=       loss.mean()
																								total_loss += seq_len * loss.item()
																								total_len += seq_len
																		lowercase__				:      Dict=       time.time() - start_time
												logger.info("Time : {:.2f}s, {:.2f}ms/segment".format(A					,		1_000 * total_time / (idx + 1)						)						)
												return total_loss / total_len

						# Run on test data.
						if args.split == "all":
												lowercase__				:      Tuple=       evaluate(A						)
												lowercase__				:      Optional[Any]=       evaluate(A						)
						elif args.split == "valid":
												lowercase__				:      Tuple=       evaluate(A						)
												lowercase__				:      Any=       None
						elif args.split == "test":
												lowercase__				:      str=       evaluate(A						)
												lowercase__				:      Any=       None

						def format_log(A					,		A						):
												lowercase__				:      List[Any]=       "| {0} loss {1:5.2f} | {0} ppl {2:9.3f} ".format(A					,		A					,		math.exp(A						)						)
												return log_str

						lowercase__				:      Optional[int]=       ""
						if valid_loss is not None:
												log_str += format_log(A					,		"valid"						)
						if test_loss is not None:
												log_str += format_log(A					,		"test"						)

						logger.info("=" * 100						)
						logger.info(A						)
						logger.info("=" * 100						)


if __name__ == "__main__":
			main()














"""simple docstring"""







import importlib.metadata
import subprocess
import sys





def 						lowercase__(A = False						)				->List[str]:




						"""simple docstring"""



						lowercase__				:      Any=       False
						if "IPython" in sys.modules:
												lowercase__				:      Tuple=       "google.colab" in str(sys.modules["IPython"].get_ipython()						)

						if in_colab:
												if upgrade:
																		lowercase__				:      Optional[int]=       ["pip", "install", "-U", "torch"]
																		subprocess.run(A					,		check=A						)
												# get the current version of torch
												lowercase__				:      Tuple=       importlib.metadata.version("torch"						)
												lowercase__				:      Dict=       torch_version[: torch_version.rindex("."						)]
												lowercase__				:      Dict=       f'''https://storage.googleapis.com/tpu-pytorch/wheels/colab/torch_xla-{torch_version_trunc}-cp37-cp37m-linux_x86_64.whl'''
												lowercase__				:      Union[str, Any]=       ["pip", "install", xla_wheel]
												subprocess.run(A					,		check=A						)
						else:
												raise RuntimeError("`install_xla` utility works only on google colab."						)






