






"""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


lowercase_    					=					5_0358
lowercase_    					=					5_0362


@require_torch
@require_torchaudio
@require_sentencepiece
class  snake_case		(							unittest.TestCase						):


   '''simple docstring'''

   def     _SCREAMING_SNAKE_CASE    (     self			:					List[Any]  ):

     '''simple docstring'''
     __A						      =			'''openai/whisper-small.en'''
     __A						      =			tempfile.mkdtemp()


   def     _SCREAMING_SNAKE_CASE    (     self			:					Tuple,	**_lowerCamelCase			:					str  ):

     '''simple docstring'''
     return WhisperTokenizer.from_pretrained(self.checkpoint,	**_lowerCamelCase  )


   def     _SCREAMING_SNAKE_CASE    (     self			:					Dict,	**_lowerCamelCase			:					List[Any]  ):

     '''simple docstring'''
     return WhisperFeatureExtractor.from_pretrained(self.checkpoint,	**_lowerCamelCase  )


   def     _SCREAMING_SNAKE_CASE    (     self			:					List[str]  ):

     '''simple docstring'''
     shutil.rmtree(self.tmpdirname  )


   def     _SCREAMING_SNAKE_CASE    (     self			:					int  ):

     '''simple docstring'''
     __A						      =			self.get_tokenizer()
     __A						      =			self.get_feature_extractor()

     __A						      =			WhisperProcessor(tokenizer=_lowerCamelCase,	feature_extractor=_lowerCamelCase  )

     processor.save_pretrained(self.tmpdirname  )
     __A						      =			WhisperProcessor.from_pretrained(self.tmpdirname  )

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

     self.assertEqual(processor.feature_extractor.to_json_string(),	feature_extractor.to_json_string()  )
     self.assertIsInstance(processor.feature_extractor,	_lowerCamelCase  )


   def     _SCREAMING_SNAKE_CASE    (     self			:					Any  ):

     '''simple docstring'''
     __A						      =			WhisperProcessor(tokenizer=self.get_tokenizer(),	feature_extractor=self.get_feature_extractor()  )
     processor.save_pretrained(self.tmpdirname  )

     __A						      =			self.get_tokenizer(bos_token='''(BOS)''',	eos_token='''(EOS)'''  )
     __A						      =			self.get_feature_extractor(do_normalize=_lowerCamelCase,	padding_value=1.0  )

     __A						      =			WhisperProcessor.from_pretrained(
         self.tmpdirname,	bos_token='''(BOS)''',	eos_token='''(EOS)''',	do_normalize=_lowerCamelCase,	padding_value=1.0  )

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

     self.assertEqual(processor.feature_extractor.to_json_string(),	feature_extractor_add_kwargs.to_json_string()  )
     self.assertIsInstance(processor.feature_extractor,	_lowerCamelCase  )


   def     _SCREAMING_SNAKE_CASE    (     self			:					Optional[int]  ):

     '''simple docstring'''
     __A						      =			self.get_feature_extractor()
     __A						      =			self.get_tokenizer()

     __A						      =			WhisperProcessor(tokenizer=_lowerCamelCase,	feature_extractor=_lowerCamelCase  )

     __A						      =			floats_list((3, 10_00)  )

     __A						      =			feature_extractor(_lowerCamelCase,	return_tensors='''np'''  )
     __A						      =			processor(_lowerCamelCase,	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     _SCREAMING_SNAKE_CASE    (     self			:					Union[str, Any]  ):

     '''simple docstring'''
     __A						      =			self.get_feature_extractor()
     __A						      =			self.get_tokenizer()

     __A						      =			WhisperProcessor(tokenizer=_lowerCamelCase,	feature_extractor=_lowerCamelCase  )

     __A						      =			'''This is a test string'''

     __A						      =			processor(text=_lowerCamelCase  )

     __A						      =			tokenizer(_lowerCamelCase  )

     for key in encoded_tok.keys():
       self.assertListEqual(encoded_tok[key],	encoded_processor[key]  )


   def     _SCREAMING_SNAKE_CASE    (     self			:					Union[str, Any]  ):

     '''simple docstring'''
     __A						      =			self.get_feature_extractor()
     __A						      =			self.get_tokenizer()

     __A						      =			WhisperProcessor(tokenizer=_lowerCamelCase,	feature_extractor=_lowerCamelCase  )

     __A						      =			[[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

     __A						      =			processor.batch_decode(_lowerCamelCase  )
     __A						      =			tokenizer.batch_decode(_lowerCamelCase  )

     self.assertListEqual(_lowerCamelCase,	_lowerCamelCase  )


   def     _SCREAMING_SNAKE_CASE    (     self			:					str  ):

     '''simple docstring'''
     __A						      =			self.get_feature_extractor()
     __A						      =			self.get_tokenizer()

     __A						      =			WhisperProcessor(tokenizer=_lowerCamelCase,	feature_extractor=_lowerCamelCase  )

     self.assertListEqual(
         processor.model_input_names,	feature_extractor.model_input_names,	msg='''`processor` and `feature_extractor` model input names do not match''',	)


   def     _SCREAMING_SNAKE_CASE    (     self			:					List[str]  ):

     '''simple docstring'''
     __A						      =			self.get_feature_extractor()
     __A						      =			self.get_tokenizer()

     __A						      =			WhisperProcessor(tokenizer=_lowerCamelCase,	feature_extractor=_lowerCamelCase  )
     __A						      =			processor.get_decoder_prompt_ids(task='''transcribe''',	no_timestamps=_lowerCamelCase  )

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

     __A						      =			[TRANSCRIBE, NOTIMESTAMPS]
     self.assertListEqual([ids[-1] for ids in forced_decoder_ids],	_lowerCamelCase  )


   def     _SCREAMING_SNAKE_CASE    (     self			:					int  ):

     '''simple docstring'''
     __A						      =			WhisperProcessor(tokenizer=self.get_tokenizer(),	feature_extractor=self.get_feature_extractor()  )
     __A						      =			processor.get_prompt_ids('''Mr. Quilter'''  )
     __A						      =			processor.tokenizer.decode(_lowerCamelCase  )

     self.assertListEqual(prompt_ids.tolist(),	[5_03_60, 17_70, 13, 22_64, 3_46, 3_53]  )
     self.assertEqual(_lowerCamelCase,	'''<|startofprev|> Mr. Quilter'''  )


   def     _SCREAMING_SNAKE_CASE    (     self			:					str  ):

     '''simple docstring'''
     __A						      =			WhisperProcessor(tokenizer=self.get_tokenizer(),	feature_extractor=self.get_feature_extractor()  )
     __A						      =			processor.get_prompt_ids(''''''  )
     __A						      =			processor.tokenizer.decode(_lowerCamelCase  )

     self.assertListEqual(prompt_ids.tolist(),	[5_03_60, 2_20]  )
     self.assertEqual(_lowerCamelCase,	'''<|startofprev|> '''  )






   def     _SCREAMING_SNAKE_CASE    (     self			:					Union[str, Any]  ):

     '''simple docstring'''
     __A						      =			WhisperProcessor(tokenizer=self.get_tokenizer(),	feature_extractor=self.get_feature_extractor()  )

     def _test_prompt_error_raised_helper(_lowerCamelCase			:					Any,	_lowerCamelCase			:					int  ):
       with pytest.raises(_lowerCamelCase  ) as excinfo:
         processor.get_prompt_ids(_lowerCamelCase  )
       __A						      =			f'Encountered text in the prompt corresponding to disallowed special token: {special_token}.'
       self.assertEqual(_lowerCamelCase,	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|>'''  )












"""simple docstring"""


# This test is meant to be run in on an instance with TPUs like this:
#
#   python examples/pytorch/xla_spawn.py --num_cores=8 tests/test_trainer_tpu.py
#
# Replace 8 with the number of TPU cores you have.
#

import sys
from typing import Dict

from transformers import EvalPrediction, HfArgumentParser, TrainingArguments, is_torch_available
from transformers.utils import logging


lowercase_    					=					logging.get_logger(__name__)


if is_torch_available():
       import torch
       from torch import nn
       from torch.utils.data import Dataset

       from transformers import Trainer


       class  snake_case		(							_lowerCAmelCase						):


          '''simple docstring'''

          def __init__(     self			:					List[Any],	_lowerCamelCase			:					int = 1_01  ):

            '''simple docstring'''
            __A						      =			length


          def __len__(     self			:					Optional[int]  ):

            '''simple docstring'''
            return self.length






          def __getitem__(     self			:					List[str],	_lowerCamelCase			:					Union[str, Any]  ):

            '''simple docstring'''
            return i





       class  snake_case		:


          '''simple docstring'''

          def __call__(     self			:					Tuple,	_lowerCamelCase			:					Dict  ):

            '''simple docstring'''
            return {"input_ids": torch.tensor(_lowerCamelCase  ), "labels": torch.tensor(_lowerCamelCase  )}





       class  snake_case		(							nn.Module						):


          '''simple docstring'''

          def __init__(     self			:					List[str]  ):

            '''simple docstring'''
            super().__init__()
            # Add some (unused) params otherwise DDP will complain.
            __A						      =			nn.Linear(1_20,	80  )






          def     _SCREAMING_SNAKE_CASE    (     self			:					Optional[Any],	_lowerCamelCase			:					Tuple,	_lowerCamelCase			:					Optional[Any]=None  ):

            '''simple docstring'''
            if labels is not None:
              return torch.tensor(0.0,	device=input_ids.device  ), input_ids
            else:
              return input_ids



def  lowerCAmelCase    (   ):




  """simple docstring"""







  __A						      =			HfArgumentParser((TrainingArguments,)			)
  sys.argv += ["--output_dir", "./examples"]
  __A						      =			parser.parse_args_into_dataclasses()[0]

  logger.warning(
      f'Process rank: {training_args.local_rank}, device: {training_args.device}, '
      f'tpu_num_cores: {training_args.tpu_num_cores}'		,	)

  # Essentially, what we want to verify in the distributed case is
  # that we get all samples back, in the right order.
  # (this is crucial for prediction for instance)
  for dataset_length in [1_0_0_1, 2_5_6, 1_5]:
    __A						      =			DummyDataset(__UpperCamelCase			)

    def compute_metrics(__UpperCamelCase			) -> Dict:
      __A						      =			list(range(len(__UpperCamelCase			)			)			)
      __A						      =			p.predictions.tolist() == sequential and p.label_ids.tolist() == sequential
      return {"success": success}

    __A						      =			Trainer(
        model=DummyModel()		,	args=__UpperCamelCase		,	data_collator=DummyDataCollator()		,	eval_dataset=__UpperCamelCase		,	compute_metrics=__UpperCamelCase		,	)
    __A						      =			trainer.evaluate()
    logger.info(__UpperCamelCase			)
    if metrics["eval_success"] is not True:
      logger.error(__UpperCamelCase			)
      exit(1			)

    __A						      =			trainer.predict(__UpperCamelCase			)
    logger.info(p.metrics			)
    if p.metrics["test_success"] is not True:
      logger.error(p.metrics			)
      exit(1			)

    __A						      =			2

    __A						      =			trainer.evaluate()
    logger.info(__UpperCamelCase			)
    if metrics["eval_success"] is not True:
      logger.error(__UpperCamelCase			)
      exit(1			)

    __A						      =			trainer.predict(__UpperCamelCase			)
    logger.info(p.metrics			)
    if p.metrics["test_success"] is not True:
      logger.error(p.metrics			)
      exit(1			)

    __A						      =			None

  logger.info('''🔥 All distributed tests successful'''			)



def  lowerCAmelCase    (   __UpperCamelCase			):




  """simple docstring"""







  main()


if __name__ == "__main__":
       main()




