


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       a					(       unittest.TestCase      ):

     '''simple docstring'''





     def   lowerCamelCase_       (     self       :      Union[str, Any]    ):
         UpperCAmelCase_										=					"| <pad> <unk> <s> </s> a b c d e f g h i j k".split()
         UpperCAmelCase_										=					dict(zip(_SCREAMING_SNAKE_CASE ,   range(len(_SCREAMING_SNAKE_CASE    )    )    )    )

         UpperCAmelCase_										=					{
             "unk_token": "<unk>",
             "bos_token": "<s>",
             "eos_token": "</s>",
         }
         UpperCAmelCase_										=					{
             "feature_size": 1,
             "padding_value": 0.0,
             "sampling_rate": 1_60_00,
             "return_attention_mask": False,
             "do_normalize": True,
         }

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

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

         # load decoder from hub
         UpperCAmelCase_										=					"hf-internal-testing/ngram-beam-search-decoder"
     def   lowerCamelCase_       (     self       :      Optional[int] ,   **__snake_case       :      Union[str, Any]    ):
         UpperCAmelCase_										=					self.add_kwargs_tokens_map.copy()
         kwargs.update(_SCREAMING_SNAKE_CASE    )
         return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname ,   **_SCREAMING_SNAKE_CASE    )
     def   lowerCamelCase_       (     self       :      Any ,   **__snake_case       :      List[Any]    ):
         return WavaVecaFeatureExtractor.from_pretrained(self.tmpdirname ,   **_SCREAMING_SNAKE_CASE    )
     def   lowerCamelCase_       (     self       :      Optional[int] ,   **__snake_case       :      int    ):
         return BeamSearchDecoderCTC.load_from_hf_hub(self.decoder_name ,   **_SCREAMING_SNAKE_CASE    )
     def   lowerCamelCase_       (     self       :      Optional[Any]    ):
         shutil.rmtree(self.tmpdirname    )
     def   lowerCamelCase_       (     self       :      int    ):
         UpperCAmelCase_										=					self.get_tokenizer()
         UpperCAmelCase_										=					self.get_feature_extractor()
         UpperCAmelCase_										=					self.get_decoder()

         UpperCAmelCase_										=					WavaVecaProcessorWithLM(tokenizer=_SCREAMING_SNAKE_CASE ,   feature_extractor=_SCREAMING_SNAKE_CASE ,   decoder=_SCREAMING_SNAKE_CASE    )

         processor.save_pretrained(self.tmpdirname    )
         UpperCAmelCase_										=					WavaVecaProcessorWithLM.from_pretrained(self.tmpdirname    )

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

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

         # 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 ,   _SCREAMING_SNAKE_CASE    )
     def   lowerCamelCase_       (     self       :      Optional[int]    ):
         UpperCAmelCase_										=					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
         UpperCAmelCase_										=					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   lowerCamelCase_       (     self       :      int    ):
         UpperCAmelCase_										=					self.get_tokenizer()
         # add token to trigger raise
         tokenizer.add_tokens(['''xx''']    )
         with self.assertRaisesRegex(_SCREAMING_SNAKE_CASE ,   '''include'''    ):
             WavaVecaProcessorWithLM(
                 tokenizer=_SCREAMING_SNAKE_CASE ,   feature_extractor=self.get_feature_extractor() ,   decoder=self.get_decoder()    )
     def   lowerCamelCase_       (     self       :      Dict    ):
         UpperCAmelCase_										=					self.get_feature_extractor()
         UpperCAmelCase_										=					self.get_tokenizer()
         UpperCAmelCase_										=					self.get_decoder()

         UpperCAmelCase_										=					WavaVecaProcessorWithLM(tokenizer=_SCREAMING_SNAKE_CASE ,   feature_extractor=_SCREAMING_SNAKE_CASE ,   decoder=_SCREAMING_SNAKE_CASE    )

         UpperCAmelCase_										=					floats_list((3, 10_00)    )

         UpperCAmelCase_										=					feature_extractor(_SCREAMING_SNAKE_CASE ,   return_tensors='''np'''    )
         UpperCAmelCase_										=					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   lowerCamelCase_       (     self       :      Optional[int]    ):
         UpperCAmelCase_										=					self.get_feature_extractor()
         UpperCAmelCase_										=					self.get_tokenizer()
         UpperCAmelCase_										=					self.get_decoder()

         UpperCAmelCase_										=					WavaVecaProcessorWithLM(tokenizer=_SCREAMING_SNAKE_CASE ,   feature_extractor=_SCREAMING_SNAKE_CASE ,   decoder=_SCREAMING_SNAKE_CASE    )

         UpperCAmelCase_										=					"This is a test string"

         UpperCAmelCase_										=					processor(text=_SCREAMING_SNAKE_CASE    )

         UpperCAmelCase_										=					tokenizer(_SCREAMING_SNAKE_CASE    )

         for key in encoded_tok.keys():
             self.assertListEqual(encoded_tok[key] ,   encoded_processor[key]    )
     def   lowerCamelCase_       (     self       :      int ,   __snake_case       :      List[Any]=(2, 10, 16) ,   __snake_case       :      str=77    ):
         np.random.seed(_SCREAMING_SNAKE_CASE    )
         return np.random.rand(*_SCREAMING_SNAKE_CASE    )
     def   lowerCamelCase_       (     self       :      Union[str, Any]    ):
         UpperCAmelCase_										=					self.get_feature_extractor()
         UpperCAmelCase_										=					self.get_tokenizer()
         UpperCAmelCase_										=					self.get_decoder()

         UpperCAmelCase_										=					WavaVecaProcessorWithLM(tokenizer=_SCREAMING_SNAKE_CASE ,   feature_extractor=_SCREAMING_SNAKE_CASE ,   decoder=_SCREAMING_SNAKE_CASE    )

         UpperCAmelCase_										=					self._get_dummy_logits(shape=(10, 16) ,   seed=13    )

         UpperCAmelCase_										=					processor.decode(_SCREAMING_SNAKE_CASE    )

         UpperCAmelCase_										=					decoder.decode_beams(_SCREAMING_SNAKE_CASE    )[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   lowerCamelCase_       (     self       :      Any ,   __snake_case       :      List[Any]    ):
         UpperCAmelCase_										=					self.get_feature_extractor()
         UpperCAmelCase_										=					self.get_tokenizer()
         UpperCAmelCase_										=					self.get_decoder()

         UpperCAmelCase_										=					WavaVecaProcessorWithLM(tokenizer=_SCREAMING_SNAKE_CASE ,   feature_extractor=_SCREAMING_SNAKE_CASE ,   decoder=_SCREAMING_SNAKE_CASE    )

         UpperCAmelCase_										=					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:
             UpperCAmelCase_										=					processor.batch_decode(_SCREAMING_SNAKE_CASE    )
         else:
             with get_context(_SCREAMING_SNAKE_CASE    ).Pool() as pool:
                 UpperCAmelCase_										=					processor.batch_decode(_SCREAMING_SNAKE_CASE ,   _SCREAMING_SNAKE_CASE    )

         UpperCAmelCase_										=					list(_SCREAMING_SNAKE_CASE    )

         with get_context('''fork'''    ).Pool() as p:
             UpperCAmelCase_										=					decoder.decode_beams_batch(_SCREAMING_SNAKE_CASE ,   _SCREAMING_SNAKE_CASE    )

         UpperCAmelCase_										=					[], [], []
         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(_SCREAMING_SNAKE_CASE ,   decoded_processor.text    )
         self.assertListEqual(['''<s> <s> </s>''', '''<s> <s> <s>'''] ,   decoded_processor.text    )
         self.assertListEqual(_SCREAMING_SNAKE_CASE ,   decoded_processor.logit_score    )
         self.assertListEqual(_SCREAMING_SNAKE_CASE ,   decoded_processor.lm_score    )
     def   lowerCamelCase_       (     self       :      Dict    ):
         UpperCAmelCase_										=					self.get_feature_extractor()
         UpperCAmelCase_										=					self.get_tokenizer()
         UpperCAmelCase_										=					self.get_decoder()

         UpperCAmelCase_										=					WavaVecaProcessorWithLM(tokenizer=_SCREAMING_SNAKE_CASE ,   feature_extractor=_SCREAMING_SNAKE_CASE ,   decoder=_SCREAMING_SNAKE_CASE    )

         UpperCAmelCase_										=					self._get_dummy_logits()

         UpperCAmelCase_										=					15
         UpperCAmelCase_										=					-20.0
         UpperCAmelCase_										=					-4.0

         UpperCAmelCase_										=					processor.batch_decode(
             _SCREAMING_SNAKE_CASE ,   beam_width=_SCREAMING_SNAKE_CASE ,   beam_prune_logp=_SCREAMING_SNAKE_CASE ,   token_min_logp=_SCREAMING_SNAKE_CASE ,   )
         UpperCAmelCase_										=					decoded_processor_out.text

         UpperCAmelCase_										=					list(_SCREAMING_SNAKE_CASE    )

         with get_context('''fork'''    ).Pool() as pool:
             UpperCAmelCase_										=					decoder.decode_beams_batch(
                 _SCREAMING_SNAKE_CASE ,   _SCREAMING_SNAKE_CASE ,   beam_width=_SCREAMING_SNAKE_CASE ,   beam_prune_logp=_SCREAMING_SNAKE_CASE ,   token_min_logp=_SCREAMING_SNAKE_CASE ,   )

         UpperCAmelCase_										=					[d[0][0] for d in decoded_decoder_out]
         UpperCAmelCase_										=					[d[0][2] for d in decoded_decoder_out]
         UpperCAmelCase_										=					[d[0][3] for d in decoded_decoder_out]

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

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

         self.assertTrue(np.array_equal(_SCREAMING_SNAKE_CASE ,   decoded_processor_out.lm_score    )    )
         self.assertTrue(np.allclose([-15.554, -13.9_474] ,   _SCREAMING_SNAKE_CASE ,   atol=1E-3    )    )
     def   lowerCamelCase_       (     self       :      Union[str, Any]    ):
         UpperCAmelCase_										=					self.get_feature_extractor()
         UpperCAmelCase_										=					self.get_tokenizer()
         UpperCAmelCase_										=					self.get_decoder()

         UpperCAmelCase_										=					WavaVecaProcessorWithLM(tokenizer=_SCREAMING_SNAKE_CASE ,   feature_extractor=_SCREAMING_SNAKE_CASE ,   decoder=_SCREAMING_SNAKE_CASE    )

         UpperCAmelCase_										=					self._get_dummy_logits()

         UpperCAmelCase_										=					2.0
         UpperCAmelCase_										=					5.0
         UpperCAmelCase_										=					-20.0
         UpperCAmelCase_										=					True

         UpperCAmelCase_										=					processor.batch_decode(
             _SCREAMING_SNAKE_CASE ,   alpha=_SCREAMING_SNAKE_CASE ,   beta=_SCREAMING_SNAKE_CASE ,   unk_score_offset=_SCREAMING_SNAKE_CASE ,   lm_score_boundary=_SCREAMING_SNAKE_CASE ,   )
         UpperCAmelCase_										=					decoded_processor_out.text

         UpperCAmelCase_										=					list(_SCREAMING_SNAKE_CASE    )
         decoder.reset_params(
             alpha=_SCREAMING_SNAKE_CASE ,   beta=_SCREAMING_SNAKE_CASE ,   unk_score_offset=_SCREAMING_SNAKE_CASE ,   lm_score_boundary=_SCREAMING_SNAKE_CASE ,   )

         with get_context('''fork'''    ).Pool() as pool:
             UpperCAmelCase_										=					decoder.decode_beams_batch(
                 _SCREAMING_SNAKE_CASE ,   _SCREAMING_SNAKE_CASE ,   )

         UpperCAmelCase_										=					[d[0][0] for d in decoded_decoder_out]

         self.assertListEqual(_SCREAMING_SNAKE_CASE ,   _SCREAMING_SNAKE_CASE    )
         self.assertListEqual(['''<s> </s> <s> </s> </s>''', '''</s> </s> <s> </s> </s>'''] ,   _SCREAMING_SNAKE_CASE    )
         UpperCAmelCase_										=					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 ,   _SCREAMING_SNAKE_CASE    )
     def   lowerCamelCase_       (     self       :      Dict    ):
         UpperCAmelCase_										=					WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''    )

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

         UpperCAmelCase_										=					os.listdir(_SCREAMING_SNAKE_CASE    )
         UpperCAmelCase_										=					["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(_SCREAMING_SNAKE_CASE ,   _SCREAMING_SNAKE_CASE    )
     def   lowerCamelCase_       (     self       :      Dict    ):
         UpperCAmelCase_										=					snapshot_download('''hf-internal-testing/processor_with_lm'''    )

         UpperCAmelCase_										=					WavaVecaProcessorWithLM.from_pretrained(_SCREAMING_SNAKE_CASE    )

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

         UpperCAmelCase_										=					os.listdir(_SCREAMING_SNAKE_CASE    )
         UpperCAmelCase_										=					os.listdir(_SCREAMING_SNAKE_CASE    )

         local_decoder_files.sort()
         expected_decoder_files.sort()

         # test that both decoder form hub and local files in cache are the same
         self.assertListEqual(_SCREAMING_SNAKE_CASE ,   _SCREAMING_SNAKE_CASE    )
     def   lowerCamelCase_       (     self       :      Any    ):
         UpperCAmelCase_										=					WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''    )
         UpperCAmelCase_										=					AutoProcessor.from_pretrained('''hf-internal-testing/processor_with_lm'''    )

         UpperCAmelCase_										=					floats_list((3, 10_00)    )

         UpperCAmelCase_										=					processor_wavaveca(_SCREAMING_SNAKE_CASE ,   return_tensors='''np'''    )
         UpperCAmelCase_										=					processor_auto(_SCREAMING_SNAKE_CASE ,   return_tensors='''np'''    )

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

         UpperCAmelCase_										=					self._get_dummy_logits()

         UpperCAmelCase_										=					processor_wavaveca.batch_decode(_SCREAMING_SNAKE_CASE    )
         UpperCAmelCase_										=					processor_auto.batch_decode(_SCREAMING_SNAKE_CASE    )

         self.assertListEqual(decoded_wavaveca.text ,   decoded_auto.text    )
     def   lowerCamelCase_       (     self       :      Optional[Any]    ):
         UpperCAmelCase_										=					self.get_feature_extractor()
         UpperCAmelCase_										=					self.get_tokenizer()
         UpperCAmelCase_										=					self.get_decoder()

         UpperCAmelCase_										=					WavaVecaProcessorWithLM(tokenizer=_SCREAMING_SNAKE_CASE ,   feature_extractor=_SCREAMING_SNAKE_CASE ,   decoder=_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''' ,   )
     @staticmethod
     def   lowerCamelCase_       (     __snake_case       :      Any ,   __snake_case       :      List[Any]    ):
         UpperCAmelCase_										=					[d[key] for d in offsets]
         return retrieved_list
     def   lowerCamelCase_       (     self       :      Tuple    ):
         UpperCAmelCase_										=					WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''    )
         UpperCAmelCase_										=					self._get_dummy_logits()[0]

         UpperCAmelCase_										=					processor.decode(_SCREAMING_SNAKE_CASE ,   output_word_offsets=_SCREAMING_SNAKE_CASE    )
         # 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(_SCREAMING_SNAKE_CASE ,   _SCREAMING_SNAKE_CASE    )    )

         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   lowerCamelCase_       (     self       :      Optional[Any]    ):
         UpperCAmelCase_										=					WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''    )
         UpperCAmelCase_										=					self._get_dummy_logits()

         UpperCAmelCase_										=					processor.batch_decode(_SCREAMING_SNAKE_CASE ,   output_word_offsets=_SCREAMING_SNAKE_CASE    )

         # 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(_SCREAMING_SNAKE_CASE ,   _SCREAMING_SNAKE_CASE    )    )

         self.assertListEqual(
             [''' '''.join(self.get_from_offsets(_SCREAMING_SNAKE_CASE ,   '''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   lowerCamelCase_       (     self       :      Union[str, Any]    ):
         import torch

         UpperCAmelCase_										=					load_dataset('''common_voice''' ,   '''en''' ,   split='''train''' ,   streaming=_SCREAMING_SNAKE_CASE    )
         UpperCAmelCase_										=					ds.cast_column('''audio''' ,   datasets.Audio(sampling_rate=1_60_00    )    )
         UpperCAmelCase_										=					iter(_SCREAMING_SNAKE_CASE    )
         UpperCAmelCase_										=					next(_SCREAMING_SNAKE_CASE    )

         UpperCAmelCase_										=					AutoProcessor.from_pretrained('''patrickvonplaten/wav2vec2-base-100h-with-lm'''    )
         UpperCAmelCase_										=					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
         UpperCAmelCase_										=					processor(sample['''audio''']['''array'''] ,   return_tensors='''pt'''    ).input_values

         with torch.no_grad():
             UpperCAmelCase_										=					model(_SCREAMING_SNAKE_CASE    ).logits.cpu().numpy()

         UpperCAmelCase_										=					processor.decode(logits[0] ,   output_word_offsets=_SCREAMING_SNAKE_CASE    )

         UpperCAmelCase_										=					model.config.inputs_to_logits_ratio / processor.feature_extractor.sampling_rate
         UpperCAmelCase_										=					[
             {
                 "start_time": d["start_offset"] * time_offset,
                 "end_time": d["end_offset"] * time_offset,
                 "word": d["word"],
             }
             for d in output["word_offsets"]
         ]

         UpperCAmelCase_										=					"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(_SCREAMING_SNAKE_CASE ,   '''word'''    )    ) ,   _SCREAMING_SNAKE_CASE    )
         self.assertEqual(''' '''.join(self.get_from_offsets(_SCREAMING_SNAKE_CASE ,   '''word'''    )    ) ,   output.text    )

         # output times
         UpperCAmelCase_										=					torch.tensor(self.get_from_offsets(_SCREAMING_SNAKE_CASE ,   '''start_time'''    )    )
         UpperCAmelCase_										=					torch.tensor(self.get_from_offsets(_SCREAMING_SNAKE_CASE ,   '''end_time'''    )    )

         # fmt: off
         UpperCAmelCase_										=					torch.tensor([1.4_199, 1.6_599, 2.2_599, 3.0, 3.24, 3.5_999, 3.7_999, 4.0_999, 4.26, 4.94, 5.28, 5.6_599, 5.78, 5.94, 6.32, 6.5_399, 6.6_599]    )
         UpperCAmelCase_										=					torch.tensor([1.5_399, 1.8_999, 2.9, 3.16, 3.5_399, 3.72, 4.0_199, 4.1_799, 4.76, 5.1_599, 5.5_599, 5.6_999, 5.86, 6.1_999, 6.38, 6.6_199, 6.94]    )
         # fmt: on

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











import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_image_inputs


if is_torch_available():
 import torch

if is_vision_available():
 from PIL import Image

 from transformers import MobileNetVaImageProcessor



class       a					(       unittest.TestCase      ):

     '''simple docstring'''





     def __init__(     self       :      Tuple ,   __snake_case       :      int ,   __snake_case       :      List[Any]=7 ,   __snake_case       :      Any=3 ,   __snake_case       :      Any=18 ,   __snake_case       :      str=30 ,   __snake_case       :      Any=4_00 ,   __snake_case       :      Optional[int]=True ,   __snake_case       :      str=None ,   __snake_case       :      Any=True ,   __snake_case       :      List[Any]=None ,   ):
         UpperCAmelCase_										=					size if size is not None else {'''shortest_edge''': 20}
         UpperCAmelCase_										=					crop_size if crop_size is not None else {'''height''': 18, '''width''': 18}
         UpperCAmelCase_										=					parent
         UpperCAmelCase_										=					batch_size
         UpperCAmelCase_										=					num_channels
         UpperCAmelCase_										=					image_size
         UpperCAmelCase_										=					min_resolution
         UpperCAmelCase_										=					max_resolution
         UpperCAmelCase_										=					do_resize
         UpperCAmelCase_										=					size
         UpperCAmelCase_										=					do_center_crop
         UpperCAmelCase_										=					crop_size




     def   lowerCamelCase_       (     self       :      Any    ):
         return {
             "do_resize": self.do_resize,
             "size": self.size,
             "do_center_crop": self.do_center_crop,
             "crop_size": self.crop_size,
         }



@require_torch
@require_vision
class       a					(       _A     ,       unittest.TestCase      ):

     '''simple docstring'''





     lowerCAmelCase    :	Union[str, Any]       		=  MobileNetVaImageProcessor if is_vision_available() else None
     def   lowerCamelCase_       (     self       :      Any    ):
         UpperCAmelCase_										=					MobileNetVaImageProcessingTester(self    )
     @property
     def   lowerCamelCase_       (     self       :      List[str]    ):
         return self.image_processor_tester.prepare_image_processor_dict()
     def   lowerCamelCase_       (     self       :      Optional[int]    ):
         UpperCAmelCase_										=					self.image_processing_class(**self.image_processor_dict    )
         self.assertTrue(hasattr(__snake_case ,   '''do_resize'''    )    )
         self.assertTrue(hasattr(__snake_case ,   '''size'''    )    )
         self.assertTrue(hasattr(__snake_case ,   '''do_center_crop'''    )    )
         self.assertTrue(hasattr(__snake_case ,   '''crop_size'''    )    )
     def   lowerCamelCase_       (     self       :      Dict    ):
         UpperCAmelCase_										=					self.image_processing_class.from_dict(self.image_processor_dict    )
         self.assertEqual(image_processor.size ,   {'''shortest_edge''': 20}    )
         self.assertEqual(image_processor.crop_size ,   {'''height''': 18, '''width''': 18}    )

         UpperCAmelCase_										=					self.image_processing_class.from_dict(self.image_processor_dict ,   size=42 ,   crop_size=84    )
         self.assertEqual(image_processor.size ,   {'''shortest_edge''': 42}    )
         self.assertEqual(image_processor.crop_size ,   {'''height''': 84, '''width''': 84}    )
     def   lowerCamelCase_       (     self       :      Optional[int]    ):
         pass
     def   lowerCamelCase_       (     self       :      Tuple    ):
         # Initialize image_processing
         UpperCAmelCase_										=					self.image_processing_class(**self.image_processor_dict    )
         # create random PIL images
         UpperCAmelCase_										=					prepare_image_inputs(self.image_processor_tester ,   equal_resolution=__snake_case    )
         for image in image_inputs:
             self.assertIsInstance(__snake_case ,   Image.Image    )

         # Test not batched input
         UpperCAmelCase_										=					image_processing(image_inputs[0] ,   return_tensors='''pt'''    ).pixel_values
         self.assertEqual(
             encoded_images.shape ,   (
                 1,
                 self.image_processor_tester.num_channels,
                 self.image_processor_tester.crop_size['''height'''],
                 self.image_processor_tester.crop_size['''width'''],
             ) ,   )

         # Test batched
         UpperCAmelCase_										=					image_processing(__snake_case ,   return_tensors='''pt'''    ).pixel_values
         self.assertEqual(
             encoded_images.shape ,   (
                 self.image_processor_tester.batch_size,
                 self.image_processor_tester.num_channels,
                 self.image_processor_tester.crop_size['''height'''],
                 self.image_processor_tester.crop_size['''width'''],
             ) ,   )
     def   lowerCamelCase_       (     self       :      str    ):
         # Initialize image_processing
         UpperCAmelCase_										=					self.image_processing_class(**self.image_processor_dict    )
         # create random numpy tensors
         UpperCAmelCase_										=					prepare_image_inputs(self.image_processor_tester ,   equal_resolution=__snake_case ,   numpify=__snake_case    )
         for image in image_inputs:
             self.assertIsInstance(__snake_case ,   np.ndarray    )

         # Test not batched input
         UpperCAmelCase_										=					image_processing(image_inputs[0] ,   return_tensors='''pt'''    ).pixel_values
         self.assertEqual(
             encoded_images.shape ,   (
                 1,
                 self.image_processor_tester.num_channels,
                 self.image_processor_tester.crop_size['''height'''],
                 self.image_processor_tester.crop_size['''width'''],
             ) ,   )

         # Test batched
         UpperCAmelCase_										=					image_processing(__snake_case ,   return_tensors='''pt'''    ).pixel_values
         self.assertEqual(
             encoded_images.shape ,   (
                 self.image_processor_tester.batch_size,
                 self.image_processor_tester.num_channels,
                 self.image_processor_tester.crop_size['''height'''],
                 self.image_processor_tester.crop_size['''width'''],
             ) ,   )




     def   lowerCamelCase_       (     self       :      int    ):
         # Initialize image_processing
         UpperCAmelCase_										=					self.image_processing_class(**self.image_processor_dict    )
         # create random PyTorch tensors
         UpperCAmelCase_										=					prepare_image_inputs(self.image_processor_tester ,   equal_resolution=__snake_case ,   torchify=__snake_case    )
         for image in image_inputs:
             self.assertIsInstance(__snake_case ,   torch.Tensor    )

         # Test not batched input
         UpperCAmelCase_										=					image_processing(image_inputs[0] ,   return_tensors='''pt'''    ).pixel_values
         self.assertEqual(
             encoded_images.shape ,   (
                 1,
                 self.image_processor_tester.num_channels,
                 self.image_processor_tester.crop_size['''height'''],
                 self.image_processor_tester.crop_size['''width'''],
             ) ,   )

         # Test batched
         UpperCAmelCase_										=					image_processing(__snake_case ,   return_tensors='''pt'''    ).pixel_values
         self.assertEqual(
             encoded_images.shape ,   (
                 self.image_processor_tester.batch_size,
                 self.image_processor_tester.num_channels,
                 self.image_processor_tester.crop_size['''height'''],
                 self.image_processor_tester.crop_size['''width'''],
             ) ,   )





