import json
import os
import shutil
import tempfile
import unittest
from multiprocessing import get_context
from pathlib import Path

import datasets
import numpy as np
from datasets import load_dataset
from parameterized import parameterized

from transformers import AutoProcessor
from transformers.models.wavaveca import WavaVecaCTCTokenizer, WavaVecaFeatureExtractor
from transformers.models.wavaveca.tokenization_wavaveca import VOCAB_FILES_NAMES
from transformers.testing_utils import require_pyctcdecode, require_torch, require_torchaudio, slow
from transformers.utils import FEATURE_EXTRACTOR_NAME, is_pyctcdecode_available, is_torch_available

from ..wavaveca.test_feature_extraction_wavaveca import floats_list


if is_pyctcdecode_available():
      from huggingface_hub import snapshot_download
      from pyctcdecode import BeamSearchDecoderCTC

      from transformers.models.wavaveca_with_lm import WavaVecaProcessorWithLM
      from transformers.models.wavaveca_with_lm.processing_wavaveca_with_lm import WavaVecaDecoderWithLMOutput

if is_torch_available():
      from transformers import WavaVecaForCTC






@require_pyctcdecode
class 		_lowerCamelCase		(   unittest.TestCase    ):
    def 						UpperCamelCase_				(  self   )       ->					Tuple:
     SCREAMING_SNAKE_CASE__:   int=      '''| <pad> <unk> <s> </s> a b c d e f g h i j k'''.split()
     SCREAMING_SNAKE_CASE__:   Any=      dict(zip(lowerCAmelCase , range(len(lowerCAmelCase   )   )   )   )

     SCREAMING_SNAKE_CASE__:   Optional[int]=      {
         '''unk_token''': '''<unk>''',
         '''bos_token''': '''<s>''',
         '''eos_token''': '''</s>''',
     }
     SCREAMING_SNAKE_CASE__:   str=      {
         '''feature_size''': 1,
         '''padding_value''': 0.0,
         '''sampling_rate''': 16000,
         '''return_attention_mask''': False,
         '''do_normalize''': True,
     }

     SCREAMING_SNAKE_CASE__:   str=      tempfile.mkdtemp()
     SCREAMING_SNAKE_CASE__:   Dict=      os.path.join(self.tmpdirname , VOCAB_FILES_NAMES['''vocab_file''']   )
     SCREAMING_SNAKE_CASE__:   Optional[int]=      os.path.join(self.tmpdirname , lowerCAmelCase   )
     with open(self.vocab_file , '''w''' , encoding='''utf-8'''   ) as fp:
      fp.write(json.dumps(lowerCAmelCase   ) + '''\n'''   )

     with open(self.feature_extraction_file , '''w''' , encoding='''utf-8'''   ) as fp:
      fp.write(json.dumps(lowerCAmelCase   ) + '''\n'''   )

     # load decoder from hub
     SCREAMING_SNAKE_CASE__:   Optional[int]=      '''hf-internal-testing/ngram-beam-search-decoder'''
    def 						UpperCamelCase_				(  self , **lowerCAmelCase   )       ->					Optional[Any]:
     SCREAMING_SNAKE_CASE__:   Tuple=      self.add_kwargs_tokens_map.copy()
     kwargs.update(lowerCAmelCase   )
     return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname , **lowerCAmelCase   )
    def 						UpperCamelCase_				(  self , **lowerCAmelCase   )       ->					Dict:
     return WavaVecaFeatureExtractor.from_pretrained(self.tmpdirname , **lowerCAmelCase   )
    def 						UpperCamelCase_				(  self , **lowerCAmelCase   )       ->					Dict:
     return BeamSearchDecoderCTC.load_from_hf_hub(self.decoder_name , **lowerCAmelCase   )
    def 						UpperCamelCase_				(  self   )       ->					Any:
     shutil.rmtree(self.tmpdirname   )
    def 						UpperCamelCase_				(  self   )       ->					Optional[Any]:
     SCREAMING_SNAKE_CASE__:   Optional[Any]=      self.get_tokenizer()
     SCREAMING_SNAKE_CASE__:   Dict=      self.get_feature_extractor()
     SCREAMING_SNAKE_CASE__:   Dict=      self.get_decoder()

     SCREAMING_SNAKE_CASE__:   Tuple=      WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase , feature_extractor=lowerCAmelCase , decoder=lowerCAmelCase   )

     processor.save_pretrained(self.tmpdirname   )
     SCREAMING_SNAKE_CASE__:   List[str]=      WavaVecaProcessorWithLM.from_pretrained(self.tmpdirname   )

     # tokenizer
     self.assertEqual(processor.tokenizer.get_vocab() , tokenizer.get_vocab()   )
     self.assertIsInstance(processor.tokenizer , lowerCAmelCase   )

     # feature extractor
     self.assertEqual(processor.feature_extractor.to_json_string() , feature_extractor.to_json_string()   )
     self.assertIsInstance(processor.feature_extractor , lowerCAmelCase   )

     # decoder
     self.assertEqual(processor.decoder._alphabet.labels , decoder._alphabet.labels   )
     self.assertEqual(
         processor.decoder.model_container[decoder._model_key]._unigram_set , decoder.model_container[decoder._model_key]._unigram_set , )
     self.assertIsInstance(processor.decoder , lowerCAmelCase   )
    def 						UpperCamelCase_				(  self   )       ->					List[str]:
     SCREAMING_SNAKE_CASE__:   int=      WavaVecaProcessorWithLM(
         tokenizer=self.get_tokenizer() , feature_extractor=self.get_feature_extractor() , decoder=self.get_decoder()   )
     processor.save_pretrained(self.tmpdirname   )

     # make sure that error is thrown when decoder alphabet doesn't match
     SCREAMING_SNAKE_CASE__:   List[str]=      WavaVecaProcessorWithLM.from_pretrained(
         self.tmpdirname , alpha=5.0 , beta=3.0 , score_boundary=-7.0 , unk_score_offset=3   )

     # decoder
     self.assertEqual(processor.language_model.alpha , 5.0   )
     self.assertEqual(processor.language_model.beta , 3.0   )
     self.assertEqual(processor.language_model.score_boundary , -7.0   )
     self.assertEqual(processor.language_model.unk_score_offset , 3   )
    def 						UpperCamelCase_				(  self   )       ->					List[str]:
     SCREAMING_SNAKE_CASE__:   int=      self.get_tokenizer()
     # add token to trigger raise
     tokenizer.add_tokens(['''xx''']   )
     with self.assertRaisesRegex(lowerCAmelCase , '''include'''   ):
      WavaVecaProcessorWithLM(
          tokenizer=lowerCAmelCase , feature_extractor=self.get_feature_extractor() , decoder=self.get_decoder()   )
    def 						UpperCamelCase_				(  self   )       ->					List[Any]:
     SCREAMING_SNAKE_CASE__:   int=      self.get_feature_extractor()
     SCREAMING_SNAKE_CASE__:   List[str]=      self.get_tokenizer()
     SCREAMING_SNAKE_CASE__:   Optional[Any]=      self.get_decoder()

     SCREAMING_SNAKE_CASE__:   Any=      WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase , feature_extractor=lowerCAmelCase , decoder=lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   Optional[int]=      floats_list((3, 1000)   )

     SCREAMING_SNAKE_CASE__:   int=      feature_extractor(lowerCAmelCase , return_tensors='''np'''   )
     SCREAMING_SNAKE_CASE__:   Tuple=      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 						UpperCamelCase_				(  self   )       ->					Union[str, Any]:
     SCREAMING_SNAKE_CASE__:   str=      self.get_feature_extractor()
     SCREAMING_SNAKE_CASE__:   str=      self.get_tokenizer()
     SCREAMING_SNAKE_CASE__:   Dict=      self.get_decoder()

     SCREAMING_SNAKE_CASE__:   Any=      WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase , feature_extractor=lowerCAmelCase , decoder=lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   str=      '''This is a test string'''

     SCREAMING_SNAKE_CASE__:   Any=      processor(text=lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   Optional[int]=      tokenizer(lowerCAmelCase   )

     for key in encoded_tok.keys():
      self.assertListEqual(encoded_tok[key] , encoded_processor[key]   )
    def 						UpperCamelCase_				(  self , lowerCAmelCase=(2, 10, 16) , lowerCAmelCase=77   )       ->					Any:
     np.random.seed(lowerCAmelCase   )
     return np.random.rand(*lowerCAmelCase   )
    def 						UpperCamelCase_				(  self   )       ->					Any:
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      self.get_feature_extractor()
     SCREAMING_SNAKE_CASE__:   Dict=      self.get_tokenizer()
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      self.get_decoder()

     SCREAMING_SNAKE_CASE__:   Optional[Any]=      WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase , feature_extractor=lowerCAmelCase , decoder=lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   Optional[Any]=      self._get_dummy_logits(shape=(10, 16) , seed=13   )

     SCREAMING_SNAKE_CASE__:   Any=      processor.decode(lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   Any=      decoder.decode_beams(lowerCAmelCase   )[0]

     self.assertEqual(decoded_decoder[0] , decoded_processor.text   )
     self.assertEqual('''</s> <s> </s>''' , decoded_processor.text   )
     self.assertEqual(decoded_decoder[-2] , decoded_processor.logit_score   )
     self.assertEqual(decoded_decoder[-1] , decoded_processor.lm_score   )
    @parameterized.expand([[None], ['''fork'''], ['''spawn''']]   )
    def 						UpperCamelCase_				(  self , lowerCAmelCase   )       ->					Union[str, Any]:
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      self.get_feature_extractor()
     SCREAMING_SNAKE_CASE__:   Tuple=      self.get_tokenizer()
     SCREAMING_SNAKE_CASE__:   Dict=      self.get_decoder()

     SCREAMING_SNAKE_CASE__:   Dict=      WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase , feature_extractor=lowerCAmelCase , decoder=lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      self._get_dummy_logits()

     # note: pool should be instantiated *after* Wav2Vec2ProcessorWithLM.
     #       otherwise, the LM won't be available to the pool's sub-processes.
     # manual logic used to allow parameterized test for both pool=None and pool=Pool(...)
     if pool_context is None:
      SCREAMING_SNAKE_CASE__:   List[str]=      processor.batch_decode(lowerCAmelCase   )
     else:
      with get_context(lowerCAmelCase   ).Pool() as pool:
       SCREAMING_SNAKE_CASE__:   Optional[int]=      processor.batch_decode(lowerCAmelCase , lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   str=      list(lowerCAmelCase   )

     with get_context('''fork'''   ).Pool() as p:
      SCREAMING_SNAKE_CASE__:   Optional[Any]=      decoder.decode_beams_batch(lowerCAmelCase , lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   Optional[Any]=      [], [], []
     for beams in decoded_beams:
      texts_decoder.append(beams[0][0]   )
      logit_scores_decoder.append(beams[0][-2]   )
      lm_scores_decoder.append(beams[0][-1]   )

     self.assertListEqual(lowerCAmelCase , decoded_processor.text   )
     self.assertListEqual(['''<s> <s> </s>''', '''<s> <s> <s>'''] , decoded_processor.text   )
     self.assertListEqual(lowerCAmelCase , decoded_processor.logit_score   )
     self.assertListEqual(lowerCAmelCase , decoded_processor.lm_score   )
    def 						UpperCamelCase_				(  self   )       ->					List[Any]:
     SCREAMING_SNAKE_CASE__:   Tuple=      self.get_feature_extractor()
     SCREAMING_SNAKE_CASE__:   str=      self.get_tokenizer()
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      self.get_decoder()

     SCREAMING_SNAKE_CASE__:   Optional[Any]=      WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase , feature_extractor=lowerCAmelCase , decoder=lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      self._get_dummy_logits()

     SCREAMING_SNAKE_CASE__:   str=      15
     SCREAMING_SNAKE_CASE__:   Tuple=      -20.0
     SCREAMING_SNAKE_CASE__:   Any=      -4.0

     SCREAMING_SNAKE_CASE__:   Tuple=      processor.batch_decode(
         lowerCAmelCase , beam_width=lowerCAmelCase , beam_prune_logp=lowerCAmelCase , token_min_logp=lowerCAmelCase , )
     SCREAMING_SNAKE_CASE__:   str=      decoded_processor_out.text

     SCREAMING_SNAKE_CASE__:   Optional[int]=      list(lowerCAmelCase   )

     with get_context('''fork'''   ).Pool() as pool:
      SCREAMING_SNAKE_CASE__:   Optional[Any]=      decoder.decode_beams_batch(
          lowerCAmelCase , lowerCAmelCase , beam_width=lowerCAmelCase , beam_prune_logp=lowerCAmelCase , token_min_logp=lowerCAmelCase , )

     SCREAMING_SNAKE_CASE__:   Optional[int]=      [d[0][0] for d in decoded_decoder_out]
     SCREAMING_SNAKE_CASE__:   str=      [d[0][2] for d in decoded_decoder_out]
     SCREAMING_SNAKE_CASE__:   str=      [d[0][3] for d in decoded_decoder_out]

     self.assertListEqual(lowerCAmelCase , lowerCAmelCase   )
     self.assertListEqual(['''</s> <s> <s>''', '''<s> <s> <s>'''] , lowerCAmelCase   )

     self.assertTrue(np.array_equal(lowerCAmelCase , decoded_processor_out.logit_score   )   )
     self.assertTrue(np.allclose([-20.054, -18.447] , lowerCAmelCase , atol=1e-3   )   )

     self.assertTrue(np.array_equal(lowerCAmelCase , decoded_processor_out.lm_score   )   )
     self.assertTrue(np.allclose([-15.554, -13.9474] , lowerCAmelCase , atol=1e-3   )   )
    def 						UpperCamelCase_				(  self   )       ->					Any:
     SCREAMING_SNAKE_CASE__:   Tuple=      self.get_feature_extractor()
     SCREAMING_SNAKE_CASE__:   List[str]=      self.get_tokenizer()
     SCREAMING_SNAKE_CASE__:   List[Any]=      self.get_decoder()

     SCREAMING_SNAKE_CASE__:   List[str]=      WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase , feature_extractor=lowerCAmelCase , decoder=lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   str=      self._get_dummy_logits()

     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      2.0
     SCREAMING_SNAKE_CASE__:   str=      5.0
     SCREAMING_SNAKE_CASE__:   Optional[Any]=      -20.0
     SCREAMING_SNAKE_CASE__:   Dict=      True

     SCREAMING_SNAKE_CASE__:   str=      processor.batch_decode(
         lowerCAmelCase , alpha=lowerCAmelCase , beta=lowerCAmelCase , unk_score_offset=lowerCAmelCase , lm_score_boundary=lowerCAmelCase , )
     SCREAMING_SNAKE_CASE__:   Dict=      decoded_processor_out.text

     SCREAMING_SNAKE_CASE__:   Any=      list(lowerCAmelCase   )
     decoder.reset_params(
         alpha=lowerCAmelCase , beta=lowerCAmelCase , unk_score_offset=lowerCAmelCase , lm_score_boundary=lowerCAmelCase , )

     with get_context('''fork'''   ).Pool() as pool:
      SCREAMING_SNAKE_CASE__:   List[str]=      decoder.decode_beams_batch(
          lowerCAmelCase , lowerCAmelCase , )

     SCREAMING_SNAKE_CASE__:   Tuple=      [d[0][0] for d in decoded_decoder_out]

     self.assertListEqual(lowerCAmelCase , lowerCAmelCase   )
     self.assertListEqual(['''<s> </s> <s> </s> </s>''', '''</s> </s> <s> </s> </s>'''] , lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   Any=      processor.decoder.model_container[processor.decoder._model_key]
     self.assertEqual(lm_model.alpha , 2.0   )
     self.assertEqual(lm_model.beta , 5.0   )
     self.assertEqual(lm_model.unk_score_offset , -20.0   )
     self.assertEqual(lm_model.score_boundary , lowerCAmelCase   )
    def 						UpperCamelCase_				(  self   )       ->					Optional[Any]:
     SCREAMING_SNAKE_CASE__:   List[Any]=      WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''   )

     SCREAMING_SNAKE_CASE__:   List[str]=      processor.decoder.model_container[processor.decoder._model_key]
     SCREAMING_SNAKE_CASE__:   Tuple=      Path(language_model._kenlm_model.path.decode('''utf-8'''   )   ).parent.parent.absolute()

     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      os.listdir(lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   int=      ['''alphabet.json''', '''language_model''']

     downloaded_decoder_files.sort()
     expected_decoder_files.sort()

     # test that only decoder relevant files from
     # https://huggingface.co/hf-internal-testing/processor_with_lm/tree/main
     # are downloaded and none of the rest (e.g. README.md, ...)
     self.assertListEqual(lowerCAmelCase , lowerCAmelCase   )
    def 						UpperCamelCase_				(  self   )       ->					Optional[Any]:
     SCREAMING_SNAKE_CASE__:   Optional[Any]=      snapshot_download('''hf-internal-testing/processor_with_lm'''   )

     SCREAMING_SNAKE_CASE__:   Tuple=      WavaVecaProcessorWithLM.from_pretrained(lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   Optional[int]=      processor.decoder.model_container[processor.decoder._model_key]
     SCREAMING_SNAKE_CASE__:   Optional[int]=      Path(language_model._kenlm_model.path.decode('''utf-8'''   )   ).parent.parent.absolute()

     SCREAMING_SNAKE_CASE__:   int=      os.listdir(lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   List[str]=      os.listdir(lowerCAmelCase   )

     local_decoder_files.sort()
     expected_decoder_files.sort()

     # test that both decoder form hub and local files in cache are the same
     self.assertListEqual(lowerCAmelCase , lowerCAmelCase   )
    def 						UpperCamelCase_				(  self   )       ->					Union[str, Any]:
     SCREAMING_SNAKE_CASE__:   List[str]=      WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''   )
     SCREAMING_SNAKE_CASE__:   Any=      AutoProcessor.from_pretrained('''hf-internal-testing/processor_with_lm'''   )

     SCREAMING_SNAKE_CASE__:   Tuple=      floats_list((3, 1000)   )

     SCREAMING_SNAKE_CASE__:   List[Any]=      processor_wavaveca(lowerCAmelCase , return_tensors='''np'''   )
     SCREAMING_SNAKE_CASE__:   List[Any]=      processor_auto(lowerCAmelCase , return_tensors='''np'''   )

     for key in input_wavaveca.keys():
      self.assertAlmostEqual(input_wavaveca[key].sum() , input_auto[key].sum() , delta=1e-2   )

     SCREAMING_SNAKE_CASE__:   Optional[Any]=      self._get_dummy_logits()

     SCREAMING_SNAKE_CASE__:   Any=      processor_wavaveca.batch_decode(lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      processor_auto.batch_decode(lowerCAmelCase   )

     self.assertListEqual(decoded_wavaveca.text , decoded_auto.text   )
    def 						UpperCamelCase_				(  self   )       ->					List[str]:
     SCREAMING_SNAKE_CASE__:   List[Any]=      self.get_feature_extractor()
     SCREAMING_SNAKE_CASE__:   str=      self.get_tokenizer()
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      self.get_decoder()

     SCREAMING_SNAKE_CASE__:   Any=      WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase , feature_extractor=lowerCAmelCase , decoder=lowerCAmelCase   )

     self.assertListEqual(
         processor.model_input_names , feature_extractor.model_input_names , msg='''`processor` and `feature_extractor` model input names do not match''' , )
    @staticmethod
    def 						UpperCamelCase_				(  lowerCAmelCase , lowerCAmelCase   )       ->					List[str]:
     SCREAMING_SNAKE_CASE__:   Optional[int]=      [d[key] for d in offsets]
     return retrieved_list
    def 						UpperCamelCase_				(  self   )       ->					Optional[Any]:
     SCREAMING_SNAKE_CASE__:   Optional[int]=      WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''   )
     SCREAMING_SNAKE_CASE__:   List[Any]=      self._get_dummy_logits()[0]

     SCREAMING_SNAKE_CASE__:   Optional[int]=      processor.decode(lowerCAmelCase , output_word_offsets=lowerCAmelCase   )
     # check Wav2Vec2CTCTokenizerOutput keys for word
     self.assertEqual(len(outputs.keys()   ) , 4   )
     self.assertTrue('''text''' in outputs   )
     self.assertTrue('''word_offsets''' in outputs   )
     self.assertTrue(isinstance(lowerCAmelCase , lowerCAmelCase   )   )

     self.assertEqual(''' '''.join(self.get_from_offsets(outputs['''word_offsets'''] , '''word'''   )   ) , outputs.text   )
     self.assertListEqual(self.get_from_offsets(outputs['''word_offsets'''] , '''word'''   ) , ['''<s>''', '''<s>''', '''</s>''']   )
     self.assertListEqual(self.get_from_offsets(outputs['''word_offsets'''] , '''start_offset'''   ) , [0, 2, 4]   )
     self.assertListEqual(self.get_from_offsets(outputs['''word_offsets'''] , '''end_offset'''   ) , [1, 3, 5]   )
    def 						UpperCamelCase_				(  self   )       ->					Dict:
     SCREAMING_SNAKE_CASE__:   Tuple=      WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''   )
     SCREAMING_SNAKE_CASE__:   Tuple=      self._get_dummy_logits()

     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      processor.batch_decode(lowerCAmelCase , output_word_offsets=lowerCAmelCase   )

     # check Wav2Vec2CTCTokenizerOutput keys for word
     self.assertEqual(len(outputs.keys()   ) , 4   )
     self.assertTrue('''text''' in outputs   )
     self.assertTrue('''word_offsets''' in outputs   )
     self.assertTrue(isinstance(lowerCAmelCase , lowerCAmelCase   )   )

     self.assertListEqual(
         [''' '''.join(self.get_from_offsets(lowerCAmelCase , '''word'''   )   ) for o in outputs['''word_offsets''']] , outputs.text   )
     self.assertListEqual(self.get_from_offsets(outputs['''word_offsets'''][0] , '''word'''   ) , ['''<s>''', '''<s>''', '''</s>''']   )
     self.assertListEqual(self.get_from_offsets(outputs['''word_offsets'''][0] , '''start_offset'''   ) , [0, 2, 4]   )
     self.assertListEqual(self.get_from_offsets(outputs['''word_offsets'''][0] , '''end_offset'''   ) , [1, 3, 5]   )



    @slow
    @require_torch
    @require_torchaudio
    def 						UpperCamelCase_				(  self   )       ->					Any:
     import torch

     SCREAMING_SNAKE_CASE__:   List[str]=      load_dataset('''common_voice''' , '''en''' , split='''train''' , streaming=lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   int=      ds.cast_column('''audio''' , datasets.Audio(sampling_rate=16000   )   )
     SCREAMING_SNAKE_CASE__:   Any=      iter(lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   int=      next(lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   Optional[int]=      AutoProcessor.from_pretrained('''patrickvonplaten/wav2vec2-base-100h-with-lm'''   )
     SCREAMING_SNAKE_CASE__:   Dict=      WavaVecaForCTC.from_pretrained('''patrickvonplaten/wav2vec2-base-100h-with-lm'''   )

     # compare to filename `common_voice_en_100038.mp3` of dataset viewer on https://huggingface.co/datasets/common_voice/viewer/en/train
     SCREAMING_SNAKE_CASE__:   str=      processor(sample['''audio''']['''array'''] , return_tensors='''pt'''   ).input_values

     with torch.no_grad():
      SCREAMING_SNAKE_CASE__:   Optional[Any]=      model(lowerCAmelCase   ).logits.cpu().numpy()

     SCREAMING_SNAKE_CASE__:   Tuple=      processor.decode(logits[0] , output_word_offsets=lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   List[Any]=      model.config.inputs_to_logits_ratio / processor.feature_extractor.sampling_rate
     SCREAMING_SNAKE_CASE__:   Optional[Any]=      [
         {
             '''start_time''': d['''start_offset'''] * time_offset,
             '''end_time''': d['''end_offset'''] * time_offset,
             '''word''': d['''word'''],
         }
         for d in output['''word_offsets''']
     ]

     SCREAMING_SNAKE_CASE__:   Dict=      '''WHY DOES MILISANDRA LOOK LIKE SHE WANTS TO CONSUME JOHN SNOW ON THE RIVER AT THE WALL'''

     # output words
     self.assertEqual(''' '''.join(self.get_from_offsets(lowerCAmelCase , '''word'''   )   ) , lowerCAmelCase   )
     self.assertEqual(''' '''.join(self.get_from_offsets(lowerCAmelCase , '''word'''   )   ) , output.text   )

     # output times
     SCREAMING_SNAKE_CASE__:   List[str]=      torch.tensor(self.get_from_offsets(lowerCAmelCase , '''start_time'''   )   )
     SCREAMING_SNAKE_CASE__:   Dict=      torch.tensor(self.get_from_offsets(lowerCAmelCase , '''end_time'''   )   )

     # fmt: off
     SCREAMING_SNAKE_CASE__:   List[Any]=      torch.tensor([1.4199, 1.6599, 2.2599, 3.0, 3.24, 3.5999, 3.7999, 4.0999, 4.26, 4.94, 5.28, 5.6599, 5.78, 5.94, 6.32, 6.5399, 6.6599]   )
     SCREAMING_SNAKE_CASE__:   Optional[int]=      torch.tensor([1.5399, 1.8999, 2.9, 3.16, 3.5399, 3.72, 4.0199, 4.1799, 4.76, 5.1599, 5.5599, 5.6999, 5.86, 6.1999, 6.38, 6.6199, 6.94]   )
     # fmt: on

     self.assertTrue(torch.allclose(lowerCAmelCase , lowerCAmelCase , atol=0.01   )   )
     self.assertTrue(torch.allclose(lowerCAmelCase , lowerCAmelCase , atol=0.01   )   )


import itertools
import os
import random
import tempfile
import unittest

import numpy as np
from datasets import load_dataset

from transformers import is_speech_available
from transformers.testing_utils import check_json_file_has_correct_format, require_torch, require_torchaudio
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_speech_available():
      from transformers import WhisperFeatureExtractor

if is_torch_available():
      import torch

lowercase_     :   Any 				=				random.Random()







def 			A__      (	snake_case_ :		int       ,				snake_case_ :		Tuple=1.0       ,				snake_case_ :		Optional[Any]=None       ,				snake_case_ :		List[Any]=None      ):
 if rng is None:
  SCREAMING_SNAKE_CASE__:   Tuple=      global_rng

 SCREAMING_SNAKE_CASE__:   List[str]=      []
 for batch_idx in range(shape[0]      ):
  values.append([]      )
  for _ in range(shape[1]      ):
   values[-1].append(rng.random() * scale      )

 return values






@require_torch
@require_torchaudio
class 		_lowerCamelCase		(   unittest.TestCase    ):
    def __init__(  self , lowerCAmelCase , lowerCAmelCase=7 , lowerCAmelCase=400 , lowerCAmelCase=2000 , lowerCAmelCase=10 , lowerCAmelCase=160 , lowerCAmelCase=8 , lowerCAmelCase=0.0 , lowerCAmelCase=4000 , lowerCAmelCase=False , lowerCAmelCase=True , )       ->					Tuple:
     SCREAMING_SNAKE_CASE__:   Any=      parent
     SCREAMING_SNAKE_CASE__:   Dict=      batch_size
     SCREAMING_SNAKE_CASE__:   List[Any]=      min_seq_length
     SCREAMING_SNAKE_CASE__:   Tuple=      max_seq_length
     SCREAMING_SNAKE_CASE__:   Optional[int]=      (self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
     SCREAMING_SNAKE_CASE__:   str=      padding_value
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      sampling_rate
     SCREAMING_SNAKE_CASE__:   List[str]=      return_attention_mask
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      do_normalize
     SCREAMING_SNAKE_CASE__:   Optional[int]=      feature_size
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      chunk_length
     SCREAMING_SNAKE_CASE__:   List[Any]=      hop_length
    def 						UpperCamelCase_				(  self   )       ->					str:
     return {
         "feature_size": self.feature_size,
         "hop_length": self.hop_length,
         "chunk_length": self.chunk_length,
         "padding_value": self.padding_value,
         "sampling_rate": self.sampling_rate,
         "return_attention_mask": self.return_attention_mask,
         "do_normalize": self.do_normalize,
     }



    def 						UpperCamelCase_				(  self , lowerCAmelCase=False , lowerCAmelCase=False   )       ->					Optional[Any]:
     def _flatten(lowerCAmelCase   ):
      return list(itertools.chain(*lowerCAmelCase   )   )

     if equal_length:
      SCREAMING_SNAKE_CASE__:   Dict=      [floats_list((self.max_seq_length, self.feature_size)   ) for _ in range(self.batch_size   )]
     else:
      # make sure that inputs increase in size
      SCREAMING_SNAKE_CASE__:   Union[str, Any]=      [
          floats_list((x, self.feature_size)   )
          for x in range(self.min_seq_length , self.max_seq_length , self.seq_length_diff   )
      ]
     if numpify:
      SCREAMING_SNAKE_CASE__:   str=      [np.asarray(lowerCAmelCase   ) for x in speech_inputs]
     return speech_inputs






@require_torch
@require_torchaudio
class 		_lowerCamelCase		(   UpperCamelCase_    ,     unittest.TestCase    ):
    __a  =      WhisperFeatureExtractor if is_speech_available() else None
    def 						UpperCamelCase_				(  self   )       ->					Dict:
     SCREAMING_SNAKE_CASE__:   Any=      WhisperFeatureExtractionTester(self   )
    def 						UpperCamelCase_				(  self   )       ->					int:
     SCREAMING_SNAKE_CASE__:   List[str]=      self.feature_extraction_class(**self.feat_extract_dict   )

     with tempfile.TemporaryDirectory() as tmpdirname:
      SCREAMING_SNAKE_CASE__:   List[str]=      feat_extract_first.save_pretrained(lowerCAmelCase   )[0]
      check_json_file_has_correct_format(lowerCAmelCase   )
      SCREAMING_SNAKE_CASE__:   str=      self.feature_extraction_class.from_pretrained(lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   int=      feat_extract_first.to_dict()
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      feat_extract_second.to_dict()
     SCREAMING_SNAKE_CASE__:   int=      feat_extract_first.mel_filters
     SCREAMING_SNAKE_CASE__:   Any=      feat_extract_second.mel_filters
     self.assertTrue(np.allclose(lowerCAmelCase , lowerCAmelCase   )   )
     self.assertEqual(lowerCAmelCase , lowerCAmelCase   )
    def 						UpperCamelCase_				(  self   )       ->					Tuple:
     SCREAMING_SNAKE_CASE__:   List[Any]=      self.feature_extraction_class(**self.feat_extract_dict   )

     with tempfile.TemporaryDirectory() as tmpdirname:
      SCREAMING_SNAKE_CASE__:   List[str]=      os.path.join(lowerCAmelCase , '''feat_extract.json'''   )
      feat_extract_first.to_json_file(lowerCAmelCase   )
      SCREAMING_SNAKE_CASE__:   Tuple=      self.feature_extraction_class.from_json_file(lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   Tuple=      feat_extract_first.to_dict()
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      feat_extract_second.to_dict()
     SCREAMING_SNAKE_CASE__:   List[str]=      feat_extract_first.mel_filters
     SCREAMING_SNAKE_CASE__:   Optional[int]=      feat_extract_second.mel_filters
     self.assertTrue(np.allclose(lowerCAmelCase , lowerCAmelCase   )   )
     self.assertEqual(lowerCAmelCase , lowerCAmelCase   )
    def 						UpperCamelCase_				(  self   )       ->					Tuple:
     # Tests that all call wrap to encode_plus and batch_encode_plus
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()   )
     # create three inputs of length 800, 1000, and 1200
     SCREAMING_SNAKE_CASE__:   str=      [floats_list((1, x)   )[0] for x in range(800 , 1400 , 200   )]
     SCREAMING_SNAKE_CASE__:   Tuple=      [np.asarray(lowerCAmelCase   ) for speech_input in speech_inputs]

     # Test feature size
     SCREAMING_SNAKE_CASE__:   Optional[Any]=      feature_extractor(lowerCAmelCase , padding='''max_length''' , return_tensors='''np'''   ).input_features
     self.assertTrue(input_features.ndim == 3   )
     self.assertTrue(input_features.shape[-1] == feature_extractor.nb_max_frames   )
     self.assertTrue(input_features.shape[-2] == feature_extractor.feature_size   )

     # Test not batched input
     SCREAMING_SNAKE_CASE__:   List[str]=      feature_extractor(speech_inputs[0] , return_tensors='''np'''   ).input_features
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      feature_extractor(np_speech_inputs[0] , return_tensors='''np'''   ).input_features
     self.assertTrue(np.allclose(lowerCAmelCase , lowerCAmelCase , atol=1e-3   )   )

     # Test batched
     SCREAMING_SNAKE_CASE__:   Optional[int]=      feature_extractor(lowerCAmelCase , return_tensors='''np'''   ).input_features
     SCREAMING_SNAKE_CASE__:   List[Any]=      feature_extractor(lowerCAmelCase , return_tensors='''np'''   ).input_features
     for enc_seq_a, enc_seq_a in zip(lowerCAmelCase , lowerCAmelCase   ):
      self.assertTrue(np.allclose(lowerCAmelCase , lowerCAmelCase , atol=1e-3   )   )

     # Test 2-D numpy arrays are batched.
     SCREAMING_SNAKE_CASE__:   Optional[int]=      [floats_list((1, x)   )[0] for x in (800, 800, 800)]
     SCREAMING_SNAKE_CASE__:   Any=      np.asarray(lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   List[str]=      feature_extractor(lowerCAmelCase , return_tensors='''np'''   ).input_features
     SCREAMING_SNAKE_CASE__:   List[Any]=      feature_extractor(lowerCAmelCase , return_tensors='''np'''   ).input_features
     for enc_seq_a, enc_seq_a in zip(lowerCAmelCase , lowerCAmelCase   ):
      self.assertTrue(np.allclose(lowerCAmelCase , lowerCAmelCase , atol=1e-3   )   )

     # Test truncation required
     SCREAMING_SNAKE_CASE__:   str=      [floats_list((1, x)   )[0] for x in range(200 , (feature_extractor.n_samples + 500) , 200   )]
     SCREAMING_SNAKE_CASE__:   str=      [np.asarray(lowerCAmelCase   ) for speech_input in speech_inputs]

     SCREAMING_SNAKE_CASE__:   str=      [x[: feature_extractor.n_samples] for x in speech_inputs]
     SCREAMING_SNAKE_CASE__:   int=      [np.asarray(lowerCAmelCase   ) for speech_input in speech_inputs_truncated]

     SCREAMING_SNAKE_CASE__:   Dict=      feature_extractor(lowerCAmelCase , return_tensors='''np'''   ).input_features
     SCREAMING_SNAKE_CASE__:   List[str]=      feature_extractor(lowerCAmelCase , return_tensors='''np'''   ).input_features
     for enc_seq_a, enc_seq_a in zip(lowerCAmelCase , lowerCAmelCase   ):
      self.assertTrue(np.allclose(lowerCAmelCase , lowerCAmelCase , atol=1e-3   )   )
    def 						UpperCamelCase_				(  self   )       ->					Dict:
     import torch

     SCREAMING_SNAKE_CASE__:   List[Any]=      self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()   )
     SCREAMING_SNAKE_CASE__:   Dict=      np.random.rand(100 , 32   ).astype(np.floataa   )
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      np_speech_inputs.tolist()

     for inputs in [py_speech_inputs, np_speech_inputs]:
      SCREAMING_SNAKE_CASE__:   Tuple=      feature_extractor.pad([{'''input_features''': inputs}] , return_tensors='''np'''   )
      self.assertTrue(np_processed.input_features.dtype == np.floataa   )
      SCREAMING_SNAKE_CASE__:   Any=      feature_extractor.pad([{'''input_features''': inputs}] , return_tensors='''pt'''   )
      self.assertTrue(pt_processed.input_features.dtype == torch.floataa   )
    def 						UpperCamelCase_				(  self , lowerCAmelCase   )       ->					str:
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      load_dataset('''hf-internal-testing/librispeech_asr_dummy''' , '''clean''' , split='''validation'''   )
     # automatic decoding with librispeech
     SCREAMING_SNAKE_CASE__:   List[str]=      ds.sort('''id'''   ).select(range(lowerCAmelCase   )   )[:num_samples]['''audio''']

     return [x["array"] for x in speech_samples]
    def 						UpperCamelCase_				(  self   )       ->					int:
     # fmt: off
     SCREAMING_SNAKE_CASE__:   Dict=      torch.tensor(
         [
             0.1193, -0.0946, -0.1098, -0.0196, 0.0225, -0.0690, -0.1736, 0.0951,
             0.0971, -0.0817, -0.0702, 0.0162, 0.0260, 0.0017, -0.0192, -0.1678,
             0.0709, -0.1867, -0.0655, -0.0274, -0.0234, -0.1884, -0.0516, -0.0554,
             -0.0274, -0.1425, -0.1423, 0.0837, 0.0377, -0.0854
         ]   )
     # fmt: on

     SCREAMING_SNAKE_CASE__:   str=      self._load_datasamples(1   )
     SCREAMING_SNAKE_CASE__:   Dict=      WhisperFeatureExtractor()
     SCREAMING_SNAKE_CASE__:   Optional[int]=      feature_extractor(lowerCAmelCase , return_tensors='''pt'''   ).input_features
     self.assertEqual(input_features.shape , (1, 80, 3000)   )
     self.assertTrue(torch.allclose(input_features[0, 0, :30] , lowerCAmelCase , atol=1e-4   )   )



    def 						UpperCamelCase_				(  self   )       ->					Optional[Any]:
     SCREAMING_SNAKE_CASE__:   List[str]=      self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()   )
     SCREAMING_SNAKE_CASE__:   Dict=      self._load_datasamples(1   )[0]
     SCREAMING_SNAKE_CASE__:   str=      ((audio - audio.min()) / (audio.max() - audio.min())) * 65535  # Rescale to [0, 65535] to show issue
     SCREAMING_SNAKE_CASE__:   Optional[int]=      feat_extract.zero_mean_unit_var_norm([audio] , attention_mask=lowerCAmelCase   )[0]

     self.assertTrue(np.all(np.mean(lowerCAmelCase   ) < 1e-3   )   )
     self.assertTrue(np.all(np.abs(np.var(lowerCAmelCase   ) - 1   ) < 1e-3   )   )

