






"""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
)
_UpperCamelCase						:    int		=						logging.getLogger(__name__)
def 		_SCREAMING_SNAKE_CASE					(       ):





  '''simple docstring'''





  lowercase				       =			argparse.ArgumentParser(description='PyTorch Transformer Language Model'       )
  parser.add_argument('--model_name'						,   type=__snake_case						,   default='transfo-xl-wt103'						,   help='pretrained model name'       )
  parser.add_argument(
      '--split'						,   type=__snake_case						,   default='test'						,   choices=['all', 'valid', 'test']						,   help='which split to evaluate'       )
  parser.add_argument('--batch_size'						,   type=__snake_case						,   default=10						,   help='batch size'       )
  parser.add_argument('--tgt_len'						,   type=__snake_case						,   default=1_28						,   help='number of tokens to predict'       )
  parser.add_argument('--ext_len'						,   type=__snake_case						,   default=0						,   help='length of the extended context'       )
  parser.add_argument('--mem_len'						,   type=__snake_case						,   default=16_00						,   help='length of the retained previous heads'       )
  parser.add_argument('--clamp_len'						,   type=__snake_case						,   default=10_00						,   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=__snake_case						,   required=__snake_case						,   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=__snake_case						,   default=''						,   help='Can be used for distant debugging.'       )
  parser.add_argument('--server_port'						,   type=__snake_case						,   default=''						,   help='Can be used for distant debugging.'       )
  lowercase				       =			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=__snake_case       )
    ptvsd.wait_for_attach()

  lowercase				       =			torch.device('cuda' if torch.cuda.is_available() and not args.no_cuda else 'cpu'       )
  logger.info('device: {}'.format(__snake_case       )       )

  # 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				       =			TransfoXLCorpus.from_pretrained(args.model_name       )

  lowercase				       =			corpus.get_iterator('valid'						,   args.batch_size						,   args.tgt_len						,   device=__snake_case						,   ext_len=args.ext_len       )
  lowercase				       =			corpus.get_iterator('test'						,   args.batch_size						,   args.tgt_len						,   device=__snake_case						,   ext_len=args.ext_len       )

  # Load a pre-trained model
  lowercase				       =			TransfoXLLMHeadModel.from_pretrained(args.model_name       )
  model.to(__snake_case       )

  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				       =			args.clamp_len
  if args.same_length:
    lowercase				       =			True

  ###############################################################################
  # Evaluation code
  ###############################################################################
  def evaluate(__snake_case     :    Any       ):
    # Turn on evaluation mode which disables dropout.
    model.eval()
    lowercase       ,	lowercase				       =			0, 0.0
    lowercase				       =			time.time()
    with torch.no_grad():
      lowercase				       =			None
      for idx, (data, target, seq_len) in enumerate(__snake_case       ):
        lowercase				       =			model(__snake_case						,   lm_labels=__snake_case						,   mems=__snake_case       )
        lowercase       ,	lowercase       ,	lowercase				       =			ret
        lowercase				       =			loss.mean()
        total_loss += seq_len * loss.item()
        total_len += seq_len
      lowercase				       =			time.time() - start_time
    logger.info('Time : {:.2f}s, {:.2f}ms/segment'.format(__snake_case						,   10_00 * total_time / (idx + 1)       )       )
    return total_loss / total_len

  # Run on test data.
  if args.split == "all":
    lowercase				       =			evaluate(__snake_case       )
    lowercase				       =			evaluate(__snake_case       )
  elif args.split == "valid":
    lowercase				       =			evaluate(__snake_case       )
    lowercase				       =			None
  elif args.split == "test":
    lowercase				       =			evaluate(__snake_case       )
    lowercase				       =			None

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

  lowercase				       =			''
  if valid_loss is not None:
    log_str += format_log(__snake_case						,   'valid'       )
  if test_loss is not None:
    log_str += format_log(__snake_case						,   'test'       )

  logger.info('=' * 1_00       )
  logger.info(__snake_case       )
  logger.info('=' * 1_00       )


if __name__ == "__main__":
    main()








from unittest.mock import patch

import datasets
from datasets import Dataset



def    UpperCAmelCase__	(  ):
  lowercase     :Any   						=    Dataset.from_dict({"col_1": [3, 2, 0, 1]} )

  with patch("tqdm.auto.tqdm" ) as mock_tqdm:
    datasets.disable_progress_bar()
    dset.map(lambda lowerCamelCase : {"col_2": x["col_1"] + 1} )
    mock_tqdm.assert_not_called()

    mock_tqdm.reset_mock()

    datasets.enable_progress_bar()
    dset.map(lambda lowerCamelCase : {"col_2": x["col_1"] + 1} )
    mock_tqdm.assert_called()



