

'''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_      =							logging.getLogger(__name__)





def 				A_					(	):
  SCREAMING_SNAKE_CASE:Any   =   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=128 ,    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=1600 ,    help="length of the retained previous heads"	)
  parser.add_argument("--clamp_len" ,    type=snake_case ,    default=1000 ,    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."	)
  SCREAMING_SNAKE_CASE:Optional[int]   =   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()

  SCREAMING_SNAKE_CASE:Optional[int]   =   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 )
  SCREAMING_SNAKE_CASE:Optional[int]   =   TransfoXLCorpus.from_pretrained(args.model_name	)

  SCREAMING_SNAKE_CASE:str   =   corpus.get_iterator("valid" ,    args.batch_size ,    args.tgt_len ,    device=snake_case ,    ext_len=args.ext_len	)
  SCREAMING_SNAKE_CASE:Any   =   corpus.get_iterator("test" ,    args.batch_size ,    args.tgt_len ,    device=snake_case ,    ext_len=args.ext_len	)

  # Load a pre-trained model
  SCREAMING_SNAKE_CASE:List[Any]   =   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:
    SCREAMING_SNAKE_CASE:Any   =   args.clamp_len
  if args.same_length:
    SCREAMING_SNAKE_CASE:int   =   True

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

  # Run on test data.
  if args.split == "all":
    SCREAMING_SNAKE_CASE:List[Any]   =   evaluate(snake_case	)
    SCREAMING_SNAKE_CASE:str   =   evaluate(snake_case	)
  elif args.split == "valid":
    SCREAMING_SNAKE_CASE:Dict   =   evaluate(snake_case	)
    SCREAMING_SNAKE_CASE:str   =   None
  elif args.split == "test":
    SCREAMING_SNAKE_CASE:List[Any]   =   evaluate(snake_case	)
    SCREAMING_SNAKE_CASE:Any   =   None

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

  SCREAMING_SNAKE_CASE:Dict   =   ""
  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("=" * 100	)
  logger.info(snake_case	)
  logger.info("=" * 100	)


if __name__ == "__main__":
    main()









'''simple docstring'''

import shutil
import tempfile
import unittest

import pytest

from transformers import WhisperTokenizer, is_speech_available
from transformers.testing_utils import require_sentencepiece, require_torch, require_torchaudio

from .test_feature_extraction_whisper import floats_list


if is_speech_available():
    from transformers import WhisperFeatureExtractor, WhisperProcessor


A_      =							5_03_58
A_      =							5_03_62





@require_torch
@require_torchaudio
@require_sentencepiece
class 							_snake_case	(    unittest.TestCase  ):
   def       __UpperCamelCase  (	self     : Any							):
     SCREAMING_SNAKE_CASE:int   =   "openai/whisper-small.en"
     SCREAMING_SNAKE_CASE:Optional[Any]   =   tempfile.mkdtemp()
   def       __UpperCamelCase  (	self     : Any       ,**SCREAMING_SNAKE_CASE__     : List[str]							):
     return WhisperTokenizer.from_pretrained(self.checkpoint       ,**SCREAMING_SNAKE_CASE__							)
   def       __UpperCamelCase  (	self     : List[str]       ,**SCREAMING_SNAKE_CASE__     : Dict							):
     return WhisperFeatureExtractor.from_pretrained(self.checkpoint       ,**SCREAMING_SNAKE_CASE__							)
   def       __UpperCamelCase  (	self     : str							):
     shutil.rmtree(self.tmpdirname							)
   def       __UpperCamelCase  (	self     : Any							):
     SCREAMING_SNAKE_CASE:Dict   =   self.get_tokenizer()
     SCREAMING_SNAKE_CASE:Optional[int]   =   self.get_feature_extractor()

     SCREAMING_SNAKE_CASE:Dict   =   WhisperProcessor(tokenizer=SCREAMING_SNAKE_CASE__       ,feature_extractor=SCREAMING_SNAKE_CASE__							)

     processor.save_pretrained(self.tmpdirname							)
     SCREAMING_SNAKE_CASE:List[Any]   =   WhisperProcessor.from_pretrained(self.tmpdirname							)

     self.assertEqual(processor.tokenizer.get_vocab()       ,tokenizer.get_vocab()							)
     self.assertIsInstance(processor.tokenizer       ,SCREAMING_SNAKE_CASE__							)

     self.assertEqual(processor.feature_extractor.to_json_string()       ,feature_extractor.to_json_string()							)
     self.assertIsInstance(processor.feature_extractor       ,SCREAMING_SNAKE_CASE__							)
   def       __UpperCamelCase  (	self     : Union[str, Any]							):
     SCREAMING_SNAKE_CASE:int   =   WhisperProcessor(tokenizer=self.get_tokenizer()       ,feature_extractor=self.get_feature_extractor()							)
     processor.save_pretrained(self.tmpdirname							)

     SCREAMING_SNAKE_CASE:List[str]   =   self.get_tokenizer(bos_token="(BOS)"       ,eos_token="(EOS)"							)
     SCREAMING_SNAKE_CASE:Tuple   =   self.get_feature_extractor(do_normalize=SCREAMING_SNAKE_CASE__       ,padding_value=1.0							)

     SCREAMING_SNAKE_CASE:Any   =   WhisperProcessor.from_pretrained(
         self.tmpdirname       ,bos_token="(BOS)"       ,eos_token="(EOS)"       ,do_normalize=SCREAMING_SNAKE_CASE__       ,padding_value=1.0							)

     self.assertEqual(processor.tokenizer.get_vocab()       ,tokenizer_add_kwargs.get_vocab()							)
     self.assertIsInstance(processor.tokenizer       ,SCREAMING_SNAKE_CASE__							)

     self.assertEqual(processor.feature_extractor.to_json_string()       ,feature_extractor_add_kwargs.to_json_string()							)
     self.assertIsInstance(processor.feature_extractor       ,SCREAMING_SNAKE_CASE__							)
   def       __UpperCamelCase  (	self     : List[Any]							):
     SCREAMING_SNAKE_CASE:Any   =   self.get_feature_extractor()
     SCREAMING_SNAKE_CASE:List[Any]   =   self.get_tokenizer()

     SCREAMING_SNAKE_CASE:Optional[Any]   =   WhisperProcessor(tokenizer=SCREAMING_SNAKE_CASE__       ,feature_extractor=SCREAMING_SNAKE_CASE__							)

     SCREAMING_SNAKE_CASE:Any   =   floats_list((3, 1_000)							)

     SCREAMING_SNAKE_CASE:Any   =   feature_extractor(SCREAMING_SNAKE_CASE__       ,return_tensors="np"							)
     SCREAMING_SNAKE_CASE:List[str]   =   processor(SCREAMING_SNAKE_CASE__       ,return_tensors="np"							)

     for key in input_feat_extract.keys():
       self.assertAlmostEqual(input_feat_extract[key].sum()       ,input_processor[key].sum()       ,delta=1e-2							)
   def       __UpperCamelCase  (	self     : List[Any]							):
     SCREAMING_SNAKE_CASE:int   =   self.get_feature_extractor()
     SCREAMING_SNAKE_CASE:Tuple   =   self.get_tokenizer()

     SCREAMING_SNAKE_CASE:Union[str, Any]   =   WhisperProcessor(tokenizer=SCREAMING_SNAKE_CASE__       ,feature_extractor=SCREAMING_SNAKE_CASE__							)

     SCREAMING_SNAKE_CASE:str   =   "This is a test string"

     SCREAMING_SNAKE_CASE:int   =   processor(text=SCREAMING_SNAKE_CASE__							)

     SCREAMING_SNAKE_CASE:str   =   tokenizer(SCREAMING_SNAKE_CASE__							)

     for key in encoded_tok.keys():
       self.assertListEqual(encoded_tok[key]       ,encoded_processor[key]							)
   def       __UpperCamelCase  (	self     : str							):
     SCREAMING_SNAKE_CASE:List[Any]   =   self.get_feature_extractor()
     SCREAMING_SNAKE_CASE:List[str]   =   self.get_tokenizer()

     SCREAMING_SNAKE_CASE:Union[str, Any]   =   WhisperProcessor(tokenizer=SCREAMING_SNAKE_CASE__       ,feature_extractor=SCREAMING_SNAKE_CASE__							)

     SCREAMING_SNAKE_CASE:Optional[Any]   =   [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

     SCREAMING_SNAKE_CASE:Any   =   processor.batch_decode(SCREAMING_SNAKE_CASE__							)
     SCREAMING_SNAKE_CASE:Union[str, Any]   =   tokenizer.batch_decode(SCREAMING_SNAKE_CASE__							)

     self.assertListEqual(SCREAMING_SNAKE_CASE__       ,SCREAMING_SNAKE_CASE__							)
   def       __UpperCamelCase  (	self     : int							):
     SCREAMING_SNAKE_CASE:Tuple   =   self.get_feature_extractor()
     SCREAMING_SNAKE_CASE:Optional[int]   =   self.get_tokenizer()

     SCREAMING_SNAKE_CASE:Union[str, Any]   =   WhisperProcessor(tokenizer=SCREAMING_SNAKE_CASE__       ,feature_extractor=SCREAMING_SNAKE_CASE__							)

     self.assertListEqual(
         processor.model_input_names       ,feature_extractor.model_input_names       ,msg="`processor` and `feature_extractor` model input names do not match"       ,)
   def       __UpperCamelCase  (	self     : int							):
     SCREAMING_SNAKE_CASE:Tuple   =   self.get_feature_extractor()
     SCREAMING_SNAKE_CASE:Any   =   self.get_tokenizer()

     SCREAMING_SNAKE_CASE:Any   =   WhisperProcessor(tokenizer=SCREAMING_SNAKE_CASE__       ,feature_extractor=SCREAMING_SNAKE_CASE__							)
     SCREAMING_SNAKE_CASE:Optional[int]   =   processor.get_decoder_prompt_ids(task="transcribe"       ,no_timestamps=SCREAMING_SNAKE_CASE__							)

     self.assertIsInstance(SCREAMING_SNAKE_CASE__       ,SCREAMING_SNAKE_CASE__							)
     for ids in forced_decoder_ids:
       self.assertIsInstance(SCREAMING_SNAKE_CASE__       ,(list, tuple)							)

     SCREAMING_SNAKE_CASE:Any   =   [TRANSCRIBE, NOTIMESTAMPS]
     self.assertListEqual([ids[-1] for ids in forced_decoder_ids]       ,SCREAMING_SNAKE_CASE__							)
   def       __UpperCamelCase  (	self     : Dict							):
     SCREAMING_SNAKE_CASE:Tuple   =   WhisperProcessor(tokenizer=self.get_tokenizer()       ,feature_extractor=self.get_feature_extractor()							)
     SCREAMING_SNAKE_CASE:Dict   =   processor.get_prompt_ids("Mr. Quilter"							)
     SCREAMING_SNAKE_CASE:int   =   processor.tokenizer.decode(SCREAMING_SNAKE_CASE__							)

     self.assertListEqual(prompt_ids.tolist()       ,[50_360, 1_770, 13, 2_264, 346, 353]							)
     self.assertEqual(SCREAMING_SNAKE_CASE__       ,"<|startofprev|> Mr. Quilter"							)
   def       __UpperCamelCase  (	self     : Union[str, Any]							):
     SCREAMING_SNAKE_CASE:Dict   =   WhisperProcessor(tokenizer=self.get_tokenizer()       ,feature_extractor=self.get_feature_extractor()							)
     SCREAMING_SNAKE_CASE:List[Any]   =   processor.get_prompt_ids(""							)
     SCREAMING_SNAKE_CASE:Tuple   =   processor.tokenizer.decode(SCREAMING_SNAKE_CASE__							)

     self.assertListEqual(prompt_ids.tolist()       ,[50_360, 220]							)
     self.assertEqual(SCREAMING_SNAKE_CASE__       ,"<|startofprev|> "							)







   def       __UpperCamelCase  (	self     : Union[str, Any]							):
     SCREAMING_SNAKE_CASE:Tuple   =   WhisperProcessor(tokenizer=self.get_tokenizer()       ,feature_extractor=self.get_feature_extractor()							)

     def _test_prompt_error_raised_helper(SCREAMING_SNAKE_CASE__     : int       ,SCREAMING_SNAKE_CASE__     : List[str]							):
       with pytest.raises(SCREAMING_SNAKE_CASE__							) as excinfo:
         processor.get_prompt_ids(SCREAMING_SNAKE_CASE__							)
       SCREAMING_SNAKE_CASE:List[Any]   =   F'''Encountered text in the prompt corresponding to disallowed special token: {special_token}.'''
       self.assertEqual(SCREAMING_SNAKE_CASE__       ,str(excinfo.value							)							)

     _test_prompt_error_raised_helper("<|startofprev|> test"       ,"<|startofprev|>"							)
     _test_prompt_error_raised_helper("test <|notimestamps|>"       ,"<|notimestamps|>"							)
     _test_prompt_error_raised_helper("test <|zh|> test <|transcribe|>"       ,"<|zh|>"							)






